diff --git a/lib/ingestor-api/runtime/src/config.py b/lib/ingestor-api/runtime/src/config.py index 5694c7d..638501e 100644 --- a/lib/ingestor-api/runtime/src/config.py +++ b/lib/ingestor-api/runtime/src/config.py @@ -1,11 +1,18 @@ import os from getpass import getuser -from typing import Optional +from typing import Annotated, Optional -from pydantic import AnyHttpUrl, Field, constr +from pydantic import ( + AfterValidator, + AnyHttpUrl, + Field, + constr, +) from pydantic_settings import BaseSettings from pydantic_ssm_settings.settings import SsmBaseSettings +HttpUrlString = Annotated[AnyHttpUrl, AfterValidator(str)] + AwsArn = constr(pattern=r"^arn:aws:iam::\d{12}:role/.+") @@ -14,11 +21,11 @@ class Settings(BaseSettings): root_path: Optional[str] = Field(description="Path from where to serve this URL.") - jwks_url: Optional[AnyHttpUrl] = Field( + jwks_url: Optional[HttpUrlString] = Field( description="URL of JWKS, e.g. https://cognito-idp.{region}.amazonaws.com/{userpool_id}/.well-known/jwks.json" # noqa ) - stac_url: AnyHttpUrl = Field(description="URL of STAC API") + stac_url: HttpUrlString = Field(description="URL of STAC API") data_access_role: AwsArn = Field( description="ARN of AWS Role used to validate access to S3 data" diff --git a/lib/ingestor-api/runtime/tests/conftest.py b/lib/ingestor-api/runtime/tests/conftest.py index a9cde2a..a54ffcf 100644 --- a/lib/ingestor-api/runtime/tests/conftest.py +++ b/lib/ingestor-api/runtime/tests/conftest.py @@ -1,5 +1,3 @@ -import os - import boto3 import pytest from fastapi.testclient import TestClient @@ -8,20 +6,20 @@ @pytest.fixture -def test_environ(): - # Mocked AWS Credentials for moto (best practice recommendation from moto) - os.environ["AWS_ACCESS_KEY_ID"] = "testing" - os.environ["AWS_SECRET_ACCESS_KEY"] = "testing" - os.environ["AWS_SECURITY_TOKEN"] = "testing" - os.environ["AWS_SESSION_TOKEN"] = "testing" +def test_environ(monkeypatch): + # Mocked AWS Credentials for moto + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing") + monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing") + monkeypatch.setenv("AWS_SESSION_TOKEN", "testing") # Config mocks - os.environ["DYNAMODB_TABLE"] = "test_table" - os.environ["JWKS_URL"] = "https://test-jwks.url" - os.environ["STAC_URL"] = "https://test-stac.url" - os.environ["DATA_ACCESS_ROLE"] = "arn:aws:iam::123456789012:role/test-role" - os.environ["DB_SECRET_ARN"] = "testing" - os.environ["ROOT_PATH"] = "testing" + monkeypatch.setenv("DYNAMODB_TABLE", "test_table") + monkeypatch.setenv("JWKS_URL", "https://test-jwks.url") + monkeypatch.setenv("STAC_URL", "https://test-stac.url") + monkeypatch.setenv("DATA_ACCESS_ROLE", "arn:aws:iam::123456789012:role/test-role") + monkeypatch.setenv("DB_SECRET_ARN", "testing") + monkeypatch.setenv("ROOT_PATH", "testing") @pytest.fixture diff --git a/lib/ingestor-api/runtime/tests/test_validators.py b/lib/ingestor-api/runtime/tests/test_validators.py new file mode 100644 index 0000000..4ae1e2e --- /dev/null +++ b/lib/ingestor-api/runtime/tests/test_validators.py @@ -0,0 +1,164 @@ +from unittest.mock import MagicMock, patch + +import pytest +import requests +from pydantic import AnyHttpUrl +from src import validators + + +@pytest.fixture +def mock_settings(): + """Fixture to instantiate and patch settings with proper types.""" + from src.config import Settings + + mock_settings = Settings( + dynamodb_table="test-table", + stac_url=AnyHttpUrl("https://test-stac.url"), + data_access_role="arn:aws:iam::123456789012:role/test-role", + requester_pays=False, + jwks_url=AnyHttpUrl("https://test-jwks.url"), + root_path="testing", + ) + + with patch("src.config.settings", mock_settings): + yield mock_settings + + +@pytest.fixture +def mock_requests(): + """Fixture to mock requests library.""" + with patch("src.validators.requests") as mock_requests: + mock_response = MagicMock() + mock_response.ok = True + mock_response.raise_for_status.return_value = None + mock_requests.get.return_value = mock_response + mock_requests.head.return_value = mock_response + + mock_requests.exceptions = requests.exceptions + + yield mock_requests + + +@pytest.fixture +def mock_boto3(): + """Fixture to mock boto3 library.""" + with patch("src.validators.boto3") as mock_boto3: + mock_client = MagicMock() + mock_client.exceptions.ClientError = Exception + + mock_sts_client = MagicMock() + mock_sts_client.assume_role.return_value = { + "Credentials": { + "AccessKeyId": "test_access_key", + "SecretAccessKey": "test_secret_key", + "SessionToken": "test_session_token", + } + } + + def mock_client_factory(service_name, **kwargs): + if service_name == "sts": + return mock_sts_client + return mock_client + + mock_boto3.client.side_effect = mock_client_factory + + yield mock_boto3, mock_client + + +class TestValidators: + def test_collection_exists_success(self, mock_settings, mock_requests): + """Test collection_exists when the collection exists.""" + validators.collection_exists.cache_clear() + + result = validators.collection_exists("test-collection") + + assert result + + expected_url = f"{mock_settings.stac_url}collections/test-collection" + mock_requests.get.assert_called_once_with(expected_url) + + def test_collection_exists_failure(self, mock_settings, mock_requests): + """Test collection_exists when the collection doesn't exist.""" + validators.collection_exists.cache_clear() + + mock_response = MagicMock() + mock_response.ok = False + mock_response.status_code = 404 + mock_requests.get.return_value = mock_response + + with pytest.raises(ValueError) as excinfo: + validators.collection_exists("nonexistent-collection") + + assert "Invalid collection 'nonexistent-collection'" in str(excinfo.value) + assert "404 response code" in str(excinfo.value) + + def test_url_is_accessible_success(self, mock_requests): + """Test url_is_accessible when the URL is accessible.""" + validators.url_is_accessible("https://example.com/asset.tif") + + mock_requests.head.assert_called_once_with("https://example.com/asset.tif") + + def test_url_is_accessible_failure(self, mock_requests): + """Test url_is_accessible when the URL is not accessible.""" + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError( + response=MagicMock(status_code=403, reason="Forbidden") + ) + mock_requests.head.return_value = mock_response + + with pytest.raises(ValueError) as excinfo: + validators.url_is_accessible("https://example.com/private.tif") + + assert "Asset not accessible" in str(excinfo.value) + + def test_s3_object_is_accessible_success(self, mock_settings, mock_boto3): + """Test s3_object_is_accessible when the object is accessible.""" + _, mock_s3_client = mock_boto3 + + validators.s3_object_is_accessible("test-bucket", "test-key") + + mock_s3_client.head_object.assert_called_once_with( + Bucket="test-bucket", Key="test-key" + ) + + def test_s3_object_is_accessible_with_requester_pays(self, mock_settings, mock_boto3): + """Test s3_object_is_accessible with requester pays enabled.""" + _, mock_s3_client = mock_boto3 + + mock_settings.requester_pays = True + + validators.s3_object_is_accessible("test-bucket", "test-key") + + mock_s3_client.head_object.assert_called_once_with( + Bucket="test-bucket", Key="test-key", RequestPayer="requester" + ) + + def test_s3_object_is_accessible_failure(self, mock_settings, mock_boto3): + """Test s3_object_is_accessible when the object is not accessible.""" + _, mock_s3_client = mock_boto3 + + _ = {"Error": {"Message": "Access Denied"}} + mock_s3_client.head_object.side_effect = Exception() + mock_s3_client.head_object.side_effect.__dict__["response"] = { + "Error": {"Message": "Access Denied"} + } + + with pytest.raises(ValueError) as excinfo: + validators.s3_object_is_accessible("test-bucket", "private-key") + + assert "Asset not accessible" in str(excinfo.value) + + def test_get_s3_credentials(self, mock_boto3): + """Test get_s3_credentials returns the expected credentials.""" + _, _ = mock_boto3 + + validators.get_s3_credentials.cache_clear() + + credentials = validators.get_s3_credentials() + + expected_credentials = { + "aws_access_key_id": "test_access_key", + "aws_secret_access_key": "test_secret_key", + "aws_session_token": "test_session_token", + } + assert credentials == expected_credentials