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.
 
 
 

812 lines
51 KiB

import json
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
p = PROJECT_ROOT / 'docs' / 'openapi-rc.json'
d = json.load(open(p, encoding='utf-8'))
print('paths:', len(d['paths']))
assert len(d['paths']) == 197, f"Expected 197 paths (184 baseline + 13 module paths), got {len(d['paths'])}"
paths = d['paths']
# Generic OpenAPI integrity checks. These catch importer-breaking structures
# before business-specific path/count assertions run.
HTTP_METHODS = {'get', 'put', 'post', 'delete', 'patch', 'options', 'head', 'trace'}
def resolve_local_ref(document, ref):
assert isinstance(ref, str) and ref.startswith('#/'), f'Invalid local $ref: {ref!r}'
node = document
for token in ref[2:].split('/'):
token = token.replace('~1', '/').replace('~0', '~')
assert isinstance(node, dict) and token in node, f'Unresolved $ref: {ref}'
node = node[token]
return node
def validate_refs(node, location='$'):
if isinstance(node, dict):
for key, value in node.items():
child = f'{location}.{key}'
if key == '$ref':
assert isinstance(value, str) and value, f'$ref must be a string at {child}'
if value.startswith('#/'):
resolve_local_ref(d, value)
validate_refs(value, child)
elif isinstance(node, list):
for index, value in enumerate(node):
validate_refs(value, f'{location}[{index}]')
validate_refs(d)
operation_ids = {}
used_tags = set()
for route, path_item in paths.items():
for method, operation in path_item.items():
if method.lower() not in HTTP_METHODS:
continue
assert isinstance(operation, dict), f'Invalid operation: {method.upper()} {route}'
assert not (method.lower() == 'get' and 'requestBody' in operation), \
f'GET must not contain requestBody: {route}'
if 'parameters' in operation:
assert isinstance(operation['parameters'], list), \
f'parameters must be an array: {method.upper()} {route}'
operation_id = operation.get('operationId')
if operation_id:
operation_ids.setdefault(operation_id, []).append(f'{method.upper()} {route}')
used_tags.update(operation.get('tags', []))
duplicates = {key: value for key, value in operation_ids.items() if len(value) > 1}
assert not duplicates, f'Duplicate operationId values: {duplicates}'
declared_tags = {tag['name'] for tag in d.get('tags', [])}
assert used_tags <= declared_tags, f'Undeclared operation tags: {sorted(used_tags - declared_tags)}'
required = [
# Phase A 企业注册与入驻聚合(2 个 path / 2 个 operation)
'/auth/company/register',
'/auth/identity/select',
'/companies/me/onboarding',
'/companies/me/jobs',
'/review-weight-templates',
'/profile/me/review-weights',
'/profile/me/privacy',
'/jobs/{jobId}/personalized-review-score',
'/company/candidates',
'/company/candidates/{applicationId}',
'/profile/me/candidate-scoring',
'/profile/me/candidate-scoring/dimensions/{dimension}',
'/profile/me/candidate-scoring/trend',
'/company/candidates/{applicationId}/scoring',
'/profile/me/candidate-scoring/appeals',
'/profile/me/candidate-scoring/appeals/{appealId}',
'/profile/me/candidate-scoring/appeals/{appealId}/cancel',
'/admin/candidate-scoring/appeals',
'/admin/candidate-scoring/appeals/{appealId}/decision',
'/admin/candidate-scoring/appeals/{appealId}/rollback',
'/admin/candidate-scoring/appeals/{appealId}/history',
'/admin/candidate-scoring/snapshots/{candidateUserId}/{dimension}/{period}/history',
# 企业评分运营看板(7 个接口)
'/company/scoring/dashboard/overview',
'/company/scoring/dashboard/job-rankings',
'/company/scoring/dashboard/job-rankings/{jobId}',
'/company/scoring/dashboard/candidate-rankings',
'/company/scoring/dashboard/candidate-rankings/{applicationId}',
'/company/scoring/dashboard/trend',
'/admin/scoring/dashboard/overview',
# Offer 进阶(5 个 path / 6 个 operation)
'/offers/{id}/counter-offers',
'/offers/{id}/counter-offers/{coId}/accept',
'/offers/{id}/counter-offers/{coId}/reject',
'/offers/compare',
'/offers/statistics',
# 账号注销(2 个 path / 3 个 operation)
'/profile/me/deletion',
'/profile/me/deletion/cancel',
# 个人数据导出(3 个 path / 4 个 operation;详情已覆盖状态查询,无需 /status)
'/profile/me/data-export-tasks',
'/profile/me/data-export-tasks/{taskId}',
'/profile/me/data-export-tasks/{taskId}/download',
# 结构化面试评分(3 个 path / 4 个 operation)
'/interviews/{interviewId}/score',
'/interviews/{interviewId}/score/submit',
'/interviews/{interviewId}/score/history',
# 同岗位竞争力排名(1 个 path / 1 个 operation)
'/jobs/{jobId}/competition-ranking',
# V34 统一求职进度聚合(1 个 path / 1 个 operation)
'/profile/me/career-progress',
# V33 举报与屏蔽(8 个 path / 9 个 operation)
'/reports',
'/reports/me',
'/reports/{id}',
'/blocks/{blockedUserId}',
'/blocks/me',
'/admin/reports',
'/admin/reports/{id}',
'/admin/reports/{id}/decision',
# V34/V35 RIASEC 职业兴趣测评(8 个 path / 8 个 operation)
'/assessments/types',
'/assessments/{type}/questions',
'/assessments/attempts',
'/assessments/attempts/{attemptId}',
'/assessments/attempts/{attemptId}/answers',
'/assessments/attempts/{attemptId}/submit',
'/assessments/attempts/{attemptId}/result',
'/assessments/me',
# V36 四维能力测评(7 个 path / 7 个 operation)
'/abilities/types',
'/abilities/attempts',
'/abilities/attempts/{attemptId}',
'/abilities/attempts/{attemptId}/answers',
'/abilities/attempts/{attemptId}/submit',
'/abilities/attempts/{attemptId}/result',
'/abilities/me',
# V37 风险概览与运营规则配置(7 个 path / 7 个 operation,仅 ADMIN)
'/admin/scoring/risk/overview',
'/admin/scoring/risk/trend',
'/admin/scoring/risk/distribution',
'/admin/scoring/risk/events',
'/admin/scoring/risk/rules',
'/admin/scoring/risk/rules/{ruleCode}',
'/admin/scoring/risk/events/{eventId}/actions',
]
for path in required:
assert path in paths, f"Missing path: {path}"
print(f" Found: {path}")
assert 'get' in paths['/review-weight-templates'], "Missing GET /review-weight-templates"
assert 'get' in paths['/profile/me/review-weights'], "Missing GET /profile/me/review-weights"
assert 'put' in paths['/profile/me/review-weights'], "Missing PUT /profile/me/review-weights"
assert 'get' in paths['/profile/me/privacy'], "Missing GET /profile/me/privacy"
assert 'put' in paths['/profile/me/privacy'], "Missing PUT /profile/me/privacy"
assert 'get' in paths['/jobs/{jobId}/personalized-review-score'], "Missing GET /jobs/{jobId}/personalized-review-score"
assert 'get' in paths['/company/candidates'], "Missing GET /company/candidates"
assert 'get' in paths['/company/candidates/{applicationId}'], "Missing GET /company/candidates/{applicationId}"
assert 'get' in paths['/profile/me/candidate-scoring'], "Missing GET /profile/me/candidate-scoring"
assert 'get' in paths['/profile/me/candidate-scoring/dimensions/{dimension}'], "Missing GET /profile/me/candidate-scoring/dimensions/{dimension}"
assert 'get' in paths['/profile/me/candidate-scoring/trend'], "Missing GET /profile/me/candidate-scoring/trend"
assert 'get' in paths['/company/candidates/{applicationId}/scoring'], "Missing GET /company/candidates/{applicationId}/scoring"
# 评分治理(9 个接口落在 8 个路径:POST+GET 共享 /appeals)
assert 'post' in paths['/profile/me/candidate-scoring/appeals'], "Missing POST /profile/me/candidate-scoring/appeals"
assert 'get' in paths['/profile/me/candidate-scoring/appeals'], "Missing GET /profile/me/candidate-scoring/appeals"
assert 'get' in paths['/profile/me/candidate-scoring/appeals/{appealId}'], "Missing GET .../appeals/{appealId}"
assert 'post' in paths['/profile/me/candidate-scoring/appeals/{appealId}/cancel'], "Missing POST .../cancel"
assert 'get' in paths['/admin/candidate-scoring/appeals'], "Missing GET /admin/candidate-scoring/appeals"
assert 'post' in paths['/admin/candidate-scoring/appeals/{appealId}/decision'], "Missing POST .../decision"
assert 'post' in paths['/admin/candidate-scoring/appeals/{appealId}/rollback'], "Missing POST .../rollback"
assert 'get' in paths['/admin/candidate-scoring/appeals/{appealId}/history'], "Missing GET .../history"
assert 'get' in paths['/admin/candidate-scoring/snapshots/{candidateUserId}/{dimension}/{period}/history'], "Missing GET snapshots history"
# 企业评分运营看板(7 个 GET)
assert 'get' in paths['/company/scoring/dashboard/overview'], "Missing GET /company/scoring/dashboard/overview"
assert 'get' in paths['/company/scoring/dashboard/job-rankings'], "Missing GET /company/scoring/dashboard/job-rankings"
assert 'get' in paths['/company/scoring/dashboard/job-rankings/{jobId}'], "Missing GET /company/scoring/dashboard/job-rankings/{jobId}"
assert 'get' in paths['/company/scoring/dashboard/candidate-rankings'], "Missing GET /company/scoring/dashboard/candidate-rankings"
assert 'get' in paths['/company/scoring/dashboard/candidate-rankings/{applicationId}'], "Missing GET /company/scoring/dashboard/candidate-rankings/{applicationId}"
assert 'get' in paths['/company/scoring/dashboard/trend'], "Missing GET /company/scoring/dashboard/trend"
assert 'get' in paths['/admin/scoring/dashboard/overview'], "Missing GET /admin/scoring/dashboard/overview"
# Offer 进阶(6 个 operation / 5 个 path)
assert 'post' in paths['/offers/{id}/counter-offers'], "Missing POST /offers/{id}/counter-offers"
assert 'get' in paths['/offers/{id}/counter-offers'], "Missing GET /offers/{id}/counter-offers"
assert 'patch' in paths['/offers/{id}/counter-offers/{coId}/accept'], "Missing PATCH .../accept"
assert 'patch' in paths['/offers/{id}/counter-offers/{coId}/reject'], "Missing PATCH .../reject"
assert 'get' in paths['/offers/compare'], "Missing GET /offers/compare"
assert 'get' in paths['/offers/statistics'], "Missing GET /offers/statistics"
# 账号注销(3 个 operation)
assert 'get' in paths['/profile/me/deletion'], "Missing GET /profile/me/deletion"
assert 'post' in paths['/profile/me/deletion'], "Missing POST /profile/me/deletion"
assert 'post' in paths['/profile/me/deletion/cancel'], "Missing POST /profile/me/deletion/cancel"
# 个人数据导出(4 个 operation / 3 个 path)
assert 'post' in paths['/profile/me/data-export-tasks'], "Missing POST /profile/me/data-export-tasks"
assert 'get' in paths['/profile/me/data-export-tasks'], "Missing GET /profile/me/data-export-tasks"
assert 'get' in paths['/profile/me/data-export-tasks/{taskId}'], "Missing GET /profile/me/data-export-tasks/{taskId}"
assert 'get' in paths['/profile/me/data-export-tasks/{taskId}/download'], "Missing GET .../download"
print(" Personal data export endpoints present (4 operations / 3 paths)")
# 同岗位竞争力排名(1 个 GET operation)
assert 'get' in paths['/jobs/{jobId}/competition-ranking'], "Missing GET /jobs/{jobId}/competition-ranking"
print(" Competition ranking endpoint present (1 GET operation)")
# V34 统一求职进度聚合(1 个 GET operation)
assert 'get' in paths['/profile/me/career-progress'], "Missing GET /profile/me/career-progress"
print(" Career progress endpoint present (1 GET operation)")
# V33 举报与屏蔽(8 个 path / 9 个 operation;PUT+DELETE 共享 /blocks/{blockedUserId})
assert 'post' in paths['/reports'], "Missing POST /reports"
assert 'get' in paths['/reports/me'], "Missing GET /reports/me"
assert 'get' in paths['/reports/{id}'], "Missing GET /reports/{id}"
assert 'put' in paths['/blocks/{blockedUserId}'], "Missing PUT /blocks/{blockedUserId}"
assert 'delete' in paths['/blocks/{blockedUserId}'], "Missing DELETE /blocks/{blockedUserId}"
assert 'get' in paths['/blocks/me'], "Missing GET /blocks/me"
assert 'get' in paths['/admin/reports'], "Missing GET /admin/reports"
assert 'get' in paths['/admin/reports/{id}'], "Missing GET /admin/reports/{id}"
assert 'post' in paths['/admin/reports/{id}/decision'], "Missing POST /admin/reports/{id}/decision"
print(" V33 report/block endpoints present (9 operations / 8 paths)")
# V34/V35 RIASEC 测评(8 个 path / 8 个 operation)
assert 'get' in paths['/assessments/types'], "Missing GET /assessments/types"
assert 'get' in paths['/assessments/{type}/questions'], "Missing GET /assessments/{type}/questions"
assert 'post' in paths['/assessments/attempts'], "Missing POST /assessments/attempts"
assert 'get' in paths['/assessments/attempts/{attemptId}'], "Missing GET /assessments/attempts/{attemptId}"
assert 'patch' in paths['/assessments/attempts/{attemptId}/answers'], "Missing PATCH .../answers"
assert 'post' in paths['/assessments/attempts/{attemptId}/submit'], "Missing POST .../submit"
assert 'get' in paths['/assessments/attempts/{attemptId}/result'], "Missing GET .../result"
assert 'get' in paths['/assessments/me'], "Missing GET /assessments/me"
print(" V34/V35 RIASEC assessment endpoints present (8 operations / 8 paths)")
# V36 四维能力测评(7 个 path / 7 个 operation)
assert 'get' in paths['/abilities/types'], "Missing GET /abilities/types"
assert 'post' in paths['/abilities/attempts'], "Missing POST /abilities/attempts"
assert 'get' in paths['/abilities/attempts/{attemptId}'], "Missing GET /abilities/attempts/{attemptId}"
assert 'patch' in paths['/abilities/attempts/{attemptId}/answers'], "Missing PATCH .../answers"
assert 'post' in paths['/abilities/attempts/{attemptId}/submit'], "Missing POST .../submit"
assert 'get' in paths['/abilities/attempts/{attemptId}/result'], "Missing GET .../result"
assert 'get' in paths['/abilities/me'], "Missing GET /abilities/me"
print(" V36 ability assessment endpoints present (7 operations / 7 paths)")
# ---- V36 严格断言:正好 7 path / 7 operation;无 /abilities/{type}/questions;requestBody+200 schema;无题库内部字段 ----
assert '/abilities/{type}/questions' not in paths, "GET /abilities/{type}/questions must NOT exist"
v36_paths = sorted([k for k in paths if k.startswith('/abilities')])
assert len(v36_paths) == 7, f"Expected exactly 7 /abilities paths, got {len(v36_paths)}"
v36_ops = sum(len(paths[k]) for k in v36_paths)
assert v36_ops == 7, f"Expected exactly 7 /abilities operations, got {v36_ops}"
print(f" V36 exactly 7 paths / 7 operations: {v36_paths}")
ABILITY_200 = {
'/abilities/types': ('get', 'ApiResponseAbilityTypeResponse'),
'/abilities/attempts': ('post', 'ApiResponseAbilityAttemptResponse'),
'/abilities/attempts/{attemptId}': ('get', 'ApiResponseAbilityAttemptResponse'),
'/abilities/attempts/{attemptId}/answers': ('patch', 'ApiResponseAbilityAttemptResponse'),
'/abilities/attempts/{attemptId}/submit': ('post', 'ApiResponseAbilityResultResponse'),
'/abilities/attempts/{attemptId}/result': ('get', 'ApiResponseAbilityResultResponse'),
'/abilities/me': ('get', 'ApiResponsePageResponseAbilityHistoryItem'),
}
def response_ref(op):
return op.get('responses', {}).get('200', {}).get('content', {}).get('application/json', {}).get('schema', {}).get('$ref', '')
for pt, (method, expect_ref) in ABILITY_200.items():
ref = response_ref(paths[pt][method])
assert ref.endswith(expect_ref), f"{method.upper()} {pt}: expected 200 ref ...{expect_ref}, got {ref}"
print(" All 7 V36 success responses reference a schema")
for pt, (method, schema_name) in {
'/abilities/attempts': ('post', 'CreateAbilityAttemptRequest'),
'/abilities/attempts/{attemptId}/answers': ('patch', 'SaveAbilityAnswersRequest')}.items():
rb = paths[pt][method].get('requestBody', {}).get('content', {}).get('application/json', {}).get('schema', {}).get('$ref', '')
assert rb.endswith(schema_name), f"{method.upper()} {pt}: expected requestBody ref ...{schema_name}, got {rb}"
print(" POST/PATCH V36 request bodies reference a schema")
# 题库内部字段不得出现在 Ability 相关 schema
for f in ['questionCode', 'competencyTag', 'answerKey', 'scoringRule']:
for name, sch in d['components']['schemas'].items():
if name.startswith('Ability') and f in json.dumps(sch):
assert False, f"schema {name} exposes forbidden field {f}"
print(" Ability schemas expose no questionCode/competencyTag/answerKey/scoringRule")
# V37 风险接口 operation 存在性
assert 'get' in paths['/admin/scoring/risk/overview'], "Missing GET overview"
assert 'get' in paths['/admin/scoring/risk/trend'], "Missing GET trend"
assert 'get' in paths['/admin/scoring/risk/distribution'], "Missing GET distribution"
assert 'get' in paths['/admin/scoring/risk/events'], "Missing GET events"
assert 'get' in paths['/admin/scoring/risk/rules'], "Missing GET rules"
assert 'put' in paths['/admin/scoring/risk/rules/{ruleCode}'], "Missing PUT rules/{ruleCode}"
assert 'post' in paths['/admin/scoring/risk/events/{eventId}/actions'], "Missing POST events/{eventId}/actions"
print(" V37 risk endpoints present (7 operations / 7 paths)")
# ---- V37 严格断言:正好 7 path / 7 operation;请求/响应 schema;无企业/学生风险接口;Risk schema 无敏感字段 ----
vk = [k for k in paths if k.startswith('/admin/scoring/risk')]
assert len(vk) == 7, f"Expected exactly 7 V37 paths, got {len(vk)}"
assert sum(len(paths[k]) for k in vk) == 7, "Expected exactly 7 V37 operations"
# 不得新增企业/学生风险接口
for k in paths:
if k.startswith('/scoring/risk') or k.startswith('/risk'):
assert False, f"unexpected student/company risk endpoint {k}"
print(f" V37 exactly 7 paths / 7 operations: {vk}")
RISK_200 = {
'/admin/scoring/risk/overview': 'ApiResponseRiskOverviewResponse',
'/admin/scoring/risk/trend': 'ApiResponseRiskTrendResponse',
'/admin/scoring/risk/distribution': 'ApiResponseRiskDistributionResponse',
'/admin/scoring/risk/events': 'ApiResponsePageResponseRiskEventResponse',
'/admin/scoring/risk/rules': 'ApiResponseListRiskRuleResponse',
'/admin/scoring/risk/rules/{ruleCode}': 'ApiResponseRiskRuleResponse',
'/admin/scoring/risk/events/{eventId}/actions': 'ApiResponseRiskEventResponse',
}
for pt, expect_ref in RISK_200.items():
method = 'put' if pt.endswith('{ruleCode}') else ('post' if pt.endswith('actions') else 'get')
ref = response_ref(paths[pt][method])
assert ref.endswith(expect_ref), f"{method.upper()} {pt}: expected 200 ref ...{expect_ref}, got {ref}"
print(" All 7 V37 success responses reference a schema")
for pt in ('/admin/scoring/risk/rules/{ruleCode}', '/admin/scoring/risk/events/{eventId}/actions'):
method = 'put' if pt.endswith('{ruleCode}') else 'post'
rb = paths[pt][method].get('requestBody', {}).get('content', {}).get('application/json', {}).get('schema', {}).get('$ref', '')
assert rb != '', f"{method.upper()} {pt} requestBody schema missing"
print(" PUT/POST V37 request bodies reference a schema")
risk_forbidden = ['subjectUserId', 'subject_user_id', 'operatorUserId', 'operator_user_id', 'phone', 'email', 'ipAddress', 'userAgent', 'details', 'token', 'password', 'secret']
for f in risk_forbidden:
for name, sch in d['components']['schemas'].items():
if name.startswith('Risk') and f in json.dumps(sch):
assert False, f"schema {name} exposes forbidden field {f}"
print(" Risk schemas expose no subjectUserId/operatorUserId/phone/email/ip/userAgent/details/credentials")
# V39 测评驱动岗位推荐:6 path / 8 operation;学生反馈、企业画像、ADMIN 审核与公开受控字典。
v39 = [
'/job-recommendation-taxonomies',
'/profile/me/job-recommendations',
'/profile/me/job-recommendations/{jobId}/feedback',
'/companies/me/jobs/{jobId}/recommendation-profile',
'/admin/job-recommendation-profiles',
'/admin/job-recommendation-profiles/{jobId}/decision',
]
assert all(k in paths for k in v39), "One or more V39 paths are missing"
assert len(v39) == 6, f"Expected exactly 6 V39 paths, got {len(v39)}"
assert sum(len(paths[k]) for k in v39) == 8, "Expected exactly 8 V39 operations"
for pt, method in [
('/job-recommendation-taxonomies', 'get'),
('/profile/me/job-recommendations', 'get'),
('/profile/me/job-recommendations/{jobId}/feedback', 'post'),
('/profile/me/job-recommendations/{jobId}/feedback', 'delete'),
('/companies/me/jobs/{jobId}/recommendation-profile', 'get'),
('/companies/me/jobs/{jobId}/recommendation-profile', 'put'),
('/admin/job-recommendation-profiles', 'get'),
('/admin/job-recommendation-profiles/{jobId}/decision', 'post'),
]:
assert method in paths[pt], f"Missing {method.upper()} {pt}"
for pt, method in [
('/profile/me/job-recommendations/{jobId}/feedback', 'post'),
('/companies/me/jobs/{jobId}/recommendation-profile', 'put'),
('/admin/job-recommendation-profiles/{jobId}/decision', 'post'),
]:
assert paths[pt][method].get('requestBody', {}).get('content', {}).get('application/json', {}).get('schema', {}).get('$ref'), f"{method.upper()} {pt} requestBody schema missing"
recommendation_sensitive = ['answerKey', 'scoringRule', 'questionCode', 'competencyTag', 'resultId', 'attemptId', 'studentUserId']
for f in recommendation_sensitive:
for name, sch in d['components']['schemas'].items():
if name.startswith(('JobRecommendation', 'Recommendation')) and f in json.dumps(sch):
assert False, f"schema {name} exposes forbidden field {f}"
print(" V39 exactly 6 paths / 8 operations; recommendation schemas are redacted")
# V40 企业评价官方回复:5 paths / 6 operations;企业、管理员和公开回复均为白名单载荷。
v40 = [
'/companies/me/job-reviews',
'/companies/me/job-reviews/{reviewId}',
'/companies/me/job-reviews/{reviewId}/reply',
'/admin/review-replies/pending',
'/admin/review-replies/{replyId}/decision',
]
assert all(k in paths for k in v40), "One or more V40 paths are missing"
assert len(v40) == 5, f"Expected exactly 5 V40 paths, got {len(v40)}"
assert sum(len(paths[k]) for k in v40) == 6, "Expected exactly 6 V40 operations"
for pt, method in [
('/companies/me/job-reviews', 'get'),
('/companies/me/job-reviews/{reviewId}', 'get'),
('/companies/me/job-reviews/{reviewId}/reply', 'post'),
('/companies/me/job-reviews/{reviewId}/reply', 'put'),
('/admin/review-replies/pending', 'get'),
('/admin/review-replies/{replyId}/decision', 'post'),
]:
assert method in paths[pt], f"Missing {method.upper()} {pt}"
for pt, method in [
('/companies/me/job-reviews/{reviewId}/reply', 'post'),
('/companies/me/job-reviews/{reviewId}/reply', 'put'),
('/admin/review-replies/{replyId}/decision', 'post'),
]:
assert paths[pt][method].get('requestBody', {}).get('content', {}).get('application/json', {}).get('schema', {}).get('$ref'), f"{method.upper()} {pt} requestBody schema missing"
for pt, method in [
('/companies/me/job-reviews', 'get'),
('/companies/me/job-reviews/{reviewId}', 'get'),
('/admin/review-replies/pending', 'get'),
('/companies/me/job-reviews/{reviewId}/reply', 'post'),
('/companies/me/job-reviews/{reviewId}/reply', 'put'),
('/admin/review-replies/{replyId}/decision', 'post'),
]:
response = paths[pt][method].get('responses', {}).get('200', {})
assert response.get('content', {}).get('application/json', {}).get('schema', {}).get('$ref'), f"{method.upper()} {pt} 200 schema missing"
v40_sensitive = ['studentUserId', 'phone', 'email', 'ipAddress', 'userAgent', 'moderationScore', 'reasonCodes']
for name in ('CompanyReviewItemResponse', 'CompanyReviewDetailResponse', 'CompanyReviewReplyResponse', 'ReviewReplyPageResponse'):
assert name in d['components']['schemas'], f"Missing V40 schema {name}"
text_schema = json.dumps(d['components']['schemas'][name])
for field in v40_sensitive:
assert field not in text_schema, f"V40 company schema {name} exposes {field}"
assert 'officialReply' not in json.dumps(d['components']['schemas']['CompanyReviewReplyResponse'])
print(" V40 exactly 5 paths / 6 operations; request/response schemas are present and redacted")
# V41 企业人才发现:2 个唯一 path / 3 个 operation(学生 GET/PUT 共享一个 path)。
v41 = [
'/profile/me/talent-discovery',
'/companies/me/jobs/{jobId}/talent-recommendations',
]
assert all(k in paths for k in v41), "One or more V41 paths are missing"
assert len(v41) == 2, f"Expected exactly 2 V41 paths, got {len(v41)}"
assert sum(len(paths[k]) for k in v41) == 3, "Expected exactly 3 V41 operations"
assert 'get' in paths[v41[0]] and 'put' in paths[v41[0]], "V41 preference GET/PUT missing"
assert 'get' in paths[v41[1]], "V41 talent recommendation GET missing"
assert paths[v41[0]]['put'].get('requestBody', {}).get('content', {}).get('application/json', {}).get('schema', {}).get('$ref'), "V41 PUT requestBody schema missing"
for pt, method in [(v41[0], 'get'), (v41[0], 'put'), (v41[1], 'get')]:
assert response_ref(paths[pt][method]), f"{method.upper()} {pt} 200 schema missing"
candidate_schema = json.dumps(d['components']['schemas']['TalentDiscoveryCandidateResponse'])
for f in ['studentUserId', 'candidateId', 'phone', 'email', 'resumeUrl', 'attemptId', 'resultId',
'answerKey', 'scoringRule', 'questionCode', 'competencyTag', 'dimensionScores',
'overallScore', 'percentile', 'fitScore', 'gender', 'birthDate', 'password', 'token']:
assert f not in candidate_schema, f"V41 candidate schema exposes {f}"
assert 'candidateKey' in candidate_schema and 'studentUserId' not in candidate_schema
print(" V41 exactly 2 paths / 3 operations; candidate card schema is redacted")
# V42 企业人才邀约:6 个唯一 path / 6 个 operation。
v42 = [
'/companies/me/jobs/{jobId}/talent-invitations',
'/companies/me/talent-invitations',
'/companies/me/talent-invitations/{invitationId}',
'/profile/me/talent-invitations',
'/profile/me/talent-invitations/{invitationId}',
'/profile/me/talent-invitations/{invitationId}/decision',
]
assert all(k in paths for k in v42), "One or more V42 paths are missing"
assert len(v42) == 6, f"Expected exactly 6 V42 paths, got {len(v42)}"
assert sum(len(paths[k]) for k in v42) == 6, "Expected exactly 6 V42 operations"
assert 'post' in paths[v42[0]], "V42 company invitation POST missing"
assert 'get' in paths[v42[1]] and 'get' in paths[v42[2]], "V42 company GET endpoints missing"
assert 'get' in paths[v42[3]] and 'get' in paths[v42[4]], "V42 student GET endpoints missing"
assert 'post' in paths[v42[5]], "V42 student decision endpoint missing"
for pt, method in [(v42[0], 'post'), (v42[5], 'post')]:
assert paths[pt][method].get('requestBody', {}).get('content', {}).get('application/json', {}).get('schema', {}).get('$ref'), f"{method.upper()} {pt} requestBody schema missing"
for pt, method in [(v42[0], 'post'), (v42[1], 'get'), (v42[2], 'get'), (v42[3], 'get'), (v42[4], 'get'), (v42[5], 'post')]:
assert response_ref(paths[pt][method]), f"{method.upper()} {pt} 200 schema missing"
for name in ('CompanyTalentInvitationResponse', 'StudentTalentInvitationResponse', 'TalentInvitationPageResponseCompanyTalentInvitationResponse', 'TalentInvitationPageResponseStudentTalentInvitationResponse'):
assert name in d['components']['schemas'], f"Missing V42 schema {name}"
company_invitation_schema = json.dumps(d['components']['schemas']['CompanyTalentInvitationResponse'])
for f in ['studentUserId', 'candidateKey', 'candidateKeyDigest', 'activeKey', 'moderationScore', 'reasonCodes', 'fitScore']:
assert f not in company_invitation_schema, f"V42 company invitation schema exposes {f}"
print(" V42 exactly 6 paths / 6 operations; invitation schemas are present and redacted")
# V43 候选评分公平性与稳定性监测:5 个 ADMIN 聚合 GET,历史漂移/排名稳定性首版明确 UNAVAILABLE。
v43 = [
'/admin/scoring/fairness/overview',
'/admin/scoring/fairness/dimensions',
'/admin/scoring/fairness/drift',
'/admin/scoring/fairness/ranking-stability',
'/admin/scoring/fairness/appeals',
]
assert all(k in paths for k in v43), "One or more V43 fairness paths are missing"
assert len(v43) == 5, f"Expected exactly 5 V43 paths, got {len(v43)}"
assert all('get' in paths[k] for k in v43), "V43 operations must all be GET"
assert sum(len(paths[k]) for k in v43) == 5, "Expected exactly 5 V43 operations"
assert paths[v43[0]]['get'].get('parameters'), "V43 overview period parameter missing"
assert paths[v43[1]]['get'].get('parameters'), "V43 dimensions parameters missing"
for pt, schema in [
(v43[0], 'ApiResponseFairnessResponseOverview'),
(v43[1], 'ApiResponseFairnessResponseDimensions'),
(v43[2], 'ApiResponseFairnessResponseUnavailable'),
(v43[3], 'ApiResponseFairnessResponseUnavailable'),
(v43[4], 'ApiResponseFairnessResponseAppeals'),
]:
assert response_ref(paths[pt]['get']).endswith(schema), f"GET {pt} 200 schema missing"
for name in ('FairnessResponseOverview', 'FairnessResponseDimensions', 'FairnessResponseUnavailable', 'FairnessResponseAppeals'):
assert name in d['components']['schemas'], f"Missing V43 schema {name}"
v43_forbidden = ['studentUserId', 'applicationId', 'candidateId', 'phone', 'email', 'resumeUrl',
'answerKey', 'scoringRule', 'questionCode', 'subScoresJson', 'auditDetails',
'token', 'password', 'ipAddress', 'userAgent', 'appealId', 'reviewerId']
def resolve_schema(value, seen=None):
"""递归展开 V43 schema 的 $ref/properties/items,避免只校验顶层漏掉嵌套字段。"""
seen = set() if seen is None else seen
if isinstance(value, dict) and '$ref' in value:
ref = value['$ref']
if ref.startswith('#/components/schemas/'):
name = ref.rsplit('/', 1)[-1]
if name in seen:
return {}
seen.add(name)
return resolve_schema(d['components']['schemas'].get(name, {}), seen)
if isinstance(value, dict):
return {k: resolve_schema(v, seen.copy()) for k, v in value.items()}
if isinstance(value, list):
return [resolve_schema(v, seen.copy()) for v in value]
return value
for name, sch in d['components']['schemas'].items():
if name.startswith('FairnessResponse'):
encoded = json.dumps(resolve_schema(sch), ensure_ascii=False)
for field in v43_forbidden:
assert field not in encoded, f"V43 schema {name} exposes {field}"
for name, required_fields in {
'FairnessResponseOverview': ['availability', 'reason', 'distributionsByRulesVersion', 'coherentVersionSubset'],
'FairnessResponseDimensions': ['availability', 'reason'],
'FairnessResponseDimensionMetric': ['availability', 'reason', 'unknownStatusCount', 'outOfRangeScoreCount'],
'FairnessResponseDistribution': ['availability', 'reason'],
'FairnessResponseAppeals': ['availability', 'reason', 'appealCount', 'statusCounts', 'statusEventCounts', 'statusCandidateCounts', 'comparableRollbackCount', 'comparableRollbackCandidateCount',
'duplicateRollbackHistoryCount', 'invalidScoreRollbackCount', 'ruleVersionChangedRollbackCount'],
}.items():
props = d['components']['schemas'][name].get('properties', {})
for field in required_fields:
assert field in props, f"V43 schema {name} missing explicit field {field}"
assert d['components']['schemas']['FairnessResponseDistribution']['properties']['availability']['type'] == 'string'
assert d['components']['schemas']['FairnessResponseAppeals']['properties']['appealCount']['type'] == 'integer'
assert '独立候选人数' in d['components']['schemas']['FairnessResponseAppeals']['properties']['comparableRollbackCandidateCount'].get('description', '')
assert '/admin/scoring/fairness/overview' in paths
print(" V43 exactly 5 paths / 5 operations; aggregate fairness schemas are redacted")
# N02 候选评分解释增强:只扩展学生评分载荷,企业评分卡保持原有白名单。
student_overview = d['components']['schemas']['CandidateScoringOverviewResponse']
overview_props = student_overview.get('properties', {})
for field in ('explanationVersion', 'coverage', 'comparabilityNotice'):
assert field in overview_props, f'CandidateScoringOverviewResponse missing {field}'
detail_props = d['components']['schemas']['DimensionScoreDetailResponse'].get('properties', {})
for field in ('displayName', 'whatItMeasures', 'interpretation', 'limitations', 'evidenceLevel',
'ruleContributions', 'improvementActions', 'explanationVersion', 'validity', 'calculatedAt'):
assert field in detail_props, f'DimensionScoreDetailResponse missing {field}'
assert overview_props['coverage'].get('$ref', '').endswith('CandidateScoringCoverage')
assert detail_props['validity'].get('$ref', '').endswith('CandidateScoringValidity')
assert detail_props['ruleContributions'].get('items', {}).get('$ref', '').endswith('CandidateScoringRuleContribution')
student_explanation_forbidden = [
'sourceResultId', 'attemptId', 'resultId', 'questionSetId', 'questionCode',
'answerKey', 'scoringRule', 'competencyTag', 'selectedOptionId', 'stem',
'studentUserId', 'phone', 'email', 'resumeUrl', 'token', 'password'
]
for name in ('CandidateScoringOverviewResponse', 'CandidateScoringCoverage',
'CandidateScoringExcludedDimension', 'DimensionScoreDetailResponse',
'CandidateScoringRuleContribution', 'CandidateScoringValidity'):
assert name in d['components']['schemas'], f'Missing N02 schema {name}'
encoded = json.dumps(resolve_schema(d['components']['schemas'][name]), ensure_ascii=False)
for field in student_explanation_forbidden:
assert field not in encoded, f'N02 schema {name} exposes forbidden field {field}'
company_schema_text = json.dumps(resolve_schema(d['components']['schemas']['CompanyCandidateScoringResponse']), ensure_ascii=False)
for field in ('explanationVersion', 'coverage', 'comparabilityNotice', 'displayName', 'ruleContributions',
'reasons', 'dataSources', 'limitations', 'improvementActions', 'evidenceLevel',
'validity', 'completedAt', 'validUntil', 'LOGIC', 'COMMUNICATION', 'EXECUTION',
'LEARNING', 'studentUserId', 'candidateId', 'resultId', 'attemptId', 'questionCode',
'answerKey', 'scoringRule', 'competencyTag', 'phone', 'email', 'resumeUrl'):
assert field not in company_schema_text, f'CompanyCandidateScoringResponse exposes N02 field {field}'
candidate_scoring_paths = [
'/profile/me/candidate-scoring',
'/profile/me/candidate-scoring/dimensions/{dimension}',
'/profile/me/candidate-scoring/trend',
'/company/candidates/{applicationId}/scoring',
]
assert all(path in paths for path in candidate_scoring_paths), 'Candidate scoring path set changed'
assert sum(len(paths[path]) for path in candidate_scoring_paths) == 4, 'Candidate scoring operation count changed'
print(' N02 explanation schemas validated; enterprise scoring schema unchanged; 4 scoring operations preserved')
# N07 消息撤回:仅新增一个 POST,消息响应为白名单 DTO。
n07_path = '/conversations/{conversationId}/messages/{messageId}/recall'
assert n07_path in paths, 'N07 recall path missing'
assert list(paths[n07_path].keys()) == ['post'], 'N07 must expose exactly one POST operation'
n07_operation = paths[n07_path]['post']
assert n07_operation.get('parameters') and len(n07_operation['parameters']) == 2, 'N07 path parameters missing'
assert n07_operation.get('responses', {}).get('200', {}).get('content', {}).get('*/*', {}).get('schema', {}).get('$ref', '').endswith('ApiResponseMessageResponse'), 'N07 200 schema missing'
assert 'MessageResponse' in d['components']['schemas'], 'MessageResponse schema missing'
assert 'ApiResponseMessageResponse' in d['components']['schemas'], 'ApiResponseMessageResponse schema missing'
assert 'MessagePageResponse' in d['components']['schemas'], 'MessagePageResponse schema missing'
assert 'ApiResponseMessagePageResponse' in d['components']['schemas'], 'ApiResponseMessagePageResponse schema missing'
# Inspect the raw array item reference here. ``resolve_schema`` intentionally
# expands local refs and therefore removes the ``$ref`` node; using the
# expanded object for this assertion made a valid MessageResponse array look
# invalid even though the static contract was correct.
message_page_schema = resolve_schema(d['components']['schemas']['MessagePageResponse'])
message_page_items = d['components']['schemas']['MessagePageResponse'].get('properties', {}).get('items', {})
assert message_page_items.get('type') == 'array', 'MessagePageResponse items missing'
assert message_page_items.get('items', {}).get('$ref', '').endswith('MessageResponse'), 'MessagePageResponse item schema missing'
assert paths['/conversations/{id}/messages']['get']['responses']['200']['content']['*/*']['schema'].get('$ref', '').endswith('ApiResponseMessagePageResponse'), 'message list schema missing'
message_props = d['components']['schemas']['MessageResponse'].get('properties', {})
for field in ('id', 'conversationId', 'content', 'status', 'createdAt'):
assert field in message_props, f'MessageResponse missing {field}'
assert message_props['status'].get('enum') == ['SENT', 'RECALLED'], 'MessageResponse status enum mismatch'
message_text = json.dumps(resolve_schema(d['components']['schemas']['MessageResponse']), ensure_ascii=False)
for forbidden in ('originalContent', 'previousContent', 'studentUserId', 'candidateId', 'phone', 'email',
'resumeUrl', 'candidateKey', 'candidateKeyDigest', 'password', 'token', 'secret'):
assert forbidden not in message_text, f'MessageResponse exposes forbidden field {forbidden}'
assert '/conversations' in paths and '/conversations/messages' in paths
assert '/conversations/{id}/messages' in paths and '/conversations/{id}/read' in paths
print(' N07 recall path/schema validated; existing messaging operations preserved')
# ── S02:Access/Refresh Token 隔离 + 密码认证 operation ─────────────
# 只允许新增 /auth/refresh 与 /auth/logout;登录原字段不得删除;
# 新 schema 不得暴露数据库字段或凭据。
assert '/auth/refresh' in paths, 'Missing /auth/refresh'
assert '/auth/logout' in paths, 'Missing /auth/logout'
assert 'post' in paths['/auth/refresh'], '/auth/refresh must be POST'
assert 'post' in paths['/auth/logout'], '/auth/logout must be POST'
_refresh_body = paths['/auth/refresh']['post']['requestBody']['content']['application/json']['schema']
_refresh_req = resolve_schema(_refresh_body)
assert sorted(_refresh_req['properties']) == ['refreshToken'], \
f"Refresh request body must only contain refreshToken, got {sorted(_refresh_req['properties'])}"
assert 'requestBody' not in paths['/auth/logout']['post'], '/auth/logout must not declare requestBody'
# 登录接口的静态契约不得被 S02 改动(请求/响应引用保持原样)
_login_post = paths['/auth/login']['post']
assert _login_post['requestBody']['content']['application/json']['schema']['$ref'] == \
'#/components/schemas/LoginRequest', 'login request schema changed'
assert _login_post['responses']['200']['content']['*/*']['schema']['$ref'] == \
'#/components/schemas/ApiResponseMapStringObject', 'login response schema changed'
assert _login_post.get('operationId') == 'mockLogin', 'login operationId changed'
_s02_text = json.dumps({
'refresh': paths['/auth/refresh'],
'logout': paths['/auth/logout'],
'schemas': {k: v for k, v in d['components']['schemas'].items()
if k in ('RefreshTokenRequest', 'TokenPairResponse', 'ApiResponseTokenPairResponse')},
}, ensure_ascii=False)
for forbidden in ('tokenHash', 'refreshTokenHash', 'sessionId', 'revocationReason',
'phone', 'secret', 'password', 'jwt'):
assert forbidden not in _s02_text, f'S02 contract exposes forbidden field {forbidden}'
# 密码认证四个 operation 的请求体必须是白名单 DTO,不能携带 userId/sessionId/角色。
_password_paths = {
'/auth/password/login': 'PasswordLoginRequest',
'/auth/password/set': 'PasswordSetRequest',
'/auth/password/change': 'PasswordChangeRequest',
'/auth/password/reset': 'PasswordResetRequest',
}
for _path, _schema_name in _password_paths.items():
assert _path in paths and set(paths[_path]) == {'post'}, f'Invalid password path: {_path}'
_body = paths[_path]['post']['requestBody']['content']['application/json']['schema']
assert _body.get('$ref', '').endswith(_schema_name), f'{_path} request schema mismatch'
_schema_text = json.dumps(resolve_schema(d['components']['schemas'][_schema_name]), ensure_ascii=False)
for _forbidden in ('userId', 'sessionId', 'role', 'passwordHash', 'token', 'secret'):
assert _forbidden not in _schema_text, f'{_path} exposes {_forbidden}'
_auth_paths = {p for p in paths if p.startswith('/auth')}
assert _auth_paths == {
'/auth/sms/send', '/auth/login', '/auth/me',
'/auth/refresh', '/auth/logout',
'/auth/password/login', '/auth/password/set', '/auth/password/change', '/auth/password/reset',
'/auth/company/register', '/auth/identity/select',
}, f'Unexpected /auth paths: {sorted(_auth_paths)}'
print(' S02 refresh/logout and password authentication operations validated')
# ── Phase A:企业注册与入驻聚合 ─────────────────────────────────────
assert set(paths['/auth/company/register']) == {'post'}
_company_register = paths['/auth/company/register']['post']
assert _company_register['requestBody']['content']['application/json']['schema']['$ref'] \
== '#/components/schemas/CompanyRegistrationRequest'
assert _company_register['responses']['200']['content']['*/*']['schema']['$ref'] \
== '#/components/schemas/ApiResponseCompanyRegistrationResponse'
_company_req = resolve_schema(d['components']['schemas']['CompanyRegistrationRequest'])
assert sorted(_company_req['properties']) == ['code', 'company', 'password', 'phone']
for _forbidden in ('userId', 'role', 'verified', 'passwordHash'):
assert _forbidden not in json.dumps(_company_req, ensure_ascii=False)
assert set(paths['/companies/me/onboarding']) == {'get'}
assert paths['/companies/me/onboarding']['get']['responses']['200']['content']['*/*']['schema']['$ref'] \
== '#/components/schemas/ApiResponseCompanyOnboardingResponse'
for _schema_name in ('CompanyOnboardingResponse', 'CompanySummary', 'VerificationSummary'):
assert _schema_name in d['components']['schemas'], f'Missing Phase A schema: {_schema_name}'
_onboarding_text = json.dumps({k: d['components']['schemas'][k] for k in
('CompanyOnboardingResponse', 'CompanySummary', 'VerificationSummary')},
ensure_ascii=False)
for _forbidden in ('materialsJson', 'decisionReason', 'reviewerId', 'storagePath'):
assert _forbidden not in _onboarding_text
print(' Phase A company registration/onboarding paths and redacted schemas validated')
# 企业岗位管理:私有全状态查询与公开岗位列表保持不同的安全边界。
assert set(paths['/companies/me/jobs']) == {'get'}
_company_jobs = paths['/companies/me/jobs']['get']
assert _company_jobs['operationId'] == 'getMyJobs'
assert {p['name'] for p in _company_jobs['parameters']} == {'status', 'keyword', 'city', 'page', 'size'}
assert _company_jobs['responses']['200']['content']['*/*']['schema']['$ref'] \
== '#/components/schemas/ApiResponsePageResponseCompanyManagedJobItem'
assert 'bearerAuth' in json.dumps(_company_jobs.get('security', []))
_company_job_properties = resolve_schema(d['components']['schemas']['CompanyManagedJobItem'])['properties']
assert {'id', 'companyId', 'title', 'city', 'salaryMin', 'salaryMax', 'status'} <= set(_company_job_properties)
print(' 企业自身全状态岗位列表路径、认证要求和强类型响应已验证')
# ── S03:唯一新增的企业简历下载接口 ────────────────────────────────
_download_path = '/company/candidates/{applicationId}/resume'
assert _download_path in paths, 'Missing company resume download endpoint'
_download_get = paths[_download_path]['get']
assert 'requestBody' not in _download_get, 'Resume download must not declare requestBody'
assert set(paths[_download_path].keys()) == {'get'}, 'Resume download must be GET only'
# 不得存在按 studentUserId / candidateKey 下载简历的入口
for _p in paths:
_lower = _p.lower()
if 'resume' in _lower or 'download' in _lower:
assert 'studentuserid' not in _lower, f'Resume path must not key on studentUserId: {_p}'
assert 'candidatekey' not in _lower, f'Resume path must not key on candidateKey: {_p}'
# 企业候选人 DTO 不得回带 resumeUrl / studentUserId / phone / email / storagePath
for _dto in ('CandidateDetailView', 'CandidateItemView'):
_node = d['components']['schemas'].get(_dto)
if _node is None:
continue
_text = json.dumps(resolve_schema(_node), ensure_ascii=False)
for _forbidden in ('resumeUrl', 'studentUserId', 'phone', 'email', 'storagePath'):
assert _forbidden not in _text, f'{_dto} exposes forbidden field {_forbidden}'
# Actuator 不得进入业务 OpenAPI
for _p in paths:
assert not _p.startswith('/actuator'), f'Actuator must not appear in the business contract: {_p}'
print(' S03 company resume download contract validated; candidate DTOs remain redacted')
# 统计 operation 总数并校验当前基线 222 个既有 operation 加 18 个模块 operation。
operations = sum(len(methods) for methods in d['paths'].values())
print('operations:', operations)
assert operations == 240, f"Expected 240 operations (222 baseline + 18 module operations), got {operations}"
# 统一账号身份选择:只允许两种业务身份,保持旧登录字段引用。
_select = paths['/auth/identity/select']['post']
assert _select['requestBody']['content']['application/json']['schema']['$ref'] == '#/components/schemas/IdentitySelectionRequest'
assert set(d['components']['schemas']['IdentitySelectionRequest']['properties']['identity']['enum']) == {'STUDENT', 'COMPANY'}
assert 'identitySelectionRequired' in d['components']['schemas']['IdentitySelectionResponse']['properties']
assert 'reauthenticationRequired' in d['components']['schemas']['IdentitySelectionResponse']['properties']
print(" Offer advance endpoints all present")
print(" 26 operations all present")
print(" Account deletion endpoints present (3 operations)")
schemas = {name.removeprefix('Lxq'): schema for name, schema in
d['components']['schemas'].items() if name.startswith('Lxq')}
onboarding_success_refs = {
('/onboarding/companies/{companyId}/checklist', 'get'): 'ApiResponseEffectiveChecklistResponse',
('/onboarding/companies/{companyId}/checklist/items/{itemId}/completion', 'put'): 'ApiResponseProgressUpdateResponse',
('/onboarding/companies/{companyId}/knowledge', 'get'): 'ApiResponseCompanyKnowledgeOverviewResponse',
('/onboarding/company/checklist-items', 'get'): 'ApiResponseListCompanyChecklistItemView',
('/onboarding/company/checklist-items', 'post'): 'ApiResponseCompanyChecklistItemView',
('/onboarding/company/checklist-items/{itemId}', 'put'): 'ApiResponseCompanyChecklistItemView',
('/onboarding/company/checklist-items/{itemId}', 'delete'): 'ApiResponseCompanyChecklistItemView',
('/onboarding/company/knowledge', 'get'): 'ApiResponseCompanyKnowledgeView',
('/onboarding/company/knowledge', 'put'): 'ApiResponseCompanyKnowledgeView',
}
for (path, method), schema_name in onboarding_success_refs.items():
response = paths[path][method]['responses']['200']
ref = response['content']['*/*']['schema']['$ref']
assert ref == f'#/components/schemas/Lxq{schema_name}', f'Wrong success schema for {method.upper()} {path}: {ref}'
for status_code in ('401', '403'):
assert status_code in paths[path][method]['responses'], f'Missing {status_code} for {method.upper()} {path}'
unauthorized = paths[path][method]['responses']['401']
assert unauthorized['content']['application/json']['schema']['$ref'] == '#/components/schemas/LxqUnauthorizedResponse'
forbidden = paths[path][method]['responses']['403']
assert forbidden['content']['application/json']['schema']['$ref'] == '#/components/schemas/LxqApiResponse'
assert forbidden['content']['application/json']['example']['code'] == 2002
onboarding_schema_fields = {
'CompanyChecklistItemCreateRequest': {'title', 'description', 'category', 'sortOrder', 'required', 'idempotencyKey'},
'CompanyChecklistItemUpdateRequest': {'title', 'description', 'category', 'sortOrder', 'required', 'active', 'version'},
'CompanyChecklistItemView': {'itemId', 'source', 'itemKey', 'title', 'description', 'category', 'required', 'sortOrder', 'active', 'version'},
'EffectiveChecklistResponse': {'companyId', 'companyName', 'customized', 'totalItems', 'completedItems', 'completionRate', 'items'},
'ChecklistItemView': {'itemId', 'source', 'itemKey', 'title', 'description', 'category', 'required', 'sortOrder', 'active', 'completed', 'completedAt', 'progressVersion'},
'ProgressUpdateRequest': {'completed', 'version'},
'ProgressUpdateResponse': {'itemId', 'completed', 'completedAt', 'progressVersion', 'totalItems', 'completedItems', 'completionRate'},
'CompanyKnowledgeUpdateRequest': {'culture', 'workPolicies', 'onboardingGuide', 'version'},
'CompanyKnowledgeView': {'configured', 'culture', 'workPolicies', 'onboardingGuide', 'updatedAt', 'version'},
'CompanyKnowledgeOverviewResponse': {'company', 'knowledgeState', 'knowledge', 'reviewSummary'},
'CompanyView': {'id', 'name', 'logoUrl', 'description', 'industry', 'website', 'address', 'verified'},
'KnowledgeView': {'culture', 'workPolicies', 'onboardingGuide', 'updatedAt', 'version'},
'ReviewSummaryView': {'state', 'totalReviews', 'averageRating', 'ratingDistribution', 'topTags', 'dimensionSummary'},
'DimensionSummaryView': {'averageScore', 'sampleCount', 'dataState'},
}
for schema_name, expected_fields in onboarding_schema_fields.items():
schema = schemas.get(schema_name)
assert schema is not None, f'Missing onboarding schema: {schema_name}'
assert expected_fields <= set(schema.get('properties', {})), (
f'Onboarding schema fields missing: {schema_name}'
)
create_request = schemas['CompanyChecklistItemCreateRequest']['properties']
assert create_request['sortOrder'].get('default') == 1000, (
'CompanyChecklistItemCreateRequest.sortOrder default must be 1000'
)
assert create_request['required'].get('default') is False, (
'CompanyChecklistItemCreateRequest.required default must be false'
)
knowledge_success_example = paths['/onboarding/company/knowledge']['put']['responses']['200']['content']['*/*']['example']
assert knowledge_success_example['data'].get('version') == 2, (
'PUT /onboarding/company/knowledge 200 example data.version must be 2'
)
onboarding_error_paths = {
10501: ('/onboarding/company/checklist-items/{itemId}', 'put', '404'),
10502: ('/onboarding/company/checklist-items/{itemId}', 'put', '409'),
10503: ('/onboarding/company/checklist-items', 'post', '409'),
10504: ('/onboarding/company/checklist-items', 'post', '409'),
10505: ('/onboarding/companies/{companyId}/checklist/items/{itemId}/completion', 'put', '409'),
10506: ('/onboarding/company/knowledge', 'put', '409'),
}
for error_code, (path, method, status_code) in onboarding_error_paths.items():
response_text = json.dumps(paths[path][method]['responses'][status_code], ensure_ascii=False)
assert str(error_code) in response_text, f'Missing onboarding error code {error_code} in {method.upper()} {path}'
print(' N04 onboarding endpoints and schemas present (9 operations / 6 paths)')
data = json.dumps(d)
assert '"" : "#/components/schemas/' not in data, "Found empty ref key"
print("No empty ref keys found")
print("Valid JSON: True")
print("All checks passed!")