codex-app-server/deploy/tests/test_verify_multiuser_helpers.py

"""deploy/verify-multiuser.py 纯 helper 单测（CI 第二阶段覆盖率：
基线 0%）。只测 check() 与 req()：

- check() 写 RESULTS 台账并打印 PASS/FAIL；
- req() 对本地回环 fake HTTP 成功/HTTPError/非 JSON 体/
  连接拒绝四路的返回值与请求头。

main() 是主机侧对真实控制面的多用户实弹验证（建
multiuser-test-* 用户/项目/任务、调 merge），按任务门禁
“不要建生产任务”，本轨不以 mock 冒充它——main() 仍为
未覆盖，如实留在缺口里。
"""

import http.server
import importlib.util
import io
import json
import os
import socket
import threading
import unittest
from contextlib import redirect_stdout
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[2]
_SPEC = importlib.util.spec_from_file_location(
    "verify_multiuser",
    os.path.join(REPO_ROOT, "deploy", "verify-multiuser.py"))
vm = importlib.util.module_from_spec(_SPEC)
_SPEC.loader.exec_module(vm)


class Handler(http.server.BaseHTTPRequestHandler):
    seen = []

    def _send(self, code, body, ctype="application/json"):
        data = body if isinstance(body, bytes) else body.encode()
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        Handler.seen.append((self.path, dict(self.headers)))
        if self.path == "/ok":
            self._send(200, json.dumps({"ok": True}))
        elif self.path == "/nf":
            self._send(404, json.dumps({"error": "nope"}))
        else:
            self._send(404, "not json", "text/plain")

    def do_POST(self):
        length = int(self.headers.get("Content-Length") or 0)
        Handler.seen.append(
            (self.path, dict(self.headers), self.rfile.read(length)))
        self._send(201, json.dumps({"created": True}))

    def log_message(self, *args):
        pass


class ReqTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.httpd = http.server.ThreadingHTTPServer(
            ("127.0.0.1", 0), Handler)
        cls.port = cls.httpd.server_address[1]
        cls.thread = threading.Thread(target=cls.httpd.serve_forever,
                                      daemon=True)
        cls.thread.start()

    @classmethod
    def tearDownClass(cls):
        cls.httpd.shutdown()
        cls.httpd.server_close()

    def setUp(self):
        Handler.seen = []

    def base(self):
        return "http://127.0.0.1:%d" % self.port

    def test_get_ok(self):
        code, body = vm.req(self.base(), "GET", "/ok")
        self.assertEqual((code, body), (200, {"ok": True}))

    def test_http_error_json_and_non_json(self):
        code, body = vm.req(self.base(), "GET", "/nf")
        self.assertEqual((code, body), (404, {"error": "nope"}))
        code, body = vm.req(self.base(), "GET", "/bad")
        self.assertEqual((code, body), (404, {}))

    def test_post_headers_and_body(self):
        code, body = vm.req(self.base(), "POST", "/ok",
                            key="k-123", body={"a": 1})
        self.assertEqual(code, 201)
        self.assertEqual(body, {"created": True})
        path, headers, raw = Handler.seen[0]
        self.assertEqual(headers.get("X-Api-Key"), "k-123")
        self.assertEqual(headers.get("Content-Type"), "application/json")
        self.assertEqual(json.loads(raw), {"a": 1})

    def test_connection_refused_returns_none(self):
        sock = socket.socket()
        sock.bind(("127.0.0.1", 0))
        dead_port = sock.getsockname()[1]
        sock.close()
        code, body = vm.req("http://127.0.0.1:%d" % dead_port,
                            "GET", "/ok")
        self.assertIsNone(code)
        self.assertIn("error", body)


class CheckTest(unittest.TestCase):
    def test_check_appends_and_prints(self):
        vm.RESULTS.clear()
        out = io.StringIO()
        with redirect_stdout(out):
            vm.check("alpha", True)
            vm.check("beta", 0)
        self.assertEqual(vm.RESULTS, [("alpha", True), ("beta", False)])
        self.assertIn("PASS alpha", out.getvalue())
        self.assertIn("FAIL beta", out.getvalue())
        vm.RESULTS.clear()


if __name__ == "__main__":
    unittest.main()