| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137 | from __future__ import unicode_literalsfrom __future__ import absolute_importimport osfrom .. import unittestimport mockfrom fig import Servicefrom fig.service import (    ConfigError,    split_port,    parse_volume_spec,    build_volume_binding,)class ServiceTest(unittest.TestCase):    def test_name_validations(self):        self.assertRaises(ConfigError, lambda: Service(name=''))        self.assertRaises(ConfigError, lambda: Service(name=' '))        self.assertRaises(ConfigError, lambda: Service(name='/'))        self.assertRaises(ConfigError, lambda: Service(name='!'))        self.assertRaises(ConfigError, lambda: Service(name='\xe2'))        self.assertRaises(ConfigError, lambda: Service(name='_'))        self.assertRaises(ConfigError, lambda: Service(name='____'))        self.assertRaises(ConfigError, lambda: Service(name='foo_bar'))        self.assertRaises(ConfigError, lambda: Service(name='__foo_bar__'))        Service('a')        Service('foo')    def test_project_validation(self):        self.assertRaises(ConfigError, lambda: Service(name='foo', project='_'))        Service(name='foo', project='bar')    def test_config_validation(self):        self.assertRaises(ConfigError, lambda: Service(name='foo', port=['8000']))        Service(name='foo', ports=['8000'])    def test_split_port(self):        internal_port, external_port = split_port("127.0.0.1:1000:2000")        self.assertEqual(internal_port, "2000")        self.assertEqual(external_port, ("127.0.0.1", "1000"))        internal_port, external_port = split_port("127.0.0.1:1000:2000/udp")        self.assertEqual(internal_port, "2000/udp")        self.assertEqual(external_port, ("127.0.0.1", "1000"))        internal_port, external_port = split_port("127.0.0.1::2000")        self.assertEqual(internal_port, "2000")        self.assertEqual(external_port, ("127.0.0.1",))        internal_port, external_port = split_port("1000:2000")        self.assertEqual(internal_port, "2000")        self.assertEqual(external_port, "1000")    def test_split_domainname_none(self):        service = Service('foo',                hostname = 'name',            )        service.next_container_name = lambda x: 'foo'        opts = service._get_container_create_options({})        self.assertEqual(opts['hostname'], 'name', 'hostname')        self.assertFalse('domainname' in opts, 'domainname')    def test_split_domainname_fqdn(self):        service = Service('foo',                hostname = 'name.domain.tld',            )        service.next_container_name = lambda x: 'foo'        opts = service._get_container_create_options({})        self.assertEqual(opts['hostname'], 'name', 'hostname')        self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')    def test_split_domainname_both(self):        service = Service('foo',                hostname = 'name',                domainname = 'domain.tld',            )        service.next_container_name = lambda x: 'foo'        opts = service._get_container_create_options({})        self.assertEqual(opts['hostname'], 'name', 'hostname')        self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')    def test_split_domainname_weird(self):        service = Service('foo',                hostname = 'name.sub',                domainname = 'domain.tld',            )        service.next_container_name = lambda x: 'foo'        opts = service._get_container_create_options({})        self.assertEqual(opts['hostname'], 'name.sub', 'hostname')        self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')class ServiceVolumesTest(unittest.TestCase):    def test_parse_volume_spec_only_one_path(self):        spec = parse_volume_spec('/the/volume')        self.assertEqual(spec, (None, '/the/volume', 'rw'))    def test_parse_volume_spec_internal_and_external(self):        spec = parse_volume_spec('external:interval')        self.assertEqual(spec, ('external', 'interval', 'rw'))    def test_parse_volume_spec_with_mode(self):        spec = parse_volume_spec('external:interval:ro')        self.assertEqual(spec, ('external', 'interval', 'ro'))    def test_parse_volume_spec_too_many_parts(self):        with self.assertRaises(ConfigError):            parse_volume_spec('one:two:three:four')    def test_parse_volume_bad_mode(self):        with self.assertRaises(ConfigError):            parse_volume_spec('one:two:notrw')    def test_build_volume_binding(self):        binding = build_volume_binding(parse_volume_spec('/outside:/inside'))        self.assertEqual(            binding,            ('/outside', dict(bind='/inside', ro=False)))    @mock.patch.dict(os.environ)    def test_build_volume_binding_with_environ(self):        os.environ['VOLUME_PATH'] = '/opt'        binding = build_volume_binding(parse_volume_spec('${VOLUME_PATH}:/opt'))        self.assertEqual(binding, ('/opt', dict(bind='/opt', ro=False)))    @mock.patch.dict(os.environ)    def test_building_volume_binding_with_home(self):        os.environ['HOME'] = '/home/user'        binding = build_volume_binding(parse_volume_spec('~:/home/user'))        self.assertEqual(            binding,            ('/home/user', dict(bind='/home/user', ro=False)))
 |