from __future__ import annotations
|
|
from collections import OrderedDict
|
import json
|
import os
|
from pathlib import Path
|
import subprocess
|
import sys
|
import tempfile
|
import unittest
|
from unittest.mock import patch
|
|
from hibor_fast_collection.manifests import canonical_json_bytes
|
from hibor_fast_collection.models import ContractError
|
from hibor_fast_collection.performance import (
|
PERFORMANCE_SUBJECT_CODEPOINTS,
|
build_execution_context,
|
validate_execution_context_bytes,
|
)
|
from hibor_fast_collection.performance_coordinator import materialize_execution_context
|
|
|
START = "2026-07-30T05:00:00Z"
|
|
|
class PerformanceCoordinatorTests(unittest.TestCase):
|
def test_memory_preimage_preserves_exact_unicode(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = (Path(temp) / "run-positive").resolve()
|
context, data = build_execution_context(root, START)
|
self.assertFalse(root.exists())
|
self.assertEqual(context["subjects"], ["三环集团", "国瓷材料", "MLCC"])
|
self.assertEqual(
|
tuple(tuple(ord(char) for char in subject) for subject in context["subjects"]),
|
PERFORMANCE_SUBJECT_CODEPOINTS,
|
)
|
self.assertIn("三环集团".encode("utf-8"), data)
|
self.assertIn("国瓷材料".encode("utf-8"), data)
|
self.assertEqual(validate_execution_context_bytes(data, root, START), context)
|
|
def test_question_mark_transport_drift_stops_before_root_creation(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = (Path(temp) / "run-question-marks").resolve()
|
context, _ = build_execution_context(root, START)
|
bad = OrderedDict(context)
|
bad["subjects"] = ["????", "????", "MLCC"]
|
with self.assertRaises(ContractError):
|
materialize_execution_context(
|
root, START, preimage=canonical_json_bytes(bad),
|
)
|
self.assertFalse(root.exists())
|
|
def test_invalid_byte_and_root_drift_stop_before_root_creation(self):
|
with tempfile.TemporaryDirectory() as temp:
|
parent = Path(temp)
|
malformed_root = (parent / "malformed").resolve()
|
with self.assertRaises(ContractError):
|
materialize_execution_context(malformed_root, START, preimage=b"\xff")
|
self.assertFalse(malformed_root.exists())
|
|
source_root = (parent / "source-root").resolve()
|
target_root = (parent / "target-root").resolve()
|
_, data = build_execution_context(source_root, START)
|
with self.assertRaises(ContractError):
|
materialize_execution_context(target_root, START, preimage=data)
|
self.assertFalse(target_root.exists())
|
|
def test_create_new_readback_and_replay_rejected_without_overwrite(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = (Path(temp) / "run-create-new").resolve()
|
receipt = materialize_execution_context(root, START)
|
path = root / "control" / "execution_context.json"
|
original = path.read_bytes()
|
self.assertEqual(receipt["path"], str(path))
|
self.assertEqual(receipt["bytes"], len(original))
|
self.assertEqual(json.loads(original.decode("utf-8"))["subjects"],
|
["三环集团", "国瓷材料", "MLCC"])
|
with self.assertRaises(ContractError):
|
materialize_execution_context(root, START)
|
self.assertEqual(path.read_bytes(), original)
|
|
def test_parent_must_be_ordinary_directory_without_writes(self):
|
with tempfile.TemporaryDirectory() as temp:
|
parent_file = Path(temp) / "not-a-directory"
|
parent_file.write_text("sentinel", encoding="ascii")
|
root = parent_file / "child"
|
with self.assertRaises(ContractError):
|
materialize_execution_context(root, START)
|
self.assertEqual(parent_file.read_text(encoding="ascii"), "sentinel")
|
|
@unittest.skipUnless(os.name == "nt", "Windows junction regression")
|
def test_windows_junction_parent_is_rejected_without_redirected_write(self):
|
with tempfile.TemporaryDirectory() as temp:
|
parent = Path(temp)
|
target = parent / "junction-target"
|
target.mkdir()
|
junction = parent / "junction-parent"
|
created = subprocess.run(
|
["cmd.exe", "/d", "/c", "mklink", "/J", str(junction), str(target)],
|
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
timeout=10, check=False,
|
)
|
if created.returncode != 0:
|
self.skipTest(created.stderr.decode("utf-8", "replace"))
|
try:
|
root = junction / "run"
|
with self.assertRaises(ContractError):
|
materialize_execution_context(root, START)
|
self.assertFalse((target / "run").exists())
|
finally:
|
os.rmdir(junction)
|
|
def test_write_failure_preserves_actual_subset(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = (Path(temp) / "run-write-failure").resolve()
|
with patch("hibor_fast_collection.performance_coordinator.os.fsync",
|
side_effect=OSError("simulated fsync failure")):
|
with self.assertRaises(OSError):
|
materialize_execution_context(root, START)
|
self.assertTrue(root.is_dir())
|
self.assertTrue((root / "control").is_dir())
|
self.assertTrue((root / "control" / "execution_context.json").exists())
|
|
def test_physical_module_windows_safe_subprocess_preserves_unicode(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = (Path(temp) / "run-subprocess").resolve()
|
package_root = Path(__file__).resolve().parents[2]
|
completed = subprocess.run(
|
[sys.executable, "-m", "hibor_fast_collection.performance_coordinator",
|
"--evidence-root", str(root), "--started-at-utc", START],
|
cwd=package_root, stdin=subprocess.DEVNULL,
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
timeout=20, check=False,
|
)
|
self.assertEqual(completed.returncode, 0, completed.stderr.decode("ascii", "replace"))
|
receipt = json.loads(completed.stdout.decode("ascii"))
|
data = (root / "control" / "execution_context.json").read_bytes()
|
self.assertEqual(receipt["bytes"], len(data))
|
context = json.loads(data.decode("utf-8"))
|
self.assertEqual(context["subjects"], ["三环集团", "国瓷材料", "MLCC"])
|
self.assertNotIn(b"????", data)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
