import json import unittest from multiprocessing import Process import requests from dateutil.parser import parse from .fixtures import APITestCase class ImageTestCase(APITestCase): def test_list(self): r = requests.get(self.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", ) images = r.json() self.assertIsInstance(images, list) for item in images: self.assertIsInstance(item, dict) for k in required_keys: self.assertIn(k, item) # Id should be prefixed with sha256: (#11645) self.assertIn("sha256:",item['Id']) def test_inspect(self): r = requests.get(self.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", "DockerVersion", "Author", "Architecture", "Os", "Size", "VirtualSize", "GraphDriver", "RootFS", "Metadata", ) image = r.json() self.assertIsInstance(image, dict) for item in required_keys: self.assertIn(item, image) _ = parse(image["Created"]) # Id should be prefixed with sha256: (#11645) self.assertIn("sha256:",image['Id']) def test_delete(self): r = requests.delete(self.compat_uri("images/alpine?force=true")) self.assertEqual(r.status_code, 409, r.text) def test_pull(self): r = requests.post(self.uri("/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") r = requests.post(self.uri("/images/pull?reference=alpine&quiet=true"), 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.assertFalse(keys["stream"], "Expected to find stream progress stanza's") def test_create(self): r = requests.post( self.podman_url + "/v1.40/images/create?fromImage=alpine&platform=linux/amd64/v8", timeout=15, ) self.assertEqual(r.status_code, 200, r.text) r = requests.post( self.podman_url + "/v1.40/images/create?fromSrc=-&repo=fedora&message=testing123&platform=linux/amd64", timeout=15, ) self.assertEqual(r.status_code, 200, r.text) def test_search_compat(self): url = self.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, f"#1: {r.text}") self.assertIsInstance(r.json(), list) def do_search2(): payload = {"term": "alpine", "limit": 1} r = requests.get(url, params=payload, timeout=5) self.assertEqual(r.status_code, 200, f"#2: {r.text}") results = r.json() self.assertIsInstance(results, list) self.assertEqual(len(results), 1) def do_search3(): # FIXME: Research if quay.io supports is-official and which image is "official" return payload = {"term": "thanos", "filters": '{"is-official":["true"]}'} r = requests.get(url, params=payload, timeout=5) self.assertEqual(r.status_code, 200, f"#3: {r.text}") results = r.json() self.assertIsInstance(results, list) # There should be only one official image self.assertEqual(len(results), 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, f"#4: {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, f"#5: {r.text}") i = 1 for fn in [do_search1, do_search2, do_search3, do_search4, do_search5]: with self.subTest(i=i): search = Process(target=fn) search.start() search.join(timeout=10) self.assertFalse(search.is_alive(), f"#{i} /images/search took too long") # 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_history(self): r = requests.get(self.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") changes = r.json() self.assertIsInstance(changes, list) for change in changes: self.assertIsInstance(change, dict) for k in required_keys: self.assertIn(k, change) def test_tree(self): r = requests.get(self.uri("/images/alpine/tree")) self.assertEqual(r.status_code, 200, r.text) tree = r.json() self.assertTrue(tree["Tree"].startswith("Image ID:"), r.text) if __name__ == "__main__": unittest.main()