feat: cheat sheet suggestion → ParseResult import adapter (Workstream F, F3)

application/tests/cheatsheets_workstream_f_test.py

@@ -1,12 +1,14 @@
-"""Tests for Cheat Sheet -> CRE mapping, Workstream F, checkpoints F1+F2.
+"""Tests for Cheat Sheet -> CRE mapping, Workstream F, checkpoints F1+F2+F3.
 
 F1: the suggestions.json data contract -- SUGGESTIONS_SCHEMA plus the
 CandidateCRE / MappingSuggestion dataclasses.
 F2: the read/write adapters -- write_suggestions_json and
 load_approved_suggestions (schema-validate -> parse -> filter approved).
+F3: the import adapter -- suggestions_to_parse_result (approved suggestions ->
+ParseResult of defs.Standard with AutomaticallyLinkedTo links). F3's tests use a
+lightweight Node_collection stub (get_CREs only) so no Postgres/Neo4j is needed.
 
-F3 (suggestions_to_parse_result) and F4/F5 (CLI) are deliberately out of scope
-here; these tests never touch defs.Standard / ParseResult / Node_collection.
+F4/F5 (CLI) are deliberately out of scope here.
 """
 
 import json
@@ -16,6 +18,8 @@
 
 import jsonschema
 
+from application.defs import cre_defs as defs
+from application.utils.external_project_parsers import base_parser_defs
 from application.utils.external_project_parsers.parsers import (
     cheatsheets_workstream_f as wf,
 )
@@ -122,5 +126,167 @@ def test_written_file_validates_against_schema(self) -> None:
         jsonschema.validate(instance=doc, schema=wf.SUGGESTIONS_SCHEMA)
 
 
+class _StubCache:
+    """Minimal ``db.Node_collection`` stand-in exposing only ``get_CREs``.
+
+    Known ids resolve to a ``defs.CRE`` (shaped like a real ``external_id`` --
+    e.g. ``"764-507"`` -- exactly what the live cheatsheets parser passes);
+    unknown ids return ``[]`` just like ``db.Node_collection.get_CREs``. No
+    Postgres/Neo4j is touched.
+    """
+
+    def __init__(self, known):
+        # known: dict of cre_id -> defs.CRE
+        self._known = known
+
+    def get_CREs(self, external_id=None, **kwargs):
+        cre = self._known.get(external_id)
+        # Return a fresh copy per call, mirroring the DB (which hydrates anew).
+        return [cre.shallow_copy()] if cre is not None else []
+
+
+def _cre(external_id):
+    return defs.CRE(id=external_id, name=f"CRE {external_id}")
+
+
+def _cand(cre_id):
+    return wf.CandidateCRE(cre_id=cre_id, score=0.9, confidence="high", reason="r")
+
+
+def _sugg(title, candidate_ids, category="authentication", status="approved"):
+    slug = title.replace(" ", "_")
+    return wf.MappingSuggestion(
+        source="owasp_cheatsheets",
+        cheatsheet_id=slug,
+        title=title,
+        hyperlink=f"https://cheatsheetseries.owasp.org/cheatsheets/{slug}.html",
+        category=category,
+        status=status,
+        candidate_cres=[_cand(c) for c in candidate_ids],
+    )
+
+
+class TestSuggestionsToParseResult(unittest.TestCase):
+    def test_builds_standards_with_links(self) -> None:
+        cache = _StubCache({"764-507": _cre("764-507")})
+        approved = [_sugg("Authentication Cheat Sheet", ["764-507"])]
+
+        result = wf.suggestions_to_parse_result(approved, cache)
+
+        self.assertIsInstance(result, base_parser_defs.ParseResult)
+        standards = result.results["OWASP Cheat Sheets"]
+        self.assertEqual(len(standards), 1)
+        std = standards[0]
+        self.assertIsInstance(std, defs.Standard)
+        self.assertEqual(std.name, "OWASP Cheat Sheets")
+        self.assertEqual(std.section, "Authentication Cheat Sheet")
+        self.assertEqual(
+            std.hyperlink,
+            "https://cheatsheetseries.owasp.org/cheatsheets/"
+            "Authentication_Cheat_Sheet.html",
+        )
+        self.assertEqual(len(std.links), 1)
+        link = std.links[0]
+        self.assertEqual(link.ltype, defs.LinkTypes.AutomaticallyLinkedTo)
+        self.assertEqual(link.document.id, "764-507")
+
+    def test_classification_tags_present_and_valid(self) -> None:
+        cache = _StubCache({"764-507": _cre("764-507")})
+        approved = [_sugg("Authentication Cheat Sheet", ["764-507"])]
+
+        result = wf.suggestions_to_parse_result(approved, cache)
+
+        # The import pipeline requires the full classification tag set.
+        base_parser_defs.validate_classification_tags(result.results)
+        std = result.results["OWASP Cheat Sheets"][0]
+        self.assertIn(base_parser_defs.Family.GUIDANCE.value, std.tags)
+        self.assertIn(base_parser_defs.Subtype.CHEATSHEET.value, std.tags)
+        self.assertIn(base_parser_defs.Audience.DEVELOPER.value, std.tags)
+        self.assertIn(base_parser_defs.Maturity.STABLE.value, std.tags)
+        self.assertIn("source:owasp_cheatsheets", std.tags)
+        # category is carried as an extra tag.
+        self.assertIn("authentication", std.tags)
+
+    def test_unknown_cre_id_skipped_sibling_still_linked(self) -> None:
+        cache = _StubCache({"764-507": _cre("764-507")})
+        # One known sibling, one unknown id on the same suggestion.
+        approved = [_sugg("Authentication Cheat Sheet", ["764-507", "999-999"])]
+
+        with self.assertLogs(wf.logger, level="WARNING") as cm:
+            result = wf.suggestions_to_parse_result(approved, cache)
+
+        standards = result.results["OWASP Cheat Sheets"]
+        self.assertEqual(len(standards), 1)
+        self.assertEqual([link.document.id for link in standards[0].links], ["764-507"])
+        # The skipped id is reported for the reviewer.
+        self.assertIn("999-999", "\n".join(cm.output))
+
+    def test_all_unknown_candidates_drops_standard(self) -> None:
+        cache = _StubCache({"764-507": _cre("764-507")})
+        approved = [_sugg("Ghost Cheat Sheet", ["111-111", "222-222"])]
+
+        result = wf.suggestions_to_parse_result(approved, cache)
+
+        # A Standard with zero resolved links is omitted entirely.
+        self.assertEqual(result.results["OWASP Cheat Sheets"], [])
+
+    def test_deterministic_output(self) -> None:
+        cache = _StubCache({"764-507": _cre("764-507"), "581-525": _cre("581-525")})
+        approved = [
+            _sugg("Authentication Cheat Sheet", ["764-507"]),
+            _sugg("Cryptographic Storage Cheat Sheet", ["581-525", "764-507"]),
+        ]
+
+        first = wf.suggestions_to_parse_result(approved, cache)
+        second = wf.suggestions_to_parse_result(approved, cache)
+
+        self.assertEqual(first.results, second.results)
+
+    def test_duplicate_candidate_ids_yield_single_link(self) -> None:
+        # Two candidates on one suggestion resolving to the SAME cre_id must not
+        # raise DuplicateLinkException -- the has_link guard collapses them.
+        cache = _StubCache({"764-507": _cre("764-507")})
+        approved = [_sugg("Authentication Cheat Sheet", ["764-507", "764-507"])]
+
+        result = wf.suggestions_to_parse_result(approved, cache)
+
+        std = result.results["OWASP Cheat Sheets"][0]
+        self.assertEqual(len(std.links), 1)
+        self.assertEqual(std.links[0].document.id, "764-507")
+
+    def test_blank_category_produces_no_extra_tag(self) -> None:
+        # Empty AND whitespace-only categories (both schema-valid) must not leak
+        # a blank/whitespace extra tag.
+        for blank in ("", "   "):
+            with self.subTest(category=repr(blank)):
+                cache = _StubCache({"764-507": _cre("764-507")})
+                approved = [
+                    _sugg("Authentication Cheat Sheet", ["764-507"], category=blank)
+                ]
+
+                result = wf.suggestions_to_parse_result(approved, cache)
+
+                # Still a valid, classifiable Standard...
+                base_parser_defs.validate_classification_tags(result.results)
+                std = result.results["OWASP Cheat Sheets"][0]
+                # ...with no blank/whitespace tag leaked in from the category.
+                self.assertNotIn("", std.tags)
+                self.assertNotIn(blank, std.tags)
+                # Only the five required classification tags remain (no extra).
+                self.assertEqual(len(std.tags), 5)
+
+    def test_category_tag_is_stripped(self) -> None:
+        cache = _StubCache({"764-507": _cre("764-507")})
+        approved = [
+            _sugg("Authentication Cheat Sheet", ["764-507"], category="  auth  ")
+        ]
+
+        result = wf.suggestions_to_parse_result(approved, cache)
+
+        std = result.results["OWASP Cheat Sheets"][0]
+        self.assertIn("auth", std.tags)
+        self.assertNotIn("  auth  ", std.tags)
+
+
 if __name__ == "__main__":
     unittest.main()

application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py

@@ -6,28 +6,29 @@
 in a later checkpoint, the conversion of approved suggestions into the import
 pipeline's ``ParseResult``.
 
-Checkpoints implemented here (F1 + F2):
+Checkpoints implemented here (F1 + F2 + F3):
 
 * F1 -- the data contract: :data:`SUGGESTIONS_SCHEMA` (JSON Schema) plus the
   :class:`CandidateCRE` / :class:`MappingSuggestion` dataclasses.
 * F2 -- the read/write adapters: :func:`write_suggestions_json` and
   :func:`load_approved_suggestions` (schema-validate -> parse -> filter to the
   reviewer-approved entries).
-
-Deliberately NOT in this module yet:
-
-* F3 -- ``suggestions_to_parse_result(approved, cache)``. When it lands it must
-  reconcile the review artifact with the *real* import types (real code wins):
+* F3 -- the import adapter: :func:`suggestions_to_parse_result`, which reconciles
+  the review artifact with the *real* import types (real code wins):
     - fixed ``defs.Standard(name="OWASP Cheat Sheets")``; the suggestion ``title``
       maps to ``Standard.section`` (matching the existing cheatsheets_parser);
     - the import pipeline REQUIRES ``family:/subtype:/audience:/maturity:/source:``
       classification tags (``base_parser_defs.validate_classification_tags``
       raises otherwise) -- built via ``build_tags`` -- even though the RFC's
-      §4 contract omits them;
+      §4 contract omits them; ``category`` rides along as an extra tag when set;
     - candidate CREs are resolved via ``cache.get_CREs(external_id=...)`` and
-      unknown ids are skipped; a Standard with zero resolved links is dropped;
-    - the advisory review fields dropped below (see the schema) have no home on
+      unknown ids are skipped (collected + logged); a Standard with zero resolved
+      links is dropped;
+    - the advisory review fields (see the schema) have no home on
       ``defs.Standard`` and are intentionally not persisted.
+
+Deliberately NOT in this module yet:
+
 * F4/F5 -- the ``generate/validate/convert`` CLI.
 
 The advisory fields ``score``, ``confidence``, ``reason`` (per candidate) and
@@ -45,11 +46,21 @@
 
 import dataclasses
 import json
+import logging
 from dataclasses import dataclass, field
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TYPE_CHECKING
 
 import jsonschema
 
+from application.defs import cre_defs as defs
+from application.utils.external_project_parsers import base_parser_defs
+from application.utils.external_project_parsers.base_parser_defs import ParseResult
+
+if TYPE_CHECKING:
+    from application.database import db
+
+logger = logging.getLogger(__name__)
+
 # --- valid values -----------------------------------------------------------
 
 #: Reviewer lifecycle for a suggestion. Only ``approved`` items are converted.
@@ -232,3 +243,87 @@ def load_approved_suggestions(path: str) -> List[MappingSuggestion]:
         doc = json.load(fh)
     _validate(doc)
     return [_parse_suggestion(raw) for raw in doc if raw["status"] == "approved"]
+
+
+# --- F3: import adapter -----------------------------------------------------
+
+#: Fixed Standard name for every imported cheat sheet, matching the live
+#: ``cheatsheets_parser`` so entries dedupe/merge against the same resource.
+STANDARD_NAME = "OWASP Cheat Sheets"
+
+
+def suggestions_to_parse_result(
+    approved: List[MappingSuggestion],
+    cache: "db.Node_collection",
+) -> ParseResult:
+    """Convert approved suggestions into a ``ParseResult`` the import flow accepts.
+
+    Per approved suggestion, builds a ``defs.Standard`` (fixed name
+    ``"OWASP Cheat Sheets"``; ``title`` -> ``section``; ``hyperlink``; the
+    required classification tags via ``build_tags`` with ``category`` as an extra
+    tag when non-empty). Each candidate ``cre_id`` is resolved via
+    ``cache.get_CREs(external_id=...)``; unknown ids are collected, logged, and
+    skipped. Resolved CREs are attached as ``AutomaticallyLinkedTo`` links (on a
+    ``shallow_copy`` so the CRE's own links don't leak in, and guarded by
+    ``has_link`` so a repeated id can't raise ``DuplicateLinkException``). A
+    Standard with zero resolved links is dropped.
+
+    The advisory review fields (``score`` / ``confidence`` / ``reason`` /
+    ``cheatsheet_id``) have no home on ``defs.Standard`` and are not persisted.
+    """
+    standards: List[defs.Document] = []
+    unknown_cre_ids: List[str] = []
+
+    for suggestion in approved:
+        category = suggestion.category.strip()
+        extra = [category] if category else []
+        standard = defs.Standard(
+            name=STANDARD_NAME,
+            section=suggestion.title,
+            hyperlink=suggestion.hyperlink,
+            tags=base_parser_defs.build_tags(
+                family=base_parser_defs.Family.GUIDANCE,
+                subtype=base_parser_defs.Subtype.CHEATSHEET,
+                audience=base_parser_defs.Audience.DEVELOPER,
+                maturity=base_parser_defs.Maturity.STABLE,
+                source="owasp_cheatsheets",
+                extra=extra,
+            ),
+        )
+
+        for candidate in suggestion.candidate_cres:
+            cres = cache.get_CREs(external_id=candidate.cre_id)
+            if not cres:
+                unknown_cre_ids.append(candidate.cre_id)
+                logger.warning(
+                    "Skipping unknown CRE id %s for cheat sheet %r; not in cache.",
+                    candidate.cre_id,
+                    suggestion.title,
+                )
+                continue
+            for cre in cres:
+                link = defs.Link(
+                    document=cre.shallow_copy(),
+                    ltype=defs.LinkTypes.AutomaticallyLinkedTo,
+                )
+                # Guard against a repeated cre_id (add_link would otherwise raise
+                # DuplicateLinkException on the second occurrence).
+                if not standard.has_link(link):
+                    standard.add_link(link)
+
+        if standard.links:
+            standards.append(standard)
+        else:
+            logger.info(
+                "Dropping cheat sheet %r: no candidate CREs resolved.",
+                suggestion.title,
+            )
+
+    if unknown_cre_ids:
+        logger.warning(
+            "suggestions_to_parse_result skipped %d unknown CRE id(s): %s",
+            len(unknown_cre_ids),
+            ", ".join(unknown_cre_ids),
+        )
+
+    return ParseResult(results={STANDARD_NAME: standards})