初始化系统融合软件代码仓库
This commit is contained in:
@@ -0,0 +1,199 @@
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import tarfile
|
||||
import tempfile
|
||||
import subprocess
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
path = Path(__file__).resolve().parents[1] / "Stage-EmbeddedImageRoute.py"
|
||||
spec = importlib.util.spec_from_file_location("route_stage", path)
|
||||
stage = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(stage)
|
||||
|
||||
|
||||
class ImageRouteStagingTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temporary = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temporary.cleanup)
|
||||
self.work = Path(self.temporary.name)
|
||||
self.root = self.work / "board"
|
||||
self.root.mkdir()
|
||||
for relative in stage.PRESERVED:
|
||||
target = self.root / relative
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text("board latest source: CollectorLogRetention::install PlatformImageSender\n", encoding="utf-8")
|
||||
(self.root / "px_Collect.pro").write_text("DEFINES += COLLECTOR_WITH_GST\nLIBS += -L/board/vendor -lhiredis\n", encoding="utf-8")
|
||||
(self.root / "bin").mkdir()
|
||||
(self.root / "bin/px_JKGL").write_bytes(b"original-test-binary")
|
||||
(self.root / "config.ini").write_text("; original comment\n[fixed_json]\nudp_host=198.51.100.1\nudp_port=19001\n"
|
||||
"[heartbeat]\ninterval_ms=3000\n[image_upload]\nenabled=true\ninterval_ms=100\n", encoding="utf-8")
|
||||
(self.root / "module_keys.json").write_text("not a real credential", encoding="utf-8")
|
||||
launcher = self.root / "start-network-simulation.sh"
|
||||
launcher.write_text("#!/bin/sh\n# current private board launcher\n", encoding="utf-8")
|
||||
launcher.chmod(0o755)
|
||||
self.package = self.work / "route.tar.gz"
|
||||
content = {}
|
||||
manifest = {"format": 1, "protocol": "COLLECTOR_IMAGE_ROUTE", "files": []}
|
||||
import hashlib
|
||||
for name in stage.PATCH_FILES + stage.TEST_FILES:
|
||||
data = b"isolated test source\n"
|
||||
content[name] = data
|
||||
manifest["files"].append({"path": name, "sha256": hashlib.sha256(data).hexdigest()})
|
||||
content["image-route-manifest.json"] = json.dumps(manifest).encode()
|
||||
with tarfile.open(str(self.package), "w:gz") as archive:
|
||||
for name, data in content.items():
|
||||
item = tarfile.TarInfo(name); item.size = len(data)
|
||||
archive.addfile(item, io.BytesIO(data))
|
||||
self.args = SimpleNamespace(root=self.root, package=self.package, gateway_host="192.0.2.10",
|
||||
gateway_port=19002, fallback_ip="192.0.2.11", fallback_port=6020)
|
||||
|
||||
def create_stage(self):
|
||||
output = io.StringIO()
|
||||
with contextlib.redirect_stdout(output):
|
||||
stage.stage(self.args)
|
||||
return Path(json.loads(output.getvalue())["stage"])
|
||||
|
||||
def test_stage_keeps_live_sources_binary_config_and_keys_untouched(self):
|
||||
original = (self.root / "config.ini").read_bytes()
|
||||
directory = self.create_stage()
|
||||
self.assertEqual(original, (self.root / "config.ini").read_bytes())
|
||||
self.assertEqual(b"original-test-binary", (self.root / "bin/px_JKGL").read_bytes())
|
||||
self.assertFalse((self.root / stage.PATCH_FILES[0]).exists())
|
||||
self.assertFalse((directory / "source/module_keys.json").exists())
|
||||
self.assertFalse((directory / "source/config.ini").exists())
|
||||
self.assertEqual((self.root / "main.cpp").read_bytes(), (directory / "source/main.cpp").read_bytes())
|
||||
text = (directory / "source/px_Collect.pro").read_text(encoding="utf-8")
|
||||
self.assertIn("LIBS += -L/board/vendor -lhiredis", text)
|
||||
self.assertIn("SOURCES += util/PlatformImageRouteClient.cpp", text)
|
||||
|
||||
def test_candidate_config_sets_independent_fallback_and_preserves_other_settings(self):
|
||||
directory = self.create_stage()
|
||||
text = (directory / "candidate-config.ini").read_text(encoding="utf-8")
|
||||
self.assertIn("; original comment", text)
|
||||
self.assertIn("[fixed_json]\nudp_host=198.51.100.1\nudp_port=19001", text)
|
||||
self.assertIn("[heartbeat]\ninterval_ms=3000", text)
|
||||
self.assertIn("interval_ms=100", text)
|
||||
self.assertIn("udp_host=192.0.2.11", text)
|
||||
self.assertIn("gateway_host=192.0.2.10", text)
|
||||
|
||||
def test_verify_rejects_source_changed_after_stage(self):
|
||||
directory = self.create_stage()
|
||||
args = SimpleNamespace(root=self.root, stage=directory)
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
stage.verify(args)
|
||||
(self.root / "main.cpp").write_text("new board modification", encoding="utf-8")
|
||||
with self.assertRaisesRegex(RuntimeError, "main.cpp"):
|
||||
stage.verify(args)
|
||||
|
||||
def test_archive_path_traversal_is_rejected_without_writing(self):
|
||||
with tarfile.open(str(self.package), "w:gz") as archive:
|
||||
item = tarfile.TarInfo("../outside.cpp"); item.size = 1
|
||||
archive.addfile(item, io.BytesIO(b"x"))
|
||||
with self.assertRaisesRegex(RuntimeError, "archive contents"):
|
||||
stage.payload(self.args)
|
||||
self.assertFalse((self.work / "outside.cpp").exists())
|
||||
|
||||
def test_missing_native_log_feature_cannot_be_silently_downgraded(self):
|
||||
(self.root / "main.cpp").write_text("old Qt application without native log", encoding="utf-8")
|
||||
with self.assertRaisesRegex(RuntimeError, "Required existing feature missing: main.cpp"):
|
||||
self.create_stage()
|
||||
|
||||
def test_bad_config_error_does_not_echo_original_line(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "unsupported or duplicate entries") as error:
|
||||
stage.candidate_config("[image_upload]\nenabled=true\nCONFIDENTIAL-TEST-LINE", "192.0.2.10", 19002, "192.0.2.11", 6020)
|
||||
self.assertNotIn("CONFIDENTIAL", str(error.exception))
|
||||
|
||||
def test_rollback_preview_validates_current_ownership_without_changing_live_files(self):
|
||||
directory = self.create_stage()
|
||||
metadata_file = directory / "stage.json"
|
||||
metadata = json.loads(metadata_file.read_text(encoding="utf-8"))
|
||||
metadata.update(built=True, candidateBinarySha256="synthetic-unused-candidate-hash")
|
||||
metadata_file.write_text(json.dumps(metadata), encoding="utf-8")
|
||||
command = [sys.executable, str(path.with_name("Rollback-EmbeddedImageRoute.py")),
|
||||
"--root", str(self.root), "--stage", str(directory)]
|
||||
original = (self.root / "config.ini").read_bytes()
|
||||
completed = subprocess.run(command, capture_output=True, text=True, check=True)
|
||||
self.assertTrue(json.loads(completed.stdout)["rollbackReady"])
|
||||
self.assertEqual(original, (self.root / "config.ini").read_bytes())
|
||||
self.assertEqual(b"original-test-binary", (self.root / "bin/px_JKGL").read_bytes())
|
||||
(self.root / "config.ini").write_text("unrelated newer configuration", encoding="utf-8")
|
||||
completed = subprocess.run(command, capture_output=True, text=True)
|
||||
self.assertNotEqual(completed.returncode, 0)
|
||||
self.assertIn("Unrelated live update detected: config.ini", completed.stderr)
|
||||
|
||||
def built_stage(self):
|
||||
directory = self.create_stage()
|
||||
binary = directory / "source/bin/px_JKGL"
|
||||
binary.parent.mkdir(); binary.write_bytes(b"isolated-candidate-binary"); binary.chmod(0o755)
|
||||
metadata_file = directory / "stage.json"
|
||||
metadata = json.loads(metadata_file.read_text(encoding="utf-8"))
|
||||
metadata.update(built=True, candidateBinarySha256=stage.digest(binary), candidateBinary=str(binary))
|
||||
metadata_file.write_text(json.dumps(metadata), encoding="utf-8")
|
||||
(directory / "logging-tests.txt").write_text("Totals: 9 passed, 0 failed, 0 skipped", encoding="utf-8")
|
||||
(directory / "image_route-tests.txt").write_text("SKIP: javaAndCppWireCodecInteroperateWithRealEncryptedFrames\n"
|
||||
"Totals: 8 passed, 0 failed, 1 skipped", encoding="utf-8")
|
||||
return directory
|
||||
|
||||
def load_script(self, name):
|
||||
spec = importlib.util.spec_from_file_location(name, path.with_name(name + ".py"))
|
||||
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
def test_activation_preview_and_success_replace_only_owned_files(self):
|
||||
activate = self.load_script("Activate-EmbeddedImageRoute")
|
||||
directory = self.built_stage()
|
||||
args = SimpleNamespace(root=self.root, stage=directory, apply=False, health_timeout=15)
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
activate.activate(args)
|
||||
self.assertEqual(b"original-test-binary", (self.root / "bin/px_JKGL").read_bytes())
|
||||
args.apply = True
|
||||
health = {"collectorPids": [4321], "nativeLogging": {"owner": "px_JKGL", "limitBytes": 1024 ** 3,
|
||||
"withinLimit": True, "failedWrites": 0}}
|
||||
with patch.object(activate, "stop_collector"), patch.object(activate, "launch") as launch, \
|
||||
patch.object(activate, "healthy", return_value=health), contextlib.redirect_stdout(io.StringIO()):
|
||||
activate.activate(args)
|
||||
launch.assert_called_once()
|
||||
self.assertEqual(b"isolated-candidate-binary", (self.root / "bin/px_JKGL").read_bytes())
|
||||
self.assertEqual((directory / "candidate-config.ini").read_bytes(), (self.root / "config.ini").read_bytes())
|
||||
self.assertEqual((directory / "source/main.cpp").read_bytes(), (self.root / "main.cpp").read_bytes())
|
||||
self.assertTrue(json.loads((directory / "activation-result.json").read_text())["activated"])
|
||||
|
||||
def test_failed_activation_invokes_real_rollback_logic_with_simulated_processes(self):
|
||||
activate = self.load_script("Activate-EmbeddedImageRoute")
|
||||
rollback = self.load_script("Rollback-EmbeddedImageRoute")
|
||||
directory = self.built_stage()
|
||||
original_config = (self.root / "config.ini").read_bytes()
|
||||
args = SimpleNamespace(root=self.root, stage=directory, apply=True, health_timeout=15)
|
||||
launched = {"old": False}
|
||||
|
||||
def simulated_run(command, **kwargs):
|
||||
if command[0] == sys.executable:
|
||||
with patch.object(sys, "argv", command[1:]):
|
||||
rollback.main()
|
||||
else:
|
||||
self.assertEqual(command, [str(self.root / "start-network-simulation.sh")])
|
||||
launched["old"] = True
|
||||
return subprocess.CompletedProcess(command, 0)
|
||||
|
||||
with patch.object(activate, "stop_collector"), patch.object(activate, "launch", side_effect=RuntimeError("simulated launch failed")), \
|
||||
patch.object(activate.subprocess, "run", side_effect=simulated_run), \
|
||||
patch.object(rollback.helper, "process_ids", side_effect=lambda root: [4321] if launched["old"] else []), \
|
||||
patch.object(rollback.time, "sleep"), contextlib.redirect_stdout(io.StringIO()):
|
||||
with self.assertRaisesRegex(RuntimeError, "rollbackSucceeded=True"):
|
||||
activate.activate(args)
|
||||
self.assertTrue(launched["old"])
|
||||
self.assertEqual(b"original-test-binary", (self.root / "bin/px_JKGL").read_bytes())
|
||||
self.assertEqual(original_config, (self.root / "config.ini").read_bytes())
|
||||
self.assertFalse((self.root / "util/PlatformImageRouteClient.cpp").exists())
|
||||
self.assertTrue(json.loads((directory / "activation-error.json").read_text())["rollbackSucceeded"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user