Add backend coverage for authority-pack config, source-host union, and registry fault isolation

Covers the patch lines codecov/patch/Backend flagged on this PR:

- test_authority_pack_config.py (new): the fail-fast validators
  (iter_shape_rules / iter_abbreviations) on every malformed shape, the
  never-raise _load_yaml / _iter_pack_mapping_files skip paths, and the
  runtime-scan fault isolation that downgrades a bad pack to log-and-skip.
- test_authority_source_hosts.py: manifest-skip branches in
  pack_declared_source_hosts (no pack.yaml, malformed YAML, non-list
  source_hosts).
- test_authority_pack.py: Command._validate_source_hosts rejects a
  non-list source_hosts declaration.
- test_authority_pack_providers.py: a broken in-pack provider module is
  logged + skipped without breaking sibling discovery; a duplicate
  supported-prefix install is warned; authority_pack_dirs skips a
  non-directory AUTHORITY_PACK_PATHS entry.
- test_safe_http.py: _resolve_allowlist fails closed to the baseline when
  no dynamic provider is registered.

Tests only; no production code change.

Author: JSv4

opencontractserver/tests/test_authority_pack.py

@@ -22,6 +22,9 @@
 
 from opencontractserver import enrichment
 from opencontractserver.annotations.models import AuthorityNamespace
