import os import tarfile import unittest from typing import cast import requests from .fixtures import APITestCase from .fixtures.api_testcase import Artifact, ArtifactFile class ArtifactTestCase(APITestCase): def test_add(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) # Assert return response is json and contains digest add_response_json = add_response.json() self.assertIn("sha256:", cast(str, add_response_json["ArtifactDigest"])) inspect_response_json = artifact.do_artifact_inspect_request().json() artifact_layer = inspect_response_json["Manifest"]["layers"][0] # Assert uploaded artifact blob is expected size self.assertEqual(artifact_layer["size"], file.size) # Assert uploaded artifact blob has expected title annotation self.assertEqual( artifact_layer["annotations"]["org.opencontainers.image.title"], file.name ) # Assert blob media type fallback detection is working self.assertEqual(artifact_layer["mediaType"], "application/octet-stream") def test_add_with_replace(self): ARTIFACT_NAME = "quay.io/myimage/newartifact:latest" # Create first artifact file1 = ArtifactFile() parameters1: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file1.name, } artifact1 = Artifact(self.uri(""), ARTIFACT_NAME, parameters1, file1) add_response1 = artifact1.add() self.assertEqual(add_response1.status_code, 201, add_response1.text) original_digest = add_response1.json()["ArtifactDigest"] # Create replacement artifact with replace=true file2 = ArtifactFile() parameters2: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file2.name, "replace": "true", } artifact2 = Artifact(self.uri(""), ARTIFACT_NAME, parameters2, file2) add_response2 = artifact2.add() self.assertEqual(add_response2.status_code, 201, add_response2.text) new_digest = add_response2.json()["ArtifactDigest"] # Verify artifacts were replaced (different digests) self.assertNotEqual(original_digest, new_digest) # Verify artifact exists and has the new content inspect_response = artifact2.do_artifact_inspect_request() self.assertEqual(inspect_response.status_code, 200) inspect_json = inspect_response.json() artifact_layer = inspect_json["Manifest"]["layers"][0] # Should have only one layer (replaced, not appended) self.assertEqual(len(inspect_json["Manifest"]["layers"]), 1) # Verify it's the second file's content self.assertEqual(artifact_layer["size"], file2.size) self.assertEqual( artifact_layer["annotations"]["org.opencontainers.image.title"], file2.name ) url = self.uri("/artifacts/" + ARTIFACT_NAME) r = requests.delete(url) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 200, r.text) # Assert return response is json and contains digest self.assertIn("sha256:", rjson["ArtifactDigests"][0]) def test_add_with_append(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" file = ArtifactFile(name="test_file_2") parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, "append": "true", } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) # Assert return response is json and contains digest add_response_json = add_response.json() self.assertIn("sha256:", cast(str, add_response_json["ArtifactDigest"])) inspect_response_json = artifact.do_artifact_inspect_request().json() artifact_layers = inspect_response_json["Manifest"]["layers"] # Assert artifact now has two layers self.assertEqual(len(artifact_layers), 2) def test_add_with_artifactMIMEType_override(self): ARTIFACT_NAME = "quay.io/myimage/myartifact_artifactType:latest" file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, "artifactMIMEType": "application/testType", } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) inspect_response_json = artifact.do_artifact_inspect_request().json() # Assert added artifact has correct mediaType self.assertEqual( inspect_response_json["Manifest"]["artifactType"], "application/testType" ) def test_add_with_annotations(self): ARTIFACT_NAME = "quay.io/myimage/myartifact_annotation:latest" file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, "annotations": ["test=test", "foo=bar"], } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) inspect_response_json = artifact.do_artifact_inspect_request().json() artifact_layer = inspect_response_json["Manifest"]["layers"][0] # Assert artifactBlobAnnotation is set correctly anno = { "foo": "bar", "org.opencontainers.image.title": artifact.file.name, "test": "test", } self.assertEqual(artifact_layer["annotations"], anno) def test_add_with_empty_file(self): ARTIFACT_NAME = "quay.io/myimage/myartifact_empty_file:latest" file = ArtifactFile(size=0) parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) # Assert return response is json and contains digest add_response_json = add_response.json() self.assertIn("sha256:", cast(str, add_response_json["ArtifactDigest"])) inspect_response_json = artifact.do_artifact_inspect_request().json() artifact_layer = inspect_response_json["Manifest"]["layers"][0] # Assert uploaded artifact blob is expected size self.assertEqual(artifact_layer["size"], file.size) # Assert uploaded artifact blob has expected title annotation self.assertEqual( artifact_layer["annotations"]["org.opencontainers.image.title"], file.name ) def test_add_with_fileMIMEType_override(self): ARTIFACT_NAME = "quay.io/myimage/myartifact_mime_type:latest" file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, "fileMIMEType": "fake/type", } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) # Assert return response is json and contains digest add_response_json = add_response.json() self.assertIn("sha256:", cast(str, add_response_json["ArtifactDigest"])) inspect_response_json = artifact.do_artifact_inspect_request().json() artifact_layer = inspect_response_json["Manifest"]["layers"][0] # Assert uploaded artifact blob is expected MIME type self.assertEqual(artifact_layer["mediaType"], "fake/type") def test_add_with_auto_fileMIMEType_discovery(self): ARTIFACT_NAME = "quay.io/myimage/myartifact_image_blob:latest" FILE_SIG = bytes([137, 80, 78, 71, 13, 10, 26, 10]) file = ArtifactFile(sig=FILE_SIG) parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) # Assert return response is json and contains digest add_response_json = add_response.json() self.assertIn("sha256:", cast(str, add_response_json["ArtifactDigest"])) inspect_response_json = artifact.do_artifact_inspect_request().json() artifact_layer = inspect_response_json["Manifest"]["layers"][0] # Assert uploaded artifact blob is automatically recognised as image self.assertEqual(artifact_layer["mediaType"], "image/png") def test_add_append_with_type_fails(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, "artifactMIMEType": "application/octet-stream", "append": "true", } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) r = artifact.add() rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 500, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "append option is not compatible with type option", ) def test_add_with_append_to_missing_artifact_fails(self): ARTIFACT_NAME = "quay.io/myimage/missing:latest" file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": ARTIFACT_NAME, "fileName": file.name, "append": "true", } artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) r = artifact.add() rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 404, r.text) # Assert return error response is json and contains correct message self.assertEqual(rjson["cause"], "artifact does not exist") def test_add_without_name_and_filename_fails(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" file = ArtifactFile() parameters: dict[str, str | list[str]] = {"fake": "fake"} artifact = Artifact(self.uri(""), ARTIFACT_NAME, parameters, file) r = artifact.add() rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 400, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "name and file parameters are required", ) def test_inspect(self): ARTIFACT_NAME = "quay.io/myimage/myartifact_mime_type:latest" url = self.uri( "/artifacts/" + ARTIFACT_NAME + "/json", ) r = requests.get(url) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 200, r.text) # Define expected layout keys expected_top_level = {"Manifest", "Name", "Digest"} expected_manifest = { "schemaVersion", "mediaType", "config", "layers", } expected_config = {"mediaType", "digest", "size", "data"} expected_layer = {"mediaType", "digest", "size", "annotations"} # Compare returned keys with expected missing_top = expected_top_level - rjson.keys() manifest = rjson.get("Manifest", {}) missing_manifest = expected_manifest - manifest.keys() config = manifest.get("config", {}) missing_config = expected_config - config.keys() layers = manifest.get("layers", []) for i, layer in enumerate(layers): missing_layer = expected_layer - layer.keys() self.assertFalse(missing_layer) # Assert all missing dicts are empty meaning all expected keys were present self.assertFalse(missing_top) self.assertFalse(missing_manifest) self.assertFalse(missing_config) def test_inspect_absent_artifact_fails(self): ARTIFACT_NAME = "fake_artifact" url = self.uri("/artifacts/" + ARTIFACT_NAME + "/json") r = requests.get(url) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 404, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "artifact does not exist", ) def test_list(self): url = self.uri("/artifacts/json") r = requests.get(url) rjson = r.json() self.assertEqual(r.status_code, 200, r.text) expected_top_level = {"Manifest", "Name"} expected_manifest = {"schemaVersion", "mediaType", "config", "layers"} expected_config = {"mediaType", "digest", "size", "data"} expected_layer = {"mediaType", "digest", "size", "annotations"} for data in rjson: missing_top = expected_top_level - data.keys() manifest = data.get("Manifest", {}) missing_manifest = expected_manifest - manifest.keys() config = manifest.get("config", {}) missing_config = expected_config - config.keys() layers = manifest.get("layers", []) for _, layer in enumerate(layers): missing_layer = expected_layer - layer.keys() self.assertFalse(missing_layer) # assert all missing dicts are empty self.assertFalse(missing_top) self.assertFalse(missing_manifest) self.assertFalse(missing_config) def test_pull(self): ARTIFACT_NAME = "quay.io/libpod/testartifact:20250206-single" url = self.uri("/artifacts/pull") parameters = { "name": ARTIFACT_NAME, } r = requests.post(url, params=parameters) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 200, r.text) # Assert return response is json and contains correct message self.assertIn("sha256:", rjson["ArtifactDigest"]) def test_pull_with_retry(self): ARTIFACT_NAME = "localhost/fake/artifact:latest" # Note: Default retry is 3 attempts with 1s delay. url = self.uri("/artifacts/pull") parameters = { "name": ARTIFACT_NAME, "retryDelay": "3s", "retry": "2", } r = requests.post(url, params=parameters) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 500, r.text) # Assert request took expected time with retries self.assertTrue(5 < r.elapsed.total_seconds() < 7) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "connection refused", ) def test_pull_unauthorised_fails(self): ARTIFACT_NAME = "quay.io/libpod_secret/testartifact:latest" url = self.uri("/artifacts/pull") parameters = { "name": ARTIFACT_NAME, } r = requests.post(url, params=parameters) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 401, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "unauthorized: access to the requested resource is not authorized", ) def test_pull_missing_fails(self): ARTIFACT_NAME = "quay.io/libpod/testartifact:superfake" url = self.uri("/artifacts/pull") parameters = { "name": ARTIFACT_NAME, } r = requests.post(url, params=parameters) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 404, r.text) # Assert return error response is json and contains correct message self.assertIn( "manifest unknown", rjson["cause"], ) def test_remove(self): ARTIFACT_NAME = "quay.io/libpod/testartifact:20250206-single" url = self.uri("/artifacts/" + ARTIFACT_NAME) r = requests.delete(url) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 200, r.text) # Assert return response is json and contains digest self.assertIn("sha256:", rjson["ArtifactDigests"][0]) def test_remove_multiple(self): # Create some artifacts to remove artifact_names = [ "quay.io/myimage/myartifact1:latest", "quay.io/myimage/myartifact2:latest", "quay.io/myimage/myartifact3:latest" ] for name in artifact_names: file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": name, "fileName": file.name, } artifact = Artifact(self.uri(""), name, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) # Test remove multiple artifacts removeparameters: dict[str, str | list[str]] = { "Artifacts": artifact_names, } url = self.uri("/artifacts/remove") r = requests.delete(url, params=removeparameters) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 200, r.text) # Assert return response is valid json and contains multiple digests self.assertEqual(len(rjson["ArtifactDigests"]), len(artifact_names)) # Test removing an artifact that doesn't exist removeparameters: dict[str, str | list[str]] = { "Artifacts": "fake_artifact", } url = self.uri("/artifacts/remove") r = requests.delete(url, params=removeparameters) rjson = r.json() print(r) # Assert correct response code self.assertEqual(r.status_code, 404, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "artifact does not exist", ) def test_remove_all(self): # Create some artifacts to remove artifact_names = [ "quay.io/myimage/myartifact1:latest", "quay.io/myimage/myartifact2:latest", "quay.io/myimage/myartifact3:latest" ] for name in artifact_names: file = ArtifactFile() parameters: dict[str, str | list[str]] = { "name": name, "fileName": file.name, } artifact = Artifact(self.uri(""), name, parameters, file) add_response = artifact.add() # Assert correct response code self.assertEqual(add_response.status_code, 201, add_response.text) # Test remove all artifacts removeparameters: dict[str, str | list[str]] = { "all": "true", } url = self.uri("/artifacts/remove") r = requests.delete(url, params=removeparameters) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 200, r.text) # Assert no artifacts remain url = self.uri("/artifacts/json") r = requests.get(url) rjson = r.json() self.assertEqual(len(rjson), 0) def test_remove_with_ignore(self): # Test remove non existent artifacts with ignore removeparameters: dict[str, str | list[str]] = { "Artifacts": "fake_artifact", "ignore": "true", } url = self.uri("/artifacts/remove") r = requests.delete(url, params=removeparameters) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 200, r.text) def test_remove_absent_artifact_fails(self): ARTIFACT_NAME = "localhost/fake/artifact:latest" url = self.uri("/artifacts/" + ARTIFACT_NAME) r = requests.delete(url) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 404, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "artifact does not exist", ) def test_push_unauthorised(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" url = self.uri( "/artifacts/" + ARTIFACT_NAME + "/push", ) r = requests.post(url) rjson = r.json() # Assert return error response is json and contains correct message self.assertEqual(r.status_code, 401, r.text) # Assert return error response is json and contains correct message self.assertIn( "authentication required", rjson["cause"], ) def test_push_bad_param(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" parameters = { "retry": "abc", } url = self.uri( "/artifacts/" + ARTIFACT_NAME + "/push", ) r = requests.post( url, params=parameters, ) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 400, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "name parameter is required", ) def test_push_missing_artifact(self): ARTIFACT_NAME = "localhost/fake/artifact:latest" url = self.uri( "/artifacts/" + ARTIFACT_NAME + "/push", ) r = requests.post( url, ) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 404, r.text) # Assert return error response is json and contains correct message self.assertIn( "no descriptor found for reference", rjson["cause"], ) def test_extract(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" url = self.uri( "/artifacts/" + ARTIFACT_NAME + "/extract", ) r = requests.get(url) # Assert correct response code self.assertEqual(r.status_code, 200, r.text) tar_file = "test.tar" tar_file_sizes = None with open(tar_file, "wb") as f: _ = f.write(r.content) with tarfile.open(tar_file, "r") as tar: tar_file_sizes = {m.name: m.size for m in tar.getmembers() if m.isfile()} self.assertEqual( tar_file_sizes, {"test_file_1": 1048576, "test_file_2": 1048576} ) os.remove(tar_file) def test_extract_with_title(self): ARTIFACT_NAME = "quay.io/myimage/myartifact:latest" parameters: dict[str, str] = { "title": "test_file_1", } url = self.uri( "/artifacts/" + ARTIFACT_NAME + "/extract", ) r = requests.get(url, parameters) # Assert correct response code self.assertEqual(r.status_code, 200, r.text) tar_file = "test.tar" tar_file_sizes = None with open(tar_file, "wb") as f: _ = f.write(r.content) with tarfile.open(tar_file, "r") as tar: tar_file_sizes = {m.name: m.size for m in tar.getmembers() if m.isfile()} self.assertEqual(tar_file_sizes, {"test_file_1": 1048576}) os.remove(tar_file) def test_extract_absent_fails(self): ARTIFACT_NAME = "localhost/fake/artifact:latest" url = self.uri( "/artifacts/" + ARTIFACT_NAME + "/extract", ) r = requests.get(url) rjson = r.json() # Assert correct response code self.assertEqual(r.status_code, 404, r.text) # Assert return error response is json and contains correct message self.assertEqual( rjson["cause"], "artifact does not exist", ) if __name__ == "__main__": unittest.main()