<s> """<STR_LIT>""" <EOL> import os <EOL> from twisted . python . filepath import FilePath <EOL> from flocker . testtools . ssh import create_ssh_agent , generate_ssh_key <EOL> from ... testtools import AsyncTestCase <EOL> from . . _ssh . _keys import ensure_agent_has_ssh_key , KeyNotFound , AgentNotFound <EOL> class EnsureKeyTests ( AsyncTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_public_key_in_agent ( self ) : <EOL> """<STR_LIT>""" <EOL> key_file = FilePath ( self . mktemp ( ) ) <EOL> key = generate_ssh_key ( key_file ) . public ( ) <EOL> create_ssh_agent ( key_file , self ) <EOL> result = ensure_agent_has_ssh_key ( self . reactor , key ) <EOL> return result <EOL> def test_private_key_in_agent ( self ) : <EOL> """<STR_LIT>""" <EOL> key_file = FilePath ( self . mktemp ( ) ) <EOL> key = generate_ssh_key ( key_file ) <EOL> create_ssh_agent ( key_file , self ) <EOL> result = ensure_agent_has_ssh_key ( self . reactor , key ) <EOL> return result <EOL> def test_public_key_not_in_agent ( self ) : <EOL> """<STR_LIT>""" <EOL> key_file = FilePath ( self . mktemp ( ) ) <EOL> generate_ssh_key ( key_file ) <EOL> create_ssh_agent ( key_file , self ) <EOL> other_key = generate_ssh_key ( FilePath ( self . mktemp ( ) ) ) . public ( ) <EOL> result = ensure_agent_has_ssh_key ( self . reactor , other_key ) <EOL> return self . assertFailure ( result , KeyNotFound ) <EOL> def test_agent_not_found ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> old_agent_socket = os . environ [ '<STR_LIT>' ] <EOL> self . addCleanup ( os . environ . __setitem__ , <EOL> '<STR_LIT>' , old_agent_socket ) <EOL> del os . environ [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> key_file = FilePath ( self . mktemp ( ) ) <EOL> key = generate_ssh_key ( key_file ) <EOL> return self . assertFailure ( <EOL> ensure_agent_has_ssh_key ( self . reactor , key ) , <EOL> AgentNotFound ) </s>
