service_test.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137
  1. from __future__ import unicode_literals
  2. from __future__ import absolute_import
  3. import os
  4. from .. import unittest
  5. import mock
  6. from fig import Service
  7. from fig.service import (
  8. ConfigError,
  9. split_port,
  10. parse_volume_spec,
  11. build_volume_binding,
  12. )
  13. class ServiceTest(unittest.TestCase):
  14. def test_name_validations(self):
  15. self.assertRaises(ConfigError, lambda: Service(name=''))
  16. self.assertRaises(ConfigError, lambda: Service(name=' '))
  17. self.assertRaises(ConfigError, lambda: Service(name='/'))
  18. self.assertRaises(ConfigError, lambda: Service(name='!'))
  19. self.assertRaises(ConfigError, lambda: Service(name='\xe2'))
  20. self.assertRaises(ConfigError, lambda: Service(name='_'))
  21. self.assertRaises(ConfigError, lambda: Service(name='____'))
  22. self.assertRaises(ConfigError, lambda: Service(name='foo_bar'))
  23. self.assertRaises(ConfigError, lambda: Service(name='__foo_bar__'))
  24. Service('a')
  25. Service('foo')
  26. def test_project_validation(self):
  27. self.assertRaises(ConfigError, lambda: Service(name='foo', project='_'))
  28. Service(name='foo', project='bar')
  29. def test_config_validation(self):
  30. self.assertRaises(ConfigError, lambda: Service(name='foo', port=['8000']))
  31. Service(name='foo', ports=['8000'])
  32. def test_split_port(self):
  33. internal_port, external_port = split_port("127.0.0.1:1000:2000")
  34. self.assertEqual(internal_port, "2000")
  35. self.assertEqual(external_port, ("127.0.0.1", "1000"))
  36. internal_port, external_port = split_port("127.0.0.1:1000:2000/udp")
  37. self.assertEqual(internal_port, "2000/udp")
  38. self.assertEqual(external_port, ("127.0.0.1", "1000"))
  39. internal_port, external_port = split_port("127.0.0.1::2000")
  40. self.assertEqual(internal_port, "2000")
  41. self.assertEqual(external_port, ("127.0.0.1",))
  42. internal_port, external_port = split_port("1000:2000")
  43. self.assertEqual(internal_port, "2000")
  44. self.assertEqual(external_port, "1000")
  45. def test_split_domainname_none(self):
  46. service = Service('foo',
  47. hostname = 'name',
  48. )
  49. service.next_container_name = lambda x: 'foo'
  50. opts = service._get_container_create_options({})
  51. self.assertEqual(opts['hostname'], 'name', 'hostname')
  52. self.assertFalse('domainname' in opts, 'domainname')
  53. def test_split_domainname_fqdn(self):
  54. service = Service('foo',
  55. hostname = 'name.domain.tld',
  56. )
  57. service.next_container_name = lambda x: 'foo'
  58. opts = service._get_container_create_options({})
  59. self.assertEqual(opts['hostname'], 'name', 'hostname')
  60. self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')
  61. def test_split_domainname_both(self):
  62. service = Service('foo',
  63. hostname = 'name',
  64. domainname = 'domain.tld',
  65. )
  66. service.next_container_name = lambda x: 'foo'
  67. opts = service._get_container_create_options({})
  68. self.assertEqual(opts['hostname'], 'name', 'hostname')
  69. self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')
  70. def test_split_domainname_weird(self):
  71. service = Service('foo',
  72. hostname = 'name.sub',
  73. domainname = 'domain.tld',
  74. )
  75. service.next_container_name = lambda x: 'foo'
  76. opts = service._get_container_create_options({})
  77. self.assertEqual(opts['hostname'], 'name.sub', 'hostname')
  78. self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')
  79. class ServiceVolumesTest(unittest.TestCase):
  80. def test_parse_volume_spec_only_one_path(self):
  81. spec = parse_volume_spec('/the/volume')
  82. self.assertEqual(spec, (None, '/the/volume', 'rw'))
  83. def test_parse_volume_spec_internal_and_external(self):
  84. spec = parse_volume_spec('external:interval')
  85. self.assertEqual(spec, ('external', 'interval', 'rw'))
  86. def test_parse_volume_spec_with_mode(self):
  87. spec = parse_volume_spec('external:interval:ro')
  88. self.assertEqual(spec, ('external', 'interval', 'ro'))
  89. def test_parse_volume_spec_too_many_parts(self):
  90. with self.assertRaises(ConfigError):
  91. parse_volume_spec('one:two:three:four')
  92. def test_parse_volume_bad_mode(self):
  93. with self.assertRaises(ConfigError):
  94. parse_volume_spec('one:two:notrw')
  95. def test_build_volume_binding(self):
  96. binding = build_volume_binding(parse_volume_spec('/outside:/inside'))
  97. self.assertEqual(
  98. binding,
  99. ('/outside', dict(bind='/inside', ro=False)))
  100. @mock.patch.dict(os.environ)
  101. def test_build_volume_binding_with_environ(self):
  102. os.environ['VOLUME_PATH'] = '/opt'
  103. binding = build_volume_binding(parse_volume_spec('${VOLUME_PATH}:/opt'))
  104. self.assertEqual(binding, ('/opt', dict(bind='/opt', ro=False)))
  105. @mock.patch.dict(os.environ)
  106. def test_building_volume_binding_with_home(self):
  107. os.environ['HOME'] = '/home/user'
  108. binding = build_volume_binding(parse_volume_spec('~:/home/user'))
  109. self.assertEqual(
  110. binding,
  111. ('/home/user', dict(bind='/home/user', ro=False)))