mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
test(image): update mock patch targets after base class refactor
Update test patches from adapter-level to base-level module paths since requests and time are now imported in base.py. Consolidate separate requests.post/requests.get patches into requests.request since all HTTP calls now go through _request_with_retry.
This commit is contained in:
@@ -30,20 +30,18 @@ class TestDockerHubAdapterInit:
|
||||
|
||||
|
||||
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
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_repos(self, mock_request):
|
||||
# Hub login (now goes through requests.request via _request_with_retry)
|
||||
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
|
||||
mock_request.side_effect = [login_resp, repos_resp]
|
||||
adapter = DockerHubAdapter("docker.io/myorg", username="u", password="p")
|
||||
repos = adapter.list_repositories()
|
||||
assert repos == ["myorg/app1", "myorg/app2"]
|
||||
@@ -53,7 +51,7 @@ class TestDockerHubListRepositories:
|
||||
with pytest.raises(ImageRegistryAuthError, match="namespace"):
|
||||
adapter.list_repositories()
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_repos_public_no_credentials(self, mock_request):
|
||||
"""When no credentials are provided, use the public /v2/repositories/{ns}/ endpoint."""
|
||||
repos_resp = MagicMock(status_code=200)
|
||||
@@ -71,39 +69,37 @@ class TestDockerHubListRepositories:
|
||||
|
||||
|
||||
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
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_tags(self, mock_request):
|
||||
# Token exchange (now goes through requests.request via _request_with_retry)
|
||||
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
|
||||
mock_request.side_effect = [token_resp, 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):
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_tags_auth_failure(self, mock_request):
|
||||
# Token exchange
|
||||
token_resp = MagicMock(status_code=200)
|
||||
token_resp.json.return_value = {"token": "tok"}
|
||||
mock_get.return_value = token_resp
|
||||
# Tag listing returns 401
|
||||
tags_resp = MagicMock(status_code=401)
|
||||
mock_request.return_value = tags_resp
|
||||
mock_request.side_effect = [token_resp, 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):
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_login_failure(self, mock_request):
|
||||
resp = MagicMock(status_code=401)
|
||||
mock_post.return_value = resp
|
||||
mock_request.return_value = resp
|
||||
adapter = DockerHubAdapter("docker.io/myorg", username="bad", password="creds")
|
||||
with pytest.raises(ImageRegistryAuthError, match="login failed"):
|
||||
adapter._hub_login()
|
||||
@@ -115,8 +111,8 @@ class TestDockerHubLogin:
|
||||
|
||||
|
||||
class TestDockerHubRetry:
|
||||
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.time.sleep")
|
||||
@patch("prowler.providers.image.lib.registry.dockerhub_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.time.sleep")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_retry_on_429(self, mock_request, mock_sleep):
|
||||
resp_429 = MagicMock(status_code=429)
|
||||
resp_200 = MagicMock(status_code=200)
|
||||
@@ -127,8 +123,8 @@ class TestDockerHubRetry:
|
||||
)
|
||||
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")
|
||||
@patch("prowler.providers.image.lib.registry.base.time.sleep")
|
||||
@patch("prowler.providers.image.lib.registry.base.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")
|
||||
|
||||
@@ -36,14 +36,14 @@ class TestOciAdapterInit:
|
||||
|
||||
|
||||
class TestOciAdapterAuth:
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.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")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_ensure_auth_anonymous_ok(self, mock_request):
|
||||
resp = MagicMock(status_code=200)
|
||||
mock_request.return_value = resp
|
||||
@@ -51,7 +51,7 @@ class TestOciAdapterAuth:
|
||||
adapter._ensure_auth()
|
||||
assert adapter._bearer_token is None
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_ensure_auth_bearer_challenge(self, mock_request):
|
||||
ping_resp = MagicMock(
|
||||
status_code=401,
|
||||
@@ -66,7 +66,7 @@ class TestOciAdapterAuth:
|
||||
adapter._ensure_auth()
|
||||
assert adapter._bearer_token == "bearer-tok"
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_ensure_auth_403_raises(self, mock_request):
|
||||
resp = MagicMock(status_code=403)
|
||||
mock_request.return_value = resp
|
||||
@@ -74,7 +74,7 @@ class TestOciAdapterAuth:
|
||||
with pytest.raises(ImageRegistryAuthError):
|
||||
adapter._ensure_auth()
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_ensure_auth_basic_challenge_with_creds(self, mock_request):
|
||||
ping_resp = MagicMock(
|
||||
status_code=401,
|
||||
@@ -86,7 +86,7 @@ class TestOciAdapterAuth:
|
||||
assert adapter._basic_auth_verified is True
|
||||
assert adapter._bearer_token is None
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_ensure_auth_basic_challenge_no_creds(self, mock_request):
|
||||
ping_resp = MagicMock(
|
||||
status_code=401,
|
||||
@@ -97,7 +97,7 @@ class TestOciAdapterAuth:
|
||||
with pytest.raises(ImageRegistryAuthError):
|
||||
adapter._ensure_auth()
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_basic_auth_used_in_requests(self, mock_request):
|
||||
ping_resp = MagicMock(
|
||||
status_code=401,
|
||||
@@ -136,7 +136,7 @@ class TestOciAdapterAuth:
|
||||
assert user == "AWS"
|
||||
assert pwd == "not!valid~base64"
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_basic_auth_decodes_ecr_token_in_request(self, mock_request):
|
||||
raw_password = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.abc"
|
||||
encoded = base64.b64encode(f"AWS:{raw_password}".encode()).decode()
|
||||
@@ -155,7 +155,7 @@ class TestOciAdapterAuth:
|
||||
|
||||
|
||||
class TestOciAdapterListRepositories:
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_repos_single_page(self, mock_request):
|
||||
ping_resp = MagicMock(status_code=200)
|
||||
catalog_resp = MagicMock(status_code=200, headers={})
|
||||
@@ -167,7 +167,7 @@ class TestOciAdapterListRepositories:
|
||||
repos = adapter.list_repositories()
|
||||
assert repos == ["app/frontend", "app/backend"]
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_repos_paginated(self, mock_request):
|
||||
ping_resp = MagicMock(status_code=200)
|
||||
page1_resp = MagicMock(
|
||||
@@ -182,7 +182,7 @@ class TestOciAdapterListRepositories:
|
||||
repos = adapter.list_repositories()
|
||||
assert repos == ["a", "b"]
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_repos_404_raises(self, mock_request):
|
||||
ping_resp = MagicMock(status_code=200)
|
||||
catalog_resp = MagicMock(status_code=404)
|
||||
@@ -193,7 +193,7 @@ class TestOciAdapterListRepositories:
|
||||
|
||||
|
||||
class TestOciAdapterListTags:
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_tags(self, mock_request):
|
||||
ping_resp = MagicMock(status_code=200)
|
||||
tags_resp = MagicMock(status_code=200, headers={})
|
||||
@@ -203,7 +203,7 @@ class TestOciAdapterListTags:
|
||||
tags = adapter.list_tags("myapp")
|
||||
assert tags == ["latest", "v1.0"]
|
||||
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_list_tags_null_tags(self, mock_request):
|
||||
ping_resp = MagicMock(status_code=200)
|
||||
tags_resp = MagicMock(status_code=200, headers={})
|
||||
@@ -215,8 +215,8 @@ class TestOciAdapterListTags:
|
||||
|
||||
|
||||
class TestOciAdapterRetry:
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.time.sleep")
|
||||
@patch("prowler.providers.image.lib.registry.oci_adapter.requests.request")
|
||||
@patch("prowler.providers.image.lib.registry.base.time.sleep")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_retry_on_429(self, mock_request, mock_sleep):
|
||||
resp_429 = MagicMock(status_code=429)
|
||||
resp_200 = MagicMock(status_code=200)
|
||||
@@ -226,8 +226,8 @@ class TestOciAdapterRetry:
|
||||
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")
|
||||
@patch("prowler.providers.image.lib.registry.base.time.sleep")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_connection_error_retries(self, mock_request, mock_sleep):
|
||||
mock_request.side_effect = requests.exceptions.ConnectionError("failed")
|
||||
adapter = OciRegistryAdapter("reg.io")
|
||||
@@ -235,7 +235,7 @@ class TestOciAdapterRetry:
|
||||
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")
|
||||
@patch("prowler.providers.image.lib.registry.base.requests.request")
|
||||
def test_timeout_raises_immediately(self, mock_request):
|
||||
mock_request.side_effect = requests.exceptions.Timeout("timeout")
|
||||
adapter = OciRegistryAdapter("reg.io")
|
||||
|
||||
@@ -200,7 +200,7 @@ class TestRegistryList:
|
||||
captured = capsys.readouterr()
|
||||
assert "v1.0" in captured.out
|
||||
assert "dev-abc" not in captured.out
|
||||
assert "1 images" in captured.out
|
||||
assert "1 image)" in captured.out
|
||||
|
||||
@patch("prowler.providers.image.image_provider.create_registry_adapter")
|
||||
def test_registry_list_skips_max_images(self, mock_factory, capsys):
|
||||
|
||||
Reference in New Issue
Block a user