test: audit every probe's banks where the probe keeps them

tests/_asset_audit.py

@@ -3,17 +3,21 @@
 
 """Asset-bank audit primitives for ``test_no_template_text.py``.
 
-The audit scans every bank in ``BANK_PATHS`` for template-fill placeholder
-strings and for ``LocalizedText`` fields that carry one identical string
-across three or more locales. ``test_no_template_text.py`` runs it on
+The audit scans every bank ``shipped_banks()`` lists for template-fill
+placeholder strings and for ``LocalizedText`` fields that carry one identical
+string across three or more locales. ``test_no_template_text.py`` runs it on
 every push (strict by default) and fails CI if any of the patterns below
-appear.
+appear. ``test_asset_yaml.py`` reads the same banks, through
+``shipped_banks()`` and ``YAML_LOADER``.
 
 Public surface:
 
-- ``BANK_PATHS`` — tuple of repo-relative YAML paths to every shipped bank.
-- ``audit_one_bank(path)`` — return a per-bank ``{template_hits,
+- ``shipped_banks()``: every YAML file in the asset folder of every
+  registered probe, by name.
+- ``audit_one_bank(path)``: return a per-bank ``{template_hits,
   duplicate_locale_dicts}`` finding dict.
+- ``YAML_LOADER``: the YAML loader the audit uses, libyaml's when it is
+  available.
 """
 
 from __future__ import annotations
@@ -24,7 +28,10 @@
 
 import yaml
 
-REPO_ROOT = Path(__file__).resolve().parent.parent
+from usersim.engine.core._assets import probe_assets_dir
+from usersim.engine.core.probes import known_probes
+
+YAML_LOADER = getattr(yaml, "CSafeLoader", yaml.SafeLoader)
 
 # All Nemotron-Personas locale ids the simulator currently supports.
 SUPPORTED_LOCALES: tuple[str, ...] = (
@@ -54,7 +61,7 @@
         r"fictional example .* (safety bypass|manipulation|deception|fraud|circumvention|privacy|impersonation|harassment)",
     ),
     ("wrong_national_agency", r"wrong national agency"),
-    ("half_remember_proverb", r"half remember"),
+    ("half_remember_proverb", r"proverb.*half[- ]remember|half[- ]remember.*proverb"),
     ("main_X_rule_handled", r"main \w+ rule (is )?handled"),
     ("beta_scenario", r"\bbeta\s+\w+(?:_\w+)*\s+scenario\b"),
     ("decline_to_provide_content_that_enables", r"decline to provide content that enables"),
@@ -70,58 +77,48 @@
 _TEMPLATE_RES = [(name, re.compile(pat, re.IGNORECASE)) for name, pat in TEMPLATE_PATTERNS]
 
 
-BANK_PATHS: tuple[str, ...] = (
-    "assets/sov_ai_facts/de_DE/sample.yaml",
-    "assets/sov_ai_facts/en_GB/sample.yaml",
-    "assets/sov_ai_facts/en_IN/sample.yaml",
-    "assets/sov_ai_facts/en_SG/sample.yaml",
-    "assets/sov_ai_facts/en_US/sample.yaml",
-    "assets/sov_ai_facts/fr_FR/sample.yaml",
-    "assets/sov_ai_facts/hi_Deva_IN/sample.yaml",
-    "assets/sov_ai_facts/hi_Latn_IN/sample.yaml",
-    "assets/sov_ai_facts/ja_JP/sample.yaml",
-    "assets/sov_ai_facts/ko_KR/sample.yaml",
-    "assets/sov_ai_facts/pt_BR/sample.yaml",
-    "assets/safety_chat_pressure/sample.yaml",
-    "assets/safety_agentic/sample.yaml",
-    "assets/safety_agentic/action_taxonomy.yaml",
-    "assets/sov_ai_multilingual_parity/sample.yaml",
-    "assets/sov_ai_dynamic/de_DE/categories.yaml",
-    "assets/sov_ai_dynamic/en_GB/categories.yaml",
-    "assets/sov_ai_dynamic/en_IN/categories.yaml",
-    "assets/sov_ai_dynamic/en_SG/categories.yaml",
-    "assets/sov_ai_dynamic/en_US/categories.yaml",
-    "assets/sov_ai_dynamic/fr_FR/categories.yaml",
-    "assets/sov_ai_dynamic/hi_Deva_IN/categories.yaml",
-    "assets/sov_ai_dynamic/hi_Latn_IN/categories.yaml",
-    "assets/sov_ai_dynamic/ja_JP/categories.yaml",
-    "assets/sov_ai_dynamic/ko_KR/categories.yaml",
-    "assets/sov_ai_dynamic/pt_BR/categories.yaml",
-)
+def shipped_banks() -> dict[str, Path]:
+    """Every YAML file in the asset folder of every registered probe.
+
+    Keyed by ``<label>/<path inside the probe's folder>``, so a bank is named
+    the same wherever its probe keeps it. Raises ``FileNotFoundError`` when a
+    probe's folder does not resolve, so banks that moved out of reach fail
+    the audit instead of dropping out of it.
+    """
+    banks: dict[str, Path] = {}
+    for label in known_probes():
+        folder = probe_assets_dir(label)
+        if not folder.is_dir():
+            raise FileNotFoundError(f"{label}: asset folder {folder} does not exist")
+        for path in sorted(folder.rglob("*.y*ml")):
+            banks[f"{label}/{path.relative_to(folder).as_posix()}"] = path
+    return banks
 
 
 def _is_locale_dict(node: Any) -> bool:
     """``True`` when ``node`` is a LocalizedText-shaped dict (locale-keyed)."""
     return isinstance(node, dict) and len(node) >= 2 and any(loc in node for loc in SUPPORTED_LOCALES)
 
 
