Fix snap promotion

backend/test_observer/data_access/models_enums.py

@@ -36,6 +36,22 @@ class StageName(str, Enum):
     candidate = "candidate"
     stable = "stable"
 
+    def _compare(self, other: str) -> int:
+        stages = list(StageName.__members__.values())
+        return stages.index(self) - stages.index(StageName(other))
+
+    def __lt__(self, other: str) -> bool:
+        return self._compare(other) < 0
+
+    def __le__(self, other: str) -> bool:
+        return self._compare(other) <= 0
+
+    def __gt__(self, other: str) -> bool:
+        return self._compare(other) > 1
+
+    def __ge__(self, other: str) -> bool:
+        return self._compare(other) >= 0
+
 
 class TestExecutionStatus(str, Enum):
     __test__ = False

backend/test_observer/external_apis/snapcraft.py

@@ -2,23 +2,21 @@
 
 import requests
 
-from .snapcraft_models import SnapInfo, rename_keys
+from .snapcraft_models import ChannelMap, SnapInfo, rename_keys
 
 logger = logging.getLogger("test-observer-backend")
 
 
-def get_channel_map_from_snapcraft(arch: str, snapstore: str, snap_name: str):
+def get_channel_map_from_snapcraft(snapstore: str, snap_name: str) -> list[ChannelMap]:
     """
     Get channel_map from snapcraft.io
 
-    :arch: architecture
     :snapstore: Snapstore name
     :snap_name: snap name
     :return: channgel map as python dict (JSON format)
     """
     headers = {
         "Snap-Device-Series": "16",
-        "Snap-Device-Architecture": arch,
         "Snap-Device-Store": snapstore,
     }
     req = requests.get(

backend/test_observer/external_apis/snapcraft_models.py

@@ -19,12 +19,18 @@
 """Mappings for json objects from snapcraft"""
 
 
+from typing import Literal
+
 from pydantic import BaseModel
 
+from test_observer.data_access.models_enums import StageName
+
 
 class Channel(BaseModel):
     architecture: str
-    risk: str
+    risk: Literal[StageName.edge] | Literal[StageName.beta] | Literal[
+        StageName.candidate
+    ] | Literal[StageName.stable]
     track: str
 
 

backend/test_observer/promotion/promoter.py

@@ -23,8 +23,7 @@
 
 from sqlalchemy.orm import Session
 
-from test_observer.data_access import queries
-from test_observer.data_access.models import Artefact, ArtefactBuild
+from test_observer.data_access.models import Artefact
 from test_observer.data_access.models_enums import FamilyName, StageName
 from test_observer.data_access.repository import get_artefacts_by_family
 from test_observer.external_apis.archive import ArchiveManager
@@ -77,76 +76,50 @@ def promoter_controller(session: Session) -> tuple[dict, dict]:
     :return: tuple of dicts, the first the processed cards and the status of execution
     the second only for the processed cards with the corresponding error message
     """
-    family_mapping = {
-        FamilyName.snap: run_snap_promoter,
-        FamilyName.deb: run_deb_promoter,
-    }
     processed_artefacts_status = {}
     processed_artefacts_error_messages = {}
-    for family, promoter_function in family_mapping.items():
+
+    for family in (FamilyName.snap, FamilyName.deb):
         artefacts = get_artefacts_by_family(session, family)
         for artefact in artefacts:
             artefact_key = f"{family} - {artefact.name} - {artefact.version}"
             try:
                 processed_artefacts_status[artefact_key] = True
-                promoter_function(session, artefact)
+                if family == FamilyName.snap:
+                    SnapPromoter(session, artefact).execute()
+                elif family == FamilyName.deb:
+                    run_deb_promoter(session, artefact)
             except Exception as exc:
                 processed_artefacts_status[artefact_key] = False
                 processed_artefacts_error_messages[artefact_key] = str(exc)
                 logger.warning("WARNING: %s", str(exc), exc_info=True)
+
     return processed_artefacts_status, processed_artefacts_error_messages
 
 
-def run_snap_promoter(session: Session, artefact: Artefact) -> None:
-    """
-    Check snap artefacts state and move/archive them if necessary
+class SnapPromoter:
+    def __init__(self, db_session: Session, snap: Artefact):
+        assert snap.family == FamilyName.snap
+        assert snap.store, f"Store is not set for the snap artefact {snap.id}"
+        self._snap = snap
+        self._db_session = db_session
 
-    :session: DB connection session
-    :artefact_build: an ArtefactBuild object
-    """
-    store = artefact.store
-    assert store is not None, f"Store is not set for the artefact {artefact.id}"
-
-    latest_builds = session.scalars(
-        queries.latest_artefact_builds.where(ArtefactBuild.artefact_id == artefact.id)
-    )
-
-    for build in latest_builds:
-        arch = build.architecture
-        channel_map = get_channel_map_from_snapcraft(
-            arch=arch,
-            snapstore=store,
-            snap_name=artefact.name,
+    def execute(self):
+        all_channel_maps = get_channel_map_from_snapcraft(
+            snapstore=self._snap.store,
+            snap_name=self._snap.name,
         )
-        track = artefact.track
-
-        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
-
-            if (
-                risk != artefact.stage
-                and version == artefact.version
-                and revision == build.revision
-            ):
-                logger.info("Move artefact '%s' to the '%s' stage", artefact, risk)
-
-                artefact.stage = StageName(risk)
-                session.commit()
+        self._snap.stage = max(
+            (
+                cm.channel.risk
+                for cm in all_channel_maps
+                if cm.channel.track == self._snap.track
+                and cm.channel.architecture in self._snap.architectures
+                and cm.version == self._snap.version
+            ),
+            default=self._snap.stage,
+        )
+        self._db_session.commit()
 
 
 def run_deb_promoter(session: Session, artefact: Artefact) -> None:

backend/tests/promotion/test_promoter.py

@@ -189,3 +189,44 @@ def test_promote_snap_from_beta_to_stable(
     promote_artefacts(db_session)
 
     assert artefact.stage == StageName.stable
+
+
+def test_snap_that_is_in_two_stages(
+    db_session: Session,
+    requests_mock: Mocker,
+    generator: DataGenerator,
+):
+    artefact = generator.gen_artefact(StageName.edge, store="ubuntu")
+    build = generator.gen_artefact_build(artefact, revision=1)
+
+    requests_mock.get(
+        f"https://api.snapcraft.io/v2/snaps/info/{artefact.name}",
+        json={
+            "channel-map": [
+                {
+                    "channel": {
+                        "architecture": build.architecture,
+                        "risk": "beta",
+                        "track": artefact.track,
+                    },
+                    "revision": build.revision,
+                    "type": "app",
+                    "version": artefact.version,
+                },
+                {
+                    "channel": {
+                        "architecture": build.architecture,
+                        "risk": "edge",
+                        "track": artefact.track,
+                    },
+                    "revision": build.revision,
+                    "type": "app",
+                    "version": artefact.version,
+                },
+            ]
+        },
+    )
+
+    promote_artefacts(db_session)
+
+    assert artefact.stage == StageName.beta