# coding=utf-8 # ------------------------------------ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. # ------------------------------------ import os import pytest import hashlib import json from datetime import datetime from io import BytesIO from unittest.mock import MagicMock from typing import Optional, AsyncIterator, Any, MutableMapping from azure.containerregistry import ( RepositoryProperties, ArtifactManifestProperties, ArtifactManifestOrder, ArtifactTagProperties, ArtifactTagOrder, ArtifactArchitecture, ArtifactOperatingSystem, DigestValidationError, ) from azure.containerregistry.aio import ContainerRegistryClient from azure.containerregistry._helpers import DOCKER_MANIFEST, OCI_IMAGE_MANIFEST, DEFAULT_CHUNK_SIZE from azure.core.exceptions import ( ResourceNotFoundError, ClientAuthenticationError, HttpResponseError, ServiceRequestError, ServiceResponseError, ) from azure.core.async_paging import AsyncItemPaged from azure.core.pipeline import PipelineRequest from azure.identity import AzureAuthorityHosts from asynctestcase import AsyncContainerRegistryTestClass, get_authority, get_audience from testcase import is_public_endpoint, is_china_endpoint from constants import HELLO_WORLD, DOES_NOT_EXIST from preparer import acr_preparer from devtools_testutils.aio import recorded_by_proxy_async class TestContainerRegistryClientAsync(AsyncContainerRegistryTestClass): @acr_preparer() @recorded_by_proxy_async async def test_list_repository_names(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: repositories = client.list_repository_names() assert isinstance(repositories, AsyncItemPaged) count = 0 prev = None async for repo in repositories: count += 1 assert isinstance(repo, str) assert prev != repo prev = repo assert count > 0 @acr_preparer() @recorded_by_proxy_async async def test_list_repository_names_by_page(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: results_per_page = 2 total_pages = 0 repository_pages = client.list_repository_names(results_per_page=results_per_page) prev = None async for page in repository_pages.by_page(): page_count = 0 async for repo in page: assert isinstance(repo, str) assert prev != repo prev = repo page_count += 1 assert page_count <= results_per_page total_pages += 1 assert total_pages >= 1 @acr_preparer() @recorded_by_proxy_async async def test_delete_repository(self, containerregistry_endpoint): repo = self.get_resource_name("repo") self.import_image(containerregistry_endpoint, repo, ["test"]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: await client.delete_repository(repo) self.sleep(5) with pytest.raises(ResourceNotFoundError): await client.get_repository_properties(repo) @acr_preparer() @recorded_by_proxy_async async def test_delete_repository_does_not_exist(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: await client.delete_repository(DOES_NOT_EXIST) @acr_preparer() @recorded_by_proxy_async async def test_get_repository_properties(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: properties = await client.get_repository_properties(HELLO_WORLD) assert isinstance(properties, RepositoryProperties) assert properties.name == HELLO_WORLD @acr_preparer() @recorded_by_proxy_async async def test_update_repository_properties(self, containerregistry_endpoint): repo = self.get_resource_name("repo") self.import_image(containerregistry_endpoint, repo, ["test"]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: properties = self.set_all_properties(RepositoryProperties(), False) received = await client.update_repository_properties(repo, properties) self.assert_all_properties(received, False) properties = self.set_all_properties(properties, True) received = await client.update_repository_properties(repo, properties) self.assert_all_properties(received, True) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_update_repository_properties_kwargs(self, containerregistry_endpoint): repo = self.get_resource_name("repo") self.import_image(containerregistry_endpoint, repo, ["test"]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: received = await client.update_repository_properties( repo, can_delete=False, can_read=False, can_write=False, can_list=False ) self.assert_all_properties(received, False) received = await client.update_repository_properties( repo, can_delete=True, can_read=True, can_write=True, can_list=True ) self.assert_all_properties(received, True) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_list_registry_artifacts(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: count = 0 async for artifact in client.list_manifest_properties(HELLO_WORLD): assert isinstance(artifact, ArtifactManifestProperties) assert isinstance(artifact.created_on, datetime) assert isinstance(artifact.last_updated_on, datetime) assert artifact.repository_name == HELLO_WORLD assert artifact.fully_qualified_reference in self.create_fully_qualified_reference( containerregistry_endpoint, HELLO_WORLD, artifact.digest ) count += 1 assert count > 0 @acr_preparer() @recorded_by_proxy_async async def test_list_registry_artifacts_by_page(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: results_per_page = 2 pages = client.list_manifest_properties(HELLO_WORLD, results_per_page=results_per_page) page_count = 0 async for page in pages.by_page(): reg_count = 0 async for tag in page: reg_count += 1 assert reg_count <= results_per_page page_count += 1 assert page_count >= 1 @acr_preparer() @recorded_by_proxy_async async def test_list_registry_artifacts_descending(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: prev_last_updated_on = None count = 0 async for artifact in client.list_manifest_properties( HELLO_WORLD, order_by=ArtifactManifestOrder.LAST_UPDATED_ON_DESCENDING ): if prev_last_updated_on: assert artifact.last_updated_on < prev_last_updated_on prev_last_updated_on = artifact.last_updated_on count += 1 assert count > 0 @acr_preparer() @recorded_by_proxy_async async def test_list_registry_artifacts_ascending(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: prev_last_updated_on = None count = 0 async for artifact in client.list_manifest_properties( HELLO_WORLD, order_by=ArtifactManifestOrder.LAST_UPDATED_ON_ASCENDING ): if prev_last_updated_on: assert artifact.last_updated_on > prev_last_updated_on prev_last_updated_on = artifact.last_updated_on count += 1 assert count > 0 @acr_preparer() @recorded_by_proxy_async async def test_get_manifest_properties(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: properties = await client.get_manifest_properties(HELLO_WORLD, "latest") assert isinstance(properties, ArtifactManifestProperties) assert properties.repository_name == HELLO_WORLD assert properties.fully_qualified_reference in self.create_fully_qualified_reference( containerregistry_endpoint, HELLO_WORLD, properties.digest ) @acr_preparer() @recorded_by_proxy_async async def test_get_manifest_properties_does_not_exist(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: manifest = await client.get_manifest_properties(HELLO_WORLD, "latest") invalid_digest = manifest.digest[:-10] + "a" * 10 with pytest.raises(ResourceNotFoundError): await client.get_manifest_properties(HELLO_WORLD, invalid_digest) with pytest.raises(ResourceNotFoundError): await client.get_manifest_properties(DOES_NOT_EXIST, DOES_NOT_EXIST) @acr_preparer() @recorded_by_proxy_async async def test_update_manifest_properties(self, containerregistry_endpoint): repo = self.get_resource_name("repo") tag = "test" self.import_image(containerregistry_endpoint, repo, [tag]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: properties = self.set_all_properties(ArtifactManifestProperties(), False) received = await client.update_manifest_properties(repo, tag, properties) self.assert_all_properties(received, False) properties = self.set_all_properties(properties, True) received = await client.update_manifest_properties(repo, tag, properties) self.assert_all_properties(received, True) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_update_manifest_properties_kwargs(self, containerregistry_endpoint): repo = self.get_resource_name("repo") tag = "test" self.import_image(containerregistry_endpoint, repo, [tag]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: received = await client.update_manifest_properties( repo, tag, can_delete=False, can_read=False, can_write=False, can_list=False ) self.assert_all_properties(received, False) received = await client.update_manifest_properties( repo, tag, can_delete=True, can_read=True, can_write=True, can_list=True ) self.assert_all_properties(received, True) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_get_tag_properties(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: properties = await client.get_tag_properties(HELLO_WORLD, "latest") assert isinstance(properties, ArtifactTagProperties) assert properties.name == "latest" @acr_preparer() @recorded_by_proxy_async async def test_get_tag_properties_does_not_exist(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: with pytest.raises(ResourceNotFoundError): await client.get_tag_properties(DOES_NOT_EXIST, DOES_NOT_EXIST) with pytest.raises(ResourceNotFoundError): await client.get_tag_properties(HELLO_WORLD, DOES_NOT_EXIST) @acr_preparer() @recorded_by_proxy_async async def test_update_tag_properties(self, containerregistry_endpoint): repo = self.get_resource_name("repo") tag = "test" self.import_image(containerregistry_endpoint, repo, [tag]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: properties = self.set_all_properties(ArtifactTagProperties(), False) received = await client.update_tag_properties(repo, tag, properties) self.assert_all_properties(received, False) properties = self.set_all_properties(properties, True) received = await client.update_tag_properties(repo, tag, properties) self.assert_all_properties(received, True) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_update_tag_properties_kwargs(self, containerregistry_endpoint): repo = self.get_resource_name("repo") tag = "test" self.import_image(containerregistry_endpoint, repo, [tag]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: received = await client.update_tag_properties( repo, tag, can_delete=False, can_read=False, can_write=False, can_list=False ) self.assert_all_properties(received, False) received = await client.update_tag_properties( repo, tag, can_delete=True, can_read=True, can_write=True, can_list=True ) self.assert_all_properties(received, True) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_list_tag_properties(self, containerregistry_endpoint): tags = ["latest", "v1"] async with self.create_registry_client(containerregistry_endpoint) as client: count = 0 async for tag in client.list_tag_properties(HELLO_WORLD): assert tag.name in tags count += 1 assert count == 2 @acr_preparer() @recorded_by_proxy_async async def test_list_tag_properties_order_descending(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: prev_last_updated_on = None count = 0 async for tag in client.list_tag_properties( HELLO_WORLD, order_by=ArtifactTagOrder.LAST_UPDATED_ON_DESCENDING ): if prev_last_updated_on: assert tag.last_updated_on < prev_last_updated_on prev_last_updated_on = tag.last_updated_on count += 1 assert count == 2 @acr_preparer() @recorded_by_proxy_async async def test_list_tag_properties_order_ascending(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: prev_last_updated_on = None count = 0 async for tag in client.list_tag_properties( HELLO_WORLD, order_by=ArtifactTagOrder.LAST_UPDATED_ON_ASCENDING ): if prev_last_updated_on: assert tag.last_updated_on > prev_last_updated_on prev_last_updated_on = tag.last_updated_on count += 1 assert count == 2 @acr_preparer() @recorded_by_proxy_async async def test_delete_tag(self, containerregistry_endpoint): repo = self.get_resource_name("repo") tag = "test" self.import_image(containerregistry_endpoint, repo, [tag]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: await client.delete_tag(repo, tag) self.sleep(10) with pytest.raises(ResourceNotFoundError): await client.get_tag_properties(repo, tag) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_delete_tag_does_not_exist(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: await client.delete_tag(DOES_NOT_EXIST, DOES_NOT_EXIST) await client.delete_tag(HELLO_WORLD, DOES_NOT_EXIST) @acr_preparer() @recorded_by_proxy_async async def test_delete_manifest(self, containerregistry_endpoint): repo = self.get_resource_name("repo") tag = "test" self.import_image(containerregistry_endpoint, repo, [tag]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: await client.delete_manifest(repo, tag) self.sleep(10) with pytest.raises(ResourceNotFoundError): await client.get_manifest_properties(repo, tag) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_delete_manifest_does_not_exist(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: manifest = await client.get_manifest_properties(HELLO_WORLD, "latest") invalid_digest = manifest.digest[:-10] + "a" * 10 await client.delete_manifest(HELLO_WORLD, invalid_digest) with pytest.raises(ResourceNotFoundError): await client.delete_manifest(HELLO_WORLD, DOES_NOT_EXIST) with pytest.raises(ResourceNotFoundError): await client.delete_manifest(DOES_NOT_EXIST, DOES_NOT_EXIST) @acr_preparer() @recorded_by_proxy_async async def test_expiration_time_parsing(self, containerregistry_endpoint): from azure.containerregistry.aio._async_authentication_policy import ContainerRegistryChallengePolicy async with self.create_registry_client(containerregistry_endpoint) as client: async for repo in client.list_repository_names(): pass for policy in client._client._client._pipeline._impl_policies: if isinstance(policy, ContainerRegistryChallengePolicy): policy._exchange_client._expiration_time = 0 break count = 0 async for repo in client.list_repository_names(): count += 1 assert count >= 1 # Live only, the fake credential doesn't check auth scope the same way @pytest.mark.live_test_only @acr_preparer() async def test_construct_container_registry_client(self, **kwargs): containerregistry_endpoint = kwargs.pop("containerregistry_endpoint") authority = get_authority(containerregistry_endpoint) credential = self.get_credential(authority=authority) async with ContainerRegistryClient( endpoint=containerregistry_endpoint, credential=credential, audience="https://microsoft.com" ) as client: with pytest.raises(ClientAuthenticationError): properties = await client.get_repository_properties(HELLO_WORLD) @acr_preparer() @recorded_by_proxy_async async def test_get_misspell_property(self, containerregistry_endpoint): async with self.create_registry_client(containerregistry_endpoint) as client: properties = await client.get_repository_properties(HELLO_WORLD) with pytest.warns(DeprecationWarning): last_udpated_on = properties.last_udpated_on last_updated_on = properties.last_updated_on assert last_udpated_on == last_updated_on # Live only, as test proxy now cannot handle spaces correctly # issue: https://github.com/Azure/azure-sdk-tools/issues/5968 @pytest.mark.live_test_only @acr_preparer() async def test_set_oci_manifest(self, **kwargs): containerregistry_endpoint = kwargs.pop("containerregistry_endpoint") repo = self.get_resource_name("repo") path = os.path.join(self.get_test_directory(), "data", "oci_artifact", "manifest.json") async with self.create_registry_client(containerregistry_endpoint) as client: await self.upload_oci_manifest_prerequisites(repo, client) with open(path, "rb") as manifest_stream: # test set oci manifest in stream format with pytest.raises(HttpResponseError): await client.set_manifest(repo, manifest_stream, tag="v1", media_type=DOCKER_MANIFEST) manifest_stream.seek(0) digest1 = await client.set_manifest(repo, manifest_stream, tag="v1") manifest_stream.seek(0) # test set oci manifest in JSON format manifest_json = json.loads(manifest_stream.read().decode()) with pytest.raises(HttpResponseError): await client.set_manifest(repo, manifest_json, tag="v2", media_type=DOCKER_MANIFEST) digest2 = await client.set_manifest(repo, manifest_json, tag="v2") assert digest1 == digest2 # test get oci manifest by digest response = await client.get_manifest(repo, digest1) assert response.media_type == OCI_IMAGE_MANIFEST # test get oci manifest by tag response = await client.get_manifest(repo, "v1") assert response.media_type == OCI_IMAGE_MANIFEST response = await client.get_manifest(repo, "v2") assert response.media_type == OCI_IMAGE_MANIFEST await client.delete_manifest(repo, digest1) await client.delete_repository(repo) # Reading data from a no space file to make this test pass in playback as test proxy cannot handle spaces correctly. # issue: https://github.com/Azure/azure-sdk-tools/issues/5968 @acr_preparer() @recorded_by_proxy_async async def test_set_oci_manifest_without_spaces(self, containerregistry_endpoint): if not is_public_endpoint(containerregistry_endpoint): pytest.skip("This test is for testing test_set_docker_manifest in playback.") repo = self.get_resource_name("repo") path = os.path.join(self.get_test_directory(), "data", "oci_artifact", "manifest_without_spaces.json") async with self.create_registry_client(containerregistry_endpoint) as client: await self.upload_oci_manifest_prerequisites(repo, client) with open(path, "rb") as manifest_stream: # test set oci manifest in stream format with pytest.raises(HttpResponseError): await client.set_manifest(repo, manifest_stream, tag="v1", media_type=DOCKER_MANIFEST) manifest_stream.seek(0) digest = await client.set_manifest(repo, manifest_stream, tag="v1") # test get oci manifest by digest response = await client.get_manifest(repo, digest) assert response.media_type == OCI_IMAGE_MANIFEST # test get oci manifest by tag response = await client.get_manifest(repo, "v1") assert response.media_type == OCI_IMAGE_MANIFEST await client.delete_manifest(repo, digest) await client.delete_repository(repo) # Live only, as test proxy now cannot handle spaces correctly # issue: https://github.com/Azure/azure-sdk-tools/issues/5968 @pytest.mark.live_test_only @acr_preparer() async def test_set_docker_manifest(self, **kwargs): containerregistry_endpoint = kwargs.pop("containerregistry_endpoint") repo = self.get_resource_name("repo") path = os.path.join(self.get_test_directory(), "data", "docker_artifact", "manifest.json") async with self.create_registry_client(containerregistry_endpoint) as client: await self.upload_docker_manifest_prerequisites(repo, client) with open(path, "rb") as manifest_stream: # test set Docker manifest in stream format with pytest.raises(HttpResponseError): # It fails as the default media type is oci image manifest media type await client.set_manifest(repo, manifest_stream, tag="v1") manifest_stream.seek(0) digest1 = await client.set_manifest(repo, manifest_stream, tag="v1", media_type=DOCKER_MANIFEST) manifest_stream.seek(0) # test set Docker manifest in JSON format manifest_json = json.loads(manifest_stream.read().decode()) with pytest.raises(HttpResponseError): # It fails as the default media type is oci image manifest media type await client.set_manifest(repo, manifest_json, tag="v2") digest2 = await client.set_manifest(repo, manifest_json, tag="v2", media_type=DOCKER_MANIFEST) assert digest1 == digest2 # test get Docker manifest by digest response = await client.get_manifest(repo, digest1) assert response.media_type == DOCKER_MANIFEST # test get Docker manifest by tag response = await client.get_manifest(repo, "v1") assert response.media_type == DOCKER_MANIFEST response = await client.get_manifest(repo, "v2") assert response.media_type == DOCKER_MANIFEST await client.delete_manifest(repo, digest1) await client.delete_repository(repo) # Reading data from a no space file to make this test pass in playback as test proxy cannot handle spaces correctly. # issue: https://github.com/Azure/azure-sdk-tools/issues/5968 @acr_preparer() @recorded_by_proxy_async async def test_set_docker_manifest_without_spaces(self, containerregistry_endpoint): if not is_public_endpoint(containerregistry_endpoint): pytest.skip("This test is for testing test_set_docker_manifest in playback.") repo = self.get_resource_name("repo") path = os.path.join(self.get_test_directory(), "data", "docker_artifact", "manifest_without_spaces.json") async with self.create_registry_client(containerregistry_endpoint) as client: await self.upload_docker_manifest_prerequisites(repo, client) with open(path, "rb") as manifest_stream: # test set Docker manifest in stream format with pytest.raises(HttpResponseError): # It fails as the default media type is oci image manifest media type await client.set_manifest(repo, manifest_stream, tag="v1") manifest_stream.seek(0) digest = await client.set_manifest(repo, manifest_stream, tag="v1", media_type=DOCKER_MANIFEST) # test get Docker manifest by digest response = await client.get_manifest(repo, digest) assert response.media_type == DOCKER_MANIFEST # test get Docker manifest by tag response = await client.get_manifest(repo, "v1") assert response.media_type == DOCKER_MANIFEST await client.delete_manifest(repo, digest) await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_upload_blob(self, containerregistry_endpoint): repo = self.get_resource_name("repo") blob = BytesIO(b"hello world") async with self.create_registry_client(containerregistry_endpoint) as client: # Act digest, blob_size = await client.upload_blob(repo, blob) # Assert blob_content = b"" stream = await client.download_blob(repo, digest) async for chunk in stream: blob_content += chunk assert len(blob_content) == blob_size await client.delete_blob(repo, digest) await client.delete_repository(repo) @pytest.mark.live_test_only @acr_preparer() async def test_upload_large_blob_in_chunk(self, **kwargs): containerregistry_endpoint = kwargs.pop("containerregistry_endpoint") if is_china_endpoint(containerregistry_endpoint): pytest.skip("Running on China cloud usually will fail due to timeout.") repo = self.get_resource_name("repo") async with self.create_registry_client(containerregistry_endpoint) as client: # Test blob upload and download in equal size chunks try: blob_size = DEFAULT_CHUNK_SIZE * 1024 # 4GB data = b"\x00" * int(blob_size) digest, size = await client.upload_blob(repo, BytesIO(data)) assert size == blob_size stream = await client.download_blob(repo, digest) size = 0 with open("text1.txt", "wb") as file: async for chunk in stream: size += file.write(chunk) assert size == blob_size await client.delete_blob(repo, digest) except (ServiceRequestError, ServiceResponseError) as err: # Service does not support resumable upload when get transient error while uploading # issue: https://github.com/Azure/azure-sdk-for-python/issues/29738 print(f"Failed to upload blob: {err.message}") except ResourceNotFoundError as err: # Service does not support resumable upload when get transient error while uploading # issue: https://github.com/Azure/azure-sdk-for-python/issues/29738 assert err.status_code == 404 assert err.response.request.method == "PATCH" assert ( err.response.text() == '{"errors":[{"code":"BLOB_UPLOAD_INVALID","message":"blob upload invalid"}]}\n' ) # Test blob upload and download in unequal size chunks try: blob_size = DEFAULT_CHUNK_SIZE * 1024 + 20 data = b"\x00" * int(blob_size) digest, size = await client.upload_blob(repo, BytesIO(data)) assert size == blob_size stream = await client.download_blob(repo, digest) size = 0 with open("text2.txt", "wb") as file: async for chunk in stream: size += file.write(chunk) assert size == blob_size await client.delete_blob(repo, digest) except (ServiceRequestError, ServiceResponseError) as err: # Service does not support resumable upload when get transient error while uploading # issue: https://github.com/Azure/azure-sdk-for-python/issues/29738 print(f"Failed to upload blob: {err.message}") except ResourceNotFoundError as err: # Service does not support resumable upload when get transient error while uploading # issue: https://github.com/Azure/azure-sdk-for-python/issues/29738 assert err.status_code == 404 assert err.response.request.method == "PATCH" assert ( err.response.text() == '{"errors":[{"code":"BLOB_UPLOAD_INVALID","message":"blob upload invalid"}]}\n' ) # Cleanup await client.delete_repository(repo) @acr_preparer() @recorded_by_proxy_async async def test_delete_blob_does_not_exist(self, containerregistry_endpoint): repo = self.get_resource_name("repo") hash_value = hashlib.sha256(b"test").hexdigest() digest = f"sha256:{hash_value}" async with self.create_registry_client(containerregistry_endpoint) as client: await client.delete_blob(repo, digest) @acr_preparer() @recorded_by_proxy_async async def test_set_audience(self, containerregistry_endpoint): authority = get_authority(containerregistry_endpoint) credential = self.get_credential(authority=authority) async with ContainerRegistryClient(endpoint=containerregistry_endpoint, credential=credential) as client: async for repo in client.list_repository_names(): pass valid_audience = get_audience(authority) async with ContainerRegistryClient( endpoint=containerregistry_endpoint, credential=credential, audience=valid_audience ) as client: async for repo in client.list_repository_names(): pass if valid_audience == get_audience(AzureAuthorityHosts.AZURE_PUBLIC_CLOUD): invalid_audience = get_audience(AzureAuthorityHosts.AZURE_GOVERNMENT) async with ContainerRegistryClient( endpoint=containerregistry_endpoint, credential=credential, audience=invalid_audience ) as client: with pytest.raises(ClientAuthenticationError): async for repo in client.list_repository_names(): pass @acr_preparer() @recorded_by_proxy_async async def test_list_in_empty_repo(self, containerregistry_endpoint): repo = self.get_resource_name("repo") self.import_image(containerregistry_endpoint, repo, ["test"]) self.sleep(5) async with self.create_registry_client(containerregistry_endpoint) as client: # cleanup tags in repo async for tag in client.list_tag_properties(repo): await client.delete_tag(repo, tag.name) # test list tags in empty repo response = client.list_tag_properties(repo) async for tag in response: # cleanup manifests in repo await client.delete_manifest(repo, tag.name) # test list manifests in empty repo response = client.list_manifest_properties(repo) async for manifest in response: pass await client.delete_repository(repo) class MyMagicMock(MagicMock): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) async def __aenter__(self): return self async def __aexit__(self, *args): return self class TestContainerRegistryClientAsyncUnitTests: containerregistry_endpoint = "https://fake_url.azurecr.io" def text(self, encoding: Optional[str] = None) -> str: return '{"hello": "world"}' @pytest.mark.asyncio async def test_manifest_digest_validation(self): JSON = MutableMapping[str, Any] async def send_in_set_manifest(request: PipelineRequest, **kwargs) -> MyMagicMock: content_digest = hashlib.sha256(b"hello world").hexdigest() return MyMagicMock( status_code=201, headers={"Docker-Content-Digest": content_digest}, content_type="application/json; charset=utf-8", text=self.text, ) async def read() -> bytes: return b'{"hello": "world"}' def json() -> JSON: return {"hello": "world"} async def send_in_get_manifest(request: PipelineRequest, **kwargs) -> MyMagicMock: content_digest = hashlib.sha256(b"hello world").hexdigest() content_type = "application/vnd.oci.image.manifest.v1+json" return MyMagicMock( status_code=200, headers={ "Docker-Content-Digest": content_digest, "Content-Type": content_type, "Content-Length": len(b"hello world"), }, read=read, json=json, ) async with ContainerRegistryClient( endpoint=self.containerregistry_endpoint, transport=MyMagicMock(send=send_in_set_manifest) ) as client: with pytest.raises(DigestValidationError) as exp: manifest = {"hello": "world"} await client.set_manifest("test-repo", manifest) assert str(exp.value) == "The server-computed digest does not match the client-computed digest." async with ContainerRegistryClient( endpoint=self.containerregistry_endpoint, transport=MyMagicMock(send=send_in_get_manifest) ) as client: with pytest.raises(DigestValidationError) as exp: digest = hashlib.sha256(b"hello world").hexdigest() await client.get_manifest("test-repo", f"sha256:{digest}") assert str(exp.value) == "The content of retrieved manifest digest does not match the requested digest." with pytest.raises(DigestValidationError) as exp: await client.get_manifest("test-repo", "test-tag") assert str(exp.value) == "The server-computed digest does not match the client-computed digest." @pytest.mark.asyncio async def test_blob_digest_validation(self): async def send_in_upload_blob(request: PipelineRequest, **kwargs) -> MyMagicMock: if request.method == "PUT": content_digest = hashlib.sha256(b"hello world").hexdigest() return MyMagicMock( status_code=201, headers={"Docker-Content-Digest": content_digest}, content_type="application/json; charset=utf-8", text=self.text, ) else: return MyMagicMock( status_code=202, headers={"Location": "/v2/test-repo/blobs/uploads/fake_location"}, content_type="application/json; charset=utf-8", text=self.text, ) async def iter_bytes() -> AsyncIterator[bytes]: yield b'{"hello": "world"}' async def send_in_download_blob(request: PipelineRequest, **kwargs) -> MyMagicMock: return MyMagicMock( status_code=206, headers={"Content-Range": "bytes 0-27/28", "Content-Length": "28"}, content_type="application/octet-stream", text=self.text, iter_bytes=iter_bytes, ) async with ContainerRegistryClient( endpoint=self.containerregistry_endpoint, transport=MyMagicMock(send=send_in_upload_blob) ) as client: with pytest.raises(DigestValidationError) as exp: await client.upload_blob("test-repo", BytesIO(b'{"hello": "world"}')) assert str(exp.value) == "The server-computed digest does not match the client-computed digest." async with ContainerRegistryClient( endpoint=self.containerregistry_endpoint, transport=MyMagicMock(send=send_in_download_blob) ) as client: digest = hashlib.sha256(b"hello world").hexdigest() stream = await client.download_blob("test-repo", f"sha256:{digest}") with pytest.raises(DigestValidationError) as exp: async for chunk in stream: pass assert str(exp.value) == "The content of retrieved blob digest does not match the requested digest." @pytest.mark.asyncio async def test_deserialize_manifest(self): def get_manifest(encoding: Optional[str] = None) -> str: manifest = { "manifests": [ { "mediaType": "application/vnd.docker.distribution.manifest.v2+json", "imageSize": 2199, "digest": "sha256:86fed9f0203a09f13cbbb9842132e9000eeff51b3de0d4ff66ee03ab0e860d1f", "architecture": "amd64", "os": "linux", }, { "mediaType": "application/vnd.docker.distribution.manifest.v2+json", "imageSize": 566, "digest": "sha256:b808af65792ab617b9032c20fb12c455dc2bf5efe1af3f0ac81a129560772d35", "annotations": { "vnd.docker.reference.digest": "sha256:86fed9f0203a09f13cbbb9842132e9000eeff51b3de0d4ff66ee03ab0e860d1f", "vnd.docker.reference.type": "attestation-manifest", }, "architecture": "unknown", "os": "unknown", }, ] } return json.dumps(manifest) async def send(request: PipelineRequest, **kwargs) -> MyMagicMock: return MyMagicMock( status_code=200, content_type="application/json; charset=utf-8", text=get_manifest, ) async with ContainerRegistryClient( endpoint=self.containerregistry_endpoint, transport=MyMagicMock(send=send) ) as client: manifests = client.list_manifest_properties(HELLO_WORLD) async for manifest in manifests: if manifest.size_in_bytes == 2199: assert isinstance(manifest.architecture, ArtifactArchitecture) assert manifest.architecture == "amd64" assert isinstance(manifest.operating_system, ArtifactOperatingSystem) assert manifest.operating_system == "linux" if manifest.size_in_bytes == 566: assert isinstance(manifest.architecture, str) assert manifest.architecture == "unknown" assert isinstance(manifest.operating_system, str) assert manifest.operating_system == "unknown"