import multiprocessing import queue import random import threading import unittest import requests import time from dateutil.parser import parse from .fixtures import APITestCase class ContainerTestCase(APITestCase): def test_list(self): r = requests.get(self.uri("/containers/json"), timeout=5) self.assertEqual(r.status_code, 200, r.text) obj = r.json() self.assertEqual(len(obj), 1) def test_list_filters(self): r = requests.get( self.podman_url + "/v1.40/containers/json?filters%3D%7B%22status%22%3A%5B%22running%22%5D%7D" ) self.assertEqual(r.status_code, 200, r.text) payload = r.json() containerAmnt = len(payload) self.assertGreater(containerAmnt, 0) def test_list_all(self): r = requests.get(self.uri("/containers/json?all=true")) self.assertEqual(r.status_code, 200, r.text) self.assertId(r.content) def test_inspect(self): r = requests.get(self.uri(self.resolve_container("/containers/{}/json"))) self.assertEqual(r.status_code, 200, r.text) self.assertId(r.content) _ = parse(r.json()["Created"]) r = requests.post( self.podman_url + "/v1.40/containers/create?name=topcontainer", json={ "Cmd": ["top"], "Image": "alpine:latest", "Healthcheck": { "Test": ["CMD", "pidof", "top"], "Interval": 5000000000, "Timeout": 2000000000, "Retries": 3, "StartPeriod": 5000000000, }, }, ) self.assertEqual(r.status_code, 201, r.text) payload = r.json() container_id = payload["Id"] self.assertIsNotNone(container_id) r = requests.get(self.podman_url + f"/v1.40/containers/{container_id}/json") self.assertEqual(r.status_code, 200, r.text) self.assertId(r.content) out = r.json() self.assertIsNotNone(out["State"].get("Health")) self.assertListEqual(["CMD", "pidof", "top"], out["Config"]["Healthcheck"]["Test"]) self.assertEqual(5000000000, out["Config"]["Healthcheck"]["Interval"]) self.assertEqual(2000000000, out["Config"]["Healthcheck"]["Timeout"]) self.assertEqual(3, out["Config"]["Healthcheck"]["Retries"]) self.assertEqual(5000000000, out["Config"]["Healthcheck"]["StartPeriod"]) r = requests.get(self.uri(f"/containers/{container_id}/json")) self.assertEqual(r.status_code, 200, r.text) self.assertId(r.content) out = r.json() hc = out["Config"]["Healthcheck"]["Test"] self.assertListEqual(["CMD", "pidof", "top"], hc) r = requests.post(self.podman_url + f"/v1.40/containers/{container_id}/start") self.assertEqual(r.status_code, 204, r.text) r = requests.get(self.podman_url + f"/v1.40/containers/{container_id}/json") self.assertEqual(r.status_code, 200, r.text) out = r.json() state = out["State"]["Health"] self.assertIsInstance(state, dict) def test_stats(self): r = requests.get(self.uri(self.resolve_container("/containers/{}/stats?stream=false"))) self.assertIn(r.status_code, (200, 409), r.text) if r.status_code == 200: self.assertId(r.content) r = requests.get( self.uri(self.resolve_container("/containers/{}/stats?stream=false&one-shot=true")) ) self.assertIn(r.status_code, (200, 409), r.text) if r.status_code == 200: self.assertId(r.content) def test_delete(self): r = requests.delete(self.uri(self.resolve_container("/containers/{}?force=true"))) self.assertEqual(r.status_code, 204, r.text) def test_stop(self): r = requests.post(self.uri(self.resolve_container("/containers/{}/start"))) self.assertIn(r.status_code, (204, 304), r.text) r = requests.post(self.uri(self.resolve_container("/containers/{}/stop"))) self.assertIn(r.status_code, (204, 304), r.text) def test_start(self): r = requests.post(self.uri(self.resolve_container("/containers/{}/stop"))) self.assertIn(r.status_code, (204, 304), r.text) r = requests.post(self.uri(self.resolve_container("/containers/{}/start"))) self.assertIn(r.status_code, (204, 304), r.text) def test_restart(self): r = requests.post(self.uri(self.resolve_container("/containers/{}/start"))) self.assertIn(r.status_code, (204, 304), r.text) r = requests.post(self.uri(self.resolve_container("/containers/{}/restart")), timeout=5) self.assertEqual(r.status_code, 204, r.text) def test_resize(self): r = requests.post(self.uri(self.resolve_container("/containers/{}/resize?h=43&w=80"))) self.assertIn(r.status_code, (200, 409), r.text) if r.status_code == 200: self.assertEqual(r.text, "", r.text) def test_attach(self): self.skipTest("FIXME: Test timeouts") r = requests.post(self.uri(self.resolve_container("/containers/{}/attach")), timeout=5) self.assertIn(r.status_code, (101, 500), r.text) def test_logs(self): r = requests.get(self.uri(self.resolve_container("/containers/{}/logs?stdout=true"))) self.assertEqual(r.status_code, 200, r.text) r = requests.post( self.podman_url + "/v1.40/containers/create?name=topcontainer", json={"Cmd": ["top", "ls"], "Image": "alpine:latest"}, ) self.assertEqual(r.status_code, 201, r.text) payload = r.json() container_id = payload["Id"] self.assertIsNotNone(container_id) r = requests.get( self.podman_url + f"/v1.40/containers/{payload['Id']}/logs?follow=false&stdout=true&until=0" ) self.assertEqual(r.status_code, 200, r.text) r = requests.get( self.podman_url + f"/v1.40/containers/{payload['Id']}/logs?follow=false&stdout=true&until=1" ) self.assertEqual(r.status_code, 200, r.text) def test_commit(self): r = requests.post(self.uri(self.resolve_container("/commit?container={}"))) self.assertEqual(r.status_code, 200, r.text) self.assertId(r.content) obj = r.json() self.assertIsInstance(obj, dict) def test_prune(self): name = f"Container_{random.getrandbits(160):x}" r = requests.post( self.podman_url + f"/v1.40/containers/create?name={name}", json={ "Cmd": ["cp", "/etc/motd", "/motd.size_test"], "Image": "alpine:latest", "NetworkDisabled": True, }, ) self.assertEqual(r.status_code, 201, r.text) create = r.json() r = requests.post(self.podman_url + f"/v1.40/containers/{create['Id']}/start") self.assertEqual(r.status_code, 204, r.text) r = requests.post(self.podman_url + f"/v1.40/containers/{create['Id']}/wait") self.assertEqual(r.status_code, 200, r.text) wait = r.json() self.assertEqual(wait["StatusCode"], 0, wait["Error"]) prune = requests.post(self.podman_url + "/v1.40/containers/prune") self.assertEqual(prune.status_code, 200, prune.status_code) prune_payload = prune.json() self.assertGreater(prune_payload["SpaceReclaimed"], 0) self.assertIn(create["Id"], prune_payload["ContainersDeleted"]) # Delete any orphaned containers r = requests.get(self.podman_url + "/v1.40/containers/json?all=true") self.assertEqual(r.status_code, 200, r.text) for self.resolve_container in r.json(): requests.delete( self.podman_url + f"/v1.40/containers/{self.resolve_container['Id']}?force=true" ) # Image prune here tied to containers freeing up prune = requests.post(self.podman_url + "/v1.40/images/prune") self.assertEqual(prune.status_code, 200, prune.text) prune_payload = prune.json() self.assertGreater(prune_payload["SpaceReclaimed"], 0) # FIXME need method to determine which image is going to be "pruned" to fix test # TODO should handler be recursive when deleting images? # self.assertIn(img["Id"], prune_payload["ImagesDeleted"][1]["Deleted"]) # FIXME (@vrothberg): I commented this line out during the `libimage` migration. # It doesn't make sense to report anything to be deleted if the reclaimed space # is zero. I think the test needs some rewrite. # self.assertIsNotNone(prune_payload["ImagesDeleted"][1]["Deleted"]) def test_status(self): r = requests.post( self.podman_url + "/v1.40/containers/create?name=topcontainer", json={"Cmd": ["top"], "Image": "alpine:latest"}, ) self.assertEqual(r.status_code, 201, r.text) payload = r.json() container_id = payload["Id"] self.assertIsNotNone(container_id) r = requests.get( self.podman_url + "/v1.40/containers/json", params={"all": "true", "filters": f'{{"id":["{container_id}"]}}'}, ) self.assertEqual(r.status_code, 200, r.text) payload = r.json() self.assertEqual(payload[0]["Status"], "Created") r = requests.post(self.podman_url + f"/v1.40/containers/{container_id}/start") self.assertEqual(r.status_code, 204, r.text) r = requests.get( self.podman_url + "/v1.40/containers/json", params={"all": "true", "filters": f'{{"id":["{container_id}"]}}'}, ) self.assertEqual(r.status_code, 200, r.text) payload = r.json() self.assertTrue(str(payload[0]["Status"]).startswith("Up")) r = requests.post(self.podman_url + f"/v1.40/containers/{container_id}/pause") self.assertEqual(r.status_code, 204, r.text) r = requests.get( self.podman_url + "/v1.40/containers/json", params={"all": "true", "filters": f'{{"id":["{container_id}"]}}'}, ) self.assertEqual(r.status_code, 200, r.text) payload = r.json() self.assertTrue(str(payload[0]["Status"]).startswith("Up")) self.assertTrue(str(payload[0]["Status"]).endswith("(Paused)")) r = requests.post(self.podman_url + f"/v1.40/containers/{container_id}/unpause") self.assertEqual(r.status_code, 204, r.text) r = requests.post(self.podman_url + f"/v1.40/containers/{container_id}/stop") self.assertEqual(r.status_code, 204, r.text) r = requests.get( self.podman_url + "/v1.40/containers/json", params={"all": "true", "filters": f'{{"id":["{container_id}"]}}'}, ) self.assertEqual(r.status_code, 200, r.text) payload = r.json() self.assertTrue(str(payload[0]["Status"]).startswith("Exited")) r = requests.delete(self.podman_url + f"/v1.40/containers/{container_id}") self.assertEqual(r.status_code, 204, r.text) def test_top_no_stream(self): uri = self.uri(self.resolve_container("/containers/{}/top")) q = queue.Queue() def _impl(fifo): fifo.put(requests.get(uri, params={"stream": False}, timeout=2)) top = threading.Thread(target=_impl, args=(q,)) top.start() time.sleep(2) self.assertFalse(top.is_alive(), f"GET {uri} failed to return in 2s") qr = q.get(False) self.assertEqual(qr.status_code, 200, qr.text) qr.close() top.join() def test_top_stream(self): uri = self.uri(self.resolve_container("/containers/{}/top")) q = queue.Queue() stop_thread = False def _impl(fifo, stop): try: with requests.get(uri, params={"stream": True, "delay": 1}, stream=True) as r: r.raise_for_status() fifo.put(r) for buf in r.iter_lines(chunk_size=None): if stop(): break fifo.put(buf) except Exception: pass top = threading.Thread(target=_impl, args=(q, (lambda: stop_thread))) top.start() time.sleep(4) self.assertTrue(top.is_alive(), f"GET {uri} exited too soon") stop_thread = True for _ in range(10): try: qr = q.get_nowait() if qr is not None: self.assertEqual(qr.status_code, 200) qr.close() break except queue.Empty: pass finally: time.sleep(1) else: self.fail("Server failed to respond in 10s") top.join() if __name__ == "__main__": unittest.main()