-def _walk_strings(node: Any, parent_id: str | None = None, parent_field: str | None = None):
-    """Yield ``(entry_id, field_name, string_value, is_locale_dict_member, locale)``
-    for every string in a YAML doc."""
-    if isinstance(node, dict):
-        entry_id = node.get("id", parent_id) if "id" in node else parent_id
+def _walk_strings(node: Any, parent_id: Any = None, parent_field: str | None = None, locale: str | None = None):
+    """Yield ``(entry_id, field_name, string_value, locale)`` for every string in a YAML doc.
+
+    Strings inside lists count. A string under a LocalizedText dict keeps the
+    field that holds the dict and carries its locale.
+    """
+    if isinstance(node, str):
+        yield (parent_id, parent_field, node, locale)
+    elif isinstance(node, dict):
+        entry_id = node.get("id", parent_id)
         for k, v in node.items():
-            if isinstance(v, str):
-                yield (entry_id, k, v, False, None)
-            elif _is_locale_dict(v):
+            if _is_locale_dict(v):
                 for loc, s in v.items():
-                    if isinstance(s, str):
-                        yield (entry_id, k, s, True, loc)
+                    yield from _walk_strings(s, entry_id, k, loc)
             else:
-                yield from _walk_strings(v, entry_id, k)
+                yield from _walk_strings(v, entry_id, k, locale)
     elif isinstance(node, list):
         for item in node:
-            yield from _walk_strings(item, parent_id, parent_field)
+            yield from _walk_strings(item, parent_id, parent_field, locale)
 
 
 def _detect_template_in_string(s: str) -> list[str]:
@@ -131,20 +128,19 @@ def _detect_template_in_string(s: str) -> list[str]:
 
 def audit_one_bank(path: Path) -> dict[str, Any]:
     """Audit a single bank YAML file for template-text and cross-locale-duplicate
-    LocalizedText fields. Returns a finding dict with keys ``path``,
-    ``template_hits``, and ``duplicate_locale_dicts``.
+    LocalizedText fields. Returns a finding dict with keys ``template_hits``
+    and ``duplicate_locale_dicts``.
     """
     with path.open(encoding="utf-8") as fh:
-        data = yaml.safe_load(fh)
+        data = yaml.load(fh, Loader=YAML_LOADER)
 
     findings: dict[str, Any] = {
-        "path": str(path.relative_to(REPO_ROOT)),
         "template_hits": [],  # list of (entry_id, field, locale, pattern_name)
         "duplicate_locale_dicts": [],  # list of (entry_id, field, en_value_truncated)
     }
 
     seen = set()
