service_test.py 2.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576
  1. from plum import Service
  2. from .testcases import DockerClientTestCase
  3. class NameTestCase(DockerClientTestCase):
  4. def test_name_validations(self):
  5. self.assertRaises(ValueError, lambda: Service(name=''))
  6. self.assertRaises(ValueError, lambda: Service(name=' '))
  7. self.assertRaises(ValueError, lambda: Service(name='/'))
  8. self.assertRaises(ValueError, lambda: Service(name='!'))
  9. self.assertRaises(ValueError, lambda: Service(name='\xe2'))
  10. Service('a')
  11. Service('foo')
  12. Service('foo_bar')
  13. Service('__foo_bar__')
  14. Service('_')
  15. Service('_____')
  16. def test_containers(self):
  17. foo = self.create_service('foo')
  18. bar = self.create_service('bar')
  19. foo.start()
  20. self.assertEqual(len(foo.containers), 1)
  21. self.assertEqual(foo.containers[0]['Names'], ['/foo_1'])
  22. self.assertEqual(len(bar.containers), 0)
  23. bar.scale(2)
  24. self.assertEqual(len(foo.containers), 1)
  25. self.assertEqual(len(bar.containers), 2)
  26. names = [c['Names'] for c in bar.containers]
  27. self.assertIn(['/bar_1'], names)
  28. self.assertIn(['/bar_2'], names)
  29. def test_up_scale_down(self):
  30. service = self.create_service('scaling_test')
  31. self.assertEqual(len(service.containers), 0)
  32. service.start()
  33. self.assertEqual(len(service.containers), 1)
  34. service.start()
  35. self.assertEqual(len(service.containers), 1)
  36. service.scale(2)
  37. self.assertEqual(len(service.containers), 2)
  38. service.scale(1)
  39. self.assertEqual(len(service.containers), 1)
  40. service.stop()
  41. self.assertEqual(len(service.containers), 0)
  42. service.stop()
  43. self.assertEqual(len(service.containers), 0)
  44. def test_start_container_passes_through_options(self):
  45. db = self.create_service('db', environment={'FOO': 'BAR'})
  46. db.start_container()
  47. self.assertEqual(db.inspect()[0]['Config']['Env'], ['FOO=BAR'])
  48. def test_start_container_creates_links(self):
  49. db = self.create_service('db')
  50. web = self.create_service('web', links=[db])
  51. db.start_container()
  52. web.start_container()
  53. self.assertIn('/web_1/db_1', db.containers[0]['Names'])
  54. db.stop()
  55. web.stop()