"""已签发授权台账测试（厂商侧 Registry）。 重点验证： 1. 由 license 文档派生记录（唯一写入口，不手工拼记录） 2. 状态判定（永久/有效/即将到期/已过期/未生效）与剩余天数 3. 检索：客户 / 机器码 / 插件 / 关键字 / 状态 4. **续期复用台账里的来源指纹** —— 台账最大的价值：不必再向客户索取申请码 5. 重新导入不覆盖人工维护的备注 6. 导出 .lic（按客户分子目录）与 CSV（UTF-8 BOM，Excel 中文不乱码） 7. 结构版本不匹配时明确报错，而不是让查询诡异失败 运行：`python test/test_license_registry.py`，或 `uv run pytest test/test_license_registry.py`。 """ import csv import sqlite3 import sys import tempfile from datetime import date, timedelta from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey # noqa: E402 from chongming_license import ( # noqa: E402 Registry, RegistryError, build_license, build_payload, machine_code, public_key_b64, public_key_fingerprint, record_from_license, ) failures: list[str] = [] def check(name, cond, detail=""): print(f"[{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else "")) if not cond: failures.append(name) PRIV = Ed25519PrivateKey.generate() PUB = public_key_b64(PRIV) FP = public_key_fingerprint(PUB) SRC = {"board_serial": "sn123456789", "emmc_cid": "1a2b3c4d", "mac": "aabbccddeeff"} CODE = machine_code(SRC) def make_doc(*, customer="某某科技", days=300, plugins=("aibox_sophon_adapter",), note="", machine=None, sources=None, lic_id="", not_before=None): payload = build_payload( machine_code=machine or CODE, plugins=list(plugins), sources=SRC if sources is None else sources, customer=customer, note=note, lic_id=lic_id, not_before=not_before, expires_at=(date.today() + timedelta(days=days)) if days is not None else None, ) return build_license(payload, PRIV, kid="vendor-2026") def test_record_derivation(): """记录完全由 license 文档派生，查询列与原始文档不会不一致。""" doc = make_doc(customer="派生测试", note="合同 X", plugins=["a", "b"]) rec = record_from_license(doc, public_key_fp=FP, source="test") payload = doc["payload"] check("lic_id 取自载荷", rec.lic_id == payload["lic_id"], rec.lic_id) check("机器码大写归一", rec.machine_code == payload["machine_code"].upper()) check("插件列表解析", rec.plugins == ["a", "b"], str(rec.plugins)) check("来源指纹解析", rec.sources == SRC, str(sorted(rec.sources))) check("kid 取自信封", rec.kid == "vendor-2026") check("公钥指纹记录在案", rec.public_key_fp == FP) check("可还原完整文档", rec.license_doc()["sig"] == doc["sig"]) # 日期只保留日期部分：否则 SQL 里的字符串比较会得出错误排序 doc2 = make_doc() doc2["payload"]["expires_at"] = "2027-09-01T12:34:56Z" check("日期裁剪到日粒度", record_from_license(doc2).expires_at == "2027-09-01", str(record_from_license(doc2).expires_at)) # 缺 lic_id 必须报错，而不是静默写一条垃圾记录 try: record_from_license({"payload": {"machine_code": "X"}}) check("缺 lic_id 时报错", False, "未抛异常") except RegistryError: check("缺 lic_id 时报错", True) def test_status_and_days_left(): """五种状态的判定（含边界：到期当天仍有效）。""" tmp = Path(tempfile.mkdtemp()) with Registry(tmp / "r.db") as reg: reg.save(make_doc(lic_id="L-FOREVER", days=None), public_key_fp=FP) reg.save(make_doc(lic_id="L-ACTIVE", days=300), public_key_fp=FP) reg.save(make_doc(lic_id="L-EXPIRING", days=10), public_key_fp=FP) reg.save(make_doc(lic_id="L-EXPIRED", days=-1), public_key_fp=FP) reg.save(make_doc(lic_id="L-TODAY", days=0), public_key_fp=FP) reg.save(make_doc(lic_id="L-FUTURE", days=100, not_before=(date.today() + timedelta(days=30)).isoformat()), public_key_fp=FP) got = {r.lic_id: r.status() for r in reg.query()} check("无到期日 → forever", got["L-FOREVER"] == "forever", got["L-FOREVER"]) check("远期 → active", got["L-ACTIVE"] == "active", got["L-ACTIVE"]) check("30 天内 → expiring", got["L-EXPIRING"] == "expiring", got["L-EXPIRING"]) check("已过 → expired", got["L-EXPIRED"] == "expired", got["L-EXPIRED"]) check("到期当天仍有效（含当天）", got["L-TODAY"] == "expiring", got["L-TODAY"]) check("未到生效日 → not_yet_valid", got["L-FUTURE"] == "not_yet_valid", got["L-FUTURE"]) check("剩余天数为负表示已过期", reg.get("L-EXPIRED").days_left() < 0, str(reg.get("L-EXPIRED").days_left())) check("永久授权剩余天数为 None", reg.get("L-FOREVER").days_left() is None) st = reg.stats() check("概览计数一致", st == {"total": 6, "expired": 1, "expiring": 2, "active": 1, "forever": 1, "not_yet_valid": 1}, str(st)) def test_query_filters(): """检索：客户 / 机器码 / 插件 / 关键字 / 状态，以及排序。""" tmp = Path(tempfile.mkdtemp()) with Registry(tmp / "r.db") as reg: reg.save(make_doc(lic_id="L-A", customer="阿尔法科技", days=300, plugins=["doubao"], note="HT-001"), public_key_fp=FP) reg.save(make_doc(lic_id="L-B", customer="贝塔公司", days=10, plugins=["ip_sound_column"], note="HT-002"), public_key_fp=FP) reg.save(make_doc(lic_id="L-C", customer="贝塔公司", days=200, plugins=["doubao_extended"], note="HT-003"), public_key_fp=FP) other = "AAAA-BBBB-CCCC-DDDD-EEEE-FFFF" reg.save(make_doc(lic_id="L-D", customer="伽马", days=300, machine=other, sources={"mac": "aabbccddeeff"}), public_key_fp=FP) check("按客户模糊匹配", {r.lic_id for r in reg.query(customer="贝塔")} == {"L-B", "L-C"}) # L-C 装的是 doubao_extended，查 doubao 时不应被命中（前缀同名不算同一插件） check("按插件匹配不误伤前缀同名", {r.lic_id for r in reg.query(plugin="doubao")} == {"L-A"}, str([r.lic_id for r in reg.query(plugin="doubao")])) check("按插件匹配到前缀同名者本身", {r.lic_id for r in reg.query(plugin="doubao_extended")} == {"L-C"}) check("按机器码匹配", {r.lic_id for r in reg.query(machine_code=CODE[:16])} == {"L-A", "L-B", "L-C"}) check("按备注关键字匹配", {r.lic_id for r in reg.query(keyword="HT-002")} == {"L-B"}) check("大小写不敏感（机器码）", {r.lic_id for r in reg.query(machine_code=other.lower())} == {"L-D"}) order = [r.lic_id for r in reg.query()] check("按到期日升序（急的在前）", order[0] == "L-B", str(order)) # 某台机器的全部历史授权 check("按机器码取全部历史", len(reg.find_by_machine_code(CODE)) == 3, str(len(reg.find_by_machine_code(CODE)))) check("limit 生效", len(reg.query(limit=2)) == 2) def test_renew_reuses_sources_without_request_code(): """核心能力：续期复用台账里的指纹，无需再向客户索取申请码。""" tmp = Path(tempfile.mkdtemp()) with Registry(tmp / "r.db") as reg: old = reg.save(make_doc(lic_id="L-OLD", customer="续期客户", days=5, plugins=["aibox_sophon_adapter", "ip_sound_column"], note="合同 HT-100"), public_key_fp=FP) check("原授权即将到期", old.status() == "expiring", old.status()) # 模拟现场：客户已联系不上，只有台账记录可用 renewed_doc = make_doc(lic_id="L-NEW", customer=old.customer, days=365, plugins=old.plugins, note=old.note, machine=old.machine_code, sources=old.sources) new = reg.save(renewed_doc, public_key_fp=FP, source="renew") check("续期成功", new.status() == "active", new.status()) check("指纹与原件一致（未索取新申请码）", new.sources == old.sources) check("机器码不变", new.machine_code == old.machine_code) check("新旧授权都在台账（可追溯）", {r.lic_id for r in reg.find_by_machine_code(old.machine_code)} == {"L-OLD", "L-NEW"}) def test_note_preserved_on_reimport(): """重新导入同一份 license 不应覆盖人工维护的备注。""" tmp = Path(tempfile.mkdtemp()) with Registry(tmp / "r.db") as reg: doc = make_doc(lic_id="L-N", note="原始备注") reg.save(doc, public_key_fp=FP) check("备注已入库", reg.get("L-N").note == "原始备注") reg.update_note("L-N", "人工补充：已电话确认续期") check("备注可更新", reg.get("L-N").note == "人工补充：已电话确认续期") created = reg.get("L-N").created_at reg.save(doc, public_key_fp=FP) # 重新导入（载荷里还是旧备注） after = reg.get("L-N") check("重新导入不覆盖人工备注", after.note == "人工补充：已电话确认续期", after.note) check("创建时间保持不变", after.created_at == created, after.created_at) check("更新时间被推进", after.updated_at >= created) def test_export_and_csv(): """导出 .lic（可按客户分目录）与 CSV（UTF-8 BOM，Excel 不乱码）。""" tmp = Path(tempfile.mkdtemp()) with Registry(tmp / "r.db") as reg: reg.save(make_doc(lic_id="L-X1", customer="甲方/科技", days=100), public_key_fp=FP) reg.save(make_doc(lic_id="L-X2", customer="乙方", days=-3), public_key_fp=FP) files = reg.export_licenses(tmp / "arc", subdir_by_customer=True) check("导出数量正确", len(files) == 2, str([f.name for f in files])) check("客户名含非法字符时安全替换", any("甲方" in str(f) for f in files), str([str(f.parent.name) for f in files])) # 导出的文件应能被校验端直接读取（验证导出内容不是"看着像"而已） # 注意顺序按到期日升序：files[0] 是已过期的 L-X2，files[1] 是有效的 L-X1 from chongming_license import read_license valid_file = next(f for f in files if "L-X1" in f.name) expired_file = next(f for f in files if "L-X2" in f.name) st = read_license(valid_file, PUB, sources=SRC) check("导出的有效授权仍可验签通过", st.ok, st.message) st2 = read_license(expired_file, PUB, sources=SRC) check("导出的已过期授权被判定为过期（到期日完整保留）", not st2.ok and st2.code == "expired", f"{st2.code} / {st2.message}") check("导出内容可原样还原", valid_file.read_text(encoding="utf-8").lstrip().startswith("{")) csv_path = reg.export_csv(tmp / "台账.csv") with csv_path.open(encoding="utf-8-sig", newline="") as f: rows = list(csv.reader(f)) check("CSV 含表头 + 两条数据", len(rows) == 3, str(len(rows))) check("CSV 表头含状态与到期日", "状态" in rows[0] and "到期日期" in rows[0], str(rows[0])) check("CSV 记录了状态文案", any("已过期" in r for r in rows[1:]), str(rows[1:])) check("CSV 以 UTF-8 BOM 写入（Excel 中文不乱码）", csv_path.read_bytes()[:3] == b"\xef\xbb\xbf") def test_schema_version_mismatch_is_explicit(): """结构版本不匹配时明确报错，而不是让后续查询诡异失败。""" tmp = Path(tempfile.mkdtemp()) path = tmp / "r.db" with Registry(path) as reg: reg.save(make_doc(lic_id="L-V"), public_key_fp=FP) # 手工把版本改成未来值，模拟"被更新版本的工具写过" conn = sqlite3.connect(str(path)) conn.execute("PRAGMA user_version = 99") conn.commit() conn.close() try: Registry(path) check("版本不匹配时拒绝打开", False, "未抛异常") except RegistryError as exc: check("版本不匹配时拒绝打开", "99" in str(exc), str(exc).splitlines()[0]) # 已存在但 user_version=0 的库（旧文件）应被平滑接管 path2 = tmp / "old.db" conn = sqlite3.connect(str(path2)) conn.executescript( "CREATE TABLE issued_licenses (lic_id TEXT PRIMARY KEY, machine_code TEXT, " "customer TEXT, note TEXT, plugins_json TEXT, sources_json TEXT, not_before TEXT, " "expires_at TEXT, issued_at TEXT, kid TEXT, public_key_fp TEXT, license_json TEXT, " "source TEXT, created_at TEXT, updated_at TEXT);" ) conn.commit() conn.close() with Registry(path2) as reg2: check("user_version=0 的旧库被平滑接管", reg2.stats()["total"] == 0) def test_concurrent_open(): """两个实例同时打开同一库（多个 GUI 窗口 / CLI 并行）不应互相报错。""" tmp = Path(tempfile.mkdtemp()) path = tmp / "r.db" a = Registry(path) b = Registry(path) try: a.save(make_doc(lic_id="L-C1", customer="并发A"), public_key_fp=FP) b.save(make_doc(lic_id="L-C2", customer="并发B"), public_key_fp=FP) check("两实例各自写入后互相可见", {r.lic_id for r in a.query()} == {"L-C1", "L-C2"}, str([r.lic_id for r in a.query()])) check("B 能看到 A 的记录", b.get("L-C1") is not None) finally: a.close() b.close() def test_delete(): tmp = Path(tempfile.mkdtemp()) with Registry(tmp / "r.db") as reg: reg.save(make_doc(lic_id="L-DEL"), public_key_fp=FP) check("删除成功", reg.delete("L-DEL") is True) check("删除后查不到", reg.get("L-DEL") is None) check("删除不存在的返回 False", reg.delete("L-DEL") is False) def main(): tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] for fn in tests: print(f"\n--- {fn.__name__} ---") fn() print("\n" + "=" * 56) if failures: print(f"❌ {len(failures)} 项未通过: {failures}") return 1 print("✅ 全部通过") return 0 if __name__ == "__main__": sys.exit(main())