refactor(image): use ImageProvider.test_connection() from API utils

- Add registry parameter to test_connection() for registry-level testing
  - Replace inline adapter logic in API utils with test_connection() call
  - Update API tests to mock test_connection instead of create_registry_adapter
This commit is contained in:
Andoni A. committed 2026-02-20 08:39:39 +01:00
1 parent 97ed67853b
commit 7ec21797cd
3 files changed
+48 -34

No files matched your search

+17 -19
View File
@@ -21,6 +21,7 @@ from prowler.providers.aws.aws_provider import AwsProvider
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHubConnection
from prowler.providers.azure.azure_provider import AzureProvider
from prowler.providers.cloudflare.cloudflare_provider import CloudflareProvider
from prowler.providers.common.models import Connection
from prowler.providers.gcp.gcp_provider import GcpProvider
from prowler.providers.github.github_provider import GithubProvider
from prowler.providers.iac.iac_provider import IacProvider
@@ -190,48 +191,45 @@ class TestProwlerProviderConnectionTest:
assert isinstance(connection.error, Provider.secret.RelatedObjectDoesNotExist)
assert str(connection.error) == "Provider has no secret."
@patch(
"prowler.providers.image.lib.registry.factory.create_registry_adapter",
)
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test_image_success(
self, mock_return_prowler_provider, mock_create_adapter
self, mock_return_prowler_provider
):
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.uid = "ghcr.io"
provider.secret.secret = {"registry_token": "tok"}
mock_adapter = MagicMock()
mock_create_adapter.return_value = mock_adapter
prowler_provider = MagicMock()
prowler_provider.test_connection.return_value = Connection(is_connected=True)
mock_return_prowler_provider.return_value = prowler_provider
connection = prowler_provider_connection_test(provider)
assert connection.is_connected is True
assert connection.error is None
mock_create_adapter.assert_called_once_with(
registry_url="ghcr.io",
username=None,
password=None,
token="tok",
prowler_provider.test_connection.assert_called_once_with(
registry="ghcr.io",
registry_username=None,
registry_password=None,
registry_token="tok",
raise_on_exception=False,
)
mock_adapter.list_repositories.assert_called_once()
@patch(
"prowler.providers.image.lib.registry.factory.create_registry_adapter",
)
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test_image_failure(
self, mock_return_prowler_provider, mock_create_adapter
self, mock_return_prowler_provider
):
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.uid = "ghcr.io"
provider.secret.secret = {"registry_token": "bad-token"}
mock_adapter = MagicMock()
mock_adapter.list_repositories.side_effect = Exception("401 Unauthorized")
mock_create_adapter.return_value = mock_adapter
prowler_provider = MagicMock()
prowler_provider.test_connection.return_value = Connection(
is_connected=False, error="401 Unauthorized"
)
mock_return_prowler_provider.return_value = prowler_provider
connection = prowler_provider_connection_test(provider)
+7 -13
View File
@@ -323,19 +323,13 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
}
return prowler_provider.test_connection(**openstack_kwargs)
elif provider.provider == Provider.ProviderChoices.IMAGE.value:
from prowler.providers.image.lib.registry.factory import create_registry_adapter
try:
adapter = create_registry_adapter(
registry_url=provider.uid,
username=prowler_provider_kwargs.get("registry_username"),
password=prowler_provider_kwargs.get("registry_password"),
token=prowler_provider_kwargs.get("registry_token"),
)
adapter.list_repositories()
return Connection(is_connected=True)
except Exception as e:
return Connection(is_connected=False, error=str(e))
return prowler_provider.test_connection(
registry=provider.uid,
registry_username=prowler_provider_kwargs.get("registry_username"),
registry_password=prowler_provider_kwargs.get("registry_password"),
registry_token=prowler_provider_kwargs.get("registry_token"),
raise_on_exception=False,
)
else:
return prowler_provider.test_connection(
**prowler_provider_kwargs,
+24 -2
View File
@@ -896,17 +896,23 @@ class ImageProvider(Provider):
image: str | None = None,
raise_on_exception: bool = True,
provider_id: str | None = None,
registry: str | None = None,
registry_username: str | None = None,
registry_password: str | None = None,
registry_token: str | None = None,
) -> "Connection":
"""
Test connection to container registry by attempting to inspect an image.
Test connection to a container registry or individual image.
When ``registry`` is provided, validates registry-level connectivity
by listing repositories through the registry adapter. Otherwise
falls back to testing access to an individual ``image`` via Trivy.
Args:
image: Container image to test
image: Container image to test (individual image mode)
raise_on_exception: Whether to raise exceptions
provider_id: Fallback for image name
registry: Registry URL for registry-level connectivity test
registry_username: Registry username for basic auth
registry_password: Registry password for basic auth
registry_token: Registry token for token-based auth
@@ -915,6 +921,22 @@ class ImageProvider(Provider):
Connection: Connection object with success status
"""
try:
# Registry-level connectivity test (used by API)
if registry:
from prowler.providers.image.lib.registry.factory import (
create_registry_adapter,
)
adapter = create_registry_adapter(
registry_url=registry,
username=registry_username,
password=registry_password,
token=registry_token,
)
adapter.list_repositories()
return Connection(is_connected=True)
# Individual image test (existing behaviour)
if provider_id and not image:
image = provider_id