fix(watchlist): vendor and product_id are the vendor source's, as inventory.py meant

docs/DESIGN.md

@@ -47,13 +47,21 @@ missing version is an absence, and a default would be a claim:
 | column | holds | set by |
 |---|---|---|
 | `cpe` | the CPE 2.3 name of the product, as NVD writes it: `cpe:2.3:a:notepad-plus-plus:notepad\+\+` | `add --cpe` |
-| `vendor` | the CPE vendor field, `notepad-plus-plus` | derived from `cpe` |
-| `product_id` | the CPE product field, escaped as NVD escapes it, `notepad\+\+` | derived from `cpe` |
+| `vendor` | the vendor source that answers by version — `inventory.py`'s *source*: `chrome`, `edge`, `msrc`, `mozilla` | the inventory import, once confirmed (§5.6) |
+| `product_id` | the product that source needs: `chrome`, `Edge` | the inventory import, once confirmed |
 | `installed_version` | the version installed, `8.5.0` | `add --version` |
-| `channel` | the release channel a vendor source answers for (stable, ESR, LTS) | reserved for the inventory import (§5.6) |
+| `channel` | the release channel the source answers for (stable, ESR, LTS) | the inventory import, once confirmed |
 
 An existing database gets them on the next start, as `cves` got its EPSS columns.
 
+Two routes to an answer by version, kept apart: the CPE, through NVD's
+configurations, which the scan judges with today; and a vendor source, which
+`inventory.py` proposes and a person confirms. The CPE's own vendor and product
+(`notepad-plus-plus`, `notepad\+\+`) are read from `cpe` when the scan needs
+them, not stored. Until 2026-10-17 they were stored in `vendor` and `product_id`,
+by a misreading of this table that `inventory.py` contradicted; no release
+carried that.
+
 ### 5.2 Nothing is guessed
 
 The CPE is typed by a person, never derived from the name. Measured on a real

src/patchradar/db/database.py

@@ -5,7 +5,7 @@
 
 import aiosqlite
 
-from patchradar.cpe import Cpe
+from patchradar.cpe import Cpe, parse_cpe
 from patchradar.priority import as_flag, sql_order
 
 logger = logging.getLogger(__name__)
@@ -209,7 +209,13 @@ async def add_to_watchlist(name: str) -> bool:
 
 @dataclass(frozen=True)
 class WatchEntry:
-    """A watchlist row with what is known of the installed product."""
+    """A watchlist row with what is known of the installed product.
+
+    `vendor`, `product_id` and `channel` are the vendor source's — the feed
+    that answers by version and the product it needs, as `inventory.py` names
+    them (docs/DESIGN.md §5.1). The CPE's vendor and product are another thing,
+    read from `cpe`: NVD by CPE is the other route.
+    """
 
     name: str
     installed_version: str | None = None
@@ -218,12 +224,21 @@ class WatchEntry:
     product_id: str | None = None
     channel: str | None = None
 
+    @property
+    def cpe_vendor(self) -> str | None:
+        return parse_cpe(self.cpe).vendor if self.cpe else None
+
+    @property
+    def cpe_product(self) -> str | None:
+        return parse_cpe(self.cpe).product if self.cpe else None
+
     def installed(self) -> tuple[str, str, str] | None:
         """(version, vendor, product) when a scan can judge CVEs against what is
-        installed — the version to compare, and the vendor and product NVD's
-        `cpeMatch` is read by — and None when any of the three is missing."""
-        if self.installed_version and self.vendor and self.product_id:
-            return self.installed_version, self.vendor, self.product_id
+        installed — the version to compare, and the CPE's vendor and product
+        NVD's `cpeMatch` is read by — and None when any of them is missing."""
+        if self.installed_version and self.cpe:
+            cpe = parse_cpe(self.cpe)
+            return self.installed_version, cpe.vendor, cpe.product
         return None
 
     @property
