Rework OWASP mapping importers as shared eval fixtures

application/cmd/cre_main.py

@@ -25,10 +25,6 @@
 from application.utils import db_backend
 from application.utils import gap_analysis
 from application.utils import cres_csv_export
-from application.utils.external_project_parsers.parsers import (
-    owasp_kubernetes_top10_2022,
-    owasp_kubernetes_top10_2025,
-)
 
 if TYPE_CHECKING:
     from application.prompt_client import prompt_client as prompt_client

application/tests/owasp_mapping_fixtures_test.py

@@ -1,10 +1,9 @@
-import json
 import re
 import unittest
-from pathlib import Path
+
+from application.utils import mapping_fixtures
 
 
-FIXTURE_DIR = Path(__file__).parent / "fixtures" / "owasp_mappings"
 EXPECTED_FIXTURES = {
     "owasp_aisvs_1_0.json",
     "owasp_api_top10_2023.json",
@@ -19,13 +18,55 @@
 
 class TestOwaspMappingFixtures(unittest.TestCase):
     def test_fixture_set_is_complete(self) -> None:
-        actual = {path.name for path in FIXTURE_DIR.glob("*.json")}
-        self.assertEqual(actual, EXPECTED_FIXTURES)
+        self.assertEqual(
+            set(mapping_fixtures.list_owasp_mapping_fixtures()), EXPECTED_FIXTURES
+        )
+
+    def test_load_all_returns_every_named_fixture(self) -> None:
+        loaded = mapping_fixtures.load_all_owasp_mapping_fixtures()
+        self.assertEqual(set(loaded), EXPECTED_FIXTURES)
+        for filename, payload in loaded.items():
+            with self.subTest(fixture=filename):
+                self.assertIsInstance(payload, list)
+                self.assertGreater(len(payload), 0)
+
+    def test_load_accepts_stem_without_json_suffix(self) -> None:
+        by_stem = mapping_fixtures.load_owasp_mapping_fixture("owasp_top10_2025")
+        by_name = mapping_fixtures.load_owasp_mapping_fixture("owasp_top10_2025.json")
+        self.assertEqual(by_stem, by_name)
+
+    def test_load_unknown_fixture_raises(self) -> None:
+        with self.assertRaises(FileNotFoundError) as ctx:
+            mapping_fixtures.load_owasp_mapping_fixture("not_a_real_mapping")
+        self.assertIn("not_a_real_mapping.json", str(ctx.exception))
+
+    def test_k8s_2025_uses_per_item_owasp_hyperlinks(self) -> None:
+        # Data from #953 (Bornunique911): section pages, not the family homepage.
+        entries = mapping_fixtures.load_owasp_mapping_fixture(
+            "owasp_kubernetes_top10_2025"
+        )
+        self.assertEqual(10, len(entries))
+        self.assertEqual("K01", entries[0]["section_id"])
+        self.assertIn("/2025/en/src/K01-", entries[0]["hyperlink"])
+        self.assertEqual(["233-748", "486-813"], entries[0]["cre_ids"])
+
+    def test_resolved_cre_ids_follow_fallback_when_own_ids_empty(self) -> None:
+        by_id = {
+            "K05": {"cre_ids": ["148-420"]},
+            "K10": {
+                "cre_ids": [],
+                "fallback_section_ids": ["K05"],
+            },
+        }
+        self.assertEqual(
+            ["148-420"],
+            mapping_fixtures.resolved_cre_ids(by_id["K10"], by_id),
+        )
 
     def test_fixtures_have_expected_mapping_shape(self) -> None:
-        for path in sorted(FIXTURE_DIR.glob("*.json")):
-            with self.subTest(fixture=path.name):
-                payload = json.loads(path.read_text(encoding="utf-8"))
+        loaded = mapping_fixtures.load_all_owasp_mapping_fixtures()
+        for filename, payload in loaded.items():
+            with self.subTest(fixture=filename):
 
                 self.assertIsInstance(payload, list)
                 self.assertGreater(len(payload), 0)
@@ -53,7 +94,7 @@ def test_fixtures_have_expected_mapping_shape(self) -> None:
                         self.assertNotIn(
                             entry["section_id"],
                             seen_section_ids,
-                            msg=f"Duplicate section_id {entry['section_id']} in {path.name}",
+                            msg=f"Duplicate section_id {entry['section_id']} in {filename}",
                         )
                         seen_section_ids.add(entry["section_id"])
 
@@ -68,7 +109,7 @@ def test_fixtures_have_expected_mapping_shape(self) -> None:
                                 known_section_ids,
                                 msg=(
                                     f"Fallback section id {fallback_section_id} "
-                                    f"in {path.name} is not a known section_id"
+                                    f"in {filename} is not a known section_id"
                                 ),
                             )
 

