You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
249 lines
13 KiB
249 lines
13 KiB
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""本批 JAR 的运行时 /v3/api-docs 与静态契约 docs/openapi-rc.json 的**严格**对照。
|
|
|
|
对照范围（整改项 2）：
|
|
* 具体路径 + 方法 + 200 响应 Schema 引用（逐 operation 核对，不只看路径数量）。
|
|
* 展示字段的 类型 / 数组元素类型 / 范围 / 长度 / 数量 / 可空性，在静态与运行时之间逐项比较；
|
|
任一必需项不一致即失败（不再把「约束缺失」判为通过，取消 “static-only 即通过”）。
|
|
* Map 型岗位响应的**文档模型**（ApiResponseJobDetailViewDoc / ApiResponseJobListViewDoc /
|
|
ApiResponseJobPublicViewDoc 及其 data）结构与静态一致。
|
|
* 公开企业摘要严格 7 字段白名单；多出敏感字段即失败。
|
|
|
|
用法：
|
|
python b01_openapi_compare.py [base-url] [--static PATH] [--runtime PATH] [--out PATH]
|
|
退出码：0=全部通过；1=存在失败。
|
|
"""
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
DEFAULT_STATIC = ROOT / "docs" / "openapi-rc.json"
|
|
ACCEPT_DIR = Path(__file__).resolve().parent
|
|
DEFAULT_RUNTIME_OUT = ACCEPT_DIR / "runtime-openapi.json"
|
|
DEFAULT_COMPARE_OUT = ACCEPT_DIR / "openapi-compare.json"
|
|
|
|
# 逐 operation 期望的响应 Schema 引用（要求为各操作专属文档模型，不得共用 MapStringObject）。
|
|
EXPECTED_REFS = {
|
|
("/jobs", "get"): "ApiResponseJobListViewDoc",
|
|
("/jobs", "post"): "ApiResponseJobDetailViewDoc",
|
|
("/jobs/{jobId}", "get"): "ApiResponseJobDetailViewDoc",
|
|
("/jobs/{jobId}", "put"): "ApiResponseJobDetailViewDoc",
|
|
("/jobs/public/{jobId}", "get"): "ApiResponseJobPublicViewDoc",
|
|
}
|
|
|
|
PUBLIC_COMPANY_FIELDS = ["companySize", "description", "id", "industry", "logoUrl", "name", "verified"]
|
|
FORBIDDEN_PUBLIC_FIELDS = ["ownerUserId", "address", "website", "phone", "email", "createdAt",
|
|
"updatedAt", "materialsJson", "decisionReason", "reviewerId"]
|
|
DOC_SCHEMAS = ["JobDocCompanySummary", "JobDetailViewDoc", "JobPublicViewDoc", "JobListViewDoc",
|
|
"ApiResponseJobDetailViewDoc", "ApiResponseJobPublicViewDoc", "ApiResponseJobListViewDoc"]
|
|
JOB_DETAIL_FIELDS = ["id", "companyId", "title", "description", "requirements", "city",
|
|
"employmentType", "salaryMin", "salaryMax", "experience", "education",
|
|
"benefits", "salaryMonths", "workLocation", "status", "publishedAt",
|
|
"createdAt", "updatedAt"]
|
|
JOB_PUBLIC_FIELDS = ["id", "companyId", "title", "description", "requirements", "city",
|
|
"employmentType", "salaryMin", "salaryMax", "experience", "education",
|
|
"benefits", "salaryMonths", "workLocation", "status", "publishedAt",
|
|
"createdAt", "company"]
|
|
JOB_LIST_FIELDS = ["items", "total", "page", "size", "totalPages"]
|
|
|
|
rows = []
|
|
|
|
|
|
def record(case, expected, actual, ok):
|
|
rows.append({"case": case, "expected": str(expected)[:200], "actual": str(actual)[:200],
|
|
"ok": bool(ok)})
|
|
print("[%s] %-62s exp=%-20s act=%s" % ("PASS" if ok else "FAIL", case,
|
|
str(expected)[:20], str(actual)[:70]))
|
|
|
|
|
|
def props(node):
|
|
return (node or {}).get("properties", {})
|
|
|
|
|
|
def canon_type(node):
|
|
value = (node or {}).get("type")
|
|
if value is None:
|
|
return None
|
|
return tuple(sorted(value)) if isinstance(value, list) else value
|
|
|
|
|
|
def is_nullable(node):
|
|
value = (node or {}).get("type")
|
|
if isinstance(value, list) and "null" in value:
|
|
return True
|
|
return bool((node or {}).get("nullable"))
|
|
|
|
|
|
def get(node, path):
|
|
cur = node
|
|
for token in path.split("."):
|
|
if not isinstance(cur, dict):
|
|
return None
|
|
cur = cur.get(token)
|
|
return cur
|
|
|
|
|
|
def cmp_field(case_prefix, s_node, r_node, aspects):
|
|
"""逐 aspect 比较静态/运行时字段节点；缺失即失败。"""
|
|
if s_node is None or r_node is None:
|
|
record("%s 存在（静态/运行时）" % case_prefix, "present", "s=%s r=%s" % (s_node is not None, r_node is not None),
|
|
s_node is not None and r_node is not None)
|
|
return
|
|
for path, kind in aspects:
|
|
sv, rv = get(s_node, path), get(r_node, path)
|
|
if kind == "type":
|
|
sv, rv = canon_type({"type": sv}), canon_type({"type": rv})
|
|
elif kind == "nullable":
|
|
sv, rv = is_nullable(s_node), is_nullable(r_node)
|
|
record("%s %s 一致" % (case_prefix, path), sv, rv, sv == rv and sv is not None)
|
|
|
|
|
|
def cmp_ref(op_path, method, static, runtime):
|
|
def ref(doc):
|
|
return doc["paths"].get(op_path, {}).get(method, {}).get(
|
|
"responses", {}).get("200", {}).get("content", {}).get("*/*", {}).get(
|
|
"schema", {}).get("$ref", "")
|
|
s_ref, r_ref = ref(static), ref(runtime)
|
|
expected = EXPECTED_REFS[(op_path, method)]
|
|
record("%s %s 运行时 200 引用 = %s" % (method.upper(), op_path, expected),
|
|
"#/components/schemas/" + expected, r_ref, r_ref.endswith(expected))
|
|
record("%s %s 静态/运行时 200 引用一致" % (method.upper(), op_path), s_ref, r_ref,
|
|
s_ref == r_ref and s_ref != "")
|
|
record("%s %s 未共用 ApiResponseMapStringObject" % (method.upper(), op_path),
|
|
"专属模型", r_ref, "MapStringObject" not in r_ref)
|
|
name = expected
|
|
in_static = name in static["components"]["schemas"]
|
|
in_runtime = name in runtime["components"]["schemas"]
|
|
record("%s %s 引用组件双方存在" % (method.upper(), op_path), True,
|
|
"static=%s runtime=%s" % (in_static, in_runtime), in_static and in_runtime)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("base", nargs="?", default="http://127.0.0.1:18080")
|
|
parser.add_argument("--static", default=str(DEFAULT_STATIC))
|
|
parser.add_argument("--runtime", default=None)
|
|
parser.add_argument("--out", default=str(DEFAULT_COMPARE_OUT))
|
|
parser.add_argument("--runtime-out", default=str(DEFAULT_RUNTIME_OUT))
|
|
args = parser.parse_args()
|
|
|
|
static = json.load(open(args.static, encoding="utf-8"))
|
|
if args.runtime:
|
|
runtime = json.load(open(args.runtime, encoding="utf-8"))
|
|
else:
|
|
url = args.base.rstrip("/") + "/api/v1/v3/api-docs"
|
|
with urllib.request.urlopen(url, timeout=90) as resp:
|
|
runtime = json.loads(resp.read().decode("utf-8"))
|
|
Path(args.runtime_out).write_text(json.dumps(runtime, ensure_ascii=False, indent=2),
|
|
encoding="utf-8")
|
|
|
|
s_schemas, r_schemas = static["components"]["schemas"], runtime["components"]["schemas"]
|
|
s_paths, r_paths = static.get("paths", {}), runtime.get("paths", {})
|
|
|
|
# 1) 基线：路径数与 operation 数一致
|
|
record("运行时路径数 == 静态契约", len(s_paths), len(r_paths), len(s_paths) == len(r_paths))
|
|
record("运行时 operation 数 == 静态契约",
|
|
sum(len(v) for v in s_paths.values()), sum(len(v) for v in r_paths.values()),
|
|
sum(len(v) for v in s_paths.values()) == sum(len(v) for v in r_paths.values()))
|
|
|
|
# 2) 逐 operation 响应引用
|
|
for (path, method) in EXPECTED_REFS:
|
|
cmp_ref(path, method, static, runtime)
|
|
|
|
# 3) 请求 DTO 展示字段约束
|
|
text_aspects = [("type", "type"), ("maxLength", "raw"), ("nullable", "nullable")]
|
|
for name in ("JobCreateRequest", "JobUpdateRequest"):
|
|
sp, rp = props(s_schemas.get(name)), props(r_schemas.get(name))
|
|
for field in ("experience", "education"):
|
|
cmp_field("%s.%s" % (name, field), sp.get(field), rp.get(field), text_aspects)
|
|
cmp_field("%s.workLocation" % name, sp.get("workLocation"), rp.get("workLocation"),
|
|
[("type", "type"), ("maxLength", "raw"), ("nullable", "nullable")])
|
|
cmp_field("%s.benefits" % name, sp.get("benefits"), rp.get("benefits"),
|
|
[("type", "type"), ("nullable", "nullable"), ("items.type", "raw"),
|
|
("x-normalization", "raw")])
|
|
for label, node in (("静态", sp.get("benefits") or {}), ("运行时", rp.get("benefits") or {})):
|
|
expected = dict(trimItems=True, deduplicate=True, rejectBlankItems=True, maxItems=20, itemMaxLength=40)
|
|
record("%s %s 福利归一化语义" % (label, name), expected, node.get("x-normalization"),
|
|
node.get("x-normalization") == expected)
|
|
record("%s %s 不限制原始数组数量或未trim长度" % (label, name), "absent",
|
|
(node.get("maxItems"), node.get("items", {}).get("maxLength")),
|
|
"maxItems" not in node and "maxLength" not in node.get("items", {}))
|
|
record("%s %s 福利可空" % (label, name), True, is_nullable(node), is_nullable(node))
|
|
cmp_field("%s.salaryMonths" % name, sp.get("salaryMonths"), rp.get("salaryMonths"),
|
|
[("type", "type"), ("minimum", "raw"), ("maximum", "raw"), ("nullable", "nullable")])
|
|
for field in ("salaryMin", "salaryMax"):
|
|
record("%s.%s 双方存在且为 integer" % (name, field), "integer",
|
|
canon_type(rp.get(field)), canon_type(sp.get(field)) == "integer"
|
|
and canon_type(rp.get(field)) == "integer")
|
|
|
|
# 4) 企业请求 / 公开摘要的 companySize
|
|
for name in ("CompanyCreateRequest", "CompanyUpdateRequest", "CompanyPublicView"):
|
|
cmp_field("%s.companySize" % name,
|
|
props(s_schemas.get(name)).get("companySize"),
|
|
props(r_schemas.get(name)).get("companySize"),
|
|
[("type", "type"), ("maxLength", "raw"), ("nullable", "nullable")])
|
|
|
|
# 5) 企业自身岗位管理摘要
|
|
for field in ("experience", "education", "workLocation"):
|
|
cmp_field("CompanyManagedJobItem.%s" % field,
|
|
props(s_schemas.get("CompanyManagedJobItem")).get(field),
|
|
props(r_schemas.get("CompanyManagedJobItem")).get(field),
|
|
[("type", "type"), ("maxLength", "raw")])
|
|
cmp_field("CompanyManagedJobItem.benefits",
|
|
props(s_schemas.get("CompanyManagedJobItem")).get("benefits"),
|
|
props(r_schemas.get("CompanyManagedJobItem")).get("benefits"),
|
|
[("type", "type"), ("maxItems", "raw"), ("items.type", "raw"), ("items.maxLength", "raw")])
|
|
cmp_field("CompanyManagedJobItem.salaryMonths",
|
|
props(s_schemas.get("CompanyManagedJobItem")).get("salaryMonths"),
|
|
props(r_schemas.get("CompanyManagedJobItem")).get("salaryMonths"),
|
|
[("type", "type"), ("minimum", "raw"), ("maximum", "raw")])
|
|
|
|
# 6) 公开企业摘要严格 7 字段白名单（静态与运行时）
|
|
for label, schemas in (("静态", s_schemas), ("运行时", r_schemas)):
|
|
node = props(schemas.get("CompanyPublicView"))
|
|
record("CompanyPublicView %s 严格 7 字段" % label, PUBLIC_COMPANY_FIELDS,
|
|
sorted(node.keys()), sorted(node.keys()) == PUBLIC_COMPANY_FIELDS)
|
|
leaked = [f for f in FORBIDDEN_PUBLIC_FIELDS if f in node]
|
|
record("CompanyPublicView %s 无敏感字段" % label, "无", leaked, not leaked)
|
|
|
|
# 7) 文档模型：双方存在、属性集合与类型一致
|
|
for name in DOC_SCHEMAS:
|
|
s_node, r_node = s_schemas.get(name), r_schemas.get(name)
|
|
record("文档模型 %s 双方存在" % name, True, "static=%s runtime=%s" % (s_node is not None, r_node is not None),
|
|
s_node is not None and r_node is not None)
|
|
if s_node is None or r_node is None:
|
|
continue
|
|
record("文档模型 %s 属性集合一致" % name, sorted(props(s_node).keys()),
|
|
sorted(props(r_node).keys()), sorted(props(s_node).keys()) == sorted(props(r_node).keys()))
|
|
mismatched = [k for k in props(s_node)
|
|
if canon_type(props(s_node)[k]) != canon_type(props(r_node).get(k))]
|
|
record("文档模型 %s 逐属性类型一致" % name, "无差异", mismatched, not mismatched)
|
|
|
|
# 8) 文档模型语义：必要 data 字段
|
|
for schema_name, fields in (("JobDetailViewDoc", JOB_DETAIL_FIELDS),
|
|
("JobPublicViewDoc", JOB_PUBLIC_FIELDS),
|
|
("JobListViewDoc", JOB_LIST_FIELDS)):
|
|
for label, schemas in (("静态", s_schemas), ("运行时", r_schemas)):
|
|
missing = [f for f in fields if f not in props(schemas.get(schema_name))]
|
|
record("%s %s 必需字段齐全" % (label, schema_name), fields, missing, not missing)
|
|
for label, schemas in (("静态", s_schemas), ("运行时", r_schemas)):
|
|
node = props(schemas.get("JobDocCompanySummary"))
|
|
record("%s JobDocCompanySummary 严格 7 字段" % label, PUBLIC_COMPANY_FIELDS,
|
|
sorted(node.keys()), sorted(node.keys()) == PUBLIC_COMPANY_FIELDS)
|
|
for label, schemas in (("静态", s_schemas), ("运行时", r_schemas)):
|
|
b = props(schemas.get("JobPublicViewDoc")).get("benefits") or {}
|
|
record("%s 响应端 benefits 恒为数组（非可空）" % label, "array", b.get("type"),
|
|
b.get("type") == "array")
|
|
|
|
bad = [r for r in rows if not r["ok"]]
|
|
print("\n== OPENAPI COMPARE SUMMARY: total=%d fail=%d ==" % (len(rows), len(bad)))
|
|
Path(args.out).write_text(json.dumps(rows, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
return 1 if bad else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|
|
|