Add post test results endpoint (#234)

backend/test_observer/controllers/test_executions/__init__.py

@@ -16,7 +16,15 @@
 
 from fastapi import APIRouter
 
-from . import end_test, get_test_results, patch, reruns, start_test, status_update
+from . import (
+    end_test,
+    get_test_results,
+    patch,
+    post_results,
+    reruns,
+    start_test,
+    status_update,
+)
 
 router = APIRouter(tags=["test-executions"])
 router.include_router(start_test.router)
@@ -25,3 +33,4 @@
 router.include_router(patch.router)
 router.include_router(reruns.router)
 router.include_router(status_update.router)
+router.include_router(post_results.router)

backend/test_observer/controllers/test_executions/end_test.py

@@ -45,7 +45,7 @@ def end_test_execution(request: EndTestExecutionRequest, db: Session = Depends(g
         raise HTTPException(status_code=404, detail="Related TestExecution not found")
 
     delete_previous_results(db, test_execution)
-    _store_test_results(db, request.test_results, test_execution)
+    _store_c3_test_results(db, request.test_results, test_execution)
 
     has_failures = test_execution.has_failures
 
@@ -82,7 +82,7 @@ def _find_related_test_execution(
     )
 
 
-def _store_test_results(
+def _store_c3_test_results(
     db: Session,
     c3_test_results: list[C3TestResult],
     test_execution: TestExecution,

backend/test_observer/controllers/test_executions/get_test_results.py

@@ -29,12 +29,12 @@
 from test_observer.data_access.setup import get_db
 
 from .logic import get_previous_test_results
-from .models import TestResultDTO
+from .models import TestResultResponse
 
-router = APIRouter()
+router = APIRouter(tags=["test-results"])
 
 
-@router.get("/{id}/test-results", response_model=list[TestResultDTO])
+@router.get("/{id}/test-results", response_model=list[TestResultResponse])
 def get_test_results(id: int, db: Session = Depends(get_db)):
     test_execution = db.get(
         TestExecution,
@@ -49,9 +49,9 @@ def get_test_results(id: int, db: Session = Depends(get_db)):
 
     previous_test_results = get_previous_test_results(db, test_execution)
 
-    test_results: list[TestResultDTO] = []
+    test_results: list[TestResultResponse] = []
     for test_result in test_execution.test_results:
-        parsed_test_result = TestResultDTO.model_validate(test_result)
+        parsed_test_result = TestResultResponse.model_validate(test_result)
         parsed_test_result.previous_results = previous_test_results.get(
             test_result.test_case_id, []
         )

backend/test_observer/controllers/test_executions/models.py

@@ -85,6 +85,15 @@ class C3TestResult(BaseModel):
     io_log: str
 
 
+class TestResultRequest(BaseModel):
+    name: str
+    status: TestResultStatus
+    template_id: str = ""
+    category: str = ""
+    comment: str = ""
+    io_log: str = ""
+
+
 class EndTestExecutionRequest(BaseModel):
     ci_link: Annotated[str, HttpUrl]
     c3_link: Annotated[str, HttpUrl] | None = None
@@ -106,7 +115,7 @@ class PreviousTestResult(BaseModel):
     artefact_id: int
 
 
-class TestResultDTO(BaseModel):
+class TestResultResponse(BaseModel):
     __test__ = False
 
     model_config = ConfigDict(from_attributes=True)

backend/test_observer/controllers/test_executions/post_results.py

@@ -0,0 +1,52 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import delete
+from sqlalchemy.orm import Session
+
+from test_observer.controllers.test_executions.models import TestResultRequest
+from test_observer.data_access.models import TestCase, TestExecution, TestResult
+from test_observer.data_access.repository import get_or_create
+from test_observer.data_access.setup import get_db
+
+router = APIRouter(tags=["test-results"])
+
+
+@router.post("/{id}/test-results")
+def post_results(
+    id: int,
+    request: list[TestResultRequest],
+    db: Session = Depends(get_db),
+):
+    test_execution = db.get(TestExecution, id)
+
+    if test_execution is None:
+        raise HTTPException(status_code=404, detail="TestExecution not found")
+
+    for result in request:
+        test_case = get_or_create(
+            db,
+            TestCase,
+            filter_kwargs={"name": result.name},
+            creation_kwargs={
+                "category": result.category,
+                "template_id": result.template_id,
+            },
+        )
+
+        db.execute(
+            delete(TestResult).where(
+                TestResult.test_execution_id == test_execution.id,
+                TestResult.test_case_id == test_case.id,
+            )
+        )
+
+        test_result = TestResult(
+            test_execution=test_execution,
+            test_case=test_case,
+            status=result.status,
+            comment=result.comment,
+            io_log=result.io_log,
+        )
+
+        db.add(test_result)
+
+    db.commit()

backend/tests/controllers/test_executions/test_post_results.py

@@ -0,0 +1,120 @@
+import pytest
+from fastapi.testclient import TestClient
+
+from test_observer.data_access.models import TestExecution, TestResult
+from tests.asserts import assert_fails_validation
+
+maximum_result = {
+    "name": "camera detect",
+    "status": "PASSED",
+    "template_id": "template",
+    "category": "camera",
+    "comment": "No comment",
+    "io_log": "test io log",
+}
+
+minimum_result = {"name": "test", "status": "FAILED"}
+
+
+def _assert_results(request: list[dict[str, str]], results: list[TestResult]) -> None:
+    assert len(request) == len(results)
+    for submitted, expected in zip(request, results, strict=True):
+        assert expected.test_case.name == submitted["name"]
+        assert expected.status == submitted["status"]
+        assert expected.test_case.template_id == submitted.get("template_id", "")
+        assert expected.test_case.category == submitted.get("category", "")
+        assert expected.comment == submitted.get("comment", "")
+        assert expected.io_log == submitted.get("io_log", "")
+
+
+def test_missing_test_execution(test_client: TestClient):
+    response = test_client.post("/v1/test-executions/1/test-results", json=[])
+    assert response.status_code == 404
+
+
+def test_one_full_test_result(test_client: TestClient, test_execution: TestExecution):
+    response = test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results", json=[maximum_result]
+    )
+
+    assert response.status_code == 200
+    _assert_results([maximum_result], test_execution.test_results)
+
+
+def test_one_minimum_result(test_client: TestClient, test_execution: TestExecution):
+    response = test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results", json=[minimum_result]
+    )
+
+    assert response.status_code == 200
+    _assert_results([minimum_result], test_execution.test_results)
+
+
+@pytest.mark.parametrize("field", ["name", "status"])
+def test_required_fields(
+    test_client: TestClient, test_execution: TestExecution, field: str
+):
+    result = minimum_result.copy()
+    result.pop(field)
+    response = test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results", json=[result]
+    )
+
+    assert_fails_validation(response, field, "missing")
+
+
+def test_batch_request(test_client: TestClient, test_execution: TestExecution):
+    request = [
+        {**maximum_result, "name": "test 1", "status": "PASSED"},
+        {**maximum_result, "name": "test 2", "status": "FAILED"},
+        {**maximum_result, "name": "test 3", "status": "SKIPPED"},
+    ]
+
+    response = test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results",
+        json=request,
+    )
+
+    assert response.status_code == 200
+    _assert_results(request, test_execution.test_results)
+
+
+def test_multiple_batch_requests(
+    test_client: TestClient, test_execution: TestExecution
+):
+    request1 = [
+        {**maximum_result, "name": "test 1", "status": "PASSED"},
+        {**maximum_result, "name": "test 2", "status": "FAILED"},
+        {**maximum_result, "name": "test 3", "status": "SKIPPED"},
+    ]
+
+    request2 = [
+        {**maximum_result, "name": "test 4", "status": "PASSED"},
+        {**maximum_result, "name": "test 5", "status": "FAILED"},
+        {**maximum_result, "name": "test 6", "status": "SKIPPED"},
+    ]
+
+    test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results", json=request1
+    )
+    test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results", json=request2
+    )
+
+    _assert_results([*request1, *request2], test_execution.test_results)
+
+
+def test_overwrites_result_if_matching_case_name(
+    test_client: TestClient, test_execution: TestExecution
+):
+    request1 = [{**minimum_result, "name": "same", "status": "FAILED"}]
+    request2 = [{**minimum_result, "name": "same", "status": "PASSED"}]
+
+    test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results", json=request1
+    )
+    test_client.post(
+        f"/v1/test-executions/{test_execution.id}/test-results", json=request2
+    )
+
+    _assert_results(request2, test_execution.test_results)