<s> """<STR_LIT>""" <EOL> from zope . interface . verify import verifyObject <EOL> from pyrsistent import pset , pvector <EOL> from docker . errors import APIError <EOL> from twisted . python . filepath import FilePath <EOL> from ... testtools import ( <EOL> AsyncTestCase , TestCase , random_name , make_with_init_tests , <EOL> ) <EOL> from . . testtools import add_with_port_collision_retry <EOL> from . . _docker import ( <EOL> IDockerClient , FakeDockerClient , AddressInUse , AlreadyExists , PortMap , <EOL> Unit , Environment , Volume , <EOL> ) <EOL> from ... control . _model import RestartAlways , RestartNever , RestartOnFailure <EOL> ANY_IMAGE = u"<STR_LIT>" <EOL> def make_idockerclient_tests ( fixture ) : <EOL> """<STR_LIT>""" <EOL> class IDockerClientTests ( AsyncTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> self . assertTrue ( verifyObject ( IDockerClient , client ) ) <EOL> def test_add_and_remove ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> d = client . add ( name , u"<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : client . remove ( name ) ) <EOL> return d <EOL> def test_no_double_add ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> self . addCleanup ( client . remove , name ) <EOL> d = client . add ( name , u"<STR_LIT>" ) <EOL> def added ( _ ) : <EOL> return client . add ( name , u"<STR_LIT>" ) <EOL> d . addCallback ( added ) <EOL> d = self . assertFailure ( d , AlreadyExists ) <EOL> d . addCallback ( lambda exc : self . assertEqual ( exc . args [ <NUM_LIT:0> ] , name ) ) <EOL> return d <EOL> def test_remove_nonexistent_is_ok ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> return client . remove ( name ) <EOL> def test_double_remove_is_ok ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> d = client . add ( name , u"<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : client . remove ( name ) ) <EOL> d . addCallback ( lambda _ : client . remove ( name ) ) <EOL> return d <EOL> def test_unknown_does_not_exist ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> d = client . exists ( name ) <EOL> d . addCallback ( self . assertFalse ) <EOL> return d <EOL> def test_added_exists ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> self . addCleanup ( client . remove , name ) <EOL> d = client . add ( name , u"<STR_LIT>" ) <EOL> def added ( _ ) : <EOL> return client . exists ( name ) <EOL> d . addCallback ( added ) <EOL> d . addCallback ( self . assertTrue ) <EOL> return d <EOL> def test_removed_does_not_exist ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> d = client . add ( name , ANY_IMAGE ) <EOL> d . addCallback ( lambda _ : client . remove ( name ) ) <EOL> d . addCallback ( lambda _ : client . exists ( name ) ) <EOL> d . addCallback ( self . assertFalse ) <EOL> return d <EOL> def test_zero_port_randomly_assigned ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> portmap = PortMap ( <EOL> internal_port = <NUM_LIT> , external_port = <NUM_LIT:0> , <EOL> ) <EOL> self . addCleanup ( client . remove , name ) <EOL> d = client . add ( name , ANY_IMAGE , ports = ( portmap , ) ) <EOL> d . addCallback ( lambda ignored : client . list ( ) ) <EOL> def check_port ( units ) : <EOL> portmap = list ( list ( units ) [ <NUM_LIT:0> ] . ports ) [ <NUM_LIT:0> ] <EOL> self . assertTrue ( <EOL> <NUM_LIT:0> < portmap . external_port < <NUM_LIT:2> ** <NUM_LIT:16> , <EOL> "<STR_LIT>" . format ( <EOL> portmap . external_port <EOL> ) , <EOL> ) <EOL> d . addCallback ( check_port ) <EOL> return d <EOL> def test_port_collision_raises_addressinuse ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> portmap = PortMap ( <EOL> internal_port = <NUM_LIT> , external_port = <NUM_LIT:0> , <EOL> ) <EOL> self . addCleanup ( client . remove , name ) <EOL> d = client . add ( name , ANY_IMAGE , ports = ( portmap , ) ) <EOL> d . addCallback ( lambda ignored : client . list ( ) ) <EOL> def extract_port ( units ) : <EOL> return list ( list ( units ) [ <NUM_LIT:0> ] . ports ) [ <NUM_LIT:0> ] . external_port <EOL> d . addCallback ( extract_port ) <EOL> def collide ( external_port ) : <EOL> self . external_port = external_port <EOL> portmap = PortMap ( <EOL> internal_port = <NUM_LIT> , external_port = external_port , <EOL> ) <EOL> name = random_name ( self ) <EOL> self . addCleanup ( client . remove , name ) <EOL> return client . add ( name , ANY_IMAGE , ports = ( portmap , ) ) <EOL> d . addCallback ( collide ) <EOL> d = self . assertFailure ( d , AddressInUse ) <EOL> def failed ( exception ) : <EOL> self . assertEqual ( <EOL> exception . address , ( b"<STR_LIT>" , self . external_port ) <EOL> ) <EOL> self . assertIsInstance ( exception . apierror , APIError ) <EOL> d . addCallback ( failed ) <EOL> return d <EOL> def test_added_is_listed ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> image = ANY_IMAGE <EOL> portmaps = [ <EOL> PortMap ( internal_port = <NUM_LIT> , external_port = <NUM_LIT:0> ) , <EOL> PortMap ( internal_port = <NUM_LIT> , external_port = <NUM_LIT:0> ) , <EOL> ] <EOL> volumes = ( <EOL> Volume ( node_path = FilePath ( self . mktemp ( ) ) , <EOL> container_path = FilePath ( b'<STR_LIT>' ) ) , <EOL> ) <EOL> environment = ( <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ) <EOL> environment = Environment ( variables = frozenset ( environment ) ) <EOL> self . addCleanup ( client . remove , name ) <EOL> d = add_with_port_collision_retry ( <EOL> client , <EOL> name , <EOL> image_name = image , <EOL> ports = portmaps , <EOL> volumes = volumes , <EOL> environment = environment , <EOL> mem_limit = <NUM_LIT> , <EOL> cpu_shares = <NUM_LIT> , <EOL> restart_policy = RestartAlways ( ) , <EOL> ) <EOL> def added ( ( app , portmaps ) ) : <EOL> d = client . list ( ) <EOL> d . addCallback ( lambda units : ( units , portmaps ) ) <EOL> return d <EOL> d . addCallback ( added ) <EOL> def got_list ( ( units , portmaps ) ) : <EOL> result = units . pop ( ) <EOL> expected = Unit ( <EOL> name = name , container_name = name , activation_state = u"<STR_LIT>" , <EOL> container_image = image , ports = frozenset ( portmaps ) , <EOL> environment = environment , volumes = frozenset ( volumes ) , <EOL> mem_limit = <NUM_LIT> , cpu_shares = <NUM_LIT> , <EOL> restart_policy = RestartAlways ( ) , <EOL> ) <EOL> result = result . set ( "<STR_LIT>" , name ) <EOL> self . assertEqual ( result , expected ) <EOL> d . addCallback ( got_list ) <EOL> return d <EOL> def test_removed_is_not_listed ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> d = client . add ( name , ANY_IMAGE ) <EOL> d . addCallback ( lambda _ : client . remove ( name ) ) <EOL> d . addCallback ( lambda _ : client . list ( ) ) <EOL> def got_list ( units ) : <EOL> self . assertNotIn ( name , [ unit . name for unit in units ] ) <EOL> d . addCallback ( got_list ) <EOL> return d <EOL> def test_container_name ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> self . addCleanup ( client . remove , name ) <EOL> d = client . add ( name , u"<STR_LIT>" ) <EOL> d . addCallback ( lambda _ : client . list ( ) ) <EOL> def got_list ( units ) : <EOL> unit = [ unit for unit in units if unit . name == name ] [ <NUM_LIT:0> ] <EOL> self . assertIsInstance ( unit . container_name , unicode ) <EOL> d . addCallback ( got_list ) <EOL> return d <EOL> def test_command_line ( self ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> self . addCleanup ( client . remove , name ) <EOL> command_line = [ u"<STR_LIT>" , u"<STR_LIT>" , u"<STR_LIT>" , u"<STR_LIT>" ] <EOL> d = client . add ( name , u"<STR_LIT>" , command_line = command_line ) <EOL> d . addCallback ( lambda _ : client . list ( ) ) <EOL> def got_list ( units ) : <EOL> unit = [ unit for unit in units if unit . name == name ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( unit . command_line , pvector ( command_line ) ) <EOL> d . addCallback ( got_list ) <EOL> return d <EOL> def assert_restart_policy_round_trips ( self , restart_policy ) : <EOL> """<STR_LIT>""" <EOL> client = fixture ( self ) <EOL> name = random_name ( self ) <EOL> self . addCleanup ( client . remove , name ) <EOL> d = client . add ( name , u"<STR_LIT>" , restart_policy = restart_policy ) <EOL> d . addCallback ( lambda _ : client . list ( ) ) <EOL> def got_list ( units ) : <EOL> unit = [ unit for unit in units if unit . name == name ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( unit . restart_policy , restart_policy ) <EOL> d . addCallback ( got_list ) <EOL> return d <EOL> def test_add_with_restart_never ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . assert_restart_policy_round_trips ( RestartNever ( ) ) <EOL> def test_add_with_restart_always ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . assert_restart_policy_round_trips ( RestartAlways ( ) ) <EOL> def test_add_with_restart_on_failure ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . assert_restart_policy_round_trips ( RestartOnFailure ( ) ) <EOL> def test_add_with_restart_on_failure_with_maximum_retry ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . assert_restart_policy_round_trips ( <EOL> RestartOnFailure ( maximum_retry_count = <NUM_LIT:5> ) ) <EOL> return IDockerClientTests <EOL> class FakeIDockerClientTests ( <EOL> make_idockerclient_tests ( <EOL> fixture = lambda test_case : FakeDockerClient ( ) , <EOL> ) <EOL> ) : <EOL> """<STR_LIT>""" <EOL> class FakeDockerClientImplementationTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_units_default ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( { } , FakeDockerClient ( ) . _units ) <EOL> def test_units_override ( self ) : <EOL> """<STR_LIT>""" <EOL> units = { u'<STR_LIT:foo>' : Unit ( name = u'<STR_LIT:foo>' , container_name = u'<STR_LIT:foo>' , <EOL> activation_state = u'<STR_LIT>' , <EOL> container_image = u'<STR_LIT>' ) } <EOL> self . assertEqual ( units , FakeDockerClient ( units = units ) . _units ) <EOL> class PortMapInitTests ( <EOL> make_with_init_tests ( <EOL> record_type = PortMap , <EOL> kwargs = dict ( <EOL> internal_port = <NUM_LIT> , <EOL> external_port = <NUM_LIT> , <EOL> ) <EOL> ) <EOL> ) : <EOL> """<STR_LIT>""" <EOL> class PortMapTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_equal ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> PortMap ( internal_port = <NUM_LIT> , external_port = <NUM_LIT> ) , <EOL> PortMap ( internal_port = <NUM_LIT> , external_port = <NUM_LIT> ) , <EOL> ) <EOL> def test_not_equal ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertNotEqual ( <EOL> PortMap ( internal_port = <NUM_LIT> , external_port = <NUM_LIT> ) , <EOL> PortMap ( internal_port = <NUM_LIT> , external_port = <NUM_LIT> ) <EOL> ) <EOL> class UnitInitTests ( <EOL> make_with_init_tests ( <EOL> record_type = Unit , <EOL> kwargs = dict ( <EOL> name = u'<STR_LIT>' , <EOL> container_name = u'<STR_LIT>' , <EOL> activation_state = u'<STR_LIT>' , <EOL> container_image = u'<STR_LIT>' , <EOL> ports = pset ( ( PortMap ( internal_port = <NUM_LIT> , external_port = <NUM_LIT> ) , ) ) , <EOL> environment = Environment ( variables = { u'<STR_LIT:foo>' : u'<STR_LIT:bar>' } ) , <EOL> restart_policy = RestartAlways ( ) , <EOL> ) , <EOL> expected_defaults = dict ( <EOL> ports = pset ( ) , container_image = None , environment = None , <EOL> restart_policy = RestartNever ( ) ) <EOL> ) <EOL> ) : <EOL> """<STR_LIT>""" <EOL> class EnvironmentInitTests ( <EOL> make_with_init_tests ( <EOL> record_type = Environment , <EOL> kwargs = dict ( <EOL> variables = dict ( foo = "<STR_LIT:bar>" ) , <EOL> ) , <EOL> ) <EOL> ) : <EOL> """<STR_LIT>""" <EOL> class EnvironmentTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> variables = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:foo>' : '<STR_LIT:bar>' } <EOL> environment = Environment ( variables = frozenset ( variables . items ( ) ) ) <EOL> self . assertEqual ( environment . to_dict ( ) , variables ) <EOL> class VolumeInitTests ( <EOL> make_with_init_tests ( <EOL> record_type = Volume , <EOL> kwargs = dict ( <EOL> node_path = FilePath ( b"<STR_LIT>" ) , <EOL> container_path = FilePath ( b"<STR_LIT>" ) , <EOL> ) , <EOL> ) <EOL> ) : <EOL> """<STR_LIT>""" </s>
