path: root/test/apiv2/rest_api/test_rest_v2_0_0.py

import json
import random
import string
import subprocess
import unittest
from multiprocessing import Process

import requests
import sys
import time
from dateutil.parser import parse

from test.apiv2.rest_api import Podman

PODMAN_URL = "http://localhost:8080"


def _url(path):
    return PODMAN_URL + "/v2.0.0/libpod" + path


def ctnr(path):
    try:
        r = requests.get(_url("/containers/json?all=true"))
        ctnrs = json.loads(r.text)
    except Exception as e:
        msg = f"Bad container response: {e}"
        if r is not None:
            msg = msg + " " + r.text
        sys.stderr.write(msg + "\n")
        raise
    return path.format(ctnrs[0]["Id"])


def validateObjectFields(buffer):
    objs = json.loads(buffer)
    if not isinstance(objs, dict):
        for o in objs:
            _ = o["Id"]
    else:
        _ = objs["Id"]
    return objs


class TestApi(unittest.TestCase):
    podman = None  # initialized podman configuration for tests
    service = None  # podman service instance

    def setUp(self):
        super().setUp()

        try:
            TestApi.podman.run("run", "alpine", "/bin/ls", check=True)
        except subprocess.CalledProcessError as e:
            if e.stdout:
                sys.stdout.write("\nRun Stdout:\n" + e.stdout.decode("utf-8"))
            if e.stderr:
                sys.stderr.write("\nRun Stderr:\n" + e.stderr.decode("utf-8"))
            raise

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        TestApi.podman = Podman()
        TestApi.service = TestApi.podman.open("system", "service", "tcp:localhost:8080", "--time=0")
        # give the service some time to be ready...
        time.sleep(2)

        returncode = TestApi.service.poll()
        if returncode is not None:
            raise subprocess.CalledProcessError(returncode, "podman system service")

        r = requests.post(_url("/images/pull?reference=docker.io%2Falpine%3Alatest"))
        if r.status_code != 200:
            raise subprocess.CalledProcessError(
                r.status_code, f"podman images pull docker.io/alpine:latest {r.text}"
            )

    @classmethod
    def tearDownClass(cls):
        TestApi.service.terminate()
        stdout, stderr = TestApi.service.communicate(timeout=0.5)
        if stdout:
            sys.stdout.write("\nService Stdout:\n" + stdout.decode("utf-8"))
        if stderr:
            sys.stderr.write("\nService Stderr:\n" + stderr.decode("utf-8"))
        return super().tearDownClass()

    def test_info(self):
        r = requests.get(_url("/info"))
        self.assertEqual(r.status_code, 200)
        self.assertIsNotNone(r.content)
        _ = json.loads(r.text)

        info = requests.get(PODMAN_URL + "/v1.40/info")
        self.assertEqual(info.status_code, 200, info.content)
        _ = json.loads(info.text)

    def test_events(self):
        r = requests.get(_url("/events?stream=false"))
        self.assertEqual(r.status_code, 200, r.text)
        self.assertIsNotNone(r.content)

        report = r.text.splitlines()
        self.assertGreater(len(report), 0, "No events found!")
        for line in report:
            obj = json.loads(line)
            # Actor.ID is uppercase for compatibility
            self.assertIn("ID", obj["Actor"])

    def test_containers(self):
        r = requests.get(_url("/containers/json"), timeout=5)
        self.assertEqual(r.status_code, 200, r.text)
        obj = json.loads(r.text)
        self.assertEqual(len(obj), 0)

    def test_containers_all(self):
        r = requests.get(_url("/containers/json?all=true"))
        self.assertEqual(r.status_code, 200, r.text)
        validateObjectFields(r.text)

    def test_inspect_container(self):
        r = requests.get(_url(ctnr("/containers/{}/json")))
        self.assertEqual(r.status_code, 200, r.text)
        obj = validateObjectFields(r.content)
        _ = parse(obj["Created"])

    def test_stats(self):
        r = requests.get(_url(ctnr("/containers/{}/stats?stream=false")))
        self.assertIn(r.status_code, (200, 409), r.text)
        if r.status_code == 200:
            validateObjectFields(r.text)

    def test_delete_containers(self):
        r = requests.delete(_url(ctnr("/containers/{}")))
        self.assertEqual(r.status_code, 204, r.text)

    def test_stop_containers(self):
        r = requests.post(_url(ctnr("/containers/{}/start")))
        self.assertIn(r.status_code, (204, 304), r.text)

        r = requests.post(_url(ctnr("/containers/{}/stop")))
        self.assertIn(r.status_code, (204, 304), r.text)

    def test_start_containers(self):
        r = requests.post(_url(ctnr("/containers/{}/stop")))
        self.assertIn(r.status_code, (204, 304), r.text)

        r = requests.post(_url(ctnr("/containers/{}/start")))
        self.assertIn(r.status_code, (204, 304), r.text)

    def test_restart_containers(self):
        r = requests.post(_url(ctnr("/containers/{}/start")))
        self.assertIn(r.status_code, (204, 304), r.text)

        r = requests.post(_url(ctnr("/containers/{}/restart")), timeout=5)
        self.assertEqual(r.status_code, 204, r.text)

    def test_resize(self):
        r = requests.post(_url(ctnr("/containers/{}/resize?h=43&w=80")))
        self.assertIn(r.status_code, (200, 409), r.text)
        if r.status_code == 200:
            self.assertIsNone(r.text)

    def test_attach_containers(self):
        self.skipTest("FIXME: Test timeouts")
        r = requests.post(_url(ctnr("/containers/{}/attach")), timeout=5)
        self.assertIn(r.status_code, (101, 500), r.text)

    def test_logs_containers(self):
        r = requests.get(_url(ctnr("/containers/{}/logs?stdout=true")))
        self.assertEqual(r.status_code, 200, r.text)

    # TODO Need to support Docker-py order of network/container creates
    def test_post_create_compat_connect(self):
        """Create network and container then connect to network"""
        net_default = requests.post(
            PODMAN_URL + "/v1.40/networks/create", json={"Name": "TestDefaultNetwork"}
        )
        self.assertEqual(net_default.status_code, 201, net_default.text)

        create = requests.post(
            PODMAN_URL + "/v1.40/containers/create?name=postCreateConnect",
            json={
                "Cmd": ["top"],
                "Image": "alpine:latest",
                "NetworkDisabled": False,
                # FIXME adding these 2 lines cause: (This is sampled from docker-py)
                #   "network already exists","message":"container
                #  01306e499df5441560d70071a54342611e422a94de20865add50a9565fd79fb9 is already connected to CNI
                #  network \"TestDefaultNetwork\": network already exists"
                # "HostConfig": {"NetworkMode": "TestDefaultNetwork"},
                # "NetworkingConfig": {"EndpointsConfig": {"TestDefaultNetwork": None}},
                # FIXME These two lines cause:
                # CNI network \"TestNetwork\" not found","message":"error configuring network namespace for container
                # 369ddfa7d3211ebf1fbd5ddbff91bd33fa948858cea2985c133d6b6507546dff: CNI network \"TestNetwork\" not
                # found"
                # "HostConfig": {"NetworkMode": "TestNetwork"},
                # "NetworkingConfig": {"EndpointsConfig": {"TestNetwork": None}},
                # FIXME no networking defined cause: (note this error is from the container inspect below)
                # "internal libpod error","message":"network inspection mismatch: asked to join 2 CNI network(s) [
                # TestDefaultNetwork podman], but have information on 1 network(s): internal libpod error"
            },
        )
        self.assertEqual(create.status_code, 201, create.text)
        payload = json.loads(create.text)
        self.assertIsNotNone(payload["Id"])

        start = requests.post(PODMAN_URL + f"/v1.40/containers/{payload['Id']}/start")
        self.assertEqual(start.status_code, 204, start.text)

        connect = requests.post(
            PODMAN_URL + "/v1.40/networks/TestDefaultNetwork/connect",
            json={"Container": payload["Id"]},
        )
        self.assertEqual(connect.status_code, 200, connect.text)
        self.assertEqual(connect.text, "OK\n")

        inspect = requests.get(f"{PODMAN_URL}/v1.40/containers/{payload['Id']}/json")
        self.assertEqual(inspect.status_code, 200, inspect.text)

        payload = json.loads(inspect.text)
        self.assertFalse(payload["Config"].get("NetworkDisabled", False))

        self.assertEqual(
            "TestDefaultNetwork",
            payload["NetworkSettings"]["Networks"]["TestDefaultNetwork"]["NetworkID"],
        )
        # TODO restore this to test, when joining multiple networks possible
        # self.assertEqual(
        #     "TestNetwork",
        #     payload["NetworkSettings"]["Networks"]["TestNetwork"]["NetworkID"],
        # )
        # TODO Need to support network aliases
        # self.assertIn(
        #     "test_post_create",
        #     payload["NetworkSettings"]["Networks"]["TestNetwork"]["Aliases"],
        # )

    def test_post_create_compat(self):
        """Create network and connect container during create"""
        net = requests.post(PODMAN_URL + "/v1.40/networks/create", json={"Name": "TestNetwork"})
        self.assertEqual(net.status_code, 201, net.text)

        create = requests.post(
            PODMAN_URL + "/v1.40/containers/create?name=postCreate",
            json={
                "Cmd": ["date"],
                "Image": "alpine:latest",
                "NetworkDisabled": False,
                "HostConfig": {"NetworkMode": "TestNetwork"},
            },
        )
        self.assertEqual(create.status_code, 201, create.text)
        payload = json.loads(create.text)
        self.assertIsNotNone(payload["Id"])

        inspect = requests.get(f"{PODMAN_URL}/v1.40/containers/{payload['Id']}/json")
        self.assertEqual(inspect.status_code, 200, inspect.text)
        payload = json.loads(inspect.text)
        self.assertFalse(payload["Config"].get("NetworkDisabled", False))
        self.assertEqual(
            "TestNetwork",
            payload["NetworkSettings"]["Networks"]["TestNetwork"]["NetworkID"],
        )

    def test_commit(self):
        r = requests.post(_url(ctnr("/commit?container={}")))
        self.assertEqual(r.status_code, 200, r.text)

        obj = json.loads(r.content)
        self.assertIsInstance(obj, dict)
        self.assertIn("Id", obj)

    def test_images_compat(self):
        r = requests.get(PODMAN_URL + "/v1.40/images/json")
        self.assertEqual(r.status_code, 200, r.text)

        # See https://docs.docker.com/engine/api/v1.40/#operation/ImageList
        required_keys = (
            "Id",
            "ParentId",
            "RepoTags",
            "RepoDigests",
            "Created",
            "Size",
            "SharedSize",
            "VirtualSize",
            "Labels",
            "Containers",
        )
        objs = json.loads(r.content)
        self.assertIn(type(objs), (list,))
        for o in objs:
            self.assertIsInstance(o, dict)
            for k in required_keys:
                self.assertIn(k, o)

    def test_inspect_image_compat(self):
        r = requests.get(PODMAN_URL + "/v1.40/images/alpine/json")
        self.assertEqual(r.status_code, 200, r.text)

        # See https://docs.docker.com/engine/api/v1.40/#operation/ImageInspect
        required_keys = (
            "Id",
            "Parent",
            "Comment",
            "Created",
            "Container",
            "DockerVersion",
            "Author",
            "Architecture",
            "Os",
            "Size",
            "VirtualSize",
            "GraphDriver",
            "RootFS",
            "Metadata",
        )

        obj = json.loads(r.content)
        self.assertIn(type(obj), (dict,))
        for k in required_keys:
            self.assertIn(k, obj)
        _ = parse(obj["Created"])

    def test_delete_image_compat(self):
        r = requests.delete(PODMAN_URL + "/v1.40/images/alpine?force=true")
        self.assertEqual(r.status_code, 200, r.text)
        obj = json.loads(r.content)
        self.assertIn(type(obj), (list,))

    def test_pull(self):
        r = requests.post(_url("/images/pull?reference=alpine"), timeout=15)
        self.assertEqual(r.status_code, 200, r.status_code)
        text = r.text
        keys = {
            "error": False,
            "id": False,
            "images": False,
            "stream": False,
        }
        # Read and record stanza's from pull
        for line in str.splitlines(text):
            obj = json.loads(line)
            key_list = list(obj.keys())
            for k in key_list:
                keys[k] = True

        self.assertFalse(keys["error"], "Expected no errors")
        self.assertTrue(keys["id"], "Expected to find id stanza")
        self.assertTrue(keys["images"], "Expected to find images stanza")
        self.assertTrue(keys["stream"], "Expected to find stream progress stanza's")

    def test_search_compat(self):
        # Had issues with this test hanging when repositories not happy
        def do_search():
            r = requests.get(PODMAN_URL + "/v1.40/images/search?term=alpine", timeout=5)
            self.assertEqual(r.status_code, 200, r.text)
            objs = json.loads(r.text)
            self.assertIn(type(objs), (list,))

        search = Process(target=do_search)
        search.start()
        search.join(timeout=10)
        self.assertFalse(search.is_alive(), "/images/search took too long")

    def test_ping(self):
        required_headers = (
            "API-Version",
            "Builder-Version",
            "Docker-Experimental",
            "Cache-Control",
            "Pragma",
            "Pragma",
        )

        def check_headers(req):
            for k in required_headers:
                self.assertIn(k, req.headers)

        r = requests.get(PODMAN_URL + "/_ping")
        self.assertEqual(r.status_code, 200, r.text)
        self.assertEqual(r.text, "OK")
        check_headers(r)

        r = requests.head(PODMAN_URL + "/_ping")
        self.assertEqual(r.status_code, 200, r.text)
        self.assertEqual(r.text, "")
        check_headers(r)

        r = requests.get(_url("/_ping"))
        self.assertEqual(r.status_code, 200, r.text)
        self.assertEqual(r.text, "OK")
        check_headers(r)

        r = requests.head(_url("/_ping"))
        self.assertEqual(r.status_code, 200, r.text)
        self.assertEqual(r.text, "")
        check_headers(r)

    def test_history_compat(self):
        r = requests.get(PODMAN_URL + "/v1.40/images/alpine/history")
        self.assertEqual(r.status_code, 200, r.text)

        # See https://docs.docker.com/engine/api/v1.40/#operation/ImageHistory
        required_keys = ("Id", "Created", "CreatedBy", "Tags", "Size", "Comment")

        objs = json.loads(r.content)
        self.assertIn(type(objs), (list,))
        for o in objs:
            self.assertIsInstance(o, dict)
            for k in required_keys:
                self.assertIn(k, o)

    def test_network_compat(self):
        name = "Network_" + "".join(random.choice(string.ascii_letters) for i in range(10))

        # Cannot test for 0 existing networks because default "podman" network always exists

        create = requests.post(PODMAN_URL + "/v1.40/networks/create", json={"Name": name})
        self.assertEqual(create.status_code, 201, create.content)
        obj = json.loads(create.content)
        self.assertIn(type(obj), (dict,))
        self.assertIn("Id", obj)
        ident = obj["Id"]
        self.assertNotEqual(name, ident)

        ls = requests.get(PODMAN_URL + "/v1.40/networks")
        self.assertEqual(ls.status_code, 200, ls.content)
        objs = json.loads(ls.content)
        self.assertIn(type(objs), (list,))

        found = False
        for network in objs:
            if network["Name"] == name:
                found = True
        self.assertTrue(found, f"Network {name} not found")

        inspect = requests.get(PODMAN_URL + f"/v1.40/networks/{ident}")
        self.assertEqual(inspect.status_code, 200, inspect.content)
        obj = json.loads(create.content)
        self.assertIn(type(obj), (dict,))

        inspect = requests.delete(PODMAN_URL + f"/v1.40/networks/{ident}")
        self.assertEqual(inspect.status_code, 204, inspect.content)
        inspect = requests.get(PODMAN_URL + f"/v1.40/networks/{ident}")
        self.assertEqual(inspect.status_code, 404, inspect.content)

        prune = requests.post(PODMAN_URL + "/v1.40/networks/prune")
        self.assertEqual(prune.status_code, 405, prune.content)

    def test_volumes_compat(self):
        name = "Volume_" + "".join(random.choice(string.ascii_letters) for i in range(10))

        ls = requests.get(PODMAN_URL + "/v1.40/volumes")
        self.assertEqual(ls.status_code, 200, ls.content)

        # See https://docs.docker.com/engine/api/v1.40/#operation/VolumeList
        required_keys = (
            "Volumes",
            "Warnings",
        )

        obj = json.loads(ls.content)
        self.assertIn(type(obj), (dict,))
        for k in required_keys:
            self.assertIn(k, obj)

        create = requests.post(PODMAN_URL + "/v1.40/volumes/create", json={"Name": name})
        self.assertEqual(create.status_code, 201, create.content)

        # See https://docs.docker.com/engine/api/v1.40/#operation/VolumeCreate
        # and https://docs.docker.com/engine/api/v1.40/#operation/VolumeInspect
        required_keys = (
            "Name",
            "Driver",
            "Mountpoint",
            "Labels",
            "Scope",
            "Options",
        )

        obj = json.loads(create.content)
        self.assertIn(type(obj), (dict,))
        for k in required_keys:
            self.assertIn(k, obj)
        self.assertEqual(obj["Name"], name)

        inspect = requests.get(PODMAN_URL + f"/v1.40/volumes/{name}")
        self.assertEqual(inspect.status_code, 200, inspect.content)

        obj = json.loads(create.content)
        self.assertIn(type(obj), (dict,))
        for k in required_keys:
            self.assertIn(k, obj)

        rm = requests.delete(PODMAN_URL + f"/v1.40/volumes/{name}")
        self.assertEqual(rm.status_code, 204, rm.content)

        prune = requests.post(PODMAN_URL + "/v1.40/volumes/prune")
        self.assertEqual(prune.status_code, 200, prune.content)

    def test_auth_compat(self):
        r = requests.post(
            PODMAN_URL + "/v1.40/auth",
            json={
                "username": "bozo",
                "password": "wedontneednopasswords",
                "serveraddress": "https://localhost/v1.40/",
            },
        )
        self.assertEqual(r.status_code, 404, r.content)

    def test_version(self):
        r = requests.get(PODMAN_URL + "/v1.40/version")
        self.assertEqual(r.status_code, 200, r.content)

        r = requests.get(_url("/version"))
        self.assertEqual(r.status_code, 200, r.content)

    def test_df_compat(self):
        r = requests.get(PODMAN_URL + "/v1.40/system/df")
        self.assertEqual(r.status_code, 200, r.content)

        obj = json.loads(r.content)
        self.assertIn("Images", obj)
        self.assertIn("Containers", obj)
        self.assertIn("Volumes", obj)
        self.assertIn("BuildCache", obj)


if __name__ == "__main__":
    unittest.main()