Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions src/azure-cli/azure/cli/command_modules/acr/_help.py
Original file line number Diff line number Diff line change
Expand Up @@ -337,9 +337,6 @@
- name: Import an image without waiting for successful completion. Failures during import will not be reflected. Run `az acr repository show-tags` to confirm that import succeeded.
text: >
az acr import -n myregistry --source sourceregistry.azurecr.io/sourcerepository:sourcetag --no-wait
- name: Import an image using a regional endpoint URI as the source.
text: >
az acr import -n myregistry --source sourceregistry.eastus.geo.azurecr.io/sourcerepository:sourcetag
"""

helps['acr list'] = """
Expand Down
8 changes: 7 additions & 1 deletion src/azure-cli/azure/cli/command_modules/acr/import.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
LOGIN_SERVER_NOT_VALID = "Login server of the registry is not valid " \
"because it is not a fully qualified domain name."
CREDENTIALS_INVALID = "Authentication failed. Please provide password."
REGIONAL_ENDPOINT_IMPORT_WARNING = "ACR imports do not support regional endpoint selection. " \
"This import will use the source registry's home region."
Comment on lines +28 to +29


def acr_import(cmd, # pylint: disable=too-many-locals
Expand Down Expand Up @@ -134,6 +136,8 @@ def _regional_endpoint_uri_to_login_server(uri, login_server_suffix):
prefix = uri_lower[:-len(geo_suffix)] # "<registry>.<region>"
prefix_parts = prefix.split('.')
if len(prefix_parts) == 2 and prefix_parts[0] and prefix_parts[1]:
# The converted login server resolves the ACR resource but no longer identifies a replica.
logger.warning(REGIONAL_ENDPOINT_IMPORT_WARNING)
return f"{prefix_parts[0]}{login_server_suffix}"

# If not a regional endpoint format, return as-is
Expand All @@ -146,8 +150,10 @@ def _get_azure_registry(cmd, source_registry):

# Try to get the pre-defined login server suffix.
login_server_suffix = get_login_server_suffix(cmd.cli_ctx)
# DNS hostnames are case-insensitive, so normalize both values before matching the suffix.
regional_suffix = f".geo{login_server_suffix}".lower() if login_server_suffix else None
# Convert regional endpoint to standard format if applicable
if login_server_suffix and source_registry.endswith(f".geo{login_server_suffix}"):
if regional_suffix and source_registry.lower().endswith(regional_suffix):
lookup_uri = _regional_endpoint_uri_to_login_server(source_registry, login_server_suffix)

# Search by login server (lookup_uri) and registry name (source_registry) to handle both URI and name inputs
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,16 @@

import unittest
import importlib
from unittest import mock

# The path contains a reserved keyword 'import', so we need a workaround here
acr_import = importlib.import_module('azure.cli.command_modules.acr.import')


class TestRegionalEndpointUriConversion(unittest.TestCase):

def test_valid_regional_endpoint_conversion(self):
@mock.patch.object(acr_import.logger, 'warning')
def test_valid_regional_endpoint_conversion(self, warning):
"""Test conversion of regional endpoint URIs to standard format."""
login_server_suffix = '.azurecr.io'

Expand All @@ -30,7 +32,11 @@ def test_valid_regional_endpoint_conversion(self):
result = acr_import._regional_endpoint_uri_to_login_server(regional_uri, login_server_suffix)
self.assertEqual(result, expected)

def test_valid_regional_endpoint_conversion_multi_label_suffix(self):
self.assertEqual(warning.call_count, len(test_cases))
warning.assert_called_with(acr_import.REGIONAL_ENDPOINT_IMPORT_WARNING)

@mock.patch.object(acr_import.logger, 'warning')
def test_valid_regional_endpoint_conversion_multi_label_suffix(self, warning):
"""Regional endpoints in sovereign clouds whose login-server suffix has more than two
labels (e.g. '.azurecr.sovcloud-azure.de') must still be converted."""
test_cases = [
Expand All @@ -47,7 +53,11 @@ def test_valid_regional_endpoint_conversion_multi_label_suffix(self):
result = acr_import._regional_endpoint_uri_to_login_server(regional_uri, suffix)
self.assertEqual(result, expected)

def test_non_regional_endpoint_uris_unchanged(self):
self.assertEqual(warning.call_count, len(test_cases))
warning.assert_called_with(acr_import.REGIONAL_ENDPOINT_IMPORT_WARNING)

@mock.patch.object(acr_import.logger, 'warning')
def test_non_regional_endpoint_uris_unchanged(self, warning):
"""Test that non-regional endpoint URIs are returned unchanged."""
login_server_suffix = '.azurecr.io'

Expand All @@ -66,6 +76,43 @@ def test_non_regional_endpoint_uris_unchanged(self):
result = acr_import._regional_endpoint_uri_to_login_server(uri, login_server_suffix)
self.assertEqual(result, uri)

warning.assert_not_called()

@mock.patch.object(acr_import.logger, 'warning')
@mock.patch.object(acr_import, 'get_registry_from_name_or_login_server')
@mock.patch.object(acr_import, 'get_login_server_suffix', return_value='.azurecr.io')
def test_get_azure_registry_matches_regional_suffix_case_insensitively(
self, get_login_server_suffix, get_registry, warning):
get_registry.return_value = mock.sentinel.registry
cmd = mock.Mock(cli_ctx=mock.sentinel.cli_ctx)
regional_endpoint = 'MyRegistry.WestUS.Geo.AzureCR.IO'

result = acr_import._get_azure_registry(cmd, regional_endpoint)

self.assertIs(result, mock.sentinel.registry)
get_login_server_suffix.assert_called_once_with(mock.sentinel.cli_ctx)
get_registry.assert_called_once_with(
mock.sentinel.cli_ctx, 'myregistry.azurecr.io', regional_endpoint)
warning.assert_called_once_with(acr_import.REGIONAL_ENDPOINT_IMPORT_WARNING)

@mock.patch.object(acr_import, '_regional_endpoint_uri_to_login_server')
@mock.patch.object(acr_import, 'get_registry_from_name_or_login_server')
@mock.patch.object(acr_import, 'get_login_server_suffix', return_value='.azurecr.io')
def test_get_azure_registry_does_not_convert_resource_id(
self, get_login_server_suffix, get_registry, convert_regional_endpoint):
get_registry.return_value = mock.sentinel.registry
cmd = mock.Mock(cli_ctx=mock.sentinel.cli_ctx)
resource_id = ('/subscriptions/000/resourceGroups/rg/providers/'
'Microsoft.ContainerRegistry/registries/source')

result = acr_import._get_azure_registry(cmd, resource_id)

self.assertIs(result, mock.sentinel.registry)
get_login_server_suffix.assert_called_once_with(mock.sentinel.cli_ctx)
convert_regional_endpoint.assert_not_called()
get_registry.assert_called_once_with(
mock.sentinel.cli_ctx, resource_id, resource_id)

@staticmethod
def _match_regional_endpoint(login_server, endpoint, regional_endpoint_host_names):
"""Replicate the matching logic from acr_login for unit testing."""
Expand Down
Loading