yuqei/ai-exchange
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ab65e8d43e7452316ca9cd0d398e2b01aa44a482
ai-exchange/tests/unit/test_training_case_docs_cli_m10.py
T

644 lines
27 KiB
Python
Raw Blame History

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_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",
),
"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_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), 14)
self.assertEqual(len(inventory_names), 15)
self.assertEqual(len(unique_inventory_names), 13)
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), 16)
self.assertEqual(len(m33_unique_inventory_names), 13)
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()
Reference in New Issue View Git Blame Copy Permalink