application/utils/external_project_parsers/parsers/cheatsheets_parser.py

@@ -8,8 +8,8 @@
 import re
 from application.utils.external_project_parsers import base_parser_defs
 import json
-from pathlib import Path
 import logging
+from application.utils.mapping_fixtures import OWASP_MAPPING_FIXTURE_DIR
 from application.utils.external_project_parsers.base_parser_defs import (
     ParserInterface,
     ParseResult,
@@ -21,11 +21,7 @@ class Cheatsheets(ParserInterface):
     name = "OWASP Cheat Sheets"
     cheatsheetseries_base_url = "https://cheatsheetseries.owasp.org/cheatsheets"
     supplement_data_file = (
-        Path(__file__).resolve().parents[3]
-        / "tests"
-        / "fixtures"
-        / "owasp_mappings"
-        / "owasp_cheatsheets_supplement.json"
+        OWASP_MAPPING_FIXTURE_DIR / "owasp_cheatsheets_supplement.json"
     )
     logger = logging.getLogger(__name__)
 

application/utils/external_project_parsers/parsers/owasp_aisvs.py

@@ -1,51 +1,8 @@
-import json
-from pathlib import Path
-
-from application.database import db
-from application.defs import cre_defs as defs
-from application.prompt_client import prompt_client
-from application.utils.external_project_parsers.base_parser_defs import (
-    ParseResult,
-    ParserInterface,
+from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import (
+    OwaspMappingFixtureParser,
 )
 
 
-class OwaspAisvs(ParserInterface):
+class OwaspAisvs(OwaspMappingFixtureParser):
     name = "OWASP AI Security Verification Standard (AISVS)"
-    data_file = (
-        Path(__file__).resolve().parents[3]
-        / "tests"
-        / "fixtures"
-        / "owasp_mappings"
-        / "owasp_aisvs_1_0.json"
-    )
-
-    def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler):
-        with self.data_file.open("r", encoding="utf-8") as handle:
-            raw_entries = json.load(handle)
-
-        entries = []
-        for entry in raw_entries:
-            standard = defs.Standard(
-                name=self.name,
-                sectionID=entry["section_id"],
-                section=entry["section"],
-                hyperlink=entry["hyperlink"],
-            )
-            for cre_id in entry.get("cre_ids", []):
-                cres = cache.get_CREs(external_id=cre_id)
-                if not cres:
-                    continue
-                standard.add_link(
-                    defs.Link(
-                        ltype=defs.LinkTypes.LinkedTo,
-                        document=cres[0].shallow_copy(),
-                    )
-                )
-            entries.append(standard)
-
-        return ParseResult(
-            results={self.name: entries},
-            calculate_gap_analysis=False,
-            calculate_embeddings=False,
-        )
+    fixture_name = "owasp_aisvs_1_0"

application/utils/external_project_parsers/parsers/owasp_api_top10_2023.py

@@ -1,51 +1,8 @@
-import json
-from pathlib import Path
-
-from application.database import db
-from application.defs import cre_defs as defs
-from application.prompt_client import prompt_client
-from application.utils.external_project_parsers.base_parser_defs import (
-    ParseResult,
-    ParserInterface,
+from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import (
+    OwaspMappingFixtureParser,
 )
 
 
-class OwaspApiTop10_2023(ParserInterface):
+class OwaspApiTop10_2023(OwaspMappingFixtureParser):
     name = "OWASP API Security Top 10 2023"
-    data_file = (
-        Path(__file__).resolve().parents[3]
-        / "tests"
-        / "fixtures"
-        / "owasp_mappings"
-        / "owasp_api_top10_2023.json"
-    )
-
-    def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler):
-        with self.data_file.open("r", encoding="utf-8") as handle:
-            raw_entries = json.load(handle)
-
-        entries = []
-        for entry in raw_entries:
-            standard = defs.Standard(
-                name=self.name,
-                sectionID=entry["section_id"],
-                section=entry["section"],
-                hyperlink=entry["hyperlink"],
-            )
-            for cre_id in entry.get("cre_ids", []):
-                cres = cache.get_CREs(external_id=cre_id)
-                if not cres:
-                    continue
-                standard.add_link(
-                    defs.Link(
-                        ltype=defs.LinkTypes.LinkedTo,
-                        document=cres[0].shallow_copy(),
-                    )
-                )
-            entries.append(standard)
-
-        return ParseResult(
-            results={self.name: entries},
-            calculate_gap_analysis=False,
-            calculate_embeddings=False,
-        )
+    fixture_name = "owasp_api_top10_2023"

application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2022.py

@@ -1,51 +1,8 @@
-import json
-from pathlib import Path
-
-from application.database import db
-from application.defs import cre_defs as defs
-from application.prompt_client import prompt_client
-from application.utils.external_project_parsers.base_parser_defs import (
-    ParseResult,
-    ParserInterface,
+from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import (
+    OwaspMappingFixtureParser,
 )
 
 
-class OwaspKubernetesTop10_2022(ParserInterface):
+class OwaspKubernetesTop10_2022(OwaspMappingFixtureParser):
     name = "OWASP Kubernetes Top Ten 2022"
-    data_file = (
-        Path(__file__).resolve().parents[3]
-        / "tests"
-        / "fixtures"
-        / "owasp_mappings"
-        / "owasp_kubernetes_top10_2022.json"
-    )
-
-    def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler):
-        with self.data_file.open("r", encoding="utf-8") as handle:
-            raw_entries = json.load(handle)
-
-        entries = []
-        for entry in raw_entries:
-            standard = defs.Standard(
-                name=self.name,
-                sectionID=entry["section_id"],
-                section=entry["section"],
-                hyperlink=entry["hyperlink"],
-            )
-            for cre_id in entry.get("cre_ids", []):
-                cres = cache.get_CREs(external_id=cre_id)
-                if not cres:
-                    continue
-                standard.add_link(
-                    defs.Link(
-                        ltype=defs.LinkTypes.LinkedTo,
-                        document=cres[0].shallow_copy(),
-                    )
-                )
-            entries.append(standard)
-
-        return ParseResult(
-            results={self.name: entries},
-            calculate_gap_analysis=False,
-            calculate_embeddings=False,
-        )
+    fixture_name = "owasp_kubernetes_top10_2022"

application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2025.py

@@ -1,82 +1,9 @@
-import json
-from pathlib import Path
-
-from application.database import db
-from application.defs import cre_defs as defs
-from application.prompt_client import prompt_client
-from application.utils.external_project_parsers.base_parser_defs import (
-    ParseResult,
-    ParserInterface,
+from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import (
+    OwaspMappingFixtureParser,
 )
 
 
-class OwaspKubernetesTop10_2025(ParserInterface):
+class OwaspKubernetesTop10_2025(OwaspMappingFixtureParser):
     name = "OWASP Kubernetes Top Ten 2025 (Draft)"
-    data_file = (
-        Path(__file__).resolve().parents[3]
-        / "tests"
-        / "fixtures"
-        / "owasp_mappings"
-        / "owasp_kubernetes_top10_2025.json"
-    )
-    fallback_data_file = (
-        Path(__file__).resolve().parents[3]
-        / "tests"
-        / "fixtures"
-        / "owasp_mappings"
-        / "owasp_kubernetes_top10_2022.json"
-    )
-
-    def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler):
-        with self.data_file.open("r", encoding="utf-8") as handle:
-            raw_entries = json.load(handle)
-        with self.fallback_data_file.open("r", encoding="utf-8") as handle:
-            fallback_entries = {
-                entry["section_id"]: entry for entry in json.load(handle)
-            }
-
-        entries = []
-        for entry in raw_entries:
-            standard = defs.Standard(
-                name=self.name,
-                sectionID=entry["section_id"],
-                section=entry["section"],
-                hyperlink=entry["hyperlink"],
-            )
-            linked_cre_ids = []
-            for cre_id in entry.get("cre_ids", []):
-                cres = cache.get_CREs(external_id=cre_id)
-                if not cres:
-                    continue
-                linked_cre_ids.append(cre_id)
-                standard.add_link(
-                    defs.Link(
-                        ltype=defs.LinkTypes.LinkedTo,
-                        document=cres[0].shallow_copy(),
-                    )
-                )
-            if not linked_cre_ids:
-                for section_id in entry.get("fallback_section_ids", []):
-                    fallback_entry = fallback_entries.get(section_id)
-                    if not fallback_entry:
-                        continue
-                    for cre_id in fallback_entry.get("cre_ids", []):
-                        if cre_id in linked_cre_ids:
-                            continue
-                        cres = cache.get_CREs(external_id=cre_id)
-                        if not cres:
-                            continue
-                        linked_cre_ids.append(cre_id)
-                        standard.add_link(
-                            defs.Link(
-                                ltype=defs.LinkTypes.LinkedTo,
-                                document=cres[0].shallow_copy(),
-                            )
-                        )
-            entries.append(standard)
-
-        return ParseResult(
-            results={self.name: entries},
-            calculate_gap_analysis=False,
-            calculate_embeddings=False,
-        )
+    fixture_name = "owasp_kubernetes_top10_2025"
+    fallback_fixture_name = "owasp_kubernetes_top10_2022"