from __future__ import annotations import unittest import xml.etree.ElementTree as ET from pathlib import Path ROOT = Path(__file__).resolve().parents[1] / "windows-node" PROJECT = ROOT / "Zcbot.WindowsNode" class WindowsNodeSourceTests(unittest.TestCase): def test_project_targets_net10_windows_forms_with_json_schema_validator(self) -> None: tree = ET.parse(PROJECT / "Zcbot.WindowsNode.csproj") root = tree.getroot() self.assertEqual(root.findtext("./PropertyGroup/TargetFramework"), "net10.0-windows") self.assertEqual(root.findtext("./PropertyGroup/UseWindowsForms"), "true") self.assertEqual(root.findtext("./PropertyGroup/OutputType"), "WinExe") packages = root.findall("./ItemGroup/PackageReference") self.assertEqual([item.attrib["Include"] for item in packages], ["JsonSchema.Net"]) def test_node_protocol_and_secret_storage_markers_are_present(self) -> None: source = "\n".join(path.read_text(encoding="utf-8") for path in PROJECT.glob("*.cs")) for marker in ( "v1/software-nodes/enroll", "v1/software-nodes/connect", 'SetRequestHeader("Authorization"', 'SetRequestHeader("X-Node-Id"', "DataProtectionScope.LocalMachine", "SetAccessRuleProtection(isProtected: true", '"adapter.json"', "NotifyIcon", "ConfigurationForm", "TrayIconFactory.Create", '"--headless"', ): self.assertIn(marker, source) def test_workspace_storage_is_global_by_workspace_not_user_directory(self) -> None: models = (PROJECT / "NodeModels.cs").read_text(encoding="utf-8") store = (PROJECT / "WorkspaceStore.cs").read_text(encoding="utf-8") runner = (PROJECT / "AdapterProcessRunner.cs").read_text(encoding="utf-8") self.assertIn('Path.Combine(root, "workspaces")', models) self.assertIn('Binding(job).WorkspaceId.ToString("D")', store) self.assertNotIn("UserId", store) self.assertIn('Path.Combine(root, "current")', store) self.assertIn('Path.Combine(root, "rollback")', store) self.assertIn("workspaceStore.Promote", runner) self.assertIn("workspaceStore.Restore", runner) def test_runtime_auto_reports_all_discovered_adapter_capabilities(self) -> None: connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") config_store = (PROJECT / "NodeConfigStore.cs").read_text(encoding="utf-8") self.assertIn( "var installedCapabilities = adapters.All.Select(item => item.Capability).ToArray()", connection, ) self.assertIn("capabilities = installedCapabilities", connection) self.assertIn("installed_capabilities = installedCapabilities", connection) self.assertNotIn("config.Capabilities.Contains", connection) self.assertNotIn("InstalledCapabilities.Contains", config_store) def test_node_does_not_expose_arbitrary_execution_primitives(self) -> None: source = "\n".join( path.read_text(encoding="utf-8") for path in PROJECT.glob("*.cs") if path.name not in {"AdapterProcessRunner.cs", "ConfigurationForm.cs"} ) for forbidden in ("Process.Start", "cmd.exe", "powershell.exe", "LabTalk"): self.assertNotIn(forbidden, source) form = (PROJECT / "ConfigurationForm.cs").read_text(encoding="utf-8") self.assertIn('"explorer.exe"', form) self.assertIn("startInfo.ArgumentList.Add(dataRoot.Text)", form) self.assertNotIn("UseShellExecute = true", form) def test_data_root_is_configurable_and_migrated_before_switching(self) -> None: settings = (PROJECT / "NodeDataRoot.cs").read_text(encoding="utf-8") models = (PROJECT / "NodeModels.cs").read_text(encoding="utf-8") form = (PROJECT / "ConfigurationForm.cs").read_text(encoding="utf-8") tray = (PROJECT / "TrayApplicationContext.cs").read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") self.assertIn('"ZCBOT_WINDOWS_NODE_DATA_DIR"', settings) self.assertLess( settings.index("Environment.GetEnvironmentVariable"), settings.index("Registry.CurrentUser.OpenSubKey"), ) self.assertIn("NodeDataRootSettings.Resolve().RootDirectory", models) self.assertIn("SHA256.HashData(stream)", settings) self.assertIn("Directory.Move(staging, target)", settings) self.assertLess( settings.index("VerifyTree(source, staging"), settings.index("Directory.Move(staging, target)"), ) self.assertIn("DataRootMigrationRequested", form) self.assertIn("jobInbox.HasPendingJobs", form) self.assertIn( "NodeDataRootSettings.SaveUserRoot(result.TargetDirectory)", tray ) self.assertIn("jobPipelines.Values.Concat(exportPipelines.Values)", connection) self.assertIn( "Path.GetPathRoot(NodePaths.ForCurrentMachine().RootDirectory)", connection ) def test_config_field_names_do_not_serialize_plain_node_token(self) -> None: models = (PROJECT / "NodeModels.cs").read_text(encoding="utf-8") stored_record = models.split("internal sealed record StoredNodeConfig", 1)[1].split(");", 1)[0] self.assertIn("ProtectedNodeToken", stored_record) self.assertNotIn("string NodeToken", stored_record) def test_configuration_ui_never_displays_or_copies_token(self) -> None: form = (PROJECT / "ConfigurationForm.cs").read_text(encoding="utf-8") self.assertIn("CreateTextBox(usePassword: true)", form) self.assertIn("UseSystemPasswordChar = usePassword", form) self.assertNotIn("NodeToken", form) self.assertIn("Clipboard.SetText(BuildDiagnosticText(currentConfig))", form) diagnostics = form.split("private string BuildDiagnosticText", 1)[1].split( "private async Task RunAnsysAcceptanceAsync", 1 )[0] self.assertNotIn("NodeToken", diagnostics) self.assertIn("内容不包含 Node Token", form) self.assertIn("清除本机身份并重新注册", form) self.assertIn("管理后台删除或禁用云端旧节点", form) def test_ansys_acceptance_ui_runs_only_fixed_bundled_suite_before_gate(self) -> None: form = (PROJECT / "ConfigurationForm.cs").read_text(encoding="utf-8") adapters = (PROJECT / "NodeAdapters.cs").read_text(encoding="utf-8") runner = (PROJECT / "AdapterProcessRunner.cs").read_text(encoding="utf-8") self.assertIn("ANSYS Mechanical 真机验收", form) self.assertIn("运行内置基准验收", form) self.assertNotIn("new OpenFileDialog", form) self.assertIn("AcceptanceReportPassed", form) self.assertIn('"ZCBOT_ANSYS_242_VALIDATED", "1"', form) self.assertIn("EnvironmentVariableTarget.Machine", form) self.assertIn("SupportsLocalAcceptance", adapters) self.assertIn('Path.Combine(descriptor.DirectoryPath, "acceptance.py")', runner) self.assertIn( 'Path.Combine(descriptor.DirectoryPath, "acceptance-coupon.step")', runner ) self.assertIn('["--work-root", root, "--repeat", "20"', runner) self.assertIn("Adapter acceptance cannot run while a dispatched job is active", runner) acceptance = form.split("private async Task RunAnsysAcceptanceAsync", 1)[1].split( "private void EnableAnsysGate", 1 )[0] self.assertNotIn("Process.Start", acceptance) def test_configuration_window_is_resizable_and_dpi_safe(self) -> None: form = (PROJECT / "ConfigurationForm.cs").read_text(encoding="utf-8") self.assertIn("ClientSize = new Size(840, 680)", form) self.assertIn("FormBorderStyle.Sizable", form) self.assertIn("AutoScaleMode.Dpi", form) self.assertIn("AutoScroll = true", form) self.assertNotIn("MaximumSize = new Size(410", form) self.assertIn("注册码默认 10 分钟有效", form) self.assertIn("成功注册一次后立即失效", form) self.assertIn("CreateCard", form) self.assertIn("注册并连接", form) self.assertIn("ContentWidth = 760", form) self.assertIn("NodeAdapterRegistry.InstalledContracts", form) self.assertIn("CreateCapabilityRow(contract.DisplayName, contract.Capability)", form) self.assertIn( 'runtime.Software == "OriginPro" ? "Origin" : runtime.Software', form ) self.assertIn('$"{softwareName} {runtime.SoftwareVersion}"', form) self.assertNotIn('$"Origin {runtime.SoftwareVersion}"', form) self.assertIn('CreateButton("立即重连", 112, primary: true)', form) self.assertIn("ReconnectRequested?.Invoke()", form) self.assertIn("registrationCard.Visible = !registered", form) self.assertIn("reconnect.Visible = registered", form) self.assertIn("resetIdentity.Visible = registered", form) self.assertIn('CreateSectionTitle("本机任务")', form) self.assertIn("DataGridView", form) self.assertIn("jobInbox.ReadJobSnapshots()", form) self.assertIn('"software_running" => "软件执行中"', form) self.assertNotIn("HttpClient", form) tray = (PROJECT / "TrayApplicationContext.cs").read_text(encoding="utf-8") self.assertIn("form.ReconnectRequested += RestartConnection", tray) def test_local_job_monitor_is_persisted_and_software_neutral(self) -> None: inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") models = (PROJECT / "JobMonitorModels.cs").read_text(encoding="utf-8") self.assertIn('"state.json"', inbox) self.assertIn("ReadJobSnapshots", inbox) self.assertIn("AtomicWrite(path, content, overwrite: true)", inbox) for stage in ( "accepted", "downloading_inputs", "ready_to_run", "software_running", "uploading_outputs", "succeeded", "failed", "cancelled", ): self.assertIn(f'"{stage}"', inbox + connection) self.assertIn("JobDisplaySnapshot", models) self.assertNotIn("origin_running", inbox + connection + models) def test_unified_installer_registers_login_startup(self) -> None: launcher = (ROOT / "install-windows-node.bat").read_text(encoding="utf-8") self.assertNotIn("powershell", launcher.lower()) self.assertIn('set "NODE_EXE=%~dp0Zcbot.WindowsNode.exe"', launcher) self.assertIn('schtasks.exe /Create /TN "Zcbot Windows Node"', launcher) self.assertIn("/SC ONLOGON", launcher) self.assertIn("/RL LIMITED", launcher) self.assertIn("/IT", launcher) self.assertIn("pause >nul", launcher) def test_unified_installer_detects_origin_and_python_312(self) -> None: script = (ROOT / "install-windows-node.bat").read_text(encoding="utf-8") self.assertIn('set "NODE_DATA_ROOT=%ZCBOT_WINDOWS_NODE_DATA_DIR%"', script) self.assertIn('reg.exe query "HKCU\\Software\\Zcbot\\WindowsNode" /v DataRoot', script) self.assertIn( 'if not defined NODE_DATA_ROOT set "NODE_DATA_ROOT=%ProgramData%\\Zcbot\\WindowsNode"', script, ) self.assertIn(":detect_origin", script) self.assertIn('reg.exe query "HKCR\\Origin.ApplicationSI\\CLSID"', script) self.assertIn(":create_python_runtime", script) self.assertIn("py.exe -3.12", script) self.assertIn("sys.version_info[:2] == (3, 12)", script) self.assertIn( 'set "ORIGIN_RUNTIME_DIR=%NODE_DATA_ROOT%\\runtimes\\origin"', script, ) self.assertIn( 'set "ANSYS_RUNTIME_DIR=%NODE_DATA_ROOT%\\runtimes\\ansys"', script, ) self.assertIn(":detect_ansys_242", script) self.assertIn("v242\\aisol\\bin\\winx64\\AnsysWBU.exe", script) self.assertIn( 'call :create_python_runtime "%ORIGIN_RUNTIME_DIR%" "%~1"', script ) self.assertIn( 'call :create_python_runtime "%ANSYS_RUNTIME_DIR%" "%~1"', script ) self.assertIn('py.exe -3.12 -m venv "%~1"', script) self.assertIn('python.exe -m venv "%~1"', script) self.assertNotIn('"%PYTHON_EXE%"', script) def test_unified_installer_detects_blender_and_creates_launcher_runtime(self) -> None: script = (ROOT / "install-windows-node.bat").read_text(encoding="utf-8") self.assertIn(":detect_blender", script) self.assertIn("ZCBOT_BLENDER_EXE", script) self.assertIn("Blender Foundation\\Blender *", script) self.assertIn( 'set "BLENDER_RUNTIME_DIR=%NODE_DATA_ROOT%\\runtimes\\blender"', script, ) self.assertIn( 'call :create_python_runtime "%BLENDER_RUNTIME_DIR%" "%~1"', script ) def test_unified_installer_uses_configurable_domestic_pypi(self) -> None: script = (ROOT / "install-windows-node.bat").read_text(encoding="utf-8") self.assertIn( 'if not defined ZCBOT_PIP_INDEX_URL set "ZCBOT_PIP_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple/"', script, ) self.assertEqual(script.count('--index-url "%ZCBOT_PIP_INDEX_URL%"'), 2) def test_publish_output_contains_the_complete_installer_payload(self) -> None: project = (PROJECT / "Zcbot.WindowsNode.csproj").read_text(encoding="utf-8") self.assertIn("..\\install-windows-node.bat", project) self.assertIn("..\\adapters\\**\\*", project) self.assertIn("adapters\\%(RecursiveDir)%(Filename)%(Extension)", project) self.assertIn("..\\..\\software-contracts\\origin.plot.v2.json", project) self.assertIn("adapters\\origin.plot@v2\\origin.plot.v2.json", project) self.assertIn( "..\\..\\software-contracts\\ansys.geometry.inspect.v1.json", project, ) self.assertIn( "adapters\\ansys.geometry.inspect@v1\\ansys.geometry.inspect.v1.json", project, ) self.assertIn( "..\\..\\software-contracts\\ansys.mechanical.static_structural.v2.json", project, ) self.assertIn( "adapters\\ansys.mechanical.static_structural@v2\\" "ansys.mechanical.static_structural.v2.json", project, ) self.assertIn( "..\\..\\software-contracts\\blender.scene.author.v1.json", project, ) self.assertIn( "adapters\\blender.scene.author@v1\\blender.scene.author.v1.json", project, ) self.assertNotIn("software-contracts\\%(Filename)%(Extension)", project) self.assertFalse((ROOT / "install-windows-node.ps1").exists()) self.assertFalse((ROOT / "install-origin-runtime.ps1").exists()) self.assertFalse((ROOT / "install-startup.ps1").exists()) def test_packager_defaults_to_framework_dependent_verified_zip(self) -> None: script = (ROOT / "package-windows-node.bat").read_text(encoding="utf-8") self.assertNotIn("powershell", script.lower()) self.assertIn('set "SELF_CONTAINED=false"', script) self.assertIn('if /I "%~1"=="--self-contained"', script) self.assertIn("dotnet publish", script) self.assertIn('"install-windows-node.bat"', script) self.assertIn('for /r "%PUBLISH_DIR%\\adapters" %%F in (adapter.json)', script) self.assertIn('do if exist "%%~fF"', script) self.assertIn("ADAPTER_CONTRACT_COUNT", script) self.assertIn("Published adapter has no colocated contract", script) self.assertNotIn('"!PUBLISH_DIR!\\software-contracts\\*.json"', script) self.assertIn("tar.exe -a -c -f", script) self.assertIn("certutil.exe -hashfile", script) self.assertFalse((ROOT / "package-windows-node.ps1").exists()) def test_auth_rejection_is_distinct_from_http_websocket_handshake_failure(self) -> None: connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") self.assertIn('socket.HttpStatusCode is HttpStatusCode.Unauthorized', connection) self.assertIn('or HttpStatusCode.Forbidden', connection) self.assertIn("WebSocket 握手被拒绝，请检查服务端或反向代理", connection) self.assertIn("catch (NodeEndpointException exception)", connection) self.assertIn("throw new NodeEndpointException", connection) self.assertIn("(int?)result.CloseStatus == 4003", connection) self.assertIn("节点身份已被服务端拒绝", connection) self.assertNotIn("Node credentials were rejected", connection) def test_adapter_runtime_probe_is_worker_owned_and_reported(self) -> None: runner = (PROJECT / "AdapterProcessRunner.cs").read_text(encoding="utf-8") worker = ( ROOT / "adapters" / "origin.plot@v2" / "worker.py" ).read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") self.assertIn('r"Origin.ApplicationSI\\CLSID"', worker) self.assertIn('["--probe"]', runner) self.assertIn('root.GetProperty("adapter_version")', runner) self.assertIn('startInfo.Environment["PYTHONUTF8"] = "1"', runner) self.assertIn('startInfo.Environment["PYTHONIOENCODING"] = "utf-8"', runner) form = (PROJECT / "ConfigurationForm.cs").read_text(encoding="utf-8") self.assertIn('Adapter {adapter.AdapterVersion}', form) self.assertIn('$"{softwareName} 版本未知"', form) self.assertIn( "&& !jobInbox.HasPendingJobs", connection, ) self.assertIn("originAdapter?.HasActiveJobs", connection) self.assertIn( "ReadRecoverableJobs().Any(item => item.Terminal is null)", (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8"), ) self.assertNotIn("CreateInstance", worker) for marker in ( "software_version = origin.SoftwareVersion", "adapter_version = origin.AdapterVersion", "health = origin.Health", "detail = origin.Detail", ): self.assertIn(marker, connection) def test_job_offer_is_persisted_before_acceptance(self) -> None: inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") adapters = (PROJECT / "NodeAdapters.cs").read_text(encoding="utf-8") self.assertIn("adapters.Find(capability)", inbox) self.assertIn("adapter.ValidateRequest(request)", inbox) self.assertIn( 'JsonSchema.Build(contractRoot.GetProperty("request_schema").Clone())', adapters, ) self.assertIn("var contractPath = ResolveFile(root, manifest.Contract)", adapters) self.assertNotIn("ResolveContractFile", adapters) self.assertNotIn('AppContext.BaseDirectory, "software-contracts"', adapters) self.assertNotIn("IsValidOriginRequest", inbox) self.assertNotIn("IsValidPlot", inbox) self.assertIn('root.TryGetProperty("input_transfers"', inbox) self.assertIn('"input", key, filename', inbox) self.assertIn("FileOptions.WriteThrough", inbox) self.assertIn("stream.Flush(flushToDisk: true)", inbox) new_record = inbox.split("var record =", 1)[1].split("private static JsonElement?", 1)[0] self.assertLess( new_record.index("AtomicWrite(requestPath, record"), new_record.index("JobOfferResult.Accept"), ) self.assertIn('offerResult.Accepted ? "job_accept" : "job_reject"', connection) self.assertIn("sendLock.WaitAsync", connection) self.assertIn("!jobInbox.HasPendingJobs", connection) self.assertIn("capability_runtime = capabilityRuntime", connection) self.assertIn("adapter.RunAsync(job)", connection) self.assertIn("ReportRecoverableJobsAsync", connection) self.assertIn("ConcurrentDictionary jobPipelines", connection) self.assertIn("StartJobPipeline(socket, acceptedJob)", connection) self.assertIn("inputDownloader.DownloadAsync(job, CancellationToken.None)", connection) self.assertIn('stage = "uploading_outputs"', connection) self.assertIn('stage = "software_running"', connection) self.assertNotIn('stage = "origin_running"', connection) self.assertIn("&& !job.UploadComplete", connection) self.assertIn("&& !job.CloudTerminal", connection) self.assertIn( "await Task.WhenAll(jobPipelines.Values.Concat(exportPipelines.Values).ToArray())", connection, ) self.assertIn("StartJobPipeline(socket, job)", connection) self.assertIn('stage = "downloading_inputs"', connection) self.assertIn('Path.Combine(jobDirectory, "terminal.json")', inbox) self.assertIn("AtomicWrite(requestPath, updated, overwrite: true)", inbox) downloader = (PROJECT / "JobInputDownloader.cs").read_text(encoding="utf-8") self.assertIn('new AuthenticationHeaderValue("Bearer", config.NodeToken)', downloader) self.assertIn('DefaultRequestHeaders.Add("X-Node-Id"', downloader) self.assertIn("HttpCompletionOption.ResponseHeadersRead", downloader) self.assertIn("IncrementalHash.CreateHash", downloader) self.assertIn("total > expectedSize", downloader) self.assertIn("File.Move(temporaryPath, destination, overwrite: false)", downloader) self.assertNotIn("Process.Start", downloader) def test_adapter_worker_launch_is_manifest_driven_and_terminal_driven(self) -> None: runner = (PROJECT / "AdapterProcessRunner.cs").read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") project = (PROJECT / "Zcbot.WindowsNode.csproj").read_text(encoding="utf-8") worker = ( ROOT / "adapters" / "origin.plot@v2" / "worker.py" ).read_text(encoding="utf-8") self.assertIn('Environment.GetEnvironmentVariable("ZCBOT_ORIGIN_PYTHON")', runner) self.assertIn( 'Path.Combine(paths.RootDirectory, "runtimes", runtimeId, "Scripts", "python.exe")', runner, ) self.assertIn("UseShellExecute = false", runner) self.assertIn("command.PrefixArguments.Concat(arguments)", runner) self.assertIn("descriptor.EntrypointPath", runner) self.assertIn('Path.Combine(jobDirectory, "terminal.json")', runner) self.assertIn('"NODE_RESTARTED_DURING_JOB"', runner) self.assertIn("CancellationTokenSource.CreateLinkedTokenSource", runner) self.assertIn("process.Kill(entireProcessTree: true)", runner) self.assertIn('type.GetString() == "job_cancel"', connection) self.assertIn('"cancelled", "USER_CANCELLED"', connection) self.assertIn("..\\adapters\\**\\*", project) self.assertIn("descriptor.Manifest.WorkerTimeoutMinutes", runner) adapters = (PROJECT / "NodeAdapters.cs").read_text(encoding="utf-8") self.assertIn( 'OptionalInteger(manifestRoot, "worker_timeout_minutes", 30, 1, 1440)', adapters, ) self.assertIn("if op.oext:", worker) self.assertIn("op.exit()", worker) self.assertIn("op.new_graph", worker) self.assertIn("layer.add_plot", worker) self.assertIn("op.save", worker) self.assertIn("graph.save_fig", worker) self.assertIn('_atomic_json(job_dir / "terminal.json"', worker) for forbidden in ("subprocess", "eval(", "exec(", "os.system"): self.assertNotIn(forbidden, worker) def test_origin_adapter_can_be_packaged_without_building_node(self) -> None: script = (ROOT / "package-origin-adapter.bat").read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") self.assertIn('"adapters\\origin.plot@v2\\adapter.json"', script) self.assertIn('"adapters\\origin.plot@v2\\worker.py"', script) self.assertIn('"..\\software-contracts\\origin.plot.v2.json"', script) self.assertNotIn("dotnet", script.lower()) form = (PROJECT / "ConfigurationForm.cs").read_text(encoding="utf-8") self.assertIn("Contract SHA-256: {adapter.ContractSha256}", form) self.assertIn("Workspace protocol:", form) self.assertIn("Workspace state:", form) self.assertIn("workspace_protocol_version", connection) self.assertIn("contract_sha256", connection) uploader = (PROJECT / "JobOutputUploader.cs").read_text(encoding="utf-8") self.assertIn('new AuthenticationHeaderValue("Bearer", config.NodeToken)', uploader) self.assertIn('DefaultRequestHeaders.Add("X-Node-Id"', uploader) self.assertIn('DefaultRequestHeaders.Add("X-Lease-Id"', uploader) self.assertIn("SHA256.HashDataAsync", uploader) self.assertIn("upload-complete.json", connection + uploader) self.assertIn("adapter.PreviewOutputIds", connection) self.assertIn("adapter.WorkspaceStateFilename", connection) self.assertIn("WorkspaceSize(jobDirectory), allowConflict: true", uploader) self.assertIn("completeResponse.StatusCode == HttpStatusCode.Conflict", uploader) self.assertIn('"JOB_ALREADY_TERMINAL"', uploader) self.assertIn('"cloud-terminal.json"', uploader) self.assertIn("OpenOutputWithRetryAsync", uploader) self.assertIn("IsSharingViolation", uploader) self.assertIn("(exception.HResult & 0xFFFF) is 32 or 33", uploader) self.assertIn("artifact={artifactId}", uploader) self.assertIn("hresult=0x{exception.HResult:X8}", connection + uploader) self.assertIn('"node-output-upload.log"', uploader) self.assertIn("MaxDiagnosticLogBytes", uploader) self.assertIn("outputUploader.RecordDeferred(current, exception)", connection) self.assertIn('completionPath + ".pending"', uploader) self.assertIn("PreparePendingCompletion(pendingPath, responseBody)", uploader) self.assertIn("File.Move(pendingPath, completionPath, overwrite: false)", uploader) self.assertIn("operation=move_pending", uploader) self.assertNotIn('".tmp-" + Guid.NewGuid()', uploader) self.assertNotIn("Process.Start", uploader) tray = (PROJECT / "TrayApplicationContext.cs").read_text(encoding="utf-8") self.assertIn("_ = form.Handle", tray) self.assertIn( 'UpdateStatus(NodeStatus.Create(NodeState.Connecting, "正在连接 zcbot…"))', tray, ) def test_blender_adapter_can_be_packaged_without_building_node(self) -> None: script = (ROOT / "package-blender-adapter.bat").read_text(encoding="utf-8") self.assertIn("adapters\\blender.scene.author@v1", script) self.assertIn("\\worker.py", script) self.assertIn("\\blender_worker.py", script) self.assertIn("..\\software-contracts\\blender.scene.author.v1.json", script) self.assertIn("tar.exe -a -c -f", script) self.assertNotIn("dotnet", script.lower()) def test_local_job_monitor_never_shows_update_before_acceptance(self) -> None: inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8") self.assertIn("state.UpdatedAt > acceptedAt", inbox) if __name__ == "__main__": unittest.main()