@@ -244,17 +259,28 @@ async def set_installed(name: str, *, version: str | None, cpe: Cpe | None) -> b
     Only what is given changes: a new version leaves the CPE as it was. False
     when the name is not on the watchlist — this never adds one.
     """
-    fields: dict[str, str] = {}
+    fields: dict[str, str | None] = {}
     if version is not None:
         fields["installed_version"] = version
     if cpe is not None:
-        fields.update(cpe=cpe.text, vendor=cpe.vendor, product_id=cpe.product)
+        fields["cpe"] = cpe.text
+    return await _update_entry(name, fields)
+
+
+async def set_source(name: str, *, vendor: str, product_id: str, channel: str | None) -> bool:
+    """Record the vendor source that answers by version for an entry already on
+    the watchlist — `inventory.py`'s source, product and channel, once a person
+    has confirmed them. False when the name is not on the watchlist."""
+    return await _update_entry(name, {"vendor": vendor, "product_id": product_id, "channel": channel})
+
+
+async def _update_entry(name: str, fields: dict[str, str | None]) -> bool:
     async with aiosqlite.connect(DB_PATH) as db:
         async with db.execute("SELECT 1 FROM watchlist WHERE name = ?", (name.lower(),)) as cursor:
             if await cursor.fetchone() is None:
                 return False
         if fields:
-            # Column names from the literal keys above, never from input.
+            # Column names from literal keys in the callers, never from input.
             assignments = ", ".join(f"{column} = ?" for column in fields)
             await db.execute(f"UPDATE watchlist SET {assignments} WHERE name = ?",
                              (*fields.values(), name.lower()))

tests/test_api_versioned.py

@@ -46,8 +46,7 @@ async def test_the_version_and_cpe_are_recorded_over_http(client):
     assert response.status_code == 200, response.text
     assert response.json()["installed_version"] == "8.5.0"
     [entry] = await database.get_watch_entries()
-    assert (entry.installed_version, entry.vendor, entry.product_id) == ("8.5.0", "notepad-plus-plus",
-                                                                          "notepad\\+\\+")
+    assert entry.installed() == ("8.5.0", "notepad-plus-plus", "notepad\\+\\+")
 
 
 async def test_a_cpe_nvd_would_match_nothing_with_is_refused_with_its_form(client):

tests/test_vendor_source_columns.py

@@ -0,0 +1,76 @@
+"""
+PatchRadar — `vendor` and `product_id` belong to the vendor source (docs/DESIGN.md §5.1)
+
+`inventory.py` named the watchlist columns before they existed: `vendor`,
+`product_id`, `installed_version`, `channel`, `cpe`. In it, a vendor source is a
+feed that answers by version — Chrome's, Edge's, MSRC, Mozilla's — and
+`product_id` is the product that feed needs ("chrome", "Edge"); the CPE is the
+separate route, NVD by CPE. #58 filled `vendor` and `product_id` with the CPE's
+fields instead, which would have had `import` write a vendor source over the
+CPE's vendor, or the other way round.
+
+The CPE's vendor and product are now read from `cpe` itself, and the two
+columns are left to the vendor source, written by `set_source`.
+"""
+from __future__ import annotations
+
+import pytest
+
+from patchradar.cpe import parse_cpe
+from patchradar.db import database
+
+NOTEPADPP = "cpe:2.3:a:notepad-plus-plus:notepad\\+\\+"
+
+
+@pytest.mark.asyncio
+async def test_a_cpe_does_not_fill_the_vendor_source_columns():
+    await database.add_to_watchlist("notepad")
+    await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP))
+
+    [entry] = await database.get_watch_entries()
+
+    assert (entry.vendor, entry.product_id) == (None, None)
+    assert (entry.cpe_vendor, entry.cpe_product) == ("notepad-plus-plus", "notepad\\+\\+")
+    assert entry.installed() == ("8.5.0", "notepad-plus-plus", "notepad\\+\\+")
+
+
+@pytest.mark.asyncio
+async def test_the_vendor_source_is_stored_beside_the_cpe():
+    await database.add_to_watchlist("google chrome")
+    await database.set_installed("google chrome", version="141.0.7390.65",
+                                 cpe=parse_cpe("cpe:2.3:a:google:chrome"))
+
+    assert await database.set_source("google chrome", vendor="chrome", product_id="chrome", channel="stable")
+
+    [entry] = await database.get_watch_entries()
+    assert (entry.vendor, entry.product_id, entry.channel) == ("chrome", "chrome", "stable")
+    assert entry.cpe == "cpe:2.3:a:google:chrome"
+    assert entry.installed() == ("141.0.7390.65", "google", "chrome")
+
+
+@pytest.mark.asyncio
+async def test_a_vendor_source_alone_does_not_make_an_entry_judgeable():
+    """The scan judges with NVD's configurations, which need the CPE."""
+    await database.add_to_watchlist("microsoft edge")
+    await database.set_installed("microsoft edge", version="141.0.3537.71", cpe=None)
+    await database.set_source("microsoft edge", vendor="edge", product_id="Edge", channel=None)
+
+    [entry] = await database.get_watch_entries()
+    assert (entry.vendor, entry.product_id) == ("edge", "Edge")
+    assert not entry.judgeable
+
+
+@pytest.mark.asyncio
+async def test_a_cpe_without_a_version_is_not_judgeable_either():
+    await database.add_to_watchlist("notepad")
+    await database.set_installed("notepad", version=None, cpe=parse_cpe(NOTEPADPP))
+
+    [entry] = await database.get_watch_entries()
+    assert entry.installed() is None
+
+
+@pytest.mark.asyncio
+async def test_set_source_never_adds_a_name():
+    assert not await database.set_source("thunderbird", vendor="mozilla", product_id="thunderbird",
+                                         channel="release")
+    assert await database.get_watchlist() == []

tests/test_watchlist_versions.py

@@ -94,16 +94,16 @@ async def test_migrating_twice_is_not_an_error(old_database):
 # ─── storing what is installed ───────────────────────────────────────────────
 
 @pytest.mark.asyncio
-async def test_version_and_cpe_are_stored_with_vendor_and_product():
+async def test_version_and_cpe_are_stored_and_the_cpe_names_vendor_and_product():
     await database.add_to_watchlist("notepad")
 
     assert await database.set_installed("notepad", version="8.5.0", cpe=parse_cpe(NOTEPADPP))
 
     [entry] = await database.get_watch_entries()
     assert entry.installed_version == "8.5.0"
     assert entry.cpe == NOTEPADPP
-    assert entry.vendor == "notepad-plus-plus"
-    assert entry.product_id == "notepad\\+\\+"
+    assert entry.cpe_vendor == "notepad-plus-plus"
+    assert entry.cpe_product == "notepad\\+\\+"
     assert entry.channel is None
     assert entry.judgeable