refactor: rename automatic_linkage module and its internal functions

Author: adekoder

README.md

@@ -17,4 +17,4 @@ By discussing issues with security team members and other maintainers, they can
 
 ## Contributing
 
-Please see the [**Contributing Guide**](CONTRIBUTING.md) for more information on how to get started. 
+Please see the [**Contributing Guide**](CONTRIBUTING.md) for more information on how to get started.

src/shared/listeners/__init__.py

@@ -1,5 +1,5 @@
 import shared.listeners.nix_channels  # noqa
 import shared.listeners.nix_evaluation  # noqa
-import shared.listeners.automatic_linkage  # noqa
+import shared.listeners.cve_derivation_matcher  # noqa
 import shared.listeners.cache_suggestions  # noqa
 import shared.listeners.notify_users  # noqa

src/shared/listeners/automatic_linkage.py …ared/listeners/cve_derivation_matcher.pysrc/shared/listeners/automatic_linkage.py renamed to src/shared/listeners/cve_derivation_matcher.py src/shared/listeners/automatic_linkage.py renamed to src/shared/listeners/cve_derivation_matcher.py

@@ -16,7 +16,7 @@
 logger = logging.getLogger(__name__)
 
 
-def produce_linkage_candidates(
+def find_linkage_candidates(
     container: Container,
 ) -> dict[NixDerivation, ProvenanceFlags]:
     latest_complete_channels = (
@@ -93,22 +93,20 @@ def produce_linkage_candidates(
     return candidates
 
 
-def build_new_links(container: Container) -> bool:
+def create_derivation_proposal(container: Container) -> bool:
     if container.cve.triaged:
         logger.info(
-            "Container received for '%s', but already triaged, skipping linkage.",
-            container.cve,
+            f"Container received for {container.cve}, but already triaged, skipping linkage.",
         )
         return False
 
     if CVEDerivationClusterProposal.objects.filter(cve=container.cve).exists():
-        logger.info("Suggestion already exists for '%s', skipping", container.cve)
+        logger.info(f"Suggestion already exists for {container.cve}, skipping")
         return False
 
     if container.tags.filter(value="exclusively-hosted-service").exists():
         logger.info(
-            "Container for '%s' is exclusively-hosted-service, rejecting without match.",
-            container.cve,
+            f"Container for {container.cve} is exclusively-hosted-service, rejecting without match.",
         )
         CVEDerivationClusterProposal.objects.create(
             cve=container.cve,
@@ -117,16 +115,14 @@ def build_new_links(container: Container) -> bool:
         )
         return True
 
-    drvs = produce_linkage_candidates(container)
+    drvs = find_linkage_candidates(container)
     if not drvs:
-        logger.info("No derivations matching '%s', ignoring", container.cve)
+        logger.info(f"No derivations matching {container.cve}, ignoring")
         return False
 
     if len(drvs) > settings.MAX_MATCHES:
         logger.warning(
-            "More than '%d' derivations matching '%s', ignoring",
-            settings.MAX_MATCHES,
-            container.cve,
+            f"More than {settings.MAX_MATCHES} derivations matching {container.cve}, ignoring",
         )
         return False
 
@@ -144,14 +140,12 @@ def build_new_links(container: Container) -> bool:
 
     if drvs_throughs:
         logger.info(
-            "Matching suggestion for '%s': %d derivations found.",
-            container.cve,
-            len(drvs_throughs),
+            f"Matching suggestion for {container.cve}: {len(drvs_throughs)} derivations found.",
         )
 
     return True
 
 
 @pgpubsub.post_insert_listener(ContainerChannel)
-def build_new_links_following_new_containers(old: Container, new: Container) -> None:
-    build_new_links(new)
+def match_derivations_on_container_insert(old: Container, new: Container) -> None:
+    create_derivation_proposal(new)

src/shared/management/commands/propose_cve_links.py

@@ -7,7 +7,7 @@
 from django.core.management.base import BaseCommand, CommandError
 
 from shared import models
-from shared.listeners.automatic_linkage import build_new_links
+from shared.listeners.cve_derivation_matcher import create_derivation_proposal
 
 logger = logging.getLogger(__name__)
 
@@ -40,15 +40,17 @@ def handle(self, *args: Any, **kwargs: Any) -> None:
         except ValueError:
             raise CommandError(f"Not a valid delta format: {_delta}")
 
-        logger.info("Proposing new CVE links starting '%s'", since_date.isoformat())
+        logger.info(f"Proposing new CVE links starting {since_date.isoformat()}")
         success = Counter()
         # Collect all containers since that delta range.
         containers = models.Container.objects.filter(date_public__gte=since_date)
 
         for container in containers.iterator():
-            success[container.cve.cve_id] += 1 if build_new_links(container) else 0
+            success[container.cve.cve_id] += (
+                1 if create_derivation_proposal(container) else 0
+            )
             print(".", end="", flush=True)
 
         for cve_id, successes in success.items():
             if successes == 0:
-                logger.warning("No derivation found for '%s', linkage failure.", cve_id)
+                logger.warning(f"No derivation found for {cve_id}, linkage failure.")

src/shared/tests/test_linkage.py

@@ -3,8 +3,8 @@
 
 import pytest
 
-from shared.listeners.automatic_linkage import build_new_links
 from shared.listeners.cache_suggestions import cache_new_suggestions
+from shared.listeners.cve_derivation_matcher import create_derivation_proposal
 from shared.models.cve import Container, Tag
 from shared.models.linkage import (
     CVEDerivationClusterProposal,
@@ -67,7 +67,7 @@ def test_link_only_latest_eval(
         )
 
     container = make_container(package_name="foo", affected_version="<3.2")
-    match = build_new_links(container)
+    match = create_derivation_proposal(container)
     assert match
     suggestion = CVEDerivationClusterProposal.objects.first()
     assert suggestion
@@ -113,7 +113,7 @@ def test_link_product_or_package_name(
     container = make_container(package_name=package_name, product=product)
     drv = make_drv(pname=drv_pname)
 
-    match = build_new_links(container)
+    match = create_derivation_proposal(container)
 
     if expected_flags:
         assert match
@@ -133,7 +133,7 @@ def test_exclusively_hosted_service_creates_rejected_proposal(
     tag, _ = Tag.objects.get_or_create(value="exclusively-hosted-service")
     container.tags.add(tag)
 
-    result = build_new_links(container)
+    result = create_derivation_proposal(container)
 
     assert result is True
     proposal = CVEDerivationClusterProposal.objects.get(cve=container.cve)