Get artefact endpoint

backend/.dockerignore

@@ -39,6 +39,7 @@ var/
 .installed.cfg
 *.egg
 .mypy_cache
+.pytest_cache
 
 # PyInstaller
 #  Usually these files are written by a python script from a template

backend/test_observer/controllers/artefacts/artefacts.py

@@ -1,205 +1,16 @@
-# 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:
-#        Nadzeya Hutsko <[email protected]>
-#        Omar Selo <[email protected]>
-
-
-import logging
 from fastapi import APIRouter, Depends
-from fastapi.responses import JSONResponse
 from sqlalchemy.orm import Session
-
-from test_observer.data_access.repository import (
-    get_stage_by_name,
-    get_artefacts_by_family_name,
-)
-from test_observer.data_access.models import Artefact
-from test_observer.data_access.models_enums import FamilyName
+from test_observer.data_access.models import ArtefactBuild
 from test_observer.data_access.setup import get_db
-from test_observer.external_apis.snapcraft import (
-    get_channel_map_from_snapcraft,
-)
-from test_observer.external_apis.archive import ArchiveManager
-
-router = APIRouter()
-
-logger = logging.getLogger("test-observer-backend")
-
-CHANNEL_PROMOTION_MAP = {
-    # channel -> next-channel
-    "edge": "beta",
-    "beta": "candidate",
-    "candidate": "stable",
-    "stable": "stable",
-}
-
-REPOSITORY_PROMOTION_MAP = {
-    # repository -> next-repository
-    "proposed": "updates",
-    "updates": "updates",
-}
-
 
-@router.put("/promote")
-def promote_artefacts(db: Session = Depends(get_db)):
-    """
-    Promote all the artefacts in all the families if it has been updated on the
-    external source
-    """
-    try:
-        processed_artefacts = promoter_controller(db)
-        logger.info("INFO: Processed artefacts %s", processed_artefacts)
-        if False in processed_artefacts.values():
-            return JSONResponse(
-                status_code=500,
-                content={
-                    "detail": (
-                        "Got some errors while processing the next artefacts: "
-                        ", ".join(
-                            [k for k, v in processed_artefacts.items() if v is False]
-                        )
-                    )
-                },
-            )
-        return JSONResponse(
-            status_code=200,
-            content={"detail": "All the artefacts have been processed successfully"},
-        )
-    except Exception as exc:
-        return JSONResponse(status_code=500, content={"detail": str(exc)})
+from .models import ArtefactBuildDTO
 
 
-def promoter_controller(session: Session) -> dict:
-    """
-    Orchestrate the snap promoter job
-
-    :session: DB connection session
-    :return: dict with the processed cards and the status of execution
-    """
-    family_mapping = {
-        FamilyName.SNAP: run_snap_promoter,
-        FamilyName.DEB: run_deb_promoter,
-    }
-    for family_name, promoter_function in family_mapping.items():
-        artefacts = get_artefacts_by_family_name(session, family_name)
-        processed_artefacts = {}
-        for artefact in artefacts:
-            try:
-                processed_artefacts[
-                    f"{family_name} - {artefact.name} - {artefact.version}"
-                ] = True
-                promoter_function(session, artefact)
-            except Exception as exc:
-                processed_artefacts[
-                    f"{family_name} - {artefact.name} - {artefact.version}"
-                ] = False
-                logger.warning("WARNING: %s", str(exc), exc_info=True)
-    return processed_artefacts
-
-
-def run_snap_promoter(session: Session, artefact: Artefact) -> None:
-    """
-    Check snap artefacts state and move/archive them if necessary
-
-    :session: DB connection session
-    :artefact_build: an ArtefactBuild object
-    """
-    for build in artefact.builds:
-        arch = build.architecture
-        channel_map = get_channel_map_from_snapcraft(
-            arch=arch,
-            snapstore=artefact.source["store"],
-            snap_name=artefact.name,
-        )
-        track = artefact.source.get("track", "latest")
-
-        for channel_info in channel_map:
-            if not (
-                channel_info.channel.track == track
-                and channel_info.channel.architecture == arch
-            ):
-                continue
-
-            risk = channel_info.channel.risk
-            try:
-                version = channel_info.version
-                revision = channel_info.revision
-            except KeyError as exc:
-                logger.warning(
-                    "No key '%s' is found. Continue processing...",
-                    str(exc),
-                )
-                continue
-
-            next_risk = CHANNEL_PROMOTION_MAP[artefact.stage.name]
-            if (
-                risk == next_risk != artefact.stage.name.lower()
-                and version == artefact.version
-                and revision == build.revision
-            ):
-                logger.info("Move artefact '%s' to the '%s' stage", artefact, next_risk)
-                stage = get_stage_by_name(
-                    session, stage_name=next_risk, family=artefact.stage.family
-                )
-                if stage:
-                    artefact.stage = stage
-                    session.commit()
-                    # The artefact was promoted, so we're done
-                    return
-
+router = APIRouter()
 
-def run_deb_promoter(session: Session, artefact: Artefact) -> None:
-    """
-    Check deb artefacts state and move/archive them if necessary
 
-    :session: DB connection session
-    :artefact: an Artefact object
-    """
-    for build in artefact.builds:
-        arch = build.architecture
-        for repo in REPOSITORY_PROMOTION_MAP:
-            with ArchiveManager(
-                arch=arch,
-                series=artefact.source["series"],
-                pocket=repo,
-                apt_repo=artefact.source["repo"],
-            ) as archivemanager:
-                deb_version = archivemanager.get_deb_version(artefact.name)
-                if deb_version is None:
-                    logger.error(
-                        "Cannot find deb_version with deb %s in package data",
-                        artefact.name,
-                    )
-                    continue
-            next_repo = REPOSITORY_PROMOTION_MAP.get(artefact.stage.name)
-            logger.debug(
-                "Artefact version: %s, deb version: %s", artefact.version, deb_version
-            )
-            if (
-                repo == next_repo != artefact.stage.name
-                and deb_version == artefact.version
-            ):
-                logger.info("Move artefact '%s' to the '%s' stage", artefact, next_repo)
-                stage = get_stage_by_name(
-                    session, stage_name=next_repo, family=artefact.stage.family
-                )
-                if stage:
-                    artefact.stage = stage
-                    session.commit()
-                    # The artefact was promoted, so we're done
-                    return
+@router.get("/{artefact_id}/builds", response_model=list[ArtefactBuildDTO])
+def get_artefact_builds(artefact_id: int, db: Session = Depends(get_db)):
+    return (
+        db.query(ArtefactBuild).filter(ArtefactBuild.artefact_id == artefact_id).all()
+    )

backend/test_observer/controllers/artefacts/models.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel
+
+
+class EnvironmentDTO(BaseModel):
+    id: int
+    name: str
+    architecture: str
+
+    class Config:
+        orm_mode = True
+
+
+class TestExecutionDTO(BaseModel):
+    id: int
+    jenkins_link: str | None
+    c3_link: str | None
+    environment: EnvironmentDTO
+
+    class Config:
+        orm_mode = True
+
+
+class ArtefactBuildDTO(BaseModel):
+    id: int
+    revision: int | None
+    test_executions: list[TestExecutionDTO]
+
+    class Config:
+        orm_mode = True