basic artifact identifier validation (eiffel-community#119)

Author: andmat900

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
 
 

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
 
 

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.