diff options
Diffstat (limited to 'test/python/docker/test_containers.py')
| -rw-r--r-- | test/python/docker/test_containers.py | 157 |
1 files changed, 64 insertions, 93 deletions
diff --git a/test/python/docker/test_containers.py b/test/python/docker/test_containers.py index 1c4c9ab53..5fb340fd4 100644 --- a/test/python/docker/test_containers.py +++ b/test/python/docker/test_containers.py @@ -3,7 +3,7 @@ import sys import time import unittest -from docker import APIClient, errors +from docker import DockerClient, errors from test.python.docker import Podman, common, constant @@ -15,7 +15,7 @@ class TestContainers(unittest.TestCase): def setUp(self): super().setUp() - self.client = APIClient(base_url="tcp://127.0.0.1:8080", timeout=15) + self.client = DockerClient(base_url="tcp://127.0.0.1:8080", timeout=15) TestContainers.podman.restore_image_from_cache(self.client) TestContainers.topContainerId = common.run_top_container(self.client) self.assertIsNotNone(TestContainers.topContainerId) @@ -52,146 +52,115 @@ class TestContainers(unittest.TestCase): TestContainers.podman.tear_down() return super().tearDownClass() - def test_inspect_container(self): - # Inspect bogus container - with self.assertRaises(errors.NotFound) as error: - self.client.inspect_container("dummy") - self.assertEqual(error.exception.response.status_code, 404) + def test_create_container(self): + # Run a container with detach mode + self.client.containers.create(image="alpine", detach=True) + self.assertEqual(len(self.client.containers.list(all=True)), 2) - # Inspect valid container by Id - container = self.client.inspect_container(TestContainers.topContainerId) - self.assertIn("top", container["Name"]) + def test_create_network(self): + net = self.client.networks.create("testNetwork", driver="bridge") + ctnr = self.client.containers.create(image="alpine", detach=True) + net.connect(ctnr) - # Inspect valid container by name - container = self.client.inspect_container("top") - self.assertIn(TestContainers.topContainerId, container["Id"]) + nets = self.client.networks.list(greedy=True) + self.assertGreaterEqual(len(nets), 1) - def test_create_container(self): - # Run a container with detach mode - container = self.client.create_container(image="alpine", detach=True) - self.assertEqual(len(container), 2) + # TODO fix endpoint to include containers + # for n in nets: + # if n.id == "testNetwork": + # self.assertEqual(ctnr.id, n.containers) + # self.assertTrue(False, "testNetwork not found") def test_start_container(self): - # Start bogus container - with self.assertRaises(errors.NotFound) as error: - self.client.start("dummy") - self.assertEqual(error.exception.response.status_code, 404) - # Podman docs says it should give a 304 but returns with no response # # Start a already started container should return 304 - # response = self.client.start(container=TestContainers.topContainerId) + # response = self.client.api.start(container=TestContainers.topContainerId) # self.assertEqual(error.exception.response.status_code, 304) # Create a new container and validate the count - self.client.create_container(image=constant.ALPINE, name="container2") - containers = self.client.containers(quiet=True, all=True) + self.client.containers.create(image=constant.ALPINE, name="container2") + containers = self.client.containers.list(all=True) self.assertEqual(len(containers), 2) def test_stop_container(self): - # Stop bogus container - with self.assertRaises(errors.NotFound) as error: - self.client.stop("dummy") - self.assertEqual(error.exception.response.status_code, 404) - - # Validate the container state - container = self.client.inspect_container("top") - self.assertEqual(container["State"]["Status"], "running") + top = self.client.containers.get("top") + self.assertEqual(top.status, "running") # Stop a running container and validate the state - self.client.stop(TestContainers.topContainerId) - container = self.client.inspect_container("top") - self.assertIn( - container["State"]["Status"], - "stopped exited", - ) + top.stop() + top.reload() + self.assertIn(top.status, ("stopped", "exited")) def test_restart_container(self): - # Restart bogus container - with self.assertRaises(errors.NotFound) as error: - self.client.restart("dummy") - self.assertEqual(error.exception.response.status_code, 404) - # Validate the container state - self.client.stop(TestContainers.topContainerId) - container = self.client.inspect_container("top") - self.assertEqual(container["State"]["Status"], "stopped") + top = self.client.containers.get(TestContainers.topContainerId) + top.stop() + top.reload() + self.assertIn(top.status, ("stopped", "exited")) # restart a running container and validate the state - self.client.restart(TestContainers.topContainerId) - container = self.client.inspect_container("top") - self.assertEqual(container["State"]["Status"], "running") + top.restart() + top.reload() + self.assertEqual(top.status, "running") def test_remove_container(self): - # Remove bogus container - with self.assertRaises(errors.NotFound) as error: - self.client.remove_container("dummy") - self.assertEqual(error.exception.response.status_code, 404) - # Remove container by ID with force - self.client.remove_container(TestContainers.topContainerId, force=True) - containers = self.client.containers() - self.assertEqual(len(containers), 0) + top = self.client.containers.get(TestContainers.topContainerId) + top.remove(force=True) + self.assertEqual(len(self.client.containers.list()), 0) def test_remove_container_without_force(self): # Validate current container count - containers = self.client.containers() - self.assertTrue(len(containers), 1) + self.assertTrue(len(self.client.containers.list()), 1) # Remove running container should throw error + top = self.client.containers.get(TestContainers.topContainerId) with self.assertRaises(errors.APIError) as error: - self.client.remove_container(TestContainers.topContainerId) + top.remove() self.assertEqual(error.exception.response.status_code, 500) - # Remove container by ID with force - self.client.stop(TestContainers.topContainerId) - self.client.remove_container(TestContainers.topContainerId) - containers = self.client.containers() - self.assertEqual(len(containers), 0) + # Remove container by ID without force + top.stop() + top.remove() + self.assertEqual(len(self.client.containers.list()), 0) def test_pause_container(self): - # Pause bogus container - with self.assertRaises(errors.NotFound) as error: - self.client.pause("dummy") - self.assertEqual(error.exception.response.status_code, 404) - # Validate the container state - container = self.client.inspect_container("top") - self.assertEqual(container["State"]["Status"], "running") + top = self.client.containers.get(TestContainers.topContainerId) + self.assertEqual(top.status, "running") # Pause a running container and validate the state - self.client.pause(container["Id"]) - container = self.client.inspect_container("top") - self.assertEqual(container["State"]["Status"], "paused") + top.pause() + top.reload() + self.assertEqual(top.status, "paused") def test_pause_stopped_container(self): # Stop the container - self.client.stop(TestContainers.topContainerId) + top = self.client.containers.get(TestContainers.topContainerId) + top.stop() # Pause exited container should trow error with self.assertRaises(errors.APIError) as error: - self.client.pause(TestContainers.topContainerId) + top.pause() self.assertEqual(error.exception.response.status_code, 500) def test_unpause_container(self): - # Unpause bogus container - with self.assertRaises(errors.NotFound) as error: - self.client.unpause("dummy") - self.assertEqual(error.exception.response.status_code, 404) + top = self.client.containers.get(TestContainers.topContainerId) # Validate the container state - self.client.pause(TestContainers.topContainerId) - container = self.client.inspect_container("top") - self.assertEqual(container["State"]["Status"], "paused") + top.pause() + top.reload() + self.assertEqual(top.status, "paused") # Pause a running container and validate the state - self.client.unpause(TestContainers.topContainerId) - container = self.client.inspect_container("top") - self.assertEqual(container["State"]["Status"], "running") + top.unpause() + top.reload() + self.assertEqual(top.status, "running") def test_list_container(self): # Add container and validate the count - self.client.create_container(image="alpine", detach=True) - containers = self.client.containers(all=True) + self.client.containers.create(image="alpine", detach=True) + containers = self.client.containers.list(all=True) self.assertEqual(len(containers), 2) def test_filters(self): @@ -199,16 +168,18 @@ class TestContainers(unittest.TestCase): # List container with filter by id filters = {"id": TestContainers.topContainerId} - ctnrs = self.client.containers(all=True, filters=filters) + ctnrs = self.client.containers.list(all=True, filters=filters) self.assertEqual(len(ctnrs), 1) # List container with filter by name filters = {"name": "top"} - ctnrs = self.client.containers(all=True, filters=filters) + ctnrs = self.client.containers.list(all=True, filters=filters) self.assertEqual(len(ctnrs), 1) def test_rename_container(self): + top = self.client.containers.get(TestContainers.topContainerId) + # rename bogus container with self.assertRaises(errors.APIError) as error: - self.client.rename(container="dummy", name="newname") + top.rename(name="newname") self.assertEqual(error.exception.response.status_code, 404) |