test: package from newest evaluation shows up in suggestion

Author: fricklerhandwerk

src/shared/tests/fixtures.py

@@ -1,3 +1,6 @@
+import secrets
+from collections.abc import Callable
+
 import pytest
 from allauth.socialaccount.models import SocialAccount
 from django.contrib.auth.models import AbstractBaseUser
@@ -19,6 +22,7 @@
     ProvenanceFlags,
 )
 from shared.models.nix_evaluation import (
+    MAJOR_CHANNELS,
     NixChannel,
     NixDerivation,
     NixDerivationMeta,
@@ -49,51 +53,75 @@ def cve(db: None) -> Container:
 
 
 @pytest.fixture
-def evaluation(db: None) -> NixEvaluation:
-    channel = NixChannel.objects.create(
-        staging_branch="release-24.05",
-        channel_branch="release-24.05",
+def channel(db: None) -> NixChannel:
+    # FIXME(@fricklerhandwerk): This will fall apart when we obtain the channel structure dynamically [ref:channel-structure]
+    release = MAJOR_CHANNELS[1]
+    return NixChannel.objects.create(
+        staging_branch=f"nixos-{release}",
+        channel_branch=f"nixos-{release}",
         head_sha1_commit="deadbeef",
         state=NixChannel.ChannelState.STABLE,
-        release_version="24.05",
+        release_version=release,
         repository="https://github.com/NixOS/nixpkgs",
     )
 
-    evaluation = NixEvaluation.objects.create(
-        channel=channel,
-        commit_sha1="deadbeef",
-        state="completed",
-    )
-    return evaluation
+
+@pytest.fixture
+def make_evaluation(channel: NixChannel) -> Callable[[], NixEvaluation]:
+    def wrapped() -> NixEvaluation:
+        return NixEvaluation.objects.create(
+            channel=channel,
+            commit_sha1=secrets.token_hex(16),
+            state="completed",
+        )
+
+    return wrapped
+
+
+@pytest.fixture
+def evaluation(make_evaluation: Callable[[], NixEvaluation]) -> NixEvaluation:
+    return make_evaluation()
 
 
 @pytest.fixture
-def drv(db: None, evaluation: NixEvaluation) -> NixDerivation:
-    maintainer = NixMaintainer.objects.create(
+def maintainer(db: None) -> NixMaintainer:
+    return NixMaintainer.objects.create(
         github_id=123, github="testuser", name="Test User", email="test@example.com"
     )
 
-    meta = NixDerivationMeta.objects.create(
-        description="First dummy derivation",
-        insecure=False,
-        available=True,
-        broken=False,
-        unfree=False,
-        unsupported=False,
-    )
 
-    meta.maintainers.add(maintainer)
+@pytest.fixture
+def make_drv(maintainer: NixMaintainer) -> Callable[[NixEvaluation], NixDerivation]:
+    def wrapped(evaluation: NixEvaluation) -> NixDerivation:
+        meta = NixDerivationMeta.objects.create(
+            description="Dummy derivation",
+            insecure=False,
+            available=True,
+            broken=False,
+            unfree=False,
+            unsupported=False,
+        )
+        meta.maintainers.add(maintainer)
+
+        return NixDerivation.objects.create(
+            attribute="foo",
+            derivation_path="/nix/store/<hash>-foo.drv",
+            name="foo-1.0",
+            metadata=meta,
+            system="x86_64-linux",
+            parent_evaluation=evaluation,
+        )
+
+    return wrapped
 
-    drv = NixDerivation.objects.create(
-        attribute="foo",
-        derivation_path="/nix/store/<hash>-foo.drv",
-        name="foo-1.0",
-        metadata=meta,
-        system="x86_64-linux",
-        parent_evaluation=evaluation,
-    )
 
-    return drv
+@pytest.fixture
+def drv(
+    db: None,
+    make_drv: Callable[[NixEvaluation], NixDerivation],
+    evaluation: NixEvaluation,
+) -> NixDerivation:
+    return make_drv(evaluation)
 
 
 @pytest.fixture

src/shared/tests/test_suggestion_caching.py

@@ -1,20 +1,75 @@
+from collections.abc import Callable
+from datetime import timedelta
+
 from shared.listeners.cache_suggestions import cache_new_suggestions
 from shared.models.cached import CachedSuggestions
 from shared.models.cve import Container
-from shared.models.linkage import CVEDerivationClusterProposal
-from shared.models.nix_evaluation import NixDerivation
+from shared.models.linkage import (
+    CVEDerivationClusterProposal,
+    DerivationClusterProposalLink,
+    ProvenanceFlags,
+)
+from shared.models.nix_evaluation import (
+    NixDerivation,
+    NixEvaluation,
+)
 
 
-# XXX(@fricklerhandwerk): This test isn't doing a lot except run the code once and assert that the result is non-empty.
 # The Pydantic class for the cached value gives us some assurance about the shape of the data, but ultimately we probabyly want property tests here.
-def test_caching(
+def test_caching_newest_package(
     cve: Container,
-    drv: NixDerivation,
+    make_evaluation: Callable[[], NixEvaluation],
+    make_drv: Callable[[NixEvaluation], NixDerivation],
     suggestion: CVEDerivationClusterProposal,
 ) -> None:
+    """
+    Check that when aggregating derivations from a suggestion, only the newest one for a given package name (== attribute path) is used.
+    """
+
+    # Order of creation matters for triggering the bug.
+    # This is brittle because it assumes things about the database, but it seems that derivations are scanned in insertion order of their evaluations.
+    eval_new = make_evaluation()
+    eval_old = make_evaluation()
+    # Overwrite the timestamp without triggering the `save()` hook that would write the current time again
+    target_time = eval_old.created_at - timedelta(hours=1)
+    NixEvaluation.objects.filter(pk=eval_old.pk).update(
+        created_at=target_time,
+        updated_at=target_time,
+    )
+    drv2 = make_drv(eval_new)
+    drv1 = make_drv(eval_old)
+
+    # Use different versions. We'll assert over those since that's what's primarily visible to users
+    drv1.name = f"{drv1.attribute}-1.0"
+    drv1.save()
+    drv2.name = f"{drv1.attribute}-2.0"
+    drv2.save()
+
+    suggestion = CVEDerivationClusterProposal.objects.create(
+        status="pending",
+        cve=cve.cve,
+    )
+
+    DerivationClusterProposalLink.objects.create(
+        proposal=suggestion,
+        derivation=drv1,
+        provenance_flags=ProvenanceFlags.PACKAGE_NAME_MATCH,
+    )
+
+    DerivationClusterProposalLink.objects.create(
+        proposal=suggestion,
+        derivation=drv2,
+        provenance_flags=ProvenanceFlags.PACKAGE_NAME_MATCH,
+    )
+
     cached = CachedSuggestions.objects.filter(proposal=suggestion)
     assert not cached.exists()
     cache_new_suggestions(suggestion)
     value = cached.get()
     assert cve.cve.cve_id == value.payload["cve_id"]
-    assert drv.attribute in value.payload["packages"]
+
+    channel, package = next(
+        iter(value.payload["packages"][drv1.attribute]["channels"].items())
+    )
+
+    assert package["major_version"] == "2.0"