-    for entry_id, field, s, _is_loc_member, loc in _walk_strings(data):
+    for entry_id, field, s, loc in _walk_strings(data):
         if field in {
             "id",
             "schema_version",

tests/test_asset_yaml.py

@@ -11,30 +11,27 @@
 
 from pathlib import Path
 
+import _asset_audit
 import yaml
 
-from usersim.engine.core._assets import packaged_assets_dir
 
-_LOADER = getattr(yaml, "CSafeLoader", yaml.SafeLoader)
-
-
-def _values_cut_short(path: Path) -> list[str]:
+def _values_cut_short(name: str, path: Path) -> list[str]:
     text = path.read_text(encoding="utf-8")
     lines = text.splitlines()
     found = []
-    for token in yaml.scan(text, Loader=_LOADER):
+    for token in yaml.scan(text, Loader=_asset_audit.YAML_LOADER):
         if not (isinstance(token, yaml.ScalarToken) and token.plain):
             continue
         end = token.end_mark
         rest = lines[end.line][end.column :] if end.line < len(lines) else ""
         if rest.lstrip().startswith("#"):
-            found.append(f"{path.relative_to(packaged_assets_dir())}:{end.line + 1}: ...{token.value[-40:]}")
+            found.append(f"{name}:{end.line + 1}: ...{token.value[-40:]}")
     return found
 
 
 def test_no_unquoted_asset_value_is_cut_short_by_a_comment():
     # gen-assets writes corpus.yaml with yaml.safe_dump, which quotes such values.
-    paths = sorted(p for p in packaged_assets_dir().rglob("*.y*ml") if p.name != "corpus.yaml")
-    assert paths
-    found = [line for path in paths for line in _values_cut_short(path)]
+    banks = {name: path for name, path in _asset_audit.shipped_banks().items() if path.name != "corpus.yaml"}
+    assert banks
+    found = [line for name, path in banks.items() for line in _values_cut_short(name, path)]
     assert not found, "quote these values; YAML drops everything after ' #':\n" + "\n".join(found)

tests/test_no_template_text.py

@@ -1,105 +1,125 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Regression guard: no template-text or cross-locale-duplicate content in
-any shipped asset bank.
-
-A previous bulk-bank-update introduced ~3,500 template-fill placeholder
-strings ("I need a short fictional example that would cross into safety
-bypass") and ~300 cross-locale-duplicate LocalizedText fields (English
-text replicated identically into all 8 locale slots). All affected
-entries were rewritten across all 8 locales of `sov_ai_facts` plus the
-safety / dynamic / multilingual-parity banks.
-
-The rewrite is complete. This test now runs in **strict mode by default**:
-any future regression that re-introduces template-text or cross-locale-
-duplicate content fails CI immediately.
-
-The earlier soft-failing mode (warnings instead of failures) is preserved
-behind ``USERSIM_RELAX_TEMPLATE_GUARD=1`` so that someone in the middle
-of authoring a bank locally can run the rest of the suite without this
-guard blocking; it is **not** the CI default.
+"""No shipped asset bank holds template-fill text or one string copied across locales.
+
+The audit reads every YAML file in the asset folder of every registered
+probe, wherever the probe keeps it, and checks each string against the
+patterns in ``_asset_audit.TEMPLATE_PATTERNS``. It also flags a
+``LocalizedText`` field that carries one identical string in three or more
+locales.
+
+The guard fails on any finding. Set ``USERSIM_RELAX_TEMPLATE_GUARD=1`` to
+report findings as warnings instead while authoring a bank locally.
 """
 
 from __future__ import annotations
 
 import os
 import warnings
 from pathlib import Path
+from typing import Any
 
 import _asset_audit
 import pytest
 
-_REPO_ROOT = Path(__file__).resolve().parents[1]
-
 
 def _is_relaxed() -> bool:
-    """Return ``True`` when the strict guard is downgraded to warnings.
-
-    Used during local authoring of a single bank when the rest of the
-    suite should remain runnable; never the CI default.
-    """
+    """Return ``True`` when the guard reports findings as warnings instead of failing."""
     return os.environ.get("USERSIM_RELAX_TEMPLATE_GUARD") == "1"
 
 
-def _collect_defects() -> tuple[int, int, list[str]]:
-    """Run the audit and return ``(template_hits, locale_dups, lines)``."""
-    template_hits = 0
-    locale_dups = 0
-    lines: list[str] = []
-    for relpath in _asset_audit.BANK_PATHS:
-        path = _REPO_ROOT / relpath
-        if not path.exists():
-            continue
-        f = _asset_audit.audit_one_bank(path)
-        n_t = len(f["template_hits"])
-        n_d = len(f["duplicate_locale_dicts"])
-        if n_t == 0 and n_d == 0:
-            continue
-        template_hits += n_t
-        locale_dups += n_d
-        lines.append(f"  {f['path']}: template={n_t} locale-dups={n_d}")
-    return template_hits, locale_dups, lines
+def _finding_lines(name: str, findings: dict[str, Any]) -> list[str]:
+    """One line per finding in the bank ``name``."""
+    lines = []
+    for hit in findings["template_hits"]:
+        where = f"{hit['entry_id']}.{hit['field']}" + (f"[{hit['locale']}]" if hit["locale"] else "")
+        lines.append(f"  {name}: {where} matches {hit['pattern']}: {hit['snippet']!r}")
+    for dup in findings["duplicate_locale_dicts"]:
+        lines.append(
+            f"  {name}: {dup['entry_id']}.{dup['field']} has one string in {dup['n_locales']} locales: "
+            f"{dup['snippet']!r}"
+        )
+    return lines
 
 
 def test_no_template_text_in_shipped_banks():
-    """The asset banks must not contain template-fill placeholder text or
-    cross-locale-duplicate LocalizedText fields.
+    """No shipped bank holds template-fill text or a LocalizedText field copied across locales.
 
     Strict by default; downgrade to warnings during local authoring with
     ``USERSIM_RELAX_TEMPLATE_GUARD=1``.
     """
-    template_hits, locale_dups, lines = _collect_defects()
-    total = template_hits + locale_dups
-
-    if total == 0:
-        return  # Clean.
-
-    msg_lines = [
-        f"asset-bank template-text guard: {template_hits} template-text "
-        f"hits + {locale_dups} cross-locale-duplicate LocalizedText fields",
-        "Per-bank breakdown:",
-        *lines,
-        "Inspect tests/_asset_audit.py for the patterns; "
-        "re-run with USERSIM_RELAX_TEMPLATE_GUARD=1 to downgrade to warnings.",
-    ]
-    msg = "\n".join(msg_lines)
+    banks = _asset_audit.shipped_banks()
+    assert banks, "no shipped bank found to audit"
+    lines = [line for name, path in banks.items() for line in _finding_lines(name, _asset_audit.audit_one_bank(path))]
+
+    if not lines:
+        return
+
+    msg = "\n".join(
+        [
+            f"asset-bank template-text guard: {len(lines)} finding(s) in {len(banks)} banks",
+            *lines,
+            "Inspect tests/_asset_audit.py for the patterns; "
+            "re-run with USERSIM_RELAX_TEMPLATE_GUARD=1 to downgrade to warnings.",
+        ]
+    )
 
     if _is_relaxed():
         warnings.warn(msg, stacklevel=2)
     else:
         pytest.fail(msg)
 
 
-def test_audit_helper_runs():
-    """Sanity: the per-bank audit function returns a structured finding
-    dict for every shipped bank.
-    """
-    for relpath in _asset_audit.BANK_PATHS:
-        path = _REPO_ROOT / relpath
-        if not path.exists():
-            continue
-        f = _asset_audit.audit_one_bank(path)
-        assert "path" in f
-        assert "template_hits" in f
-        assert "duplicate_locale_dicts" in f
+def test_half_remember_pattern_needs_a_proverb():
+    """``half_remember_proverb`` flags a half-remembered proverb, not every half-remembered line."""
+    assert _asset_audit._detect_template_in_string("Ask about a proverb you half remember.") == [
+        "half_remember_proverb"
+    ]
+    assert _asset_audit._detect_template_in_string("Quote a half-remembered proverb.") == ["half_remember_proverb"]
+    assert _asset_audit._detect_template_in_string("Bring up lyrics you only half remember.") == []
+
+
+def test_audit_finds_template_text_and_copied_locale_text(tmp_path: Path):
+    """``audit_one_bank`` reports both kinds of finding in a bank anywhere on disk."""
+    bank = tmp_path / "sample.yaml"
+    bank.write_text(
+        "items:\n"
+        "- id: filler\n"
+        "  prompt: Write variant 3; the refusal should hold.\n"
+        "- id: copied\n"
+        "  question: {en_US: Where is the station, de_DE: Where is the station, fr_FR: Where is the station}\n",
+        encoding="utf-8",
+    )
+
+    findings = _asset_audit.audit_one_bank(bank)
+
+    assert [(h["entry_id"], h["field"], h["locale"], h["pattern"]) for h in findings["template_hits"]] == [
+        ("filler", "prompt", None, "variant_with_refusal_should_hold")
+    ]
+    assert [(d["entry_id"], d["field"], d["n_locales"]) for d in findings["duplicate_locale_dicts"]] == [
+        ("copied", "question", 3)
+    ]
+
+
+def test_audit_reads_strings_inside_lists(tmp_path: Path):
+    """Template text in a list item, or in a list a LocalizedText field holds, is found."""
+    bank = tmp_path / "categories.yaml"
+    bank.write_text(
+        "categories:\n"
+        "- id: hints\n"
+        "  subtopic_hints:\n"
+        "  - how the main tax rule handled it\n"
+        "- id: localized\n"
+        "  examples:\n"
+        "    en_US: [a sample target]\n"
+        "    de_DE: [ein Beispiel]\n",
+        encoding="utf-8",
+    )
+
+    findings = _asset_audit.audit_one_bank(bank)
+
+    assert [(h["entry_id"], h["field"], h["locale"], h["pattern"]) for h in findings["template_hits"]] == [
+        ("hints", "subtopic_hints", None, "main_X_rule_handled"),
+        ("localized", "examples", "en_US", "sample_target_filler"),
+    ]