+from opencontractserver.corpuses.management.commands.load_authority_pack import (
+    Command as LoadAuthorityPackCommand,
+)
 from opencontractserver.corpuses.models import Corpus
 from opencontractserver.corpuses.services.corpus_documents import (
     CorpusDocumentService,
@@ -42,6 +45,32 @@ def _load_yaml(path: Path) -> dict:
     return yaml.safe_load(path.read_text(encoding="utf-8"))
 
 
+class ValidateSourceHostsTests(SimpleTestCase):
+    """``_validate_source_hosts`` fails fast on a malformed manifest declaration."""
+
+    def test_missing_source_hosts_is_ok(self):
+        # Absent / null source_hosts is valid (a non-scraping pack).
+        LoadAuthorityPackCommand._validate_source_hosts({})
+        LoadAuthorityPackCommand._validate_source_hosts({"source_hosts": None})
+
+    def test_source_hosts_not_a_list_raises(self):
+        with self.assertRaisesMessage(CommandError, "must be a list of hostnames"):
+            LoadAuthorityPackCommand._validate_source_hosts(
+                {"source_hosts": "tcpbolivia.bo"}
+            )
+
+    def test_source_hosts_entry_not_a_bare_host_raises(self):
+        with self.assertRaisesMessage(CommandError, "is not a bare hostname"):
+            LoadAuthorityPackCommand._validate_source_hosts(
+                {"source_hosts": ["https://tcpbolivia.bo"]}
+            )
+
+    def test_valid_source_hosts_pass(self):
+        LoadAuthorityPackCommand._validate_source_hosts(
+            {"source_hosts": ["tcpbolivia.bo", "gacetaoficialdebolivia.gob.bo"]}
+        )
+
+
 class BoliviaPackContentTests(SimpleTestCase):
     """Static integrity of the shipped pack — runs with no DB."""
 

opencontractserver/tests/test_authority_pack_config.py

@@ -0,0 +1,226 @@
+"""Validation + fault-isolation of pack-declared citation vocabulary.
+
+``enrichment/services/authority_pack_config.py`` reads each installed pack's
+authority-mappings YAML and merges its ``shape_rules`` / ``abbreviations`` onto
+the Python baseline. The happy path (a well-formed pack actually changing
+classification / extraction) is covered by ``test_authority_pack_taxonomy``; this
+module pins the *defensive* contract:
+
+* the fail-fast validators (``iter_shape_rules`` / ``iter_abbreviations``) raise
+  ``ValueError`` on every malformed shape so ``load_authority_pack`` aborts a bad
+  install loudly, and
+* the runtime scan (``pack_declared_shape_rules`` / ``pack_declared_abbreviations``
+  / ``_iter_pack_mapping_files`` / ``_load_yaml``) downgrades those raises to
+  log-and-skip so a single broken pack can never break extraction for every
+  jurisdiction.
+
+No database required — the config is read from pack files on disk.
+"""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+from unittest import mock
+
+from django.test import SimpleTestCase
+
+from opencontractserver.enrichment.services import authority_pack_config as apc
+
+_MODULE = "opencontractserver.enrichment.services.authority_pack_config"
+
+# A reliably-unparseable YAML document (unterminated flow sequence).
+_BAD_YAML = "name: oops\nsource: [unterminated"
+
+
+class IterShapeRulesValidationTests(SimpleTestCase):
+    """``iter_shape_rules`` — every malformed shape raises ``ValueError``."""
+
+    def test_none_returns_empty(self):
+        self.assertEqual(apc.iter_shape_rules({}), [])
+        self.assertEqual(apc.iter_shape_rules({"shape_rules": None}), [])
+
+    def test_not_a_list_raises(self):
+        with self.assertRaisesMessage(ValueError, "must be a list"):
+            apc.iter_shape_rules({"shape_rules": {"pattern": "^x$"}})
+
+    def test_entry_without_pattern_raises(self):
+        with self.assertRaisesMessage(ValueError, "needs a 'pattern'"):
+            apc.iter_shape_rules({"shape_rules": [{"jurisdiction": "bo"}]})
+
+    def test_uncompilable_pattern_raises(self):
+        with self.assertRaisesMessage(ValueError, "bad regex"):
+            apc.iter_shape_rules({"shape_rules": [{"pattern": "([unclosed"}]})
+
+    def test_unknown_authority_type_raises(self):
+        with self.assertRaisesMessage(ValueError, "not in "):
+            apc.iter_shape_rules(
+                {
+                    "shape_rules": [
+                        {"pattern": "^bo-ley-\\d+$", "authority_type": "not_a_type"}
+                    ]
+                }
+            )
+
+    def test_valid_entry_normalised(self):
+        out = apc.iter_shape_rules(
+            {"shape_rules": [{"pattern": "^bo-ley-\\d+$", "authority_type": "statute"}]}
+        )
+        self.assertEqual(len(out), 1)
+        self.assertEqual(out[0]["authority_type"], "statute")
+        self.assertIsNone(out[0]["jurisdiction"])  # absent -> None
+
+
+class IterAbbreviationsValidationTests(SimpleTestCase):
+    """``iter_abbreviations`` — every malformed shape raises ``ValueError``."""
+
+    def test_none_returns_empty_groups(self):
+        self.assertEqual(apc.iter_abbreviations({}), {"state": {}, "municipal": {}})
+
+    def test_not_a_mapping_raises(self):
+        with self.assertRaisesMessage(ValueError, "must be a mapping"):
+            apc.iter_abbreviations({"abbreviations": ["nope"]})
+
+    def test_group_not_a_mapping_raises(self):
+        with self.assertRaisesMessage(ValueError, "must be a mapping"):
+            apc.iter_abbreviations({"abbreviations": {"state": ["nope"]}})
+
+    def test_entry_without_prefix_raises(self):
+        with self.assertRaisesMessage(ValueError, "needs 'prefix'"):
+            apc.iter_abbreviations(
+                {"abbreviations": {"state": {"Bol. Civ. Code": {"jurisdiction": "bo"}}}}
+            )
+
+    def test_unknown_authority_type_raises(self):
+        with self.assertRaisesMessage(ValueError, "not in "):
+            apc.iter_abbreviations(
+                {
+                    "abbreviations": {
+                        "state": {
+                            "Bol. Civ. Code": {
+                                "prefix": "bo-civ",
+                                "authority_type": "not_a_type",
+                            }
+                        }
+                    }
+                }
+            )
+
+    def test_valid_entry_normalised(self):
+        out = apc.iter_abbreviations(
+            {
+                "abbreviations": {
+                    "municipal": {
+                        "Some Ord.": {"prefix": "bo-ord", "authority_type": "statute"}
+                    }
+                }
+            }
+        )
+        self.assertEqual(out["municipal"]["Some Ord."]["prefix"], "bo-ord")
+        self.assertEqual(out["state"], {})
+
+
+class LoadYamlTests(SimpleTestCase):
+    """``_load_yaml`` never raises — bad / non-mapping YAML degrades to ``{}``."""
+
+    def test_malformed_yaml_returns_empty(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "m.yaml"
+            path.write_text(_BAD_YAML, encoding="utf-8")
+            self.assertEqual(apc._load_yaml(path), {})
+
+    def test_non_mapping_yaml_returns_empty(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "m.yaml"
+            path.write_text("- just\n- a\n- list\n", encoding="utf-8")
+            self.assertEqual(apc._load_yaml(path), {})
+
+    def test_validate_pack_taxonomy_extensions_passes_for_valid(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "m.yaml"
+            path.write_text(
+                "shape_rules:\n  - pattern: '^bo-ley-\\d+$'\n", encoding="utf-8"
+            )
+            # Should not raise.
+            apc.validate_pack_taxonomy_extensions(path)
+
+
+class IterPackMappingFilesSkipTests(SimpleTestCase):
+    """``_iter_pack_mapping_files`` skips packs it cannot use, never raises."""
+
+    def _patch_dirs(self, *dirs: Path):
+        return mock.patch.object(apc, "authority_pack_dirs", return_value=list(dirs))
+
+    def test_pack_without_manifest_is_skipped(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            pack = Path(tmp) / "no-manifest"
+            pack.mkdir()
+            with self._patch_dirs(pack):
+                self.assertEqual(list(apc._iter_pack_mapping_files()), [])
+
+    def test_pack_with_malformed_manifest_is_skipped(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            pack = Path(tmp) / "bad-manifest"
+            pack.mkdir()
+            (pack / "pack.yaml").write_text(_BAD_YAML, encoding="utf-8")
+            with self._patch_dirs(pack):
+                with self.assertLogs(_MODULE, level="WARNING"):
+                    self.assertEqual(list(apc._iter_pack_mapping_files()), [])
+
+    def test_pack_without_mappings_key_is_skipped(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            pack = Path(tmp) / "no-mappings"
+            pack.mkdir()
+            (pack / "pack.yaml").write_text("name: x\n", encoding="utf-8")
+            with self._patch_dirs(pack):
+                self.assertEqual(list(apc._iter_pack_mapping_files()), [])
+
+    def test_well_formed_pack_is_yielded(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            pack = Path(tmp) / "ok"
+            pack.mkdir()
+            (pack / "pack.yaml").write_text(
+                "name: x\nmappings: m.yaml\n", encoding="utf-8"
+            )
+            (pack / "m.yaml").write_text("shape_rules: []\n", encoding="utf-8")
+            with self._patch_dirs(pack):
+                yielded = list(apc._iter_pack_mapping_files())
+            self.assertEqual([p for p, _ in yielded], [pack])
+
+
+class RuntimeScanFaultIsolationTests(SimpleTestCase):
+    """A pack whose mappings are malformed is logged + skipped, not raised."""
+
+    def setUp(self):
+        self.addCleanup(apc.reset_pack_config_cache)
+
+    def _patch_dirs(self, *dirs: Path):
+        return mock.patch.object(apc, "authority_pack_dirs", return_value=list(dirs))
+
+    @staticmethod
+    def _write_pack(root: Path, mappings_body: str) -> Path:
+        pack = root / "broken-vocab"
+        pack.mkdir()
+        (pack / "pack.yaml").write_text("name: x\nmappings: m.yaml\n", encoding="utf-8")
+        (pack / "m.yaml").write_text(mappings_body, encoding="utf-8")
+        return pack
+
+    def test_malformed_shape_rules_skipped_not_raised(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            # shape_rules is a string, not a list -> iter_shape_rules raises;
+            # the scan must downgrade to a logged skip and return no rules.
+            pack = self._write_pack(Path(tmp), "shape_rules: not-a-list\n")
+            with self._patch_dirs(pack):
+                apc.reset_pack_config_cache()
+                with self.assertLogs(_MODULE, level="WARNING"):
+                    rules = apc.pack_declared_shape_rules()
+            self.assertEqual(rules, ())
+
+    def test_malformed_abbreviations_skipped_not_raised(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            pack = self._write_pack(Path(tmp), "abbreviations: not-a-mapping\n")
+            with self._patch_dirs(pack):
+                apc.reset_pack_config_cache()
+                with self.assertLogs(_MODULE, level="WARNING"):
+                    state, municipal = apc.pack_declared_abbreviations()
+            self.assertEqual((state, municipal), ({}, {}))

opencontractserver/tests/test_authority_pack_providers.py

@@ -17,10 +17,13 @@
 from django.test import SimpleTestCase, override_settings
 
 from opencontractserver.pipeline.registry import (
+    authority_pack_dirs,
     get_all_authority_source_providers_cached,
     reset_registry,
 )
 
+_REGISTRY_LOGGER = "opencontractserver.pipeline.registry"
+
 # A minimal, importable provider shipped "inside a pack". Imported by file path
 # under a synthetic module name, so its real OpenContracts imports must resolve.
 _DEMO_PROVIDER_SRC = """
@@ -91,3 +94,61 @@ def test_provider_absent_without_pack_path(self):
             self.assertNotIn("DemoPackProvider", names)
             # The shipped core providers are still discovered.
             self.assertIn("CFRAuthoritySourceProvider", names)
+
+    def test_broken_pack_provider_is_logged_and_skipped(self):
+        # A provider module that fails to import must be logged + skipped without
+        # breaking discovery of the pack's other (valid) providers — one bad file
+        # never crashes registry build.
+        with tempfile.TemporaryDirectory() as tmp:
+            pack = self._write_pack(Path(tmp))
+            (pack / "providers" / "broken.py").write_text(
+                "raise RuntimeError('boom in pack provider')\n", encoding="utf-8"
+            )
+            with override_settings(AUTHORITY_PACK_PATHS=[str(pack)]):
+                reset_registry()
+                with self.assertLogs(_REGISTRY_LOGGER, level="WARNING") as cm:
+                    names = {
+                        p.name for p in get_all_authority_source_providers_cached()
+                    }
+            # The valid sibling still loads despite the broken module.
+            self.assertIn("DemoPackProvider", names)
+            self.assertTrue(
+                any("Failed to import pack provider" in m for m in cm.output),
+                cm.output,
+            )
+
+    def test_duplicate_provider_prefix_is_warned(self):
+        # Two providers claiming the same supported_prefixes family resolve
+        # non-deterministically; the registry makes the shadowing install loud.
+        with tempfile.TemporaryDirectory() as tmp:
+            pack = self._write_pack(Path(tmp))
+            dup_src = _DEMO_PROVIDER_SRC.replace("DemoPackProvider", "DupPackProvider")
+            (pack / "providers" / "dup_provider.py").write_text(
+                dup_src, encoding="utf-8"
+            )
+            with override_settings(AUTHORITY_PACK_PATHS=[str(pack)]):
+                reset_registry()
+                with self.assertLogs(_REGISTRY_LOGGER, level="WARNING") as cm:
+                    get_all_authority_source_providers_cached()
+            self.assertTrue(
+                any(
+                    "Duplicate authority-source-provider prefix" in m for m in cm.output
+                ),
+                cm.output,
+            )
+
+
+class AuthorityPackDirsTests(SimpleTestCase):
+    """``authority_pack_dirs`` never raises on a misconfigured setting entry."""
+
+    def setUp(self):
+        self.addCleanup(reset_registry)
+
+    def test_non_directory_path_entry_is_warned_and_skipped(self):
+        with override_settings(AUTHORITY_PACK_PATHS=["/no/such/authority/pack/dir"]):
+            with self.assertLogs(_REGISTRY_LOGGER, level="WARNING") as cm:
+                dirs = authority_pack_dirs()
+            self.assertTrue(
+                any("is not a directory" in m for m in cm.output), cm.output
+            )
+            self.assertNotIn(Path("/no/such/authority/pack/dir"), dirs)