guanlan-src/app_qualityGate/app_qualityGate_guanlan/audits/http_contract_audit.py

# -*- coding: utf-8 -*-
r"""HTTP 契约复核门（P11-A）：把"重构前的接口行为"变成可自动复核的判据。

规则（比结构与可达性，不比数据值）：
  · 方法一致：契约登记 GET/POST，实测必须 200；
  · 状态码一致：200 ⇒ 必须 200；
  · JSON 顶层键：契约登记的键**必须都在**（允许新增键 —— 但少一个就是回归）；
  · HTML：`<title>` 与契约一致（页面呈现的锚点之一）。
用法：
    python scripts/http_contract_audit.py            # 复核（要求服务在跑）
    python scripts/http_contract_audit.py --json     # 机读输出
退出码：0 全一致 · 9 有不一致 · 2 契约文件缺失 · 3 服务不可达。
"""
from __future__ import annotations

import argparse
import json
import pathlib
import re
import sys
import urllib.error
import urllib.request

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                       # 包结构异常时按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]

ROOT = _install_root(__file__)
CONTRACT = ROOT / 'app_backEnd/app_backEnd_guanlan/contract/http_api_v1.json'
BASES = {'gateway': 'http://127.0.0.1:28084', 'frontend': 'http://127.0.0.1:28084',
         'detail_api': 'http://127.0.0.1:18033', 'detail_page': 'http://127.0.0.1:18033',
         'cms': 'http://127.0.0.1:28084', 'algorithm': 'http://127.0.0.1:18050'}
SLOW = ('/api/curves', '/api/fleet', '/api/rpt_compose', '/api/snapshot')


def probe(url: str, method: str, timeout: int) -> dict:
    data = b'{}' if method == 'POST' else None
    try:
        req = urllib.request.Request(url, data=data, method=method,
                                     headers={'Content-Type': 'application/json',
                                              'User-Agent': 'guanlan-http-contract-audit'})
        with urllib.request.urlopen(req, timeout=timeout) as r:
            body = r.read(600_000)
            return dict(status=r.status, ctype=r.headers.get('content-type', ''), body=body)
    except urllib.error.HTTPError as e:
        return dict(status=e.code, ctype='', body=b'')
    except Exception as e:                 # noqa: BLE001
        return dict(status=0, ctype='', body=b'', note=type(e).__name__)


def audit(as_json: bool = False) -> int:
    if not CONTRACT.is_file():
        print('[X] 缺契约正本', CONTRACT.relative_to(ROOT).as_posix())
        return 2
    c = json.loads(CONTRACT.read_text(encoding='utf-8'))
    rows, bad, unreachable = [], [], 0
    for e in c['endpoints']:
        url = BASES.get(e['kind'], '') + e['path']
        to = 120 if any(s in e['path'] for s in SLOW) else 60
        got = probe(url, e['method'], to)
        if got['status'] == 0:
            unreachable += 1
            rows.append(dict(path=e['path'], method=e['method'], want=e['status'], got=0,
                             ok=False, why='不可达(%s)' % got.get('note', '')))
            continue
        ok, why = True, ''
        if got['status'] != e['status']:
            ok, why = False, '状态码 %s ≠ %s' % (got['status'], e['status'])
        elif e.get('json_keys_any'):
            try:
                o = json.loads(got['body'].decode('utf-8', 'replace'))
            except Exception:              # noqa: BLE001
                o = None
            if not isinstance(o, dict):
                ok, why = False, '响应不是 JSON 对象'
            else:
                sets = [set(s) for s in e['json_keys_any']]
                if not any(s <= set(o.keys()) for s in sets):
                    ok, why = False, ('响应键集不在契约允许的任一种里（实测: %s）'
                                      % ','.join(sorted(o.keys())[:8]))
        elif e.get('json_keys'):
            try:
                o = json.loads(got['body'].decode('utf-8', 'replace'))
            except Exception:              # noqa: BLE001
                o = None
            if not isinstance(o, dict):
                ok, why = False, '响应不是 JSON 对象'
            else:
                miss = [k for k in e['json_keys'] if k not in o]
                if miss:
                    ok, why = False, '缺顶层键: ' + ','.join(miss[:6])
        elif e.get('html_title'):
            m = re.search(r'<title>(.*?)</title>', got['body'].decode('utf-8', 'replace'), re.S)
            title = (m.group(1).strip() if m else '')
            if title != e['html_title']:
                ok, why = False, '标题 %r ≠ 契约 %r' % (title[:40], e['html_title'][:40])
        rows.append(dict(path=e['path'], method=e['method'], want=e['status'], got=got['status'],
                         ok=ok, why=why))
        if not ok:
            bad.append(rows[-1])
    if as_json:
        print(json.dumps(dict(total=len(rows), bad=len(bad), unreachable=unreachable, rows=rows),
                         ensure_ascii=False, indent=1))
    else:
        print('== HTTP 契约复核（%d 条端点）==' % len(rows))
        for r in rows:
            if not r['ok']:
                print('   [X] %-42s %-4s 实测 %s  %s' % (r['path'], r['method'], r['got'], r['why']))
        print('   一致 %d · 不一致 %d · 不可达 %d' % (len(rows) - len(bad), len(bad), unreachable))
        print('结论:', '全部与重构前一致' if not bad else '存在不一致（见上）')
    if bad and len(bad) == unreachable:
        return 3
    return 0 if not bad else 9


def main() -> int:
    ap = argparse.ArgumentParser(description='HTTP 契约复核门（重构前行为对拍）')
    ap.add_argument('--json', action='store_true')
    a = ap.parse_args()
    return audit(a.json)


if __name__ == '__main__':
    sys.exit(main())