text: import json
  import os
  import random
  import shutil
  import string
  import subprocess
  import sys
  import time
  import unittest
  from multiprocessing import Process
  
  import requests
  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):
          url = PODMAN_URL + "/v1.40/images/search"
          # Had issues with this test hanging when repositories not happy
          def do_search1():
              payload = {'term': 'alpine'}
              r = requests.get(url, params=payload, timeout=5)
              self.assertEqual(r.status_code, 200, r.text)
              objs = json.loads(r.text)
              self.assertIn(type(objs), (list,))
  
          def do_search2():
              payload = {'term': 'alpine', 'limit': 1}
              r = requests.get(url, params=payload, timeout=5)
              self.assertEqual(r.status_code, 200, r.text)
              objs = json.loads(r.text)
              self.assertIn(type(objs), (list,))
              self.assertEqual(len(objs), 1)
  
          def do_search3():
              payload = {'term': 'alpine', 'filters': {'is-official': True}}
              r = requests.get(url, params=payload, timeout=5)
              self.assertEqual(r.status_code, 200, r.text)
              objs = json.loads(r.text)
              self.assertIn(type(objs), (list,))
  # TODO: Request should return only one item, but it returns more. For now this check is commented out.
  #            self.assertEqual(len(objs), 1)
  
          def do_search4():
              headers = {'X-Registry-Auth': 'null'}
              payload = {'term': 'alpine'}
              r = requests.get(url, params=payload, headers=headers, timeout=5)
              self.assertEqual(r.status_code, 200, r.text)
  
          def do_search5():
              headers = {'X-Registry-Auth': 'invalid value'}
              payload = {'term': 'alpine'}
              r = requests.get(url, params=payload, headers=headers, timeout=5)
              self.assertEqual(r.status_code, 400, r.text)
  
          search_methods = [do_search1, do_search2, do_search3, do_search4, do_search5]
          for search_method in search_methods:
              search = Process(target=search_method)
              search.start()
              search.join(timeout=10)
              self.assertFalse(search.is_alive(), "/images/search took too long")
  
      def test_search_compat_with_(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, 404, 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)
  
          # recreate volume with data and then prune it
          r = requests.post(PODMAN_URL + "/v1.40/volumes/create", json={"Name": name})
          self.assertEqual(create.status_code, 201, create.content)
          create = json.loads(r.content)
          with open(os.path.join(create["Mountpoint"], "test_prune"), "w") as file:
              file.writelines(["This is a test\n", "This is a good test\n"])
  
          prune = requests.post(PODMAN_URL + "/v1.40/volumes/prune")
          self.assertEqual(prune.status_code, 200, prune.content)
          payload = json.loads(prune.content)
          self.assertIn(name, payload["VolumesDeleted"])
          self.assertGreater(payload["SpaceReclaimed"], 0)
  
      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)
  
      def test_prune_compat(self):
          name = "Ctnr_" + "".join(random.choice(string.ascii_letters) for i in range(10))
  
          r = requests.post(
              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 = json.loads(r.text)
  
          r = requests.post(PODMAN_URL + f"/v1.40/containers/{create['Id']}/start")
          self.assertEqual(r.status_code, 204, r.text)
  
          r = requests.post(PODMAN_URL + f"/v1.40/containers/{create['Id']}/wait")
          self.assertEqual(r.status_code, 200, r.text)
          wait = json.loads(r.text)
          self.assertEqual(wait["StatusCode"], 0, wait["Error"]["Message"])
  
          prune = requests.post(PODMAN_URL + "/v1.40/containers/prune")
          self.assertEqual(prune.status_code, 200, prune.status_code)
          prune_payload = json.loads(prune.text)
          self.assertGreater(prune_payload["SpaceReclaimed"], 0)
          self.assertIn(create["Id"], prune_payload["ContainersDeleted"])
  
          # Delete any orphaned containers
          r = requests.get(PODMAN_URL + "/v1.40/containers/json?all=true")
          self.assertEqual(r.status_code, 200, r.text)
          for ctnr in json.loads(r.text):
              requests.delete(PODMAN_URL + f"/v1.40/containers/{ctnr['Id']}?force=true")
  
          prune = requests.post(PODMAN_URL + "/v1.40/images/prune")
          self.assertEqual(prune.status_code, 200, prune.text)
          prune_payload = json.loads(prune.text)
          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"])
          self.assertIsNotNone(prune_payload["ImagesDeleted"][1]["Deleted"])
  
  
  if __name__ == "__main__":
      unittest.main()
  