From aa284b8b24ba7b61ae093e79ccdd30884719abcd Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Mon, 16 Feb 2026 15:42:02 +0100 Subject: [PATCH] test(image): add comprehensive registry scan mode unit tests 69 tests covering: - Factory auto-detection (OCI, Docker Hub, ECR URL patterns) - OCI adapter (auth flows, pagination, retry, error handling) - Docker Hub adapter (namespace extraction, Hub API, token exchange) - Argument validation (registry flags, Docker Hub namespace) - Provider integration (enumeration, filtering, max-images, dedup) --- .../image/lib/registry/test_arguments.py | 79 +++++++++ .../lib/registry/test_dockerhub_adapter.py | 117 ++++++++++++ .../image/lib/registry/test_factory.py | 63 +++++++ .../image/lib/registry/test_oci_adapter.py | 166 ++++++++++++++++++ .../lib/registry/test_provider_registry.py | 149 ++++++++++++++++ 5 files changed, 574 insertions(+) create mode 100644 tests/providers/image/lib/registry/test_arguments.py create mode 100644 tests/providers/image/lib/registry/test_dockerhub_adapter.py create mode 100644 tests/providers/image/lib/registry/test_factory.py create mode 100644 tests/providers/image/lib/registry/test_oci_adapter.py create mode 100644 tests/providers/image/lib/registry/test_provider_registry.py diff --git a/tests/providers/image/lib/registry/test_arguments.py b/tests/providers/image/lib/registry/test_arguments.py new file mode 100644 index 0000000000..416c3792da --- /dev/null +++ b/tests/providers/image/lib/registry/test_arguments.py @@ -0,0 +1,79 @@ +from argparse import Namespace + +import pytest + +from prowler.providers.image.lib.arguments.arguments import validate_arguments + + +class TestValidateArguments: + def test_no_source_fails(self): + args = Namespace(images=[], image_list_file=None, registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, msg = validate_arguments(args) + assert not ok + assert "--image" in msg + + def test_image_only_passes(self): + args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, _ = validate_arguments(args) + assert ok + + def test_image_list_only_passes(self): + args = Namespace(images=[], image_list_file="images.txt", registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, _ = validate_arguments(args) + assert ok + + def test_registry_only_passes(self): + args = Namespace(images=[], image_list_file=None, registry="myregistry.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, _ = validate_arguments(args) + assert ok + + def test_image_filter_without_registry_fails(self): + args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter="^prod", tag_filter=None, max_images=0, registry_insecure=False) + ok, msg = validate_arguments(args) + assert not ok + assert "--image-filter requires --registry" in msg + + def test_tag_filter_without_registry_fails(self): + args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter=None, tag_filter="^v", max_images=0, registry_insecure=False) + ok, msg = validate_arguments(args) + assert not ok + assert "--tag-filter requires --registry" in msg + + def test_max_images_without_registry_fails(self): + args = Namespace(images=["nginx:latest"], image_list_file=None, registry=None, image_filter=None, tag_filter=None, max_images=50, registry_insecure=False) + ok, msg = validate_arguments(args) + assert not ok + assert "--max-images requires --registry" in msg + + def test_registry_insecure_without_registry_fails(self): + args = Namespace(images=[], image_list_file="i.txt", registry=None, image_filter=None, tag_filter=None, max_images=0, registry_insecure=True) + ok, msg = validate_arguments(args) + assert not ok + assert "--registry-insecure requires --registry" in msg + + def test_docker_hub_no_namespace_fails(self): + args = Namespace(images=[], image_list_file=None, registry="docker.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, msg = validate_arguments(args) + assert not ok + assert "namespace" in msg.lower() + + def test_docker_hub_with_namespace_passes(self): + args = Namespace(images=[], image_list_file=None, registry="docker.io/myorg", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, _ = validate_arguments(args) + assert ok + + def test_docker_hub_https_no_namespace_fails(self): + args = Namespace(images=[], image_list_file=None, registry="https://docker.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, msg = validate_arguments(args) + assert not ok + assert "namespace" in msg.lower() + + def test_registry_with_filters_passes(self): + args = Namespace(images=[], image_list_file=None, registry="myregistry.io", image_filter="^prod", tag_filter="^v", max_images=100, registry_insecure=True) + ok, _ = validate_arguments(args) + assert ok + + def test_combined_registry_and_image_passes(self): + args = Namespace(images=["nginx:latest"], image_list_file=None, registry="myregistry.io", image_filter=None, tag_filter=None, max_images=0, registry_insecure=False) + ok, _ = validate_arguments(args) + assert ok diff --git a/tests/providers/image/lib/registry/test_dockerhub_adapter.py b/tests/providers/image/lib/registry/test_dockerhub_adapter.py new file mode 100644 index 0000000000..d8a68300a4 --- /dev/null +++ b/tests/providers/image/lib/registry/test_dockerhub_adapter.py @@ -0,0 +1,117 @@ +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from prowler.providers.image.exceptions.exceptions import ( + ImageRegistryAuthError, + ImageRegistryNetworkError, +) +from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter + + +class TestDockerHubAdapterInit: + def test_extract_namespace_simple(self): + assert DockerHubAdapter._extract_namespace("docker.io/myorg") == "myorg" + + def test_extract_namespace_https(self): + assert DockerHubAdapter._extract_namespace("https://docker.io/myorg") == "myorg" + + def test_extract_namespace_registry1(self): + assert DockerHubAdapter._extract_namespace("registry-1.docker.io/myorg") == "myorg" + + def test_extract_namespace_empty(self): + assert DockerHubAdapter._extract_namespace("docker.io") == "" + + def test_extract_namespace_with_slash(self): + assert DockerHubAdapter._extract_namespace("docker.io/myorg/") == "myorg" + + +class TestDockerHubListRepositories: + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request") + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.post") + def test_list_repos(self, mock_post, mock_request): + # Hub login + login_resp = MagicMock(status_code=200) + login_resp.json.return_value = {"token": "jwt"} + mock_post.return_value = login_resp + # Repo listing + repos_resp = MagicMock(status_code=200) + repos_resp.json.return_value = { + "results": [{"name": "app1"}, {"name": "app2"}], + "next": None, + } + mock_request.return_value = repos_resp + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + repos = adapter.list_repositories() + assert repos == ["myorg/app1", "myorg/app2"] + + def test_list_repos_no_namespace_raises(self): + adapter = DockerHubAdapter("docker.io") + with pytest.raises(ImageRegistryAuthError, match="namespace"): + adapter.list_repositories() + + +class TestDockerHubListTags: + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request") + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.get") + def test_list_tags(self, mock_get, mock_request): + # Token exchange + token_resp = MagicMock(status_code=200) + token_resp.json.return_value = {"token": "registry-token"} + mock_get.return_value = token_resp + # Tag listing + tags_resp = MagicMock(status_code=200, headers={}) + tags_resp.json.return_value = {"tags": ["latest", "v1.0"]} + mock_request.return_value = tags_resp + adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p") + tags = adapter.list_tags("myorg/myapp") + assert tags == ["latest", "v1.0"] + + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request") + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.get") + def test_list_tags_auth_failure(self, mock_get, mock_request): + token_resp = MagicMock(status_code=200) + token_resp.json.return_value = {"token": "tok"} + mock_get.return_value = token_resp + tags_resp = MagicMock(status_code=401) + mock_request.return_value = tags_resp + adapter = DockerHubAdapter("docker.io/myorg") + with pytest.raises(ImageRegistryAuthError): + adapter.list_tags("myorg/myapp") + + +class TestDockerHubLogin: + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.post") + def test_login_failure(self, mock_post): + resp = MagicMock(status_code=401) + mock_post.return_value = resp + adapter = DockerHubAdapter("docker.io/myorg", username="bad", password="creds") + with pytest.raises(ImageRegistryAuthError, match="login failed"): + adapter._hub_login() + + def test_login_skipped_without_credentials(self): + adapter = DockerHubAdapter("docker.io/myorg") + adapter._hub_login() # Should not raise + assert adapter._hub_jwt is None + + +class TestDockerHubRetry: + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.time.sleep") + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request") + def test_retry_on_429(self, mock_request, mock_sleep): + resp_429 = MagicMock(status_code=429) + resp_200 = MagicMock(status_code=200) + mock_request.side_effect = [resp_429, resp_200] + adapter = DockerHubAdapter("docker.io/myorg") + result = adapter._request_with_retry("GET", "https://hub.docker.com/v2/namespaces/myorg/repositories") + assert result.status_code == 200 + + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.time.sleep") + @patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request") + def test_connection_error_retries(self, mock_request, mock_sleep): + mock_request.side_effect = requests.exceptions.ConnectionError("fail") + adapter = DockerHubAdapter("docker.io/myorg") + with pytest.raises(ImageRegistryNetworkError): + adapter._request_with_retry("GET", "https://hub.docker.com") + assert mock_request.call_count == 3 diff --git a/tests/providers/image/lib/registry/test_factory.py b/tests/providers/image/lib/registry/test_factory.py new file mode 100644 index 0000000000..1e674ef087 --- /dev/null +++ b/tests/providers/image/lib/registry/test_factory.py @@ -0,0 +1,63 @@ +from unittest.mock import patch + +import pytest + +from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter +from prowler.providers.image.lib.registry.factory import ( + create_registry_adapter, + detect_registry_type, +) +from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter + + +class TestDetectRegistryType: + def test_generic_oci(self): + assert detect_registry_type("myregistry.io") == "oci" + + def test_generic_oci_with_https(self): + assert detect_registry_type("https://myregistry.io") == "oci" + + def test_docker_hub(self): + assert detect_registry_type("docker.io/myorg") == "dockerhub" + + def test_docker_hub_registry1(self): + assert detect_registry_type("registry-1.docker.io/myorg") == "dockerhub" + + def test_docker_hub_with_https(self): + assert detect_registry_type("https://docker.io/myorg") == "dockerhub" + + def test_ecr(self): + assert detect_registry_type("123456789.dkr.ecr.us-east-1.amazonaws.com") == "ecr" + + def test_ecr_with_https(self): + assert detect_registry_type("https://123456789.dkr.ecr.us-east-1.amazonaws.com") == "ecr" + + def test_harbor(self): + assert detect_registry_type("harbor.example.com") == "oci" + + +class TestCreateRegistryAdapter: + def test_docker_hub_returns_dockerhub_adapter(self): + adapter = create_registry_adapter("docker.io/myorg") + assert isinstance(adapter, DockerHubAdapter) + + def test_oci_returns_oci_adapter(self): + adapter = create_registry_adapter("myregistry.io") + assert isinstance(adapter, OciRegistryAdapter) + + def test_ecr_returns_oci_adapter(self): + adapter = create_registry_adapter("123456789.dkr.ecr.us-east-1.amazonaws.com") + assert isinstance(adapter, OciRegistryAdapter) + + def test_passes_credentials(self): + adapter = create_registry_adapter( + "myregistry.io", username="user", password="pass", token="tok", verify_ssl=False + ) + assert adapter.username == "user" + assert adapter.password == "pass" + assert adapter.token == "tok" + assert adapter.verify_ssl is False + + def test_registry_1_docker_io(self): + adapter = create_registry_adapter("registry-1.docker.io/myorg") + assert isinstance(adapter, DockerHubAdapter) diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py new file mode 100644 index 0000000000..385da5cea9 --- /dev/null +++ b/tests/providers/image/lib/registry/test_oci_adapter.py @@ -0,0 +1,166 @@ +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from prowler.providers.image.exceptions.exceptions import ( + ImageRegistryAuthError, + ImageRegistryCatalogError, + ImageRegistryNetworkError, +) +from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter + + +class TestOciAdapterInit: + def test_normalise_url_adds_https(self): + adapter = OciRegistryAdapter("myregistry.io") + assert adapter._base_url == "https://myregistry.io" + + def test_normalise_url_keeps_http(self): + adapter = OciRegistryAdapter("http://myregistry.io") + assert adapter._base_url == "http://myregistry.io" + + def test_normalise_url_strips_trailing_slash(self): + adapter = OciRegistryAdapter("https://myregistry.io/") + assert adapter._base_url == "https://myregistry.io" + + def test_stores_credentials(self): + adapter = OciRegistryAdapter("reg.io", username="u", password="p", token="t", verify_ssl=False) + assert adapter.username == "u" + assert adapter.password == "p" + assert adapter.token == "t" + assert adapter.verify_ssl is False + + +class TestOciAdapterAuth: + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_ensure_auth_with_token(self, mock_request): + adapter = OciRegistryAdapter("reg.io", token="my-token") + adapter._ensure_auth() + assert adapter._bearer_token == "my-token" + mock_request.assert_not_called() + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_ensure_auth_anonymous_ok(self, mock_request): + resp = MagicMock(status_code=200) + mock_request.return_value = resp + adapter = OciRegistryAdapter("reg.io") + adapter._ensure_auth() + assert adapter._bearer_token is None + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_ensure_auth_bearer_challenge(self, mock_request): + ping_resp = MagicMock(status_code=401, headers={"Www-Authenticate": 'Bearer realm="https://auth.example.com/token",service="registry"'}) + token_resp = MagicMock(status_code=200) + token_resp.json.return_value = {"token": "bearer-tok"} + mock_request.side_effect = [ping_resp, token_resp] + adapter = OciRegistryAdapter("reg.io", username="u", password="p") + adapter._ensure_auth() + assert adapter._bearer_token == "bearer-tok" + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_ensure_auth_403_raises(self, mock_request): + resp = MagicMock(status_code=403) + mock_request.return_value = resp + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryAuthError): + adapter._ensure_auth() + + +class TestOciAdapterListRepositories: + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_list_repos_single_page(self, mock_request): + ping_resp = MagicMock(status_code=200) + catalog_resp = MagicMock(status_code=200, headers={}) + catalog_resp.json.return_value = {"repositories": ["app/frontend", "app/backend"]} + mock_request.side_effect = [ping_resp, catalog_resp] + adapter = OciRegistryAdapter("reg.io") + repos = adapter.list_repositories() + assert repos == ["app/frontend", "app/backend"] + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_list_repos_paginated(self, mock_request): + ping_resp = MagicMock(status_code=200) + page1_resp = MagicMock(status_code=200, headers={"Link": '; rel="next"'}) + page1_resp.json.return_value = {"repositories": ["a"]} + page2_resp = MagicMock(status_code=200, headers={}) + page2_resp.json.return_value = {"repositories": ["b"]} + mock_request.side_effect = [ping_resp, page1_resp, page2_resp] + adapter = OciRegistryAdapter("reg.io") + repos = adapter.list_repositories() + assert repos == ["a", "b"] + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_list_repos_404_raises(self, mock_request): + ping_resp = MagicMock(status_code=200) + catalog_resp = MagicMock(status_code=404) + mock_request.side_effect = [ping_resp, catalog_resp] + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryCatalogError): + adapter.list_repositories() + + +class TestOciAdapterListTags: + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_list_tags(self, mock_request): + ping_resp = MagicMock(status_code=200) + tags_resp = MagicMock(status_code=200, headers={}) + tags_resp.json.return_value = {"tags": ["latest", "v1.0"]} + mock_request.side_effect = [ping_resp, tags_resp] + adapter = OciRegistryAdapter("reg.io") + tags = adapter.list_tags("myapp") + assert tags == ["latest", "v1.0"] + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_list_tags_null_tags(self, mock_request): + ping_resp = MagicMock(status_code=200) + tags_resp = MagicMock(status_code=200, headers={}) + tags_resp.json.return_value = {"tags": None} + mock_request.side_effect = [ping_resp, tags_resp] + adapter = OciRegistryAdapter("reg.io") + tags = adapter.list_tags("myapp") + assert tags == [] + + +class TestOciAdapterRetry: + @patch("prowler.providers.image.lib.registry.oci_adapter.time.sleep") + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_retry_on_429(self, mock_request, mock_sleep): + resp_429 = MagicMock(status_code=429) + resp_200 = MagicMock(status_code=200) + mock_request.side_effect = [resp_429, resp_200] + adapter = OciRegistryAdapter("reg.io") + result = adapter._request_with_retry("GET", "https://reg.io/v2/") + assert result.status_code == 200 + mock_sleep.assert_called_once() + + @patch("prowler.providers.image.lib.registry.oci_adapter.time.sleep") + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_connection_error_retries(self, mock_request, mock_sleep): + mock_request.side_effect = requests.exceptions.ConnectionError("failed") + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryNetworkError): + adapter._request_with_retry("GET", "https://reg.io/v2/") + assert mock_request.call_count == 3 + + @patch("prowler.providers.image.lib.registry.oci_adapter.requests.request") + def test_timeout_raises_immediately(self, mock_request): + mock_request.side_effect = requests.exceptions.Timeout("timeout") + adapter = OciRegistryAdapter("reg.io") + with pytest.raises(ImageRegistryNetworkError): + adapter._request_with_retry("GET", "https://reg.io/v2/") + assert mock_request.call_count == 1 + + +class TestOciAdapterNextPageUrl: + def test_no_link_header(self): + resp = MagicMock(headers={}) + assert OciRegistryAdapter._next_page_url(resp) is None + + def test_link_header_with_next(self): + resp = MagicMock(headers={"Link": '; rel="next"'}) + assert OciRegistryAdapter._next_page_url(resp) == "https://reg.io/v2/_catalog?n=200&last=b" + + def test_link_header_no_next(self): + resp = MagicMock(headers={"Link": '; rel="prev"'}) + assert OciRegistryAdapter._next_page_url(resp) is None diff --git a/tests/providers/image/lib/registry/test_provider_registry.py b/tests/providers/image/lib/registry/test_provider_registry.py new file mode 100644 index 0000000000..3f5d2fc8b3 --- /dev/null +++ b/tests/providers/image/lib/registry/test_provider_registry.py @@ -0,0 +1,149 @@ +import os +from argparse import Namespace +from unittest.mock import MagicMock, patch + +import pytest + +from prowler.providers.image.exceptions.exceptions import ( + ImageInvalidFilterError, + ImageMaxImagesExceededError, +) +from prowler.providers.image.image_provider import ImageProvider + +_CLEAN_ENV = { + "PATH": os.environ.get("PATH", ""), + "HOME": os.environ.get("HOME", ""), +} + + +def _build_provider(**overrides): + defaults = dict( + images=[], + registry="myregistry.io", + image_filter=None, + tag_filter=None, + max_images=0, + registry_insecure=False, + config_content={"image": {}}, + ) + defaults.update(overrides) + with patch.dict(os.environ, _CLEAN_ENV, clear=True): + return ImageProvider(**defaults) + + +class TestRegistryEnumeration: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_enumerate_oci_registry(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app/frontend", "app/backend"] + adapter.list_tags.side_effect = [["latest", "v1.0"], ["latest"]] + mock_factory.return_value = adapter + + provider = _build_provider() + assert "myregistry.io/app/frontend:latest" in provider.images + assert "myregistry.io/app/frontend:v1.0" in provider.images + assert "myregistry.io/app/backend:latest" in provider.images + assert len(provider.images) == 3 + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_image_filter(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["prod/app", "dev/app", "staging/app"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + provider = _build_provider(image_filter="^prod/") + assert len(provider.images) == 1 + assert "myregistry.io/prod/app:latest" in provider.images + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_tag_filter(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["myapp"] + adapter.list_tags.return_value = ["latest", "v1.0", "v2.0", "dev-abc123"] + mock_factory.return_value = adapter + + provider = _build_provider(tag_filter=r"^v\d+\.\d+$") + assert len(provider.images) == 2 + assert "myregistry.io/myapp:v1.0" in provider.images + assert "myregistry.io/myapp:v2.0" in provider.images + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_combined_filters(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["prod/app", "dev/app"] + adapter.list_tags.return_value = ["latest", "v1.0"] + mock_factory.return_value = adapter + + provider = _build_provider(image_filter="^prod/", tag_filter="^v") + assert len(provider.images) == 1 + assert "myregistry.io/prod/app:v1.0" in provider.images + + +class TestMaxImages: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_max_images_exceeded(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app1", "app2", "app3"] + adapter.list_tags.return_value = ["latest", "v1.0"] + mock_factory.return_value = adapter + + with pytest.raises(ImageMaxImagesExceededError): + _build_provider(max_images=2) + + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_max_images_not_exceeded(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app1"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + provider = _build_provider(max_images=10) + assert len(provider.images) == 1 + + +class TestDeduplication: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_deduplication_with_explicit_images(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["myapp"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + provider = _build_provider(images=["myregistry.io/myapp:latest"]) + assert provider.images.count("myregistry.io/myapp:latest") == 1 + + +class TestInvalidFilters: + def test_invalid_image_filter_regex(self): + with pytest.raises(ImageInvalidFilterError): + _build_provider(image_filter="[invalid") + + def test_invalid_tag_filter_regex(self): + with pytest.raises(ImageInvalidFilterError): + _build_provider(tag_filter="(unclosed") + + +class TestRegistryInsecure: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_insecure_passes_verify_false(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = ["app"] + adapter.list_tags.return_value = ["latest"] + mock_factory.return_value = adapter + + _build_provider(registry_insecure=True) + mock_factory.assert_called_once() + call_kwargs = mock_factory.call_args[1] + assert call_kwargs["verify_ssl"] is False + + +class TestEmptyRegistry: + @patch("prowler.providers.image.image_provider.create_registry_adapter") + def test_empty_catalog_with_explicit_images(self, mock_factory): + adapter = MagicMock() + adapter.list_repositories.return_value = [] + mock_factory.return_value = adapter + + provider = _build_provider(images=["nginx:latest"]) + assert provider.images == ["nginx:latest"]