103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
"""gapfill G5：机制 item 5 端点快照接线定向测试。
|
||
|
||
接线方式：GET /audit-scorecard 可选查询串
|
||
``?current_main_head=<头>``，端点用 check_baseline_stale()
|
||
对返回的卡做一次漂移判定，结果放在响应顶层
|
||
``baseline_staleness`` 里。锁定点：
|
||
- 未传参数时默认不做判定：响应里没有 baseline_staleness，
|
||
其余字段与以前完全一致，查询绝不被挡住（永远 200）。
|
||
- 传了参数时附加判定字段，但判定绝不改写卡本身的
|
||
overall / overall_pass（示例卡 baseline_head 为 None，
|
||
故任何传入的头都判 stale=True、原因 baseline_head_missing，
|
||
这是失败关闭，不是把旧分改写）。
|
||
- 空串参数按未传入处理：不做判定、不标 stale。
|
||
- 端点本身不读 git、不起 subprocess 查版本：main 头只能
|
||
由调用方传入（本文件只测 HTTP 入参一路，不测任何
|
||
版本控制探测——check_baseline_stale 本身已在机制基线
|
||
测试里钉住“只认调用方传入参数”）。
|
||
|
||
只加测试；生产改动在 cloud/server.py 的 _audit_scorecard。
|
||
"""
|
||
|
||
import http.client
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.auth import AuthStore
|
||
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.server import create_server
|
||
|
||
|
||
class BaselineSnapshotEndpointTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
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.port = self.server.server_address[1]
|
||
self.thread = threading.Thread(
|
||
target=self.server.serve_forever, daemon=True)
|
||
self.thread.start()
|
||
|
||
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()
|
||
|
||
def _req(self, path):
|
||
conn = http.client.HTTPConnection(
|
||
"127.0.0.1", self.port, timeout=10)
|
||
try:
|
||
conn.request("GET", path,
|
||
headers={"X-API-Key": self.owner_key})
|
||
resp = conn.getresponse()
|
||
data = resp.read()
|
||
finally:
|
||
conn.close()
|
||
return resp.status, json.loads(data.decode("utf-8"))
|
||
|
||
def test_no_param_means_no_staleness_field(self):
|
||
status, report = self._req("/audit-scorecard")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(report["overall"], "not_pass")
|
||
self.assertNotIn("baseline_staleness", report)
|
||
|
||
def test_param_attaches_staleness_without_rewriting_overall(self):
|
||
status, report = self._req(
|
||
"/audit-scorecard?current_main_head=82d0e84")
|
||
self.assertEqual(status, 200)
|
||
# 示例卡无冻结基线：显式要求判定时失败关闭为 stale，
|
||
# 但卡本身的 overall / overall_pass 原样不动。
|
||
self.assertEqual(report["overall"], "not_pass")
|
||
self.assertFalse(report["overall_pass"])
|
||
stale = report["baseline_staleness"]
|
||
self.assertTrue(stale["stale"])
|
||
self.assertFalse(stale["usable_as_pass_basis"])
|
||
self.assertTrue(stale["overall_must_not_be_used_as_pass_basis"])
|
||
self.assertIn("baseline_head_missing", stale["reason"])
|
||
self.assertEqual(stale["current_main_head"], "82d0e84")
|
||
|
||
def test_empty_param_means_no_staleness_field(self):
|
||
status, report = self._req("/audit-scorecard?current_main_head=")
|
||
self.assertEqual(status, 200)
|
||
self.assertNotIn("baseline_staleness", report)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|