"""任务级选项（model/effort/approval）端到端单测。 覆盖：POST /tasks 校验与落库回读、GET /relay/models（注入取数， 不联网不碰真 key）、LocalExecutor/DockerExecutor 命令行里 的 --model/--effort/--approval-policy/--sandbox 位置与映射。 不跑真实模型，不启动 worker。 """ import http.client import json import os import sys import tempfile import threading import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.auth import AuthStore from cloud.relay import _reset_cache_for_tests from cloud.sandbox import DockerExecutor, LocalExecutor from cloud.sandbox import _approval_cli_args from cloud.server import create_server except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.relay import _reset_cache_for_tests from cloud.sandbox import DockerExecutor, LocalExecutor from cloud.sandbox import _approval_cli_args from cloud.server import create_server class TaskOptionsTestCase(unittest.TestCase): def setUp(self): _reset_cache_for_tests() self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.server = create_server(self.db_path, "127.0.0.1", 0) self.server.relay_fetcher = lambda u, h, t: { "data": [{"id": "gpt-6.1-sol"}, {"id": "gpt-6-astra"}]} self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, self.project = self._req( "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201) def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() _reset_cache_for_tests() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) headers = {"Content-Type": "application/json"} if key: headers["X-API-Key"] = key conn.request(method, path, json.dumps(body) if body is not None else None, headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) except ValueError: data = None return resp.status, data def test_create_task_with_options_roundtrip(self): status, task = self._req("POST", "/tasks", { "project_id": self.project["id"], "prompt": "hi", "model": "gpt-6-astra", "effort": "high", "approval": "full"}, self.owner_key) self.assertEqual(status, 201) self.assertEqual(task["model"], "gpt-6-astra") self.assertEqual(task["effort"], "high") self.assertEqual(task["approval"], "full") status, tasks = self._req( "GET", "/tasks?project_id=" + self.project["id"], key=self.owner_key) self.assertEqual(status, 200) self.assertEqual(tasks["tasks"][0]["model"], "gpt-6-astra") def test_create_task_defaults_are_null(self): status, task = self._req("POST", "/tasks", { "project_id": self.project["id"], "prompt": "hi"}, self.owner_key) self.assertEqual(status, 201) self.assertIsNone(task["model"]) self.assertIsNone(task["effort"]) self.assertIsNone(task["approval"]) def test_create_task_invalid_options_rejected(self): for body in ({"effort": "ultra"}, {"approval": "yes"}, {"model": ""}, {"model": 42}): payload = {"project_id": self.project["id"], "prompt": "hi"} payload.update(body) status, data = self._req("POST", "/tasks", payload, self.owner_key) self.assertEqual(status, 400, body) def test_relay_models_endpoint(self): status, data = self._req("GET", "/relay/models", key=self.owner_key) self.assertEqual(status, 200) self.assertEqual(data["models"], ["gpt-6.1-sol", "gpt-6-astra"]) # 响应里只有模型 id，绝无密钥字样。 self.assertNotIn("Bearer", json.dumps(data)) self.assertNotIn("XCM_API_KEY", json.dumps(data)) def test_relay_models_requires_auth(self): status, _ = self._req("GET", "/relay/models") self.assertEqual(status, 401) def test_relay_models_failure_is_502(self): from cloud.relay import RelayError self.server.relay_fetcher = lambda u, h, t: (_ for _ in ()).throw( RelayError("relay down")) _reset_cache_for_tests() status, data = self._req("GET", "/relay/models", key=self.owner_key) self.assertEqual(status, 502) self.assertIn("relay down", data["error"]) def test_approval_mapping(self): self.assertEqual(_approval_cli_args("ask"), ["--approval-policy", "untrusted"]) self.assertEqual(_approval_cli_args("auto"), ["--approval-policy", "never"]) self.assertEqual(_approval_cli_args("full"), ["--approval-policy", "never", "--sandbox", "danger-full-access"]) self.assertEqual(_approval_cli_args(None), []) self.assertEqual(_approval_cli_args("bogus"), []) def test_local_dispatch_command_carries_options(self): seen = [] def fake_runner(cmd): seen.append(cmd) return SimpleNamespace(returncode=0, stdout="{}") ex = LocalExecutor("/fake/host.py", config_path="/fake/cfg.json", runner=fake_runner) ex.dispatch_task({"id": "t1", "prompt": "hi", "model": "gpt-6-astra", "effort": "max", "approval": "full"}) cmd = seen[0] # 顶层全局参数必须在子命令之前。 self.assertLess(cmd.index("--model"), cmd.index("dispatch")) self.assertLess(cmd.index("--effort"), cmd.index("dispatch")) self.assertLess(cmd.index("--approval-policy"), cmd.index("dispatch")) self.assertLess(cmd.index("--sandbox"), cmd.index("dispatch")) self.assertEqual(cmd[cmd.index("--model") + 1], "gpt-6-astra") self.assertEqual(cmd[cmd.index("--effort") + 1], "max") self.assertEqual(cmd[cmd.index("--approval-policy") + 1], "never") self.assertEqual(cmd[cmd.index("--sandbox") + 1], "danger-full-access") def test_local_dispatch_without_options_passes_nothing(self): seen = [] def fake_runner(cmd): seen.append(cmd) return SimpleNamespace(returncode=0, stdout="{}") ex = LocalExecutor("/fake/host.py", runner=fake_runner) ex.dispatch_task({"id": "t1", "prompt": "hi"}) cmd = seen[0] self.assertNotIn("--effort", cmd) self.assertNotIn("--approval-policy", cmd) self.assertNotIn("--sandbox", cmd) def test_docker_inner_command_carries_options(self): ex = DockerExecutor(config_path="/fake/cfg.json") cmd = ex._inner_host_command( "dispatch", "--name", "x", "--task", "hi", model="gpt-6-astra", effort="high", approval="ask") self.assertLess(cmd.index("--model"), cmd.index("dispatch")) self.assertLess(cmd.index("--effort"), cmd.index("dispatch")) self.assertEqual(cmd[cmd.index("--approval-policy") + 1], "untrusted") self.assertNotIn("--sandbox", cmd) # -- G9：worker 路径默认忽略 host 预算 ------------------------------- def test_local_command_ignore_host_budget_flag(self): seen = [] def fake_runner(cmd): seen.append(cmd) return SimpleNamespace(returncode=0, stdout="{}") ex = LocalExecutor("/fake/host.py", runner=fake_runner, ignore_host_budget=True) ex.dispatch_task({"id": "t1", "prompt": "hi"}) cmd = seen[0] self.assertIn("--ignore-host-budget", cmd) # 顶层全局参数，必须在子命令之前。 self.assertLess(cmd.index("--ignore-host-budget"), cmd.index("dispatch")) ex2 = LocalExecutor("/fake/host.py", runner=fake_runner) ex2.dispatch_task({"id": "t1", "prompt": "hi"}) self.assertNotIn("--ignore-host-budget", seen[1]) def test_docker_inner_command_ignore_host_budget_flag(self): ex = DockerExecutor(config_path="/fake/cfg.json", ignore_host_budget=True) cmd = ex._inner_host_command("dispatch", "--name", "x", "--task", "hi") self.assertIn("--ignore-host-budget", cmd) self.assertLess(cmd.index("--ignore-host-budget"), cmd.index("dispatch")) ex2 = DockerExecutor(config_path="/fake/cfg.json") cmd2 = ex2._inner_host_command("dispatch", "--name", "x", "--task", "hi") self.assertNotIn("--ignore-host-budget", cmd2) if __name__ == "__main__": unittest.main()