import io import multiprocessing import queue import random import subprocess import tarfile import threading import unittest import requests import os 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_status(self): r = requests.get( self.podman_url + "/v1.40/containers/json?filters=%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_filters_label(self): r = requests.get( self.podman_url + "/v1.40/containers/json?filters=%7B%22label%22%3A%5B%22nonexistlabel%22%5D%7D" ) self.assertEqual(r.status_code, 200, r.text) payload = r.json() containerAmnt = len(payload) self.assertEqual(containerAmnt, 0) def test_list_filters_label_not(self): r = requests.get( self.podman_url + "/v1.40/containers/json?filters=%7B%22label%21%22%3A%5B%22nonexistlabel%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, 200, 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): r = requests.post( self.podman_url + "/v1.40/containers/create?name=topcontainer", json={"Cmd": ["sh", "-c", "echo podman; sleep 100"], "Image": "alpine:latest"}, ) self.assertEqual(r.status_code, 201, r.text) payload = r.json() r = requests.post( self.podman_url + f"/v1.40/containers/{payload['Id']}/start" ) self.assertEqual(r.status_code, 204, r.text) # wait for the log message to appear to avoid flakes on slow systems # with the /attach?logs=true test below for _ in range(5): r = requests.get( self.podman_url + f"/v1.40/containers/{payload['Id']}/logs?stdout=true" ) self.assertIn(r.status_code, (101, 200), r.text) if r.content == b"\x01\x00\x00\x00\x00\x00\x00\x07podman\n": break time.sleep(1) r = requests.post( self.podman_url + f"/v1.40/containers/{payload['Id']}/attach?logs=true&stream=false" ) self.assertIn(r.status_code, (101, 200), r.text) # see the attach format docs, stdout = 1, length = 7, message = podman\n self.assertEqual(r.content, b"\x01\x00\x00\x00\x00\x00\x00\x07podman\n", r.text) r = requests.post( self.podman_url + f"/v1.40/containers/{payload['Id']}/stop?t=0" ) self.assertEqual(r.status_code, 204, r.text) requests.delete( self.podman_url + f"/v1.40/containers/{payload['Id']}?force=true" ) 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() def test_memory(self): r = requests.post( self.podman_url + "/v1.4.0/libpod/containers/create", json={ "Name": "memory", "Cmd": ["top"], "Image": "alpine:latest", "Resource_Limits": { "Memory":{ "Limit": 1000, }, "CPU":{ "Shares": 200, }, }, }, ) 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.assertEqual(2000, out["HostConfig"]["MemorySwap"]) self.assertEqual(1000, out["HostConfig"]["Memory"]) def test_host_config_port_bindings(self): # create a container with two ports exposed, but only one of the ports bound r = requests.post( self.podman_url + "/v1.40/containers/create", json={ "Name": "memory", "Cmd": ["top"], "Image": "alpine:latest", "HostConfig": { "PortBindings": { "8080": [{"HostPort": "87634"}] } }, "ExposedPorts": { "8080": {}, "8081": {} } }, ) 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) inspect_response = r.json() # both ports are in the config self.assertEqual(2, len(inspect_response["Config"]["ExposedPorts"])) self.assertTrue("8080/tcp" in inspect_response["Config"]["ExposedPorts"]) self.assertTrue("8081/tcp" in inspect_response["Config"]["ExposedPorts"]) # only 8080 one port is bound self.assertEqual(1, len(inspect_response["HostConfig"]["PortBindings"])) self.assertTrue("8080/tcp" in inspect_response["HostConfig"]["PortBindings"]) self.assertFalse("8081/tcp" in inspect_response["HostConfig"]["PortBindings"]) def test_host_config_cgroupns_mode(self): for mode in ["private", "host"]: r = requests.post( self.podman_url + "/v1.40/containers/create", json={ "Name": "cgroupns_" + mode, "Cmd": ["top"], "Image": "alpine:latest", "HostConfig": { "CgroupnsMode": mode } }, ) 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) inspect_response = r.json() self.assertEqual(mode, inspect_response["HostConfig"]["CgroupnsMode"]) r = requests.delete(self.podman_url + f"/v1.40/containers/{container_id}") self.assertEqual(r.status_code, 204, r.text) def execute_process(cmd): return subprocess.run( cmd, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) def create_named_network_ns(network_ns_name): execute_process(f"ip netns add {network_ns_name}") execute_process(f"ip netns exec {network_ns_name} ip link add enp2s0 type veth peer name eth0") execute_process(f"ip netns exec {network_ns_name} ip addr add 10.0.1.0/24 dev eth0") execute_process(f"ip netns exec {network_ns_name} ip link set eth0 up") execute_process(f"ip netns exec {network_ns_name} ip link add enp2s1 type veth peer name eth1") execute_process(f"ip netns exec {network_ns_name} ip addr add 10.0.2.0/24 dev eth1") execute_process(f"ip netns exec {network_ns_name} ip link set eth1 up") def delete_named_network_ns(network_ns_name): execute_process(f"ip netns delete {network_ns_name}") class ContainerCompatibleAPITestCase(APITestCase): def test_inspect_network(self): if os.getuid() != 0: self.skipTest("test needs to be executed as root!") try: network_ns_name = "test-compat-api" create_named_network_ns(network_ns_name) self.podman.run("rm", "--all", "--force", check=True) self.podman.run("run", "--net", f"ns:/run/netns/{network_ns_name}", "-d", "alpine", "top", check=True) r = requests.post(self.uri(self.resolve_container("/containers/{}/start"))) self.assertIn(r.status_code, (204, 304), r.text) r = requests.get(self.compat_uri(self.resolve_container("/containers/{}/json"))) self.assertEqual(r.status_code, 200, r.text) self.assertId(r.content) out = r.json() self.assertEqual("10.0.2.0", out["NetworkSettings"]["SecondaryIPAddresses"][0]["Addr"]) self.assertEqual(24, out["NetworkSettings"]["SecondaryIPAddresses"][0]["PrefixLen"]) finally: delete_named_network_ns(network_ns_name) if __name__ == "__main__": unittest.main()