wip(k1): checkpoint connection recovery rewrite
Capture the current unreleased K1 connection, recovery, lifecycle, viewer, and test work as a single known-bad baseline for subsequent fixes.
This commit is contained in:
@@ -0,0 +1,50 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import atexit
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# ``k1link.web.app`` builds the installed plugin environment at import time.
|
||||
# Pytest imports test modules during collection and future tests may lazy-load
|
||||
# it during execution or in a spawned child. Keep every mutable Mission Core
|
||||
# store used by that composition inside one process-scoped test root for the
|
||||
# entire pytest lifetime. Restoring the caller environment after collection
|
||||
# would reopen the operator's real stores for lazy imports and child processes.
|
||||
_ISOLATED_ENVIRONMENT_KEYS = (
|
||||
"MISSIONCORE_DATA_DIR",
|
||||
"MISSIONCORE_EVIDENCE_DIR",
|
||||
"MISSIONCORE_LEGACY_SESSIONS_DIR",
|
||||
)
|
||||
_ORIGINAL_ENVIRONMENT = {
|
||||
key: os.environ.get(key) for key in _ISOLATED_ENVIRONMENT_KEYS
|
||||
}
|
||||
_TEST_RUNTIME_ROOT = Path(
|
||||
tempfile.mkdtemp(prefix="mission-core-pytest-runtime-")
|
||||
).resolve()
|
||||
_TEST_RUNTIME_ROOT.chmod(0o700)
|
||||
|
||||
_ISOLATED_ENVIRONMENT = {
|
||||
"MISSIONCORE_DATA_DIR": _TEST_RUNTIME_ROOT / "data",
|
||||
"MISSIONCORE_EVIDENCE_DIR": _TEST_RUNTIME_ROOT / "evidence",
|
||||
"MISSIONCORE_LEGACY_SESSIONS_DIR": _TEST_RUNTIME_ROOT / "legacy-sessions",
|
||||
}
|
||||
for _key, _path in _ISOLATED_ENVIRONMENT.items():
|
||||
_path.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
_path.chmod(0o700)
|
||||
os.environ[_key] = str(_path)
|
||||
|
||||
|
||||
def _restore_environment() -> None:
|
||||
for key, original_value in _ORIGINAL_ENVIRONMENT.items():
|
||||
if original_value is None:
|
||||
os.environ.pop(key, None)
|
||||
else:
|
||||
os.environ[key] = original_value
|
||||
|
||||
|
||||
@atexit.register
|
||||
def _remove_test_runtime() -> None:
|
||||
_restore_environment()
|
||||
shutil.rmtree(_TEST_RUNTIME_ROOT, ignore_errors=True)
|
||||
Reference in New Issue
Block a user