Automatic approval of test executions (#152)

* Refactor tests in test_test_executions.py to use DataGenerator

* Small bug fix

* Refactor test_executions controller into multiple files

* Implement logic to automatically approve test executions

* Remove unused import

* Remove unused id in end test execution request

* Fix corner case

* Simplify get previous artefact logic

* Tag test executions endpoint to improve docs

* Include more data in test results report

* Add copyright notices

backend/scripts/seed_data.py

@@ -172,11 +172,9 @@
 
 END_TEST_EXECUTION_REQUESTS = [
     EndTestExecutionRequest(
-        id=1,
         ci_link="http://example1",
         test_results=[
             C3TestResult(
-                id=1,
                 name="docker/compose-and-basic_armhf",
                 status=C3TestResultStatus.PASS,
                 category="Docker containers",
@@ -218,15 +216,13 @@
                 ),
             ),
             C3TestResult(
-                id=2,
                 name="after-suspend-audio/alsa-loopback-automated",
                 status=C3TestResultStatus.SKIP,
                 category="Audio tests",
                 comment="""job cannot be started: resource expression "manifest.has_audio_loopback_connector == 'True'" evaluates to false, required dependency 'com.canonical.certification::audio/alsa-loopback-automated' has failed, required dependency 'com.canonical.certification::audio/detect-capture-devices' has failed, required dependency 'com.canonical.certification::audio/detect-playback-devices' has failed, required dependency 'com.canonical.certification::suspend/suspend_advanced_auto' has failed""",
                 io_log="",
             ),
             C3TestResult(
-                id=3,
                 name="bluetooth4/beacon_eddystone_url_hci0",
                 status=C3TestResultStatus.FAIL,
                 category="Bluetooth tests",
@@ -266,19 +262,16 @@
         ],
     ),
     EndTestExecutionRequest(
-        id=2,
         ci_link="http://example2",
         test_results=[
             C3TestResult(
-                id=4,
                 name="test7",
                 status=C3TestResultStatus.PASS,
                 category="",
                 comment="",
                 io_log="",
             ),
             C3TestResult(
-                id=5,
                 name="test8",
                 status=C3TestResultStatus.SKIP,
                 category="",

backend/test_observer/controllers/reports/reports.py

@@ -23,12 +23,15 @@
 
 TESTRESULTS_REPORT_COLUMNS: list[InstrumentedAttribute] = [
     Family.name,
+    Artefact.id,
     Artefact.name,
     Artefact.version,
     Artefact.status,
     Artefact.track,
     Artefact.series,
     Artefact.repo,
+    Artefact.created_at,
+    TestExecution.id,
     TestExecution.status,
     TestExecution.review_decision,
     TestExecution.review_comment,
@@ -37,6 +40,7 @@
     TestCase.name,
     TestCase.category,
     TestResult.status,
+    TestResult.created_at,
 ]
 
 

backend/test_observer/controllers/router.py

@@ -23,11 +23,11 @@
 
 from test_observer.data_access.setup import get_db
 
+from . import test_executions
 from .application import version
 from .artefacts import artefacts
 from .promoter import promoter
 from .reports import reports
-from .test_executions import test_executions
 
 router = APIRouter()
 router.include_router(promoter.router)

backend/test_observer/controllers/test_executions/__init__.py

@@ -0,0 +1,25 @@
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+from fastapi import APIRouter
+
+from . import end_test, get_test_results, patch, start_test
+
+router = APIRouter(tags=["test-executions"])
+router.include_router(start_test.router)
+router.include_router(get_test_results.router)
+router.include_router(end_test.router)
+router.include_router(patch.router)

backend/test_observer/controllers/test_executions/end_test.py

@@ -0,0 +1,191 @@
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.orm import Session, joinedload
+
+from test_observer.data_access.models import (
+    Artefact,
+    ArtefactBuild,
+    TestCase,
+    TestExecution,
+    TestResult,
+)
+from test_observer.data_access.models_enums import (
+    TestExecutionReviewDecision,
+    TestExecutionStatus,
+    TestResultStatus,
+)
+from test_observer.data_access.repository import get_or_create
+from test_observer.data_access.setup import get_db
+
+from .logic import delete_previous_results
+from .models import C3TestResult, C3TestResultStatus, EndTestExecutionRequest
+
+router = APIRouter()
+
+
+@router.put("/end-test")
+def end_test_execution(request: EndTestExecutionRequest, db: Session = Depends(get_db)):
+    test_execution = _find_related_test_execution(request, db)
+
+    if test_execution is None:
+        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)
+
+    has_failures = test_execution.has_failures
+
+    test_execution.status = (
+        TestExecutionStatus.FAILED if has_failures else TestExecutionStatus.PASSED
+    )
+
+    prev_test_execution = _get_previous_test_execution(db, test_execution)
+
+    if (
+        prev_test_execution
+        and not has_failures
+        and prev_test_execution.is_approved
+        and _ran_all_previously_run_cases(prev_test_execution, test_execution)
+    ):
+        test_execution.review_decision = [
+            TestExecutionReviewDecision.APPROVED_ALL_TESTS_PASS
+        ]
+
+    if request.c3_link is not None:
+        test_execution.c3_link = request.c3_link
+
+    db.commit()
+
+
+def _ran_all_previously_run_cases(
+    prev_test_execution: TestExecution, test_execution: TestExecution
+) -> bool:
+    prev_test_cases = {
+        tr.test_case.name
+        for tr in prev_test_execution.test_results
+        if tr.status != TestResultStatus.SKIPPED
+    }
+
+    test_cases = {
+        tr.test_case.name
+        for tr in test_execution.test_results
+        if tr.status != TestResultStatus.SKIPPED
+    }
+
+    return prev_test_cases.issubset(test_cases)
+
+
+def _get_previous_test_execution(
+    db: Session, test_execution: TestExecution
+) -> TestExecution | None:
+    artefact = test_execution.artefact_build.artefact
+    prev_artefact = _get_previous_artefact(db, artefact)
+
+    if prev_artefact is None:
+        return None
+
+    query = (
+        select(TestExecution)
+        .join(ArtefactBuild)
+        .where(
+            ArtefactBuild.artefact_id == prev_artefact.id,
+            TestExecution.environment_id == test_execution.environment_id,
+        )
+        .order_by(ArtefactBuild.revision.desc())
+        .limit(1)
+        .options(
+            joinedload(TestExecution.test_results).joinedload(TestResult.test_case)
+        )
+    )
+
+    return db.execute(query).unique().scalar_one_or_none()
+
+
+def _get_previous_artefact(db: Session, artefact: Artefact) -> Artefact | None:
+    query = (
+        select(Artefact)
+        .where(
+            Artefact.id < artefact.id,
+            Artefact.name == artefact.name,
+            Artefact.track == artefact.track,
+            Artefact.series == artefact.series,
+            Artefact.repo == artefact.repo,
+        )
+        .order_by(Artefact.id.desc())
+        .limit(1)
+    )
+
+    return db.execute(query).scalar_one_or_none()
+
+
+def _find_related_test_execution(
+    request: EndTestExecutionRequest, db: Session
+) -> TestExecution | None:
+    return (
+        db.execute(
+            select(TestExecution)
+            .where(TestExecution.ci_link == request.ci_link)
+            .options(
+                joinedload(TestExecution.artefact_build).joinedload(
+                    ArtefactBuild.artefact
+                )
+            )
+            .options(
+                joinedload(TestExecution.test_results).joinedload(TestResult.test_case)
+            )
+        )
+        .unique()
+        .scalar_one_or_none()
+    )
+
+
+def _store_test_results(
+    db: Session,
+    c3_test_results: list[C3TestResult],
+    test_execution: TestExecution,
+) -> None:
+    for r in c3_test_results:
+        test_case = get_or_create(
+            db,
+            TestCase,
+            filter_kwargs={"name": r.name},
+            creation_kwargs={"category": r.category},
+        )
+
+        test_result = TestResult(
+            test_case=test_case,
+            test_execution=test_execution,
+            status=_parse_c3_test_result_status(r.status),
+            comment=r.comment,
+            io_log=r.io_log,
+        )
+
+        db.add(test_result)
+
+    db.commit()
+
+
+def _parse_c3_test_result_status(status: C3TestResultStatus) -> TestResultStatus:
+    match status:
+        case C3TestResultStatus.PASS:
+            return TestResultStatus.PASSED
+        case C3TestResultStatus.FAIL:
+            return TestResultStatus.FAILED
+        case C3TestResultStatus.SKIP:
+            return TestResultStatus.SKIPPED

backend/test_observer/controllers/test_executions/get_test_results.py

@@ -0,0 +1,64 @@
+# Copyright 2023 Canonical Ltd.
+# All rights reserved.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# Written by:
+#        Omar Selo <[email protected]>
+#        Nadzeya Hutsko <[email protected]>
+
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session, joinedload
+
+from test_observer.controllers.test_executions.helpers import (
+    parse_previous_test_results,
+)
+from test_observer.data_access.models import (
+    TestExecution,
+    TestResult,
+)
+from test_observer.data_access.setup import get_db
+
+from .logic import get_previous_test_results
+from .models import TestResultDTO
+
+router = APIRouter()
+
+
+@router.get("/{id}/test-results", response_model=list[TestResultDTO])
+def get_test_results(id: int, db: Session = Depends(get_db)):
+    test_execution = db.get(
+        TestExecution,
+        id,
+        options=[
+            joinedload(TestExecution.test_results).joinedload(TestResult.test_case),
+        ],
+    )
+
+    if test_execution is None:
+        raise HTTPException(status_code=404, detail="TestExecution not found")
+
+    previous_test_results = get_previous_test_results(db, test_execution)
+    parsed_previous_test_results = parse_previous_test_results(previous_test_results)
+
+    test_results: list[TestResultDTO] = []
+    for test_result in test_execution.test_results:
+        parsed_test_result = TestResultDTO.model_validate(test_result)
+        parsed_test_result.previous_results = parsed_previous_test_results.get(
+            test_result.test_case_id, []
+        )
+        test_results.append(parsed_test_result)
+
+    return test_results

backend/test_observer/controllers/test_executions/helpers.py

@@ -1,7 +1,25 @@
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
 from itertools import groupby
-from .models import PreviousTestResult
+
 from test_observer.data_access.models import TestResult
 
+from .models import PreviousTestResult
+
 
 def parse_previous_test_results(
     previous_test_results: list[TestResult],