XFE.SpaceEngineers.AgentBridge

"""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()