diff --git a/python/src/etos_api/routers/v0/schemas.py b/python/src/etos_api/routers/v0/schemas.py index 19817c2..5864992 100644 --- a/python/src/etos_api/routers/v0/schemas.py +++ b/python/src/etos_api/routers/v0/schemas.py @@ -49,7 +49,7 @@ class StartEtosRequest(EtosRequest): @field_validator("artifact_id") def validate_id_or_identity(cls, artifact_id, info): - """Validate that at least one and only one of id and identity are set. + """Validate that id/identity is set correctly. :param artifact_id: The value of 'artifact_id' to validate. :value artifact_id: str or None @@ -59,10 +59,23 @@ def validate_id_or_identity(cls, artifact_id, info): :rtype: str or None """ values = info.data - if values.get("artifact_identity") is None and not artifact_id: + artifact_identity = values.get("artifact_identity") + + # Check that at least one is provided + if artifact_identity is None and not artifact_id: raise ValueError("At least one of 'artifact_identity' or 'artifact_id' is required.") - if values.get("artifact_identity") is not None and artifact_id: + + # Check that only one is provided + if artifact_identity is not None and artifact_id: raise ValueError("Only one of 'artifact_identity' or 'artifact_id' is required.") + + # Validate artifact_identity format if provided + if artifact_identity is not None: + if not isinstance(artifact_identity, str) or not artifact_identity.startswith("pkg:"): + raise ValueError("artifact_identity must be a string starting with 'pkg:'") + + # Note: artifact_id UUID validation is handled by Pydantic's built-in UUID type validation + return artifact_id diff --git a/python/src/etos_api/routers/v1alpha/schemas.py b/python/src/etos_api/routers/v1alpha/schemas.py index 9741603..17a24ce 100644 --- a/python/src/etos_api/routers/v1alpha/schemas.py +++ b/python/src/etos_api/routers/v1alpha/schemas.py @@ -48,7 +48,7 @@ class StartTestrunRequest(TestrunRequest): @field_validator("artifact_id") def validate_id_or_identity(cls, artifact_id, info): - """Validate that at least one and only one of id and identity are set. + """Validate that id/identity is set correctly. :param artifact_id: The value of 'artifact_id' to validate. :value artifact_id: str or None @@ -58,10 +58,23 @@ def validate_id_or_identity(cls, artifact_id, info): :rtype: str or None """ values = info.data - if values.get("artifact_identity") is None and not artifact_id: + artifact_identity = values.get("artifact_identity") + + # Check that at least one is provided + if artifact_identity is None and not artifact_id: raise ValueError("At least one of 'artifact_identity' or 'artifact_id' is required.") - if values.get("artifact_identity") is not None and artifact_id: + + # Check that only one is provided + if artifact_identity is not None and artifact_id: raise ValueError("Only one of 'artifact_identity' or 'artifact_id' is required.") + + # Validate artifact_identity format if provided + if artifact_identity is not None: + if not isinstance(artifact_identity, str) or not artifact_identity.startswith("pkg:"): + raise ValueError("artifact_identity must be a string starting with 'pkg:'") + + # Note: artifact_id UUID validation is handled by Pydantic's built-in UUID type validation + return artifact_id diff --git a/python/tests/test_routers.py b/python/tests/test_routers.py index 89448c8..ced2169 100644 --- a/python/tests/test_routers.py +++ b/python/tests/test_routers.py @@ -232,6 +232,103 @@ def test_start_etos_empty_suite(self, download_suite_mock, digest_mock): break assert tercc is None + def test_start_etos_missing_artifact_identity_and_id(self): + """Test that POST requests to /etos with missing artifact identity and ID fail validation. + + Approval criteria: + - POST requests to ETOS with missing artifact_identity and artifact_id shall return 422. + - The error message shall indicate that at least one is required. + + Test steps:: + 1. Send a POST request to etos without artifact_identity or artifact_id. + 2. Verify that the status code is 422. + 3. Verify that the error message indicates missing required field. + """ + self.logger.info( + "STEP: Send a POST request to etos without artifact_identity or artifact_id." + ) + response = self.client.post( + "/api/etos", + json={ + "test_suite_url": "http://localhost/my_test.json", + "artifact_id": None, # Explicitly set to None to trigger validation + }, + ) + self.logger.info("STEP: Verify that the status code is 422.") + assert response.status_code == 422 + + self.logger.info("STEP: Verify that the error message indicates missing required field.") + error_detail = response.json() + assert "detail" in error_detail + error_messages = [error["msg"] for error in error_detail["detail"]] + expected_message = "At least one of 'artifact_identity' or 'artifact_id' is required." + assert any(expected_message in msg for msg in error_messages) + + def test_start_etos_empty_artifact_identity_and_none_artifact_id(self): + """Test that POST requests to /etos with empty artifact_identity fail validation. + + Approval criteria: + - POST requests to ETOS with empty artifact_identity shall return 422. + - The error message shall indicate invalid format (empty doesn't start with 'pkg:'). + + Test steps:: + 1. Send a POST request to etos with empty artifact_identity and None artifact_id. + 2. Verify that the status code is 422. + 3. Verify that the error message indicates invalid format. + """ + self.logger.info( + "STEP: Send a POST request to etos with empty artifact_identity and None artifact_id." + ) + response = self.client.post( + "/api/etos", + json={ + "artifact_identity": "", + "artifact_id": None, + "test_suite_url": "http://localhost/my_test.json", + }, + ) + self.logger.info("STEP: Verify that the status code is 422.") + assert response.status_code == 422 + + self.logger.info("STEP: Verify that the error message indicates invalid format.") + error_detail = response.json() + assert "detail" in error_detail + error_messages = [error["msg"] for error in error_detail["detail"]] + expected_message = "artifact_identity must be a string starting with 'pkg:'" + assert any(expected_message in msg for msg in error_messages) + + def test_start_etos_both_artifact_identity_and_id_provided(self): + """ + Approval criteria: + - POST requests to ETOS with both artifact_identity and artifact_id returns 422. + - The error message shall indicate that only one is required. + + Test steps:: + 1. Send a POST request to etos with both artifact_identity and artifact_id. + 2. Verify that the status code is 422. + 3. Verify that the error message indicates only one is required. + """ + self.logger.info( + "STEP: Send a POST request to etos with both artifact_identity and artifact_id." + ) + response = self.client.post( + "/api/etos", + json={ + "artifact_identity": "pkg:testing/etos", + "artifact_id": "123e4567-e89b-12d3-a456-426614174000", + "test_suite_url": "http://localhost/my_test.json", + }, + ) + self.logger.info("STEP: Verify that the status code is 422.") + assert response.status_code == 422 + + self.logger.info("STEP: Verify that the error message indicates only one is required.") + error_detail = response.json() + assert "detail" in error_detail + error_messages = [error["msg"] for error in error_detail["detail"]] + expected_message = "Only one of 'artifact_identity' or 'artifact_id' is required." + assert any(expected_message in msg for msg in error_messages) + def test_selftest_get_ping(self): """Test that selftest ping with HTTP GET pings the system.