"""Isolated integration checks for tools/prepare_debug_world.py; no real saves are accessed.""" import hashlib import json import os from pathlib import Path import subprocess import sys import tempfile import unittest import xml.etree.ElementTree as ET SCRIPT = Path(__file__).resolve().parents[1] / "tools" / "prepare_debug_world.py" XSI_TYPE = "{http://www.w3.org/2001/XMLSchema-instance}type" class DebugWorldCopyTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(prefix="xfe-world-copy-tests-") self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name).resolve() self.roaming = self.root / "Roaming" self.local = self.root / "Local" self.account = self.roaming / "SpaceEngineers" / "Saves" / "76561198000000001" self.account.mkdir(parents=True) self.environment = os.environ.copy() self.environment.update( APPDATA=str(self.roaming), LOCALAPPDATA=str(self.local), PYTHONIOENCODING="utf-8", ) self.report = self.root / "artifacts" / "preparation.json" self.config = self.local / "XFE" / "SpaceEngineersAgent" / "config.json" def world(self, name, modified=1000, ams=True, program_data=False, name_tag=False): world = self.account / name world.mkdir() for filename in ("Sandbox.sbc", "Sandbox_config.sbc"): root = ET.Element("MyObjectBuilder_Checkpoint") ET.SubElement(root, "SessionName").text = name settings = ET.SubElement(root, "Settings") ET.SubElement(settings, "OnlineMode").text = "PUBLIC" ET.SubElement(settings, "AutoSaveInMinutes").text = "5" ET.SubElement(settings, "InventorySizeMultiplier").text = "10" ET.SubElement(root, "Unrelated").text = "保留其他世界设置" ET.ElementTree(root).write(world / filename, encoding="utf-8", xml_declaration=True) sector = ET.Element("MyObjectBuilder_Sector") entities = ET.SubElement(sector, "SectorObjects") grid = ET.SubElement(entities, "MyObjectBuilder_EntityBase", {XSI_TYPE: "MyObjectBuilder_CubeGrid"}) ET.SubElement(grid, "EntityId").text = "900000000000000001" ET.SubElement(grid, "DisplayName").text = "临时矿机测试网格" blocks = ET.SubElement(grid, "CubeBlocks") block = ET.SubElement(blocks, "MyObjectBuilder_CubeBlock", {XSI_TYPE: "MyObjectBuilder_MyProgrammableBlock"}) ET.SubElement(block, "EntityId").text = "900000000000000002" ET.SubElement(block, "CustomName").text = "XFEAMS 测试矿机" if name_tag else "临时可编程方块" code = "// XFE AMS 2.0\npublic void Main() { Echo(\"测试\"); }" if ams else "public void Main() { }" ET.SubElement(block, "ProgramData" if program_data else "Program").text = code ET.SubElement(block, "CustomData").text = "目标矿种=\n优先矿种=Iron,Cobalt,Nickel,Silicon" ET.ElementTree(sector).write(world / "SANDBOX_0_0_0_.sbs", encoding="utf-8", xml_declaration=True) (world / "Asteroid.vx2").write_bytes(b"\x00\x01fixture-voxel\xff\x80") (world / "Extra").mkdir() (world / "Extra" / "settings.json").write_text('{"unchanged":"metadata"}', encoding="utf-8") os.utime(world / "Sandbox.sbc", (modified, modified)) return world @staticmethod def snapshot(world): return { str(path.relative_to(world)): ( hashlib.sha256(path.read_bytes()).hexdigest(), path.stat().st_size, path.stat().st_mtime_ns, ) for path in world.rglob("*") if path.is_file() } def run_copy(self, source=None, output=None, succeeds=True): command = [sys.executable, str(SCRIPT), "--out", str(output or self.report)] if source is not None: command += ["--source", str(source)] completed = subprocess.run( command, cwd=self.root, env=self.environment, text=True, encoding="utf-8", stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) if succeeds: self.assertEqual(completed.returncode, 0, completed.stderr) response = json.loads(completed.stdout) self.assertTrue(Path(response["debugWorld"]).is_relative_to(self.root)) return response, json.loads((output or self.report).read_text(encoding="utf-8")) self.assertNotEqual(completed.returncode, 0, "Copy command unexpectedly succeeded") return completed def test_automatic_selection_uses_newest_ams_and_skips_debug_copies(self): old = self.world("旧 AMS", modified=1000) newest_ams = self.world("新 AMS", modified=2000) newer_plain = self.world("更晚但没有 AMS", modified=3000, ams=False) previous_debug = self.world("XFE Agent Debug AMS previous", modified=4000) snapshots = {path: self.snapshot(path) for path in (old, newest_ams, newer_plain, previous_debug)} response, report = self.run_copy() self.assertEqual(Path(response["original"]), newest_ams) self.assertEqual(Path(report["original"]), newest_ams) self.assertEqual(response["programCount"], 1) for path, snapshot in snapshots.items(): self.assertEqual(self.snapshot(path), snapshot) def test_copy_changes_session_settings_only_in_new_world(self): source = self.world("Original AMS") original = self.snapshot(source) response, report = self.run_copy(source) target = Path(response["debugWorld"]) self.assertNotEqual(target, source) self.assertEqual(target.parent, source.parent) self.assertEqual(self.snapshot(source), original) for filename in ("Sandbox.sbc", "Sandbox_config.sbc"): with self.subTest(filename=filename): original_xml = ET.parse(source / filename).getroot() copy_xml = ET.parse(target / filename).getroot() self.assertEqual(original_xml.findtext("Settings/OnlineMode"), "PUBLIC") self.assertEqual(original_xml.findtext("Settings/AutoSaveInMinutes"), "5") self.assertEqual(original_xml.findtext("SessionName"), source.name) self.assertEqual(copy_xml.findtext("Settings/OnlineMode"), "OFFLINE") self.assertEqual(copy_xml.findtext("Settings/AutoSaveInMinutes"), "0") self.assertEqual(copy_xml.findtext("SessionName"), target.name) self.assertEqual(copy_xml.findtext("Unrelated"), original_xml.findtext("Unrelated")) self.assertEqual(copy_xml.findtext("Settings/InventorySizeMultiplier"), "10") for filename in ("SANDBOX_0_0_0_.sbs", "Asteroid.vx2", "Extra/settings.json"): self.assertEqual((target / filename).read_bytes(), (source / filename).read_bytes()) self.assertEqual(report["originalFiles"], {name: state[0] for name, state in original.items()}) block = report["programmableBlocks"][0] self.assertEqual(block["entityId"], "900000000000000002") self.assertEqual(block["gridId"], "900000000000000001") self.assertIn("优先矿种=", block["customData"]) self.assertEqual(len(block["sourceSha256"]), 64) def test_authorization_appends_copy_and_retains_previous_settings(self): source = self.world("Authorization AMS") previous = [str(self.root / "previous authorized test world")] self.config.parent.mkdir(parents=True) self.config.write_text(json.dumps({"allowedWorldPaths": previous, "extraSetting": {"retain": True}}), encoding="utf-8-sig") response, _ = self.run_copy(source) config = json.loads(self.config.read_text(encoding="utf-8")) self.assertEqual(config["allowedWorldPaths"], previous + [response["debugWorld"]]) self.assertNotIn(str(source), config["allowedWorldPaths"]) self.assertEqual(config["extraSetting"], {"retain": True}) self.assertFalse(self.config.with_suffix(".tmp").exists()) def test_backups_are_excluded_from_copy_and_manifest_but_preserved_in_source(self): source = self.world("Backups AMS") for relative in ("Backup/older-world/old.sbs", "Extra/backup/old.txt"): path = source / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"preserve original backup") original = self.snapshot(source) response, report = self.run_copy(source) target = Path(response["debugWorld"]) self.assertFalse((target / "Backup").exists()) self.assertFalse((target / "Extra/backup").exists()) self.assertTrue(all("backup" not in [part.lower() for part in Path(name).parts] for name in report["originalFiles"])) self.assertEqual(self.snapshot(source), original) def test_explicit_source_takes_precedence_over_newer_world(self): selected = self.world("Explicit AMS", modified=1000) self.world("Newer AMS", modified=2000) response, _ = self.run_copy(selected) self.assertEqual(Path(response["original"]), selected) def test_program_data_and_named_ams_block_are_recognized(self): source = self.world("ProgramData AMS", ams=False, program_data=True, name_tag=True) response, report = self.run_copy(source) self.assertEqual(response["programCount"], 1) self.assertEqual(report["programmableBlocks"][0]["name"], "XFEAMS 测试矿机") def test_explicit_source_without_ams_fails_without_writes(self): source = self.world("No AMS", ams=False) original = self.snapshot(source) self.config.parent.mkdir(parents=True) self.config.write_text('{"allowedWorldPaths":[],"retain":true}', encoding="utf-8") old_config = self.config.read_bytes() failed = self.run_copy(source, succeeds=False) self.assertIn("No saved world containing AMS", failed.stderr) self.assertEqual(self.snapshot(source), original) self.assertEqual(self.config.read_bytes(), old_config) self.assertEqual([path.name for path in self.account.iterdir()], [source.name]) self.assertFalse(self.report.exists()) def test_no_matching_automatic_source_fails_without_authorization(self): source = self.world("Plain world", ams=False) original = self.snapshot(source) self.run_copy(succeeds=False) self.assertFalse(self.config.exists()) self.assertFalse(self.report.exists()) self.assertEqual(self.snapshot(source), original) def test_report_cannot_overwrite_original_world_file(self): source = self.world("Protected original AMS") original = self.snapshot(source) self.run_copy(source, output=source / "Sandbox.sbc", succeeds=False) self.assertEqual(self.snapshot(source), original) self.assertFalse(self.config.exists()) self.assertEqual([path.name for path in self.account.iterdir()], [source.name]) if __name__ == "__main__": unittest.main()