328 lines
15 KiB
Python
328 lines
15 KiB
Python
"""已签发授权台账测试（厂商侧 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())
|