175 lines
6.1 KiB
Python
175 lines
6.1 KiB
Python
"""覆盖率 78%->85% 第 2 步：deploy/verify-multiuser.py 的 main()。
|
||
|
||
main() 是对真实控制面的多用户验证：本文件用本地临时 SQLite +
|
||
本地临时端口的真实控制面（test_server.py 同款 create_server）跑
|
||
完整流程，绝不连生产、不建生产任务；失败分支用本地 stub HTTP
|
||
服务与死端口驱动。临时库里的 multiuser-test-* 行跑完必须清零。
|
||
"""
|
||
|
||
import http.server
|
||
import importlib.util
|
||
import io
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from contextlib import redirect_stdout
|
||
from pathlib import Path
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
_SPEC = importlib.util.spec_from_file_location(
|
||
"verify_multiuser",
|
||
os.path.join(REPO_ROOT, "deploy", "verify-multiuser.py"))
|
||
assert _SPEC is not None and _SPEC.loader is not None
|
||
vm = importlib.util.module_from_spec(_SPEC)
|
||
_SPEC.loader.exec_module(vm)
|
||
|
||
from cloud.server import create_server, init_db # noqa: E402
|
||
|
||
|
||
class StubHandler(http.server.BaseHTTPRequestHandler):
|
||
mode = "wrong-id"
|
||
|
||
def _send(self, code, body):
|
||
data = json.dumps(body).encode()
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def do_POST(self):
|
||
length = int(self.headers.get("Content-Length") or 0)
|
||
self.rfile.read(length)
|
||
if self.mode == "wrong-id":
|
||
self._send(201, {"id": "wrong"})
|
||
else:
|
||
self._send(500, {"error": "boom"})
|
||
|
||
def do_GET(self):
|
||
self._send(404, {"error": "not found"})
|
||
|
||
def log_message(self, *args):
|
||
pass
|
||
|
||
|
||
def _start_stub(mode):
|
||
StubHandler.mode = mode
|
||
srv = http.server.HTTPServer(("127.0.0.1", 0), StubHandler)
|
||
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||
return srv
|
||
|
||
|
||
class VerifyMultiuserMainTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self._old_argv = sys.argv
|
||
|
||
def tearDown(self):
|
||
sys.argv = self._old_argv
|
||
self.tmp.cleanup()
|
||
|
||
def _run_main(self, base):
|
||
vm.RESULTS.clear()
|
||
sys.argv = ["verify-multiuser.py", "--base", base,
|
||
"--db", self.db_path]
|
||
out = io.StringIO()
|
||
with redirect_stdout(out):
|
||
rc = vm.main()
|
||
return rc, out.getvalue(), list(vm.RESULTS)
|
||
|
||
def _leftover_counts(self):
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
counts = {}
|
||
for table, col in (("users", "username"),
|
||
("projects", "id"),
|
||
("tasks", "project_id")):
|
||
counts[table] = conn.execute(
|
||
"SELECT COUNT(*) FROM %s WHERE %s LIKE ?"
|
||
% (table, col), ("multiuser-test-%",)).fetchone()[0]
|
||
return counts
|
||
finally:
|
||
conn.close()
|
||
|
||
def test_happy_path_against_real_local_server(self):
|
||
server = create_server(self.db_path, "127.0.0.1", 0)
|
||
port = server.server_address[1]
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
try:
|
||
rc, out, results = self._run_main(
|
||
"http://127.0.0.1:%d" % port)
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
thread.join(timeout=5)
|
||
self.assertEqual(rc, 0, out)
|
||
self.assertEqual(len(results), 13, results)
|
||
self.assertTrue(all(ok for _, ok in results), results)
|
||
self.assertIn("13/13", out)
|
||
self.assertEqual(self._leftover_counts(),
|
||
{"users": 0, "projects": 0, "tasks": 0})
|
||
|
||
def test_dead_base_fails_early(self):
|
||
init_db(self.db_path)
|
||
rc, out, results = self._run_main("http://127.0.0.1:1")
|
||
self.assertEqual(rc, 1)
|
||
self.assertEqual(len(results), 2)
|
||
self.assertFalse(any(ok for _, ok in results))
|
||
self.assertIn("项目创建失败", out)
|
||
self.assertEqual(self._leftover_counts(),
|
||
{"users": 0, "projects": 0, "tasks": 0})
|
||
|
||
def test_cleanup_error_reports_leftover(self):
|
||
server = create_server(self.db_path, "127.0.0.1", 0)
|
||
port = server.server_address[1]
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
try:
|
||
# 删用户时触发器直接 ABORT：清理抛错 -> leftover -> 返回 1，
|
||
# 但 13 项检查本身全部通过（区分“检查失败”与“清理失败”）。
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute(
|
||
"CREATE TRIGGER no_user_delete BEFORE DELETE ON users "
|
||
"BEGIN SELECT RAISE(ABORT, 'trigger blocks delete'); END")
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
rc, out, results = self._run_main(
|
||
"http://127.0.0.1:%d" % port)
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
thread.join(timeout=5)
|
||
self.assertEqual(rc, 1)
|
||
self.assertTrue(all(ok for _, ok in results), results)
|
||
self.assertIn("清理未完成", out)
|
||
self.assertIn("cleanup-error", out)
|
||
|
||
def test_stub_project_responses_fail_early(self):
|
||
init_db(self.db_path)
|
||
for mode, want in [("wrong-id", [True, True]),
|
||
("error-500", [False, False])]:
|
||
with self.subTest(mode=mode):
|
||
stub = _start_stub(mode)
|
||
try:
|
||
rc, out, results = self._run_main(
|
||
"http://127.0.0.1:%d" % stub.server_address[1])
|
||
finally:
|
||
stub.shutdown()
|
||
stub.server_close()
|
||
self.assertEqual(rc, 1)
|
||
self.assertEqual([ok for _, ok in results], want)
|
||
self.assertIn("项目创建失败", out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|