#!/usr/bin/env python3 # -*- coding: utf-8 -*- """岗位与企业展示字段批次：HTTP 黑盒验收（针对本批隔离 JAR）。 仅访问隔离环境（默认 http://127.0.0.1:18080/api/v1），写操作只发生在 b01-app-mysql 隔离库。 不改认证/安全策略；不打印 Token 原文。 """ import argparse import json import subprocess import sys import time import urllib.error import urllib.request CONTAINER = "b01-app-mysql" DB = "mydreamjob_b01_demo" DB_USER = "b01demo" DB_PASS = "b01-demo-only" PHONE_PREFIX = "19902" CODE = "246810" API = "/api/v1" rows = [] def record(section, case, expected, actual, ok, note=""): rows.append({"section": section, "case": case, "expected": expected, "actual": str(actual)[:300], "ok": bool(ok), "note": note}) print("[%s] %-8s %-52s exp=%-26s act=%s %s" % ( "PASS" if ok else "FAIL", section, case, expected, str(actual)[:80], note)) return ok def http(base, method, path, token=None, body=None, raw=None, timeout=30): headers = {} data = None if raw is not None: data = raw headers["Content-Type"] = "application/json" elif body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" if token: headers["Authorization"] = "Bearer " + token req = urllib.request.Request(base + path, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", "replace") except Exception as e: return -1, "CONNECTION_ERROR: %s" % type(e).__name__ def mysql(sql, db=DB): p = subprocess.run(["docker", "exec", "-i", CONTAINER, "mysql", "-u" + DB_USER, "-p" + DB_PASS, "--default-character-set=utf8mb4", "-N", "-B", db, "-e", sql], capture_output=True, text=True, encoding="utf-8", errors="replace") if p.returncode != 0: raise RuntimeError("mysql failed: %s" % (p.stderr or "").strip()[:300]) return [ln.split("\t") for ln in p.stdout.strip().splitlines() if ln != ""] def scalar(sql): r = mysql(sql) return r[0][0] if r else None def phone(n): return "%s%06d" % (PHONE_PREFIX, n) def seed_sms(p): mysql("INSERT INTO sms_verification_codes (phone, code, purpose, attempts, max_attempts, " "expired_at, used, created_at) VALUES ('%s','%s','login',0,5," "DATE_ADD(NOW(), INTERVAL 1 DAY),0,NOW());" % (p, CODE)) def login(base, p): seed_sms(p) st, body = http(base, "POST", API + "/auth/login", body={"phone": p, "code": CODE, "purpose": "login"}) if st != 200: raise RuntimeError("login failed for %s: %s %s" % (p, st, body[:200])) return json.loads(body)["data"] def as_company(base, p): data = login(base, p) uid = scalar("SELECT id FROM users WHERE phone='%s'" % p) mysql("UPDATE users SET role='COMPANY' WHERE id=%s; INSERT IGNORE INTO user_roles(user_id, role) " "VALUES (%s,'COMPANY');" % (uid, uid)) return data["accessToken"], int(uid) def jbody(body): try: return json.loads(body) except Exception: return {} def data_of(body): return jbody(body).get("data") def run(base, api_prefix): api = api_prefix seq = int(time.time()) % 100000 # ── A. 企业资料 + companySize 归一化 ───────────────────────── tok_a, uid_a = as_company(base, phone(seq)) st, body = http(base, "POST", api + "/companies/me", token=tok_a, body={"name": "黑盒企业A", "companySize": " 100-499人 ", "logoUrl": "https://logo.example/b01.png", "description": "黑盒企业简介", "industry": "互联网", "website": "https://b01.example"}) d = data_of(body) or {} record("A", "创建 companySize 去空白", "200/100-499人", "%s/%s" % (st, d.get("companySize")), st == 200 and d.get("companySize") == "100-499人") cid = d.get("id") st, body = http(base, "PUT", api + "/companies/me", token=tok_a, body={"name": "黑盒企业A"}) record("A", "更新缺省保持原值", "100-499人", (data_of(body) or {}).get("companySize"), (data_of(body) or {}).get("companySize") == "100-499人") st, body = http(base, "GET", api + "/companies/" + str(cid)) record("A", "公开读取 companySize", "100-499人", (data_of(body) or {}).get("companySize"), (data_of(body) or {}).get("companySize") == "100-499人") st, _ = http(base, "PUT", api + "/companies/me", token=tok_a, body={"name": "黑盒企业A", "companySize": "x" * 41}) record("A", "companySize 超长 400", 400, st, st == 400) st, body = http(base, "PUT", api + "/companies/me", token=tok_a, body={"name": "黑盒企业A", "companySize": " "}) record("A", "空白清空为 null", "None", (data_of(body) or {}).get("companySize"), (data_of(body) or {}).get("companySize") in (None, "")) mysql("UPDATE companies SET verified=1, company_size='100-499人' WHERE id=%s;" % cid) # ── B. 岗位全字段 + 各接口一致 ─────────────────────────────── city = "B01CITY" + str(seq) full = {"title": "黑盒岗位", "description": "岗位描述", "requirements": "岗位要求", "city": city, "employmentType": "FULL_TIME", "salaryMin": 20000, "salaryMax": 40000, "experience": "3-5年", "education": "本科", "benefits": ["五险一金", "弹性工作"], "salaryMonths": 14, "workLocation": "上海市浦东新区"} st, body = http(base, "POST", api + "/jobs", token=tok_a, body=full) jid = (data_of(body) or {}).get("id") record("B", "全字段创建岗位", "200", st, st == 200) st, _ = http(base, "PATCH", api + "/jobs/%s/status" % jid, token=tok_a, body={"status": "PUBLISHED"}) record("B", "发布岗位", "200", st, st == 200) def check_fields(label, node): ok = (node.get("experience") == "3-5年" and node.get("education") == "本科" and node.get("salaryMonths") == 14 and node.get("workLocation") == "上海市浦东新区" and node.get("benefits") == ["五险一金", "弹性工作"]) record("B", label, "5 字段一致", "exp=%s ben=%s sm=%s" % ( node.get("experience"), node.get("benefits"), node.get("salaryMonths")), ok) st, body = http(base, "GET", api + "/jobs?city=" + city) check_fields("GET /jobs 字段一致", (data_of(body) or {}).get("items", [{}])[0]) st, body = http(base, "GET", api + "/jobs/search?city=" + city + "&companyId=" + str(cid)) check_fields("GET /jobs/search 字段一致", (data_of(body) or {}).get("items", [{}])[0]) st, body = http(base, "GET", api + "/jobs/public/" + str(jid)) check_fields("GET /jobs/public/{id} 字段一致", data_of(body) or {}) st, body = http(base, "GET", api + "/companies/%s/jobs" % cid) check_fields("GET /companies/{id}/jobs 字段一致", (data_of(body) or {}).get("items", [{}])[0]) st, body = http(base, "GET", api + "/companies/me/jobs", token=tok_a) check_fields("GET /companies/me/jobs 字段一致", (data_of(body) or {}).get("items", [{}])[0]) st, body = http(base, "GET", api + "/jobs/" + str(jid), token=tok_a) check_fields("GET /jobs/{id} 字段一致", data_of(body) or {}) st, body = http(base, "PUT", api + "/jobs/" + str(jid), token=tok_a, body={"title": "黑盒岗位-更新"}) check_fields("PUT /jobs/{id} 响应字段一致", data_of(body) or {}) # ── C. 缺省 / null / 空白 / benefits=[] ───────────────────── st, body = http(base, "POST", api + "/jobs", token=tok_a, body={ "title": "最小岗位", "description": "d", "city": city, "employmentType": "FULL_TIME"}) d2 = data_of(body) or {} record("C", "缺省新字段读为 null/[]", "all None + []", "exp=%s sm=%s ben=%s" % (d2.get("experience"), d2.get("salaryMonths"), d2.get("benefits")), d2.get("experience") is None and d2.get("salaryMonths") is None and d2.get("benefits") == []) st, body = http(base, "PUT", api + "/jobs/%s" % jid, token=tok_a, body={"benefits": []}) record("C", "benefits=[] 清空", "[]", (data_of(body) or {}).get("benefits"), (data_of(body) or {}).get("benefits") == []) st, body = http(base, "PUT", api + "/jobs/%s" % jid, token=tok_a, body={"experience": " 5-10年 ", "workLocation": " "}) d3 = data_of(body) or {} record("C", "空白清空 workLocation / trim experience", "5-10年 / None", "%s / %s" % (d3.get("experience"), d3.get("workLocation")), d3.get("experience") == "5-10年" and d3.get("workLocation") is None) # ── D. salaryMonths 类型与范围 ────────────────────────────── base_job = {"title": "月数岗位", "description": "d", "city": city, "employmentType": "FULL_TIME"} for raw in ["1", "24"]: payload = json.dumps({**dict(base_job, title="月数OK" + raw), "salaryMonths": int(raw)}).encode() st, body = http(base, "POST", api + "/jobs", token=tok_a, raw=payload) record("D", "salaryMonths=%s 成功" % raw, 200, st, st == 200) for raw in ["0", "-1", "25", "1.9", "24.9", "1.0", '"12"', '""', "true", "[]", "{}", "99999999999"]: body = ('{"title":"月数非法","description":"d","city":"%s","employmentType":"FULL_TIME",' '"salaryMonths":%s}' % (city, raw)).encode() st, resp = http(base, "POST", api + "/jobs", token=tok_a, raw=body) record("D", "salaryMonths=%s -> 400" % raw, 400, st, st == 400) # ── E. 超长/非法 benefits -> 400 且无部分更新 ──────────────── st, _ = http(base, "POST", api + "/jobs", token=tok_a, body=dict(base_job, title="超长经验", experience="e" * 41)) record("E", "experience 41 字符 -> 400", 400, st, st == 400) st, _ = http(base, "POST", api + "/jobs", token=tok_a, body=dict(base_job, title="超多福利", benefits=["b%d" % i for i in range(21)])) record("E", "benefits 21 项 -> 400", 400, st, st == 400) before = int(scalar("SELECT count(*) FROM jobs WHERE title='B01PARTIAL'")) st, _ = http(base, "POST", api + "/jobs", token=tok_a, body=dict(base_job, title="B01PARTIAL", salaryMonths=25)) after = int(scalar("SELECT count(*) FROM jobs WHERE title='B01PARTIAL'")) record("E", "失败创建无部分写入", 400, "before=%d after=%d" % (before, after), st == 400 and after == before) st, _ = http(base, "PUT", api + "/jobs/" + str(jid), token=tok_a, body={"title": "B01PARTIAL_TITLE", "salaryMonths": 25}) title_now = scalar("SELECT title FROM jobs WHERE id=%s" % jid) record("E", "失败更新无部分写入", 400, "%s / %s" % (st, title_now), st == 400 and title_now == "黑盒岗位-更新") # ── F. 公开企业摘要白名单 ─────────────────────────────────── st, body = http(base, "GET", api + "/companies/" + str(cid)) node = data_of(body) or {} keys = sorted(node.keys()) record("F", "公开企业摘要 7 字段", 7, keys, keys == sorted(["id", "name", "logoUrl", "description", "industry", "verified", "companySize"])) raw = body leak_keys = [k for k in ("ownerUserId", "address", "website", "phone", "email", "materialsJson", "reviewerId", "decisionReason") if k in node] record("F", "无 ownerUserId/地址/联系方式", "no leak", "裸响应含敏感字段=%s" % leak_keys, not leak_keys and "ownerUserId" not in raw) # ── G. 草稿不可公开 / 跨企业不可改 ────────────────────────── st, body = http(base, "POST", api + "/jobs", token=tok_a, body=dict(base_job, title="黑盒草稿")) draft = (data_of(body) or {}).get("id") st, _ = http(base, "GET", api + "/jobs/public/" + str(draft)) record("G", "草稿公开详情 404", 404, st, st == 404) tok_b, _ = as_company(base, phone(seq + 1)) st, _ = http(base, "PUT", api + "/jobs/%s" % jid, token=tok_b, body={"title": "越权"}) record("G", "跨企业修改 404", 404, st, st == 404) # ── H. 历史 NULL 记录读取 ─────────────────────────────────── mysql("INSERT INTO jobs (company_id,title,description,city,employment_type,status,created_at,updated_at) " "VALUES (%s,'B01LEGACY','desc','%s','FULL_TIME','PUBLISHED',NOW(),NOW());" % (cid, city)) legacy = scalar("SELECT id FROM jobs WHERE title='B01LEGACY'") st, body = http(base, "GET", api + "/jobs/public/" + str(legacy)) node = data_of(body) or {} record("H", "历史记录 benefits=[] 且新字段 null", "[] / None", "%s / %s" % (node.get("benefits"), node.get("experience")), node.get("benefits") == [] and node.get("experience") is None) # ── J. 福利限制必须在 trim、去重之后判断（真实 HTTP） ────── padded = " " * 25 + "A" * 40 + " " * 25 st, body = http(base, "POST", api + "/jobs", token=tok_a, body=dict(base_job, title="福利归一化", benefits=[padded] * 21)) normal = data_of(body) or {} normal_id = normal.get("id") record("J", "创建允许原始超20项及trim前超40字", ["A" * 40], normal.get("benefits"), st == 200 and normal.get("benefits") == ["A" * 40]) st, body = http(base, "PUT", api + "/jobs/" + str(normal_id), token=tok_a, body={"benefits": [padded] * 21}) record("J", "更新同样按归一化后限制", ["A" * 40], (data_of(body) or {}).get("benefits"), st == 200 and (data_of(body) or {}).get("benefits") == ["A" * 40]) st, _ = http(base, "PUT", api + "/jobs/" + str(normal_id), token=tok_a, body={"benefits": ["A" * 41]}) record("J", "trim后41字拒绝", 400, st, st == 400) twenty = ["B" + str(i) for i in range(20)] st, body = http(base, "PUT", api + "/jobs/" + str(normal_id), token=tok_a, body={"benefits": twenty}) record("J", "20项不同福利允许", twenty, (data_of(body) or {}).get("benefits"), st == 200 and (data_of(body) or {}).get("benefits") == twenty) st, _ = http(base, "PUT", api + "/jobs/" + str(normal_id), token=tok_a, body={"title": "不可部分更新", "benefits": twenty + ["B20"]}) record("J", "21项不同福利拒绝", 400, st, st == 400) st, body = http(base, "GET", api + "/jobs/" + str(normal_id), token=tok_a) preserved = data_of(body) or {} record("J", "拒绝后无部分更新", "原题及20项福利", preserved.get("title"), st == 200 and preserved.get("title") == "福利归一化" and preserved.get("benefits") == twenty) st, body = http(base, "PUT", api + "/jobs/" + str(normal_id), token=tok_a, body={"benefits": None}) record("J", "null保留福利", twenty, (data_of(body) or {}).get("benefits"), st == 200 and (data_of(body) or {}).get("benefits") == twenty) st, body = http(base, "PUT", api + "/jobs/" + str(normal_id), token=tok_a, body={"benefits": []}) record("J", "空数组清空福利", [], (data_of(body) or {}).get("benefits"), st == 200 and (data_of(body) or {}).get("benefits") == []) # ── 持久化用夹具（供重启后核对） ─────────────────────────── mysql("UPDATE companies SET description='B01PERSIST' WHERE id=%s" % cid) return {"companyId": cid, "jobId": jid, "city": city, "token": tok_a, "uid": uid_a} def persist_create(base, api): seq = int(time.time()) % 100000 tok, uid = as_company(base, phone(seq)) st, body = http(base, "POST", api + "/companies/me", token=tok, body={"name": "B01PERSIST企业", "companySize": "500-999人", "description": "B01PERSIST"}) cid = (data_of(body) or {}).get("id") mysql("UPDATE companies SET verified=1 WHERE id=%s" % cid) st, body = http(base, "POST", api + "/jobs", token=tok, body={ "title": "B01PERSIST岗位", "description": "d", "city": "B01PERSISTCITY", "employmentType": "FULL_TIME", "experience": "5-10年", "salaryMonths": 16, "benefits": ["五险一金"], "workLocation": "上海市"}) jid = (data_of(body) or {}).get("id") print("PERSIST_CREATED companyId=%s jobId=%s" % (cid, jid)) return cid, jid def persist_verify(base, api): cid = scalar("SELECT id FROM companies WHERE name='B01PERSIST企业'") jid = scalar("SELECT id FROM jobs WHERE title='B01PERSIST岗位'") st, body = http(base, "GET", api + "/companies/" + str(cid)) record("I", "重启后企业资料仍在", "B01PERSIST企业/500-999人", "%s/%s" % ((data_of(body) or {}).get("name"), (data_of(body) or {}).get("companySize")), (data_of(body) or {}).get("name") == "B01PERSIST企业" and (data_of(body) or {}).get("companySize") == "500-999人") mysql("UPDATE jobs SET status='PUBLISHED', published_at=NOW() WHERE id=%s" % jid) st, body = http(base, "GET", api + "/jobs/public/" + str(jid)) node = data_of(body) or {} record("I", "重启后岗位展示字段仍在", "5-10年/16/[五险一金]", "%s/%s/%s" % (node.get("experience"), node.get("salaryMonths"), node.get("benefits")), node.get("experience") == "5-10年" and node.get("salaryMonths") == 16 and node.get("benefits") == ["五险一金"]) def main(): parser = argparse.ArgumentParser() parser.add_argument("--base", default="http://127.0.0.1:18080") parser.add_argument("--api", default="/api/v1") parser.add_argument("--mode", default="full", choices=["full", "persist-create", "persist-verify"]) parser.add_argument("--out") args = parser.parse_args() global API API = args.api base = args.base.rstrip("/") if args.mode == "persist-create": persist_create(base, args.api) return if args.mode == "persist-verify": persist_verify(base, args.api) else: summary = run(base, args.api) print("FIXTURE companyId=%s jobId=%s" % (summary["companyId"], summary["jobId"])) bad = [r for r in rows if not r["ok"]] print("\n== BLACKBOX SUMMARY: total=%d fail=%d ==" % (len(rows), len(bad))) if args.out: with open(args.out, "w", encoding="utf-8") as fh: json.dump(rows, fh, ensure_ascii=False, indent=1) sys.exit(1 if bad else 0) if __name__ == "__main__": main()