681 lines
29 KiB
Python
681 lines
29 KiB
Python
import ast
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import re
|
|
import shlex
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from scripts import run_offline_checks
|
|
from scripts import run_training_case_package as training_cli
|
|
from src.services.training import TrainingCaseService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
DOCUMENTATION_CONSISTENCY_TEST = ROOT / "tests" / "unit" / "test_documentation_consistency.py"
|
|
DOC_EXAMPLE = ROOT / "docs" / "examples" / "training_case_package_smoke_m6.md"
|
|
PRODUCT_README = ROOT / "docs" / "training_case_engine_m1_m6_product_readme.md"
|
|
README = ROOT / "README.md"
|
|
FIXTURE = ROOT / "tests" / "fixtures" / "training_case_package_m6.json"
|
|
FIXTURE_REF = "tests/fixtures/training_case_package_m6.json"
|
|
OFFLINE_CHECK_COMMAND = ".venv/bin/python scripts/run_offline_checks.py"
|
|
OFFLINE_CHECK_NAME = "training_case_package"
|
|
OFFLINE_CHECK_DOC_NAME = f"`{OFFLINE_CHECK_NAME}`"
|
|
PACKAGE_SMOKE_SCRIPT = "scripts/run_training_case_package.py"
|
|
PACKAGE_SMOKE_COMMAND = f".venv/bin/python {PACKAGE_SMOKE_SCRIPT}"
|
|
DASHBOARD_SYNTAX_CHECK_NAME = "dashboard_js_syntax"
|
|
DASHBOARD_SYNTAX_CHECK_PATH = "src/web/app.js"
|
|
OFFLINE_CHECK_REGISTRY_NAMES = (
|
|
"unit_tests",
|
|
OFFLINE_CHECK_NAME,
|
|
"migration_static_checks",
|
|
"script_static_checks",
|
|
DASHBOARD_SYNTAX_CHECK_NAME,
|
|
)
|
|
CURRENT_TRAINING_DOC_MATURITY_END = 28
|
|
CURRENT_TRAINING_DOC_MATURITY_MARKER = f"M1-M{CURRENT_TRAINING_DOC_MATURITY_END}"
|
|
CURRENT_TRAINING_DOC_MATURITY_SPAN = f"through M{CURRENT_TRAINING_DOC_MATURITY_END}"
|
|
TRAINING_DOCS_CLI_TEST_CLASS = "TrainingCaseDocsCliM10Tests"
|
|
DOCUMENTATION_CONSISTENCY_TEST_CLASS = "DocumentationConsistencyTests"
|
|
PUBLIC_FIELD = "answer" + "_key"
|
|
PRIVATE_MATERIAL_PHRASES = ("private solution material", "internal solution details")
|
|
PUBLIC_PACKAGE_DOCS = (DOC_EXAMPLE, README)
|
|
PACKAGE_SMOKE_SURFACE_DOCS = (DOC_EXAMPLE, README, PRODUCT_README)
|
|
PUBLIC_PACKAGE_REQUIRED_TEXT = (
|
|
".venv/bin/python scripts/run_training_case_package.py",
|
|
FIXTURE_REF,
|
|
"--input",
|
|
"--output",
|
|
"import/export smoke",
|
|
"public export",
|
|
"public package",
|
|
"public output",
|
|
"internal solution details",
|
|
)
|
|
PACKAGE_IMPORT_EXPORT_SURFACE_TEXT = (
|
|
"import",
|
|
"export",
|
|
"public package",
|
|
"public export",
|
|
"public output",
|
|
)
|
|
PRODUCT_PACKAGE_WORDING = (
|
|
CURRENT_TRAINING_DOC_MATURITY_MARKER,
|
|
"offline checks registry smoke",
|
|
"public package",
|
|
"public export",
|
|
"public output",
|
|
"private solution material",
|
|
"internal solution details",
|
|
)
|
|
PRIVATE_DOC_DRIFT_PHRASES = ("private " + "answer", "answer " + "detail")
|
|
PRIVATE_FIXTURE_DRIFT_PHRASES = PRIVATE_MATERIAL_PHRASES + PRIVATE_DOC_DRIFT_PHRASES
|
|
RESTRICTED_PUBLIC_TEXT = (
|
|
"data" + "base",
|
|
"data" + "base_url",
|
|
"post" + "gresql",
|
|
"psy" + "copg",
|
|
"pass" + "word",
|
|
"pass" + "wd",
|
|
"tok" + "en",
|
|
"sec" + "ret",
|
|
"api" + "_key",
|
|
)
|
|
TRAINING_M24_M29_PACKAGE_DOCS_TEST_INVENTORY = {
|
|
"package_smoke_privacy": (
|
|
"test_public_privacy_guard_rejects_solution_wording_drift",
|
|
"test_public_fixture_schema_contract_is_minimal_and_demo_safe",
|
|
"test_fixture_is_compatible_with_public_service_and_cli_export",
|
|
"test_smoke_docs_and_product_readme_share_fixture_and_public_export_contract",
|
|
"test_example_command_uses_supported_cli_flags_and_fixture",
|
|
"test_example_documents_cli_summary_keys_and_output_semantics",
|
|
"test_cli_error_summary_uses_public_output_target_code_without_runtime_leaks",
|
|
"test_readme_and_product_docs_advertise_public_offline_smoke_contract",
|
|
"test_product_readme_uses_current_maturity_and_public_private_package_language",
|
|
"test_public_package_smoke_docs_lock_import_export_privacy_language",
|
|
"test_m1_m28_package_smoke_docs_cover_static_import_export_surface",
|
|
),
|
|
"offline_registry_coverage": (
|
|
"test_offline_check_registry_name_matches_package_smoke_docs",
|
|
"test_offline_checks_registry_runs_training_package_smoke_entry",
|
|
"test_training_docs_name_broader_offline_check_path",
|
|
),
|
|
"m1_m28_maturity_wording": (
|
|
"test_product_readme_uses_current_maturity_and_public_private_package_language",
|
|
"test_m1_m28_package_smoke_docs_cover_static_import_export_surface",
|
|
),
|
|
}
|
|
TRAINING_M33_REQUIRED_INVENTORY_GROUPS = (
|
|
"package_smoke_docs_command_help",
|
|
"fixture_public_shape",
|
|
"fixture_private_solution_absence",
|
|
"service_import_export_public_output",
|
|
"cli_smoke_export_import_summary",
|
|
"readme_product_docs_offline_registry_refs",
|
|
)
|
|
TRAINING_M33_PACKAGE_FIXTURE_IMPORT_EXPORT_PRIVACY_INVENTORY = {
|
|
"package_smoke_docs_command_help": (
|
|
"test_example_command_uses_supported_cli_flags_and_fixture",
|
|
"test_public_package_smoke_docs_lock_import_export_privacy_language",
|
|
"test_m1_m28_package_smoke_docs_cover_static_import_export_surface",
|
|
),
|
|
"fixture_public_shape": (
|
|
"test_public_fixture_schema_contract_is_minimal_and_demo_safe",
|
|
),
|
|
"fixture_private_solution_absence": (
|
|
"test_public_privacy_guard_rejects_solution_wording_drift",
|
|
"test_public_fixture_schema_contract_is_minimal_and_demo_safe",
|
|
"test_fixture_is_compatible_with_public_service_and_cli_export",
|
|
),
|
|
"service_import_export_public_output": (
|
|
"test_fixture_is_compatible_with_public_service_and_cli_export",
|
|
),
|
|
"cli_smoke_export_import_summary": (
|
|
"test_fixture_is_compatible_with_public_service_and_cli_export",
|
|
"test_example_documents_cli_summary_keys_and_output_semantics",
|
|
"test_cli_error_summary_uses_public_output_target_code_without_runtime_leaks",
|
|
),
|
|
"readme_product_docs_offline_registry_refs": (
|
|
"test_smoke_docs_and_product_readme_share_fixture_and_public_export_contract",
|
|
"test_readme_and_product_docs_advertise_public_offline_smoke_contract",
|
|
"test_offline_check_registry_name_matches_package_smoke_docs",
|
|
"test_offline_checks_registry_runs_training_package_smoke_entry",
|
|
"test_training_docs_name_broader_offline_check_path",
|
|
"test_training_readmes_name_dashboard_privacy_and_offline_registry_coverage",
|
|
),
|
|
}
|
|
|
|
|
|
def _example_command_parts() -> list[str]:
|
|
text = DOC_EXAMPLE.read_text(encoding="utf-8")
|
|
match = re.search(r"```bash\n(?P<command>.*?)\n```", text, flags=re.DOTALL)
|
|
if match is None:
|
|
raise AssertionError("missing bash command block")
|
|
command = re.sub(r"\\\n\s*", " ", match.group("command").strip())
|
|
return shlex.split(command)
|
|
|
|
|
|
def _fixture_payload() -> dict:
|
|
return json.loads(FIXTURE.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _package_doc_text(path: Path) -> str:
|
|
text = path.read_text(encoding="utf-8")
|
|
if path != README:
|
|
return text
|
|
|
|
lines = [line for line in text.splitlines() if "run_training_case_package.py" in line]
|
|
if not lines:
|
|
raise AssertionError("README is missing the package smoke command")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _package_smoke_flags_from_source() -> tuple[str, ...]:
|
|
source = Path(training_cli.__file__).read_text(encoding="utf-8")
|
|
return tuple(re.findall(r'add_argument\("(?P<flag>--[a-z-]+)"', source))
|
|
|
|
|
|
def _test_method_names_from_file(path: Path, class_name: str) -> tuple[str, ...]:
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
for node in tree.body:
|
|
if isinstance(node, ast.ClassDef) and node.name == class_name:
|
|
return tuple(
|
|
item.name
|
|
for item in node.body
|
|
if isinstance(item, ast.FunctionDef) and item.name.startswith("test_")
|
|
)
|
|
raise AssertionError(f"{path} is missing {class_name}")
|
|
|
|
|
|
def _walk_public_text(value) -> list[str]:
|
|
if isinstance(value, dict):
|
|
entries = []
|
|
for key, item in value.items():
|
|
entries.append(str(key))
|
|
entries.extend(_walk_public_text(item))
|
|
return entries
|
|
if isinstance(value, list):
|
|
entries = []
|
|
for item in value:
|
|
entries.extend(_walk_public_text(item))
|
|
return entries
|
|
if isinstance(value, str):
|
|
return [value]
|
|
return []
|
|
|
|
|
|
class TrainingCaseDocsCliM10Tests(unittest.TestCase):
|
|
def assertNoPrivateFixtureMaterial(self, payload: dict) -> None:
|
|
serialized = json.dumps(payload, sort_keys=True)
|
|
lower_serialized = serialized.lower()
|
|
|
|
self.assertNotIn(PUBLIC_FIELD, lower_serialized)
|
|
self.assertIsNone(re.search(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", serialized))
|
|
for phrase in PRIVATE_FIXTURE_DRIFT_PHRASES:
|
|
self.assertNotIn(phrase, lower_serialized)
|
|
for text in _walk_public_text(payload):
|
|
lower_text = text.lower()
|
|
for fragment in RESTRICTED_PUBLIC_TEXT:
|
|
with self.subTest(text=text, fragment=fragment):
|
|
self.assertNotIn(fragment, lower_text)
|
|
|
|
def test_public_privacy_guard_rejects_solution_wording_drift(self) -> None:
|
|
polluted_payloads = [
|
|
{"items": [{"prompt": "Do not expose " + ("private " + "answer") + " wording."}]},
|
|
{"items": [{"prompt": "Do not expose internal solution details wording."}]},
|
|
]
|
|
|
|
for payload in polluted_payloads:
|
|
with self.subTest(payload=payload):
|
|
with self.assertRaises(AssertionError):
|
|
self.assertNoPrivateFixtureMaterial(payload)
|
|
|
|
def test_public_fixture_schema_contract_is_minimal_and_demo_safe(self) -> None:
|
|
payload = _fixture_payload()
|
|
|
|
self.assertEqual(set(payload), {"version", "metadata", "items"})
|
|
self.assertEqual(payload["version"], 1)
|
|
self.assertEqual(payload["metadata"], {"source": "m6_public_fixture", "purpose": "offline_smoke"})
|
|
self.assertEqual(len(payload["items"]), 1)
|
|
|
|
item = payload["items"][0]
|
|
self.assertEqual(
|
|
set(item),
|
|
{
|
|
"external_id",
|
|
"title",
|
|
"scenario",
|
|
"skill_tags",
|
|
"difficulty",
|
|
"answer",
|
|
"rubric",
|
|
"tasks",
|
|
"metadata",
|
|
},
|
|
)
|
|
self.assertEqual(item["external_id"], "m6-sweep-public-001")
|
|
self.assertEqual(item["title"], "Liquidity sweep reading practice")
|
|
self.assertEqual(item["skill_tags"], ["sweep", "context_reading"])
|
|
self.assertEqual(item["difficulty"], "easy")
|
|
self.assertEqual(item["answer"], {})
|
|
self.assertEqual(
|
|
item["rubric"],
|
|
[
|
|
{"category": "context", "max_score": 3},
|
|
{"category": "risk_story", "max_score": 2},
|
|
],
|
|
)
|
|
self.assertEqual(len(item["tasks"]), 1)
|
|
self.assertEqual(item["metadata"], {"chapter_ref": "public_training_m6", "demo": True})
|
|
|
|
scenario = item["scenario"]
|
|
self.assertEqual(
|
|
set(scenario),
|
|
{
|
|
"instrument_id",
|
|
"timeframe",
|
|
"case_type",
|
|
"start_ts",
|
|
"decision_ts",
|
|
"end_ts",
|
|
"source_event_type",
|
|
"source_event_id",
|
|
"source_event_refs",
|
|
"chapter_ref",
|
|
},
|
|
)
|
|
self.assertEqual(scenario["instrument_id"], 1)
|
|
self.assertEqual(scenario["timeframe"], "15m")
|
|
self.assertEqual(scenario["case_type"], "sweep")
|
|
self.assertEqual(scenario["start_ts"], "2026-04-20T13:00:00Z")
|
|
self.assertEqual(scenario["decision_ts"], "2026-04-20T14:00:00Z")
|
|
self.assertEqual(scenario["end_ts"], "2026-04-20T15:00:00Z")
|
|
self.assertEqual(scenario["source_event_type"], "sweep_event")
|
|
self.assertEqual(scenario["source_event_id"], 61)
|
|
self.assertEqual(scenario["source_event_refs"], [{"kind": "sweep_event", "id": 61}])
|
|
self.assertEqual(scenario["chapter_ref"], "public_training_m6")
|
|
|
|
task = item["tasks"][0]
|
|
self.assertEqual(
|
|
set(task),
|
|
{
|
|
"task_type",
|
|
"skill_tag",
|
|
"prompt",
|
|
"required_annotation_types",
|
|
"max_score",
|
|
"sort_order",
|
|
"meta",
|
|
},
|
|
)
|
|
self.assertEqual(task["task_type"], "short_answer")
|
|
self.assertEqual(task["skill_tag"], "sweep")
|
|
self.assertEqual(task["prompt"], "Describe what needs confirmation before marking this as a usable sweep.")
|
|
self.assertEqual(task["required_annotation_types"], ["text"])
|
|
self.assertEqual(task["max_score"], 5)
|
|
self.assertEqual(task["sort_order"], 0)
|
|
self.assertEqual(task["meta"], {"public_hint": "Focus on context and confirmation."})
|
|
self.assertNoPrivateFixtureMaterial(payload)
|
|
|
|
def test_fixture_is_compatible_with_public_service_and_cli_export(self) -> None:
|
|
payload = _fixture_payload()
|
|
service = TrainingCaseService(repository=training_cli._OfflineTrainingRepository())
|
|
|
|
import_result = service.import_cases(payload)
|
|
export_result = service.export_cases(limit=100)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
output_path = Path(tmpdir) / "training_case_public_export.json"
|
|
summary = training_cli.run_fixture_smoke(FIXTURE, output_path=output_path)
|
|
cli_export = json.loads(output_path.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(import_result["version"], 1)
|
|
self.assertEqual(import_result["imported"], 1)
|
|
self.assertEqual(len(import_result["items"]), 1)
|
|
self.assertEqual(set(import_result["items"][0]), {"id", "title", "scenario", "skill_tags", "difficulty", "status", "external_id"})
|
|
self.assertEqual(summary["input"], FIXTURE_REF)
|
|
self.assertEqual(summary["imported"], 1)
|
|
self.assertEqual(summary["exported"], 1)
|
|
self.assertEqual(export_result, cli_export)
|
|
self.assertEqual(set(export_result), {"version", "metadata", "items"})
|
|
self.assertEqual(export_result["metadata"], {"item_count": 1})
|
|
|
|
exported_item = export_result["items"][0]
|
|
fixture_item = payload["items"][0]
|
|
self.assertEqual(
|
|
set(exported_item),
|
|
{
|
|
"id",
|
|
"title",
|
|
"scenario",
|
|
"skill_tags",
|
|
"difficulty",
|
|
"status",
|
|
"rubric",
|
|
"tasks",
|
|
"metadata",
|
|
},
|
|
)
|
|
self.assertEqual(exported_item["title"], fixture_item["title"])
|
|
self.assertEqual(set(exported_item["skill_tags"]), set(fixture_item["skill_tags"]))
|
|
self.assertEqual(
|
|
exported_item["metadata"],
|
|
{"chapter_ref": "public_training_m6", "demo": True, "external_id": fixture_item["external_id"]},
|
|
)
|
|
self.assertEqual(exported_item["rubric"], fixture_item["rubric"])
|
|
self.assertEqual(exported_item["scenario"], fixture_item["scenario"])
|
|
self.assertEqual(set(exported_item["tasks"][0]), set(fixture_item["tasks"][0]) | {"status"})
|
|
self.assertEqual(exported_item["tasks"][0]["status"], "active")
|
|
for key, value in fixture_item["tasks"][0].items():
|
|
with self.subTest(task_field=key):
|
|
self.assertEqual(exported_item["tasks"][0][key], value)
|
|
self.assertNotIn("answer", exported_item)
|
|
self.assertNoPrivateFixtureMaterial({"imported": import_result, "exported": export_result})
|
|
|
|
def test_smoke_docs_and_product_readme_share_fixture_and_public_export_contract(self) -> None:
|
|
for path in (DOC_EXAMPLE, PRODUCT_README):
|
|
text = path.read_text(encoding="utf-8")
|
|
lower_text = text.lower()
|
|
fixture_refs = set(re.findall(r"tests/fixtures/[A-Za-z0-9_./-]+\.json", text))
|
|
|
|
with self.subTest(path=path.name):
|
|
self.assertEqual(fixture_refs, {FIXTURE_REF})
|
|
self.assertIn(".venv/bin/python scripts/run_training_case_package.py", text)
|
|
self.assertIn("public export", lower_text)
|
|
self.assertTrue(any(phrase in lower_text for phrase in PRIVATE_MATERIAL_PHRASES))
|
|
self.assertNotIn(PUBLIC_FIELD, text)
|
|
|
|
def test_example_command_uses_supported_cli_flags_and_fixture(self) -> None:
|
|
parts = _example_command_parts()
|
|
self.assertEqual(parts[:2], [".venv/bin/python", "scripts/run_training_case_package.py"])
|
|
|
|
parser = training_cli._build_parser()
|
|
supported_flags = {flag for action in parser._actions for flag in action.option_strings}
|
|
used_flags = [part for part in parts[2:] if part.startswith("--")]
|
|
|
|
self.assertEqual(used_flags, ["--input", "--output"])
|
|
self.assertLessEqual(set(used_flags), supported_flags)
|
|
|
|
args = parser.parse_args(parts[2:])
|
|
expected_fixture = training_cli.DEFAULT_FIXTURE.relative_to(ROOT)
|
|
self.assertEqual(Path(args.input), expected_fixture)
|
|
self.assertEqual(Path(args.input), FIXTURE.relative_to(ROOT))
|
|
self.assertEqual(Path(args.output), Path("/tmp/training_case_public_export.json"))
|
|
|
|
def test_example_documents_cli_summary_keys_and_output_semantics(self) -> None:
|
|
doc_text = DOC_EXAMPLE.read_text(encoding="utf-8")
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
output_path = Path(tmpdir) / "training_case_public_export.json"
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
exit_code = training_cli.main(
|
|
["--input", str(FIXTURE.relative_to(ROOT)), "--output", str(output_path)]
|
|
)
|
|
|
|
summary = json.loads(stdout.getvalue())
|
|
exported = json.loads(output_path.read_text(encoding="utf-8"))
|
|
|
|
expected_keys = {"exported", "imported", "input", "output", "passed"}
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertEqual(set(summary), expected_keys)
|
|
self.assertTrue(summary["passed"])
|
|
self.assertEqual(summary["imported"], 1)
|
|
self.assertEqual(summary["exported"], 1)
|
|
self.assertTrue(summary["output"].endswith("training_case_public_export.json"))
|
|
self.assertEqual(summary["input"], FIXTURE_REF)
|
|
self.assertEqual(exported["metadata"], {"item_count": 1})
|
|
self.assertNotIn(PUBLIC_FIELD, json.dumps(exported, sort_keys=True))
|
|
|
|
for key in sorted(expected_keys):
|
|
with self.subTest(key=key):
|
|
self.assertIn(f"`{key}`", doc_text)
|
|
|
|
def test_cli_error_summary_uses_public_output_target_code_without_runtime_leaks(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
output_path = Path(tmpdir) / "training_case_public_export_dir"
|
|
output_path.mkdir()
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
|
exit_code = training_cli.main(
|
|
["--input", str(FIXTURE.relative_to(ROOT)), "--output", str(output_path)]
|
|
)
|
|
|
|
raw_output = stdout.getvalue()
|
|
line = raw_output.strip()
|
|
summary = json.loads(line)
|
|
|
|
self.assertEqual(exit_code, 2)
|
|
self.assertEqual(stderr.getvalue(), "")
|
|
self.assertEqual(raw_output.count("\n"), 1)
|
|
self.assertEqual(list(summary), ["passed", "input", "output", "error"])
|
|
self.assertFalse(summary["passed"])
|
|
self.assertEqual(summary["input"], FIXTURE_REF)
|
|
self.assertEqual(summary["output"], output_path.name)
|
|
self.assertEqual(summary["error"], "output_path_is_directory")
|
|
self.assertNotIn(str(output_path), line)
|
|
self.assertNotIn(str(Path(tmpdir)), line)
|
|
self.assertIsNone(re.search(r'(?<!\\)"(?:input|output)": "/', line))
|
|
for fragment in ("[Errno", "Is a directory", "Traceback"):
|
|
with self.subTest(fragment=fragment):
|
|
self.assertNotIn(fragment.lower(), line.lower())
|
|
self.assertNotIn(PUBLIC_FIELD, line)
|
|
for phrase in PRIVATE_FIXTURE_DRIFT_PHRASES:
|
|
with self.subTest(phrase=phrase):
|
|
self.assertNotIn(phrase, line.lower())
|
|
|
|
def test_readme_and_product_docs_advertise_public_offline_smoke_contract(self) -> None:
|
|
for path in (README, PRODUCT_README):
|
|
text = path.read_text(encoding="utf-8")
|
|
lower_text = text.lower()
|
|
|
|
with self.subTest(path=path.name):
|
|
self.assertIn(".venv/bin/python scripts/run_training_case_package.py", text)
|
|
self.assertIn(FIXTURE_REF, text)
|
|
self.assertIn("offline package smoke", lower_text)
|
|
self.assertIn("public export", lower_text)
|
|
self.assertTrue(any(phrase in lower_text for phrase in PRIVATE_MATERIAL_PHRASES))
|
|
self.assertNotIn(PUBLIC_FIELD, text)
|
|
|
|
def test_product_readme_uses_current_maturity_and_public_private_package_language(self) -> None:
|
|
text = PRODUCT_README.read_text(encoding="utf-8")
|
|
lower_text = text.lower()
|
|
|
|
for snippet in PRODUCT_PACKAGE_WORDING:
|
|
with self.subTest(snippet=snippet):
|
|
self.assertIn(snippet, text if snippet.startswith("M1-") else lower_text)
|
|
for phrase in PRIVATE_DOC_DRIFT_PHRASES:
|
|
with self.subTest(phrase=phrase):
|
|
self.assertNotIn(phrase, lower_text)
|
|
self.assertNotIn("M1-M21", text)
|
|
self.assertNotIn(PUBLIC_FIELD, text)
|
|
|
|
def test_m24_m29_package_docs_privacy_test_inventory_is_current(self) -> None:
|
|
test_names = set(unittest.TestLoader().getTestCaseNames(type(self)))
|
|
inventory_names = tuple(
|
|
name
|
|
for names in TRAINING_M24_M29_PACKAGE_DOCS_TEST_INVENTORY.values()
|
|
for name in names
|
|
)
|
|
unique_inventory_names = set(inventory_names)
|
|
product_text = PRODUCT_README.read_text(encoding="utf-8")
|
|
|
|
self.assertEqual(len(test_names), 15)
|
|
self.assertEqual(len(inventory_names), 16)
|
|
self.assertEqual(len(unique_inventory_names), 14)
|
|
self.assertEqual(sorted(TRAINING_M24_M29_PACKAGE_DOCS_TEST_INVENTORY), [
|
|
"m1_m28_maturity_wording",
|
|
"offline_registry_coverage",
|
|
"package_smoke_privacy",
|
|
])
|
|
self.assertEqual(
|
|
test_names - unique_inventory_names,
|
|
{"test_m24_m29_package_docs_privacy_test_inventory_is_current"},
|
|
)
|
|
for surface, names in TRAINING_M24_M29_PACKAGE_DOCS_TEST_INVENTORY.items():
|
|
with self.subTest(surface=surface):
|
|
self.assertTrue(names)
|
|
for name in names:
|
|
self.assertIn(name, test_names)
|
|
|
|
self.assertIn(CURRENT_TRAINING_DOC_MATURITY_MARKER, product_text)
|
|
self.assertIn(CURRENT_TRAINING_DOC_MATURITY_SPAN, product_text)
|
|
|
|
docs_cli_test_names = set(
|
|
_test_method_names_from_file(Path(__file__), TRAINING_DOCS_CLI_TEST_CLASS)
|
|
)
|
|
documentation_consistency_test_names = set(
|
|
_test_method_names_from_file(
|
|
DOCUMENTATION_CONSISTENCY_TEST,
|
|
DOCUMENTATION_CONSISTENCY_TEST_CLASS,
|
|
)
|
|
)
|
|
m33_available_test_names = docs_cli_test_names | documentation_consistency_test_names
|
|
m33_inventory_names = tuple(
|
|
name
|
|
for names in TRAINING_M33_PACKAGE_FIXTURE_IMPORT_EXPORT_PRIVACY_INVENTORY.values()
|
|
for name in names
|
|
)
|
|
m33_unique_inventory_names = set(m33_inventory_names)
|
|
|
|
self.assertEqual(
|
|
sorted(TRAINING_M33_PACKAGE_FIXTURE_IMPORT_EXPORT_PRIVACY_INVENTORY),
|
|
sorted(TRAINING_M33_REQUIRED_INVENTORY_GROUPS),
|
|
)
|
|
self.assertEqual(len(m33_inventory_names), 17)
|
|
self.assertEqual(len(m33_unique_inventory_names), 14)
|
|
self.assertIn(
|
|
"test_training_readmes_name_dashboard_privacy_and_offline_registry_coverage",
|
|
documentation_consistency_test_names,
|
|
)
|
|
for surface, names in TRAINING_M33_PACKAGE_FIXTURE_IMPORT_EXPORT_PRIVACY_INVENTORY.items():
|
|
with self.subTest(m33_surface=surface):
|
|
self.assertTrue(names)
|
|
for name in names:
|
|
self.assertIn(name, m33_available_test_names)
|
|
|
|
def test_public_package_smoke_docs_lock_import_export_privacy_language(self) -> None:
|
|
help_text = training_cli._build_parser().format_help()
|
|
|
|
self.assertIn("--input", help_text)
|
|
self.assertIn("--output", help_text)
|
|
self.assertIn("public export", help_text.lower())
|
|
|
|
for path in PUBLIC_PACKAGE_DOCS:
|
|
text = _package_doc_text(path)
|
|
lower_text = text.lower()
|
|
|
|
with self.subTest(path=path.name):
|
|
for snippet in PUBLIC_PACKAGE_REQUIRED_TEXT:
|
|
with self.subTest(snippet=snippet):
|
|
self.assertIn(snippet, text if snippet.startswith(".") or snippet.startswith("--") else lower_text)
|
|
for phrase in PRIVATE_DOC_DRIFT_PHRASES:
|
|
with self.subTest(phrase=phrase):
|
|
self.assertNotIn(phrase, lower_text)
|
|
self.assertNotIn(PUBLIC_FIELD, text)
|
|
|
|
def test_m1_m28_package_smoke_docs_cover_static_import_export_surface(self) -> None:
|
|
product_text = PRODUCT_README.read_text(encoding="utf-8")
|
|
cli_flags = _package_smoke_flags_from_source()
|
|
|
|
self.assertIn(CURRENT_TRAINING_DOC_MATURITY_MARKER, product_text)
|
|
self.assertEqual(cli_flags, ("--input", "--output"))
|
|
|
|
for path in PACKAGE_SMOKE_SURFACE_DOCS:
|
|
text = _package_doc_text(path)
|
|
lower_text = text.lower()
|
|
|
|
with self.subTest(path=path.name):
|
|
self.assertIn(PACKAGE_SMOKE_SCRIPT, text)
|
|
self.assertIn(FIXTURE_REF, text)
|
|
for flag in cli_flags:
|
|
with self.subTest(flag=flag):
|
|
self.assertIn(flag, text)
|
|
for snippet in PACKAGE_IMPORT_EXPORT_SURFACE_TEXT:
|
|
with self.subTest(snippet=snippet):
|
|
self.assertIn(snippet, lower_text)
|
|
|
|
def test_offline_check_registry_name_matches_package_smoke_docs(self) -> None:
|
|
commands = run_offline_checks.build_check_commands(env={})
|
|
names = tuple(name for name, _command in commands)
|
|
commands_by_name = dict(commands)
|
|
matching_commands = [
|
|
(name, command) for name, command in commands if name == OFFLINE_CHECK_NAME
|
|
]
|
|
registry_text = Path(run_offline_checks.__file__).read_text(encoding="utf-8")
|
|
|
|
self.assertEqual(names, OFFLINE_CHECK_REGISTRY_NAMES)
|
|
for check_name in OFFLINE_CHECK_REGISTRY_NAMES:
|
|
with self.subTest(check_name=check_name):
|
|
self.assertEqual(names.count(check_name), 1)
|
|
self.assertEqual(
|
|
matching_commands,
|
|
[(OFFLINE_CHECK_NAME, [sys.executable, PACKAGE_SMOKE_SCRIPT])],
|
|
)
|
|
self.assertEqual(
|
|
commands_by_name[OFFLINE_CHECK_NAME],
|
|
[sys.executable, PACKAGE_SMOKE_SCRIPT],
|
|
)
|
|
self.assertEqual(
|
|
commands_by_name[DASHBOARD_SYNTAX_CHECK_NAME],
|
|
["node", "--check", DASHBOARD_SYNTAX_CHECK_PATH],
|
|
)
|
|
self.assertEqual(registry_text.count(f'"{OFFLINE_CHECK_NAME}"'), 1)
|
|
self.assertIn(f'"{PACKAGE_SMOKE_SCRIPT}"', registry_text)
|
|
|
|
for path in (README, DOC_EXAMPLE, PRODUCT_README):
|
|
text = path.read_text(encoding="utf-8")
|
|
lower_text = text.lower()
|
|
|
|
with self.subTest(path=path.name):
|
|
self.assertIn(OFFLINE_CHECK_DOC_NAME, text)
|
|
self.assertIn(PACKAGE_SMOKE_COMMAND, text)
|
|
self.assertIn(OFFLINE_CHECK_COMMAND, text)
|
|
self.assertIn("package smoke", lower_text)
|
|
|
|
def test_offline_checks_registry_runs_training_package_smoke_entry(self) -> None:
|
|
commands = run_offline_checks.build_check_commands(env={})
|
|
names = [name for name, _command in commands]
|
|
command = dict(commands)[OFFLINE_CHECK_NAME]
|
|
calls = []
|
|
|
|
def fake_runner(command_parts, **kwargs):
|
|
calls.append((command_parts, kwargs["cwd"]))
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_offline_checks.run_checks(
|
|
commands=[(OFFLINE_CHECK_NAME, command)],
|
|
runner=fake_runner,
|
|
node_available=True,
|
|
env={},
|
|
)
|
|
|
|
self.assertEqual(names.count(OFFLINE_CHECK_NAME), 1)
|
|
self.assertEqual(command, [sys.executable, PACKAGE_SMOKE_SCRIPT])
|
|
self.assertEqual((ROOT / command[1]).resolve(), Path(training_cli.__file__).resolve())
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertEqual(calls, [(command, run_offline_checks.ROOT)])
|
|
self.assertIn(f"Offline check passed: {OFFLINE_CHECK_NAME}", output.getvalue())
|
|
|
|
def test_training_docs_name_broader_offline_check_path(self) -> None:
|
|
for path in (DOC_EXAMPLE, README, PRODUCT_README):
|
|
text = path.read_text(encoding="utf-8")
|
|
lower_text = text.lower()
|
|
|
|
with self.subTest(path=path.name):
|
|
self.assertIn(OFFLINE_CHECK_COMMAND, text)
|
|
self.assertIn("broader offline", lower_text)
|
|
self.assertIn("training case", lower_text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|