NODEDC_MISSION_CORE/tests/test_frontend_plugin_bounda...

159 lines
6.3 KiB
Python

from __future__ import annotations
from pathlib import Path
def test_mission_core_frontend_has_single_plugin_composition_root() -> None:
repository_root = Path(__file__).resolve().parents[1]
source_root = repository_root / "apps" / "control-station" / "src"
composition = source_root / "composition" / "devicePlugins.ts"
forbidden_import = "@xgrids-k1/frontend"
imports = [
path.relative_to(source_root).as_posix()
for path in source_root.rglob("*.ts*")
if forbidden_import in path.read_text("utf-8")
]
assert imports == [composition.relative_to(source_root).as_posix()]
def test_mission_core_shell_does_not_know_xgrids_wire_fields() -> None:
repository_root = Path(__file__).resolve().parents[1]
source_root = repository_root / "apps" / "control-station" / "src"
core_paths = [
source_root / "App.tsx",
source_root / "presentation.ts",
source_root / "productModel.ts",
source_root / "components",
source_root / "core",
source_root / "workspaces",
]
forbidden_tokens = ("k1_ip", "likely_k1", "useK1Console", "rerun_grpc_url")
checked_files: list[Path] = []
for path in core_paths:
if path.is_dir():
checked_files.extend(path.rglob("*.ts*"))
else:
checked_files.append(path)
violations = {
path.relative_to(source_root).as_posix(): token
for path in checked_files
for token in forbidden_tokens
if token in path.read_text("utf-8")
}
assert violations == {}
def test_xgrids_client_uses_manifest_identity_and_plugin_scoped_events() -> None:
repository_root = Path(__file__).resolve().parents[1]
plugin_root = repository_root / "plugins" / "xgrids-k1" / "frontend" / "src"
api_source = (plugin_root / "api.ts").read_text("utf-8")
manifest_source = (plugin_root / "manifest.ts").read_text("utf-8")
assert "nodedc.device.xgrids-lixelkity-k1" not in api_source
assert 'new URL("/api/events"' not in api_source
assert "/api/v1/device-plugins/" in api_source
assert "parseDevicePluginManifest" in manifest_source
def test_backend_host_has_no_concrete_xgrids_imports() -> None:
repository_root = Path(__file__).resolve().parents[1]
app_source = (repository_root / "src" / "k1link" / "web" / "app.py").read_text("utf-8")
assert "xgrids" not in app_source.lower()
assert "k1_ip" not in app_source
assert "Bleak" not in app_source
def test_model_switch_is_guarded_by_plugin_deactivation() -> None:
repository_root = Path(__file__).resolve().parents[1]
frontend_root = repository_root / "apps" / "control-station" / "src"
plugin_root = repository_root / "plugins" / "xgrids-k1" / "frontend" / "src"
host_source = (frontend_root / "core" / "device-plugins" / "DevicePluginHost.tsx").read_text(
"utf-8"
)
xgrids_runtime = (plugin_root / "runtimeContext.tsx").read_text("utf-8")
deactivation_guard = "if (!(await deactivate()))"
assert "if (current && !deactivate)" in host_source
assert deactivation_guard in host_source
assert host_source.index(deactivation_guard) < host_source.index(
"setSelectedModelId(nextModelId)"
)
assert "catch" in host_source
assert "selectionTransitionError" in host_source
assert "return controller.stop();" in xgrids_runtime
def test_xgrids_frontend_uses_semantic_acquisition_actions_and_stable_identity() -> None:
repository_root = Path(__file__).resolve().parents[1]
plugin_root = repository_root / "plugins" / "xgrids-k1" / "frontend" / "src"
manifest_source = (plugin_root / "manifest.ts").read_text("utf-8")
api_source = (plugin_root / "api.ts").read_text("utf-8")
hook_source = (plugin_root / "useXgridsK1Runtime.ts").read_text("utf-8")
runtime_source = (plugin_root / "runtimeContext.tsx").read_text("utf-8")
for action in ("acquisition.prepare", "acquisition.start", "acquisition.stop"):
assert action in manifest_source
assert "prepareAcquisition" in api_source
assert "startAcquisition" in api_source
assert hook_source.index("xgridsK1Api.prepareAcquisition") < hook_source.index(
"xgridsK1Api.startAcquisition"
)
assert "isSoftwareCommandedAcquisition(state)" in hook_source
assert '? "graceful" : "capture-only"' in hook_source
assert "state.device_ref" in runtime_source
assert "instanceId: deviceRef.device_id" in runtime_source
assert "acquisition?.acquisition_id" in runtime_source
assert "instanceId: state.selected_device_id" not in runtime_source
assert 'id: "xgrids-k1-rerun-live"' not in runtime_source
def test_xgrids_live_copy_does_not_claim_software_controls_the_physical_scanner() -> None:
repository_root = Path(__file__).resolve().parents[1]
connection_source = (
repository_root
/ "plugins"
/ "xgrids-k1"
/ "frontend"
/ "src"
/ "components"
/ "K1AcquisitionPipeline.tsx"
).read_text("utf-8")
assert "Подготовить локальный приём данных" in connection_source
assert "Программная команда запуска на K1 не отправляется" in connection_source
assert "Остановить локальный приём" in connection_source
assert "Физическое состояние сканера остаётся неизвестным" in connection_source
def test_xgrids_start_uses_atomic_automatic_source_transition() -> None:
repository_root = Path(__file__).resolve().parents[1]
pipeline_source = (
repository_root
/ "plugins"
/ "xgrids-k1"
/ "frontend"
/ "src"
/ "components"
/ "K1AcquisitionPipeline.tsx"
).read_text("utf-8")
transition_source = (
repository_root / "plugins" / "xgrids-k1" / "frontend" / "src" / "automaticSourceStart.ts"
).read_text("utf-8")
assert pipeline_source.count("runAutomaticSpatialSourceStart(") == 2
assert transition_source.index("const started = await start();") < (
transition_source.index("activateAutomaticSpatialSource();")
)
assert transition_source.index("if (!started) return false;") < (
transition_source.index("activateAutomaticSpatialSource();")
)
assert transition_source.index("activateAutomaticSpatialSource();") < (
transition_source.index("openSpatialScene();")
)