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_without_third_party_packages(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") self.assertEqual(root.findall("./ItemGroup/PackageReference"), []) 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", '"origin.plot@v1"', "NotifyIcon", "ConfigurationForm", "TrayIconFactory.Create", '"--headless"', ): self.assertIn(marker, source) 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 != "OriginWorkerRunner.cs" ) for forbidden in ("Process.Start", "cmd.exe", "powershell.exe", "LabTalk"): self.assertNotIn(forbidden, source) 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.assertNotIn("Clipboard", form) self.assertIn("清除本机身份并重新注册", form) self.assertIn("管理后台删除或禁用云端旧节点", form) 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("Origin 绘图", form) self.assertIn('CreateCapabilityRow("Origin 绘图", "origin.plot@v1")', 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) tray = (PROJECT / "TrayApplicationContext.cs").read_text(encoding="utf-8") self.assertIn("form.ReconnectRequested += RestartConnection", tray) def test_startup_task_is_login_scoped_and_runs_the_fixed_node_executable(self) -> None: script = (ROOT / "install-startup.ps1").read_text(encoding="utf-8") self.assertIn('GetFileName($resolvedExecutable) -ne "Zcbot.WindowsNode.exe"', script) self.assertIn("New-ScheduledTaskTrigger -AtLogOn -User $currentUser", script) self.assertIn("-LogonType Interactive", script) self.assertIn("-RunLevel Limited", script) self.assertNotIn("-RunLevel Highest", script) 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_origin_runtime_probe_is_read_only_and_reported(self) -> None: probe = (PROJECT / "OriginRuntimeProbe.cs").read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") self.assertIn('AutomationProgId = @"Origin.ApplicationSI\\CLSID"', probe) self.assertIn("RegistryHive.LocalMachine", probe) self.assertIn("RegistryHive.CurrentUser", probe) self.assertIn('new("OriginPro", version, "0.2.0", health, detail)', probe) self.assertIn( 'available_slots = origin.Health == "ready" && !jobInbox.HasPendingJobs ? 1 : 0', connection, ) self.assertNotIn("CreateInstance", probe) self.assertNotIn("Process.Start", probe) 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") self.assertIn('capabilityValue.GetString() != "origin.plot@v1"', inbox) self.assertIn("PlotTypes.Contains", inbox) self.assertIn("OutputFormats.Contains", 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 ? 1 : 0", 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("&& !job.UploadComplete", connection) self.assertIn("StartJobPipeline(socket, job)", connection) self.assertIn('stage = "waiting_input"', 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_origin_worker_launch_is_fixed_and_terminal_driven(self) -> None: runner = (PROJECT / "OriginWorkerRunner.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 / "origin-worker" / "worker.py").read_text(encoding="utf-8") self.assertIn('Environment.GetEnvironmentVariable("ZCBOT_ORIGIN_PYTHON")', runner) self.assertIn('Path.Combine(paths.RootDirectory, "runtimes", "origin", "python.exe")', runner) self.assertIn("UseShellExecute = false", runner) self.assertIn("startInfo.ArgumentList.Add(workerScript)", runner) self.assertIn("startInfo.ArgumentList.Add(jobDirectory)", runner) self.assertIn('Path.Combine(jobDirectory, "terminal.json")', runner) self.assertIn('"NODE_RESTARTED_DURING_JOB"', runner) self.assertNotIn("RunAsync(RecoverableJob job, CancellationToken", runner) self.assertIn("origin-worker\\worker.py", project) 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) 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.assertNotIn("Process.Start", uploader) if __name__ == "__main__": unittest.main()