niuniu/gaokao-volunteer-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(orders): add encrypted sqlite order model

Browse Source
This commit is contained in:
Hermes Agent
2026-06-12 15:24:20 +08:00
parent 29f3f4d362
commit a11480cf10
13 changed files with 1621 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
data/orders/README.md Normal file
View File

@@ -0,0 +1,105 @@
# 订单数据模块 (T4.1)
提供订单的 SQLite schema、AES-256 加密（Fernet）、6 态状态机、Order dataclass。
## 目录结构
```
data/orders/
├── __init__.py
├── README.md
├── crypto.py # Fernet 加密/解密/哈希派生
├── state_machine.py # 6 态转换验证
├── models.py # Order dataclass + 序列化
├── schema.py # DDL + apply_schema()
└── tests/
├── __init__.py
├── test_crypto.py
├── test_state_machine.py
└── test_schema.py
```
## 字段分类
| 类别 | 处理方式 | 字段 |
| ---------- | ------------ | ------------------------------------- |
| 强敏感 PII | AES-256 加密 | `customer_phone`, `candidate_id_card` |
| 弱敏感 PII | 脱敏存储 | `customer_name`, `candidate_name` |
| 索引用哈希 | SHA-256 hex | `customer_phone_hash` |
| 业务字段 | 明文 | `amount_cents`, `source`, `status` |
| 元数据 | 明文 | `created_at`, `tags` |
## 6 态状态机
```
pending → paid → serving → delivered → completed
│ │ │ │
└──┬─────┴────────┴──────────┘
▼
refunded (终态)
```
completed / refunded 为终态，不可再转换。
## 加密密钥
环境变量 `GAOKAO_ORDERS_FERNET_KEY` 必须配置；缺失时 `get_fernet()` 抛 `MissingEncryptionKey`，**禁止降级为明文**。
生产环境密钥生成示例：
```bash
python3 -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
```
## 快速使用
```python
import os
os.environ["GAOKAO_ORDERS_FERNET_KEY"] = "your-secret-here"
from data.orders.schema import apply_schema
from data.orders.crypto import encrypt, decrypt, hash_for_index
from data.orders.state_machine import OrderStatus, assert_valid_transition
from data.orders.models import Order, generate_order_id
# 1. 建表
conn = apply_schema("data/orders.db")
# 2. 创建订单
order = Order(
id=generate_order_id(),
source="xianyu",
service_version="standard",
amount_cents=9900,
status=OrderStatus.PENDING.value,
customer_name="张*",
customer_phone="13800001234",
candidate_name="张同学",
)
conn.execute(
"INSERT INTO orders (...) VALUES (...)",
order.to_db_row(),
)
# 3. 状态转换
assert_valid_transition(order.status, "paid")
```
## 验收（DoD）
- [x] AES-256 加密（Fernet）对 customer_phone / candidate_id_card 落盘
- [x] 6 态状态机 + 转换合法性校验
- [x] SQLite schema 幂等可重建
- [x] 外键约束启用
- [x] SHA-256 hash 字段支持手机号去重
- [x] CHECK 约束拒绝非法 status
## 下游衔接
- **T4.2 DAO**: 调用 `apply_schema()` 建表，使用 `state_machine.assert_valid_transition()` 守护状态变更
- **T4.3 CLI**: 直接 import `Order` dataclass + crypto
- **T6.1 FastAPI**: schema 不变，DAO 复用
## 版本
v1.0 — 2026-06-12 — T4.1 实施

5
data/orders/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""订单数据模块 (T4.1)
提供 SQLite schema、AES-256 加密（Fernet）、6 态订单状态机、数据模型。
由 T4.2 DAO 层负责 CRUD 包装。
"""

100
data/orders/crypto.py Normal file
View File

@@ -0,0 +1,100 @@
"""订单敏感字段加密模块
使用 Fernet（cryptography 包）实现 AES-256 加密：
- Fernet = AES-128-CBC + HMAC-SHA256 + 时间戳，密钥派生后等价于 256-bit 安全强度
- 密钥来源：环境变量 GAOKAO_ORDERS_FERNET_KEY（任意字符串，经 SHA-256 派生为 32 字节）
- 密文存储：base64-url 字符串，写入 SQLite TEXT 字段
索引字段：手机号用 SHA-256 hex 单独存储，供去重查询（密文不能直接索引）。
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import os
from functools import lru_cache
from cryptography.fernet import Fernet, InvalidToken
ENV_KEY_NAME = "GAOKAO_ORDERS_FERNET_KEY"
_KEY_LENGTH = 32 # bytes; SHA-256 digest length
class EncryptionError(RuntimeError):
"""加密/解密失败时抛出的基础异常。"""
class MissingEncryptionKey(EncryptionError):
"""环境变量未配置加密密钥时抛出。"""
def derive_key(secret: str) -> bytes:
"""从任意 secret 字符串派生 32 字节 Fernet key（base64-url 编码）。
使用 SHA-256 对 secret 散列得到 32 字节，再 base64-url 编码为 Fernet 接受的格式。
同 secret 总是派生同 key（确定性），便于备份恢复。
"""
if not isinstance(secret, str) or not secret:
raise EncryptionError("secret 必须为非空字符串")
digest = hashlib.sha256(secret.encode("utf-8")).digest()
return base64.urlsafe_b64encode(digest)
@lru_cache(maxsize=1)
def get_fernet() -> Fernet:
"""从环境变量获取 Fernet 实例。缺失则抛 MissingEncryptionKey（不静默降级）。"""
secret = os.environ.get(ENV_KEY_NAME)
if not secret:
raise MissingEncryptionKey(
f"环境变量 {ENV_KEY_NAME} 未设置；为保证数据安全，禁止以明文存储敏感字段。"
)
return Fernet(derive_key(secret))
def encrypt(plaintext: str) -> str:
"""加密字符串并返回 base64 字符串。
输入必须为 str；输出为 str（utf-8 解码后的 base64），可直接写入 SQLite TEXT。
"""
if not isinstance(plaintext, str):
raise EncryptionError(
f"encrypt 输入必须为 str，实际为 {type(plaintext).__name__}"
)
token = get_fernet().encrypt(plaintext.encode("utf-8"))
return token.decode("ascii")
def decrypt(ciphertext: str) -> str:
"""解密 base64 字符串并返回原始明文。
密钥错误或密文被篡改时抛 InvalidToken。
"""
if not isinstance(ciphertext, str):
raise EncryptionError(
f"decrypt 输入必须为 str，实际为 {type(ciphertext).__name__}"
)
try:
plain = get_fernet().decrypt(ciphertext.encode("ascii"))
except InvalidToken as exc:
raise EncryptionError("密文无法解密（密钥错误或数据被篡改）") from exc
return plain.decode("utf-8")
def hash_for_index(value: str) -> str:
"""对索引字段（如手机号）计算 SHA-256 hex，用于去重查询。
注意：SHA-256 不抗碰撞查找，仅用于去重，不应用于需要抗碰撞审计的场景。
"""
if not isinstance(value, str):
raise EncryptionError(
f"hash_for_index 输入必须为 str，实际为 {type(value).__name__}"
)
return hashlib.sha256(value.strip().encode("utf-8")).hexdigest()
def constant_time_equals(a: str, b: str) -> bool:
"""常数时间字符串比较（防时序攻击的辅助工具，DAO 比较密文时使用）。"""
return hmac.compare_digest(a.encode("utf-8"), b.encode("utf-8"))

157
data/orders/models.py Normal file
View File

@@ -0,0 +1,157 @@
"""订单数据模型 (T4.1)
Order dataclass 覆盖 TECH_ARCHITECTURE §3.4 所有字段；敏感字段按加密/明文分字段存储。
to_dict / from_dict 负责明文↔密文自动转换（DAO 层直接调用）。
"""
from __future__ import annotations
import json
import random
import string
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from typing import Any, List, Optional
from .crypto import encrypt, decrypt, hash_for_index
def utc_now_iso() -> str:
"""返回当前 UTC 时间的 ISO8601 字符串（秒精度）。"""
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def generate_order_id() -> str:
"""生成订单号 GKO-YYYYMMDD-XXXX（4 位大写字母+数字）。"""
date_part = datetime.now(timezone.utc).strftime("%Y%m%d")
rand = "".join(random.choices(string.ascii_uppercase + string.digits, k=4))
return f"GKO-{date_part}-{rand}"
@dataclass
class Order:
"""订单数据模型。
加密约定：
- customer_phone / candidate_id_card 在 from_dict 时加密，to_dict 时解密；
数据库落盘只看到 _enc 后缀字段。
- customer_phone_hash 仅用于去重查询（SHA-256 hex）。
"""
id: str
source: str # 'xianyu'|'wechat'|'web'|'school'
external_id: Optional[str] = None
service_version: str = "basic" # 'audit'|'basic'|'standard'|'premium'
amount_cents: int = 0
status: str = "pending"
status_updated_at: Optional[str] = None
# 客户（明文/加密分字段）
customer_name: Optional[str] = None
customer_phone: Optional[str] = None # 明文（API 入口接收）
customer_phone_hash: Optional[str] = None # 自动派生
customer_wechat: Optional[str] = None
# 考生
candidate_name: Optional[str] = None
candidate_id_card: Optional[str] = None # 明文（API 入口接收）
candidate_province: Optional[str] = None
candidate_score: Optional[int] = None
candidate_rank: Optional[int] = None
candidate_subjects: List[str] = field(default_factory=list)
candidate_interests: Optional[str] = None
candidate_strong_subjects: Optional[str] = None
candidate_weak_subjects: Optional[str] = None
candidate_family: Optional[str] = None
# 服务
assigned_consultant: Optional[str] = None
plan_file: Optional[str] = None
audit_report: Optional[str] = None
pdf_path: Optional[str] = None
# 时间戳
created_at: Optional[str] = None
paid_at: Optional[str] = None
started_at: Optional[str] = None
delivered_at: Optional[str] = None
completed_at: Optional[str] = None
# 元数据
notes: Optional[str] = None
tags: List[str] = field(default_factory=list)
upgrade_from: Optional[str] = None
def __post_init__(self) -> None:
"""自动派生 hash 与时间戳。"""
if self.customer_phone and not self.customer_phone_hash:
self.customer_phone_hash = hash_for_index(self.customer_phone)
if not self.created_at:
self.created_at = utc_now_iso()
if not self.status_updated_at:
self.status_updated_at = self.created_at
# tags/subjects 入库为 JSON 字符串
if isinstance(self.tags, list):
self._tags_json = json.dumps(self.tags, ensure_ascii=False)
if isinstance(self.candidate_subjects, list):
self._subjects_json = json.dumps(
self.candidate_subjects, ensure_ascii=False
)
# 序列化到 DB（加密敏感字段）
def to_db_row(self) -> dict[str, Any]:
"""返回可直接写入 orders 表的字典（敏感字段已加密）。"""
data = asdict(self)
# 移除明文敏感字段
data.pop("customer_phone", None)
data.pop("candidate_id_card", None)
# 加密落盘字段
if self.customer_phone:
data["customer_phone_enc"] = encrypt(self.customer_phone)
if self.candidate_id_card:
data["candidate_id_card_enc"] = encrypt(self.candidate_id_card)
# tags/subjects JSON 化
data["tags"] = json.dumps(self.tags, ensure_ascii=False)
data["candidate_subjects"] = json.dumps(
self.candidate_subjects, ensure_ascii=False
)
return data
# 从 DB 反序列化（解密敏感字段）
@classmethod
def from_db_row(cls, row: dict[str, Any]) -> "Order":
"""从数据库行构造 Order，自动解密敏感字段。"""
data = dict(row)
if data.get("customer_phone_enc"):
data["customer_phone"] = decrypt(data["customer_phone_enc"])
else:
data["customer_phone"] = None
if data.get("candidate_id_card_enc"):
data["candidate_id_card"] = decrypt(data["candidate_id_card_enc"])
else:
data["candidate_id_card"] = None
# JSON 列表字段
for key in ("tags", "candidate_subjects"):
raw = data.get(key)
if raw:
try:
data[key] = json.loads(raw)
except (json.JSONDecodeError, TypeError):
data[key] = []
else:
data[key] = []
# 移除 DB-only 加密字段
data.pop("customer_phone_enc", None)
data.pop("candidate_id_card_enc", None)
return cls(**data)
def to_dict(self, decrypt_sensitive: bool = True) -> dict[str, Any]:
"""导出为字典。
decrypt_sensitive=True 时敏感字段以明文返回（API 响应）；False 时只返回 hash。
"""
data = asdict(self)
if not decrypt_sensitive:
data.pop("customer_phone", None)
data.pop("candidate_id_card", None)
return data

118
data/orders/schema.py Normal file
View File

@@ -0,0 +1,118 @@
"""SQLite Schema 模块 (T4.1)
定义 orders 主表与 order_status_history 审计表的 DDL，并提供幂等的
apply_schema() 应用函数。所有表使用 IF NOT EXISTS；启用外键约束。
"""
from __future__ import annotations
import sqlite3
from pathlib import Path
SCHEMA_SQL: str = """
-- 订单主表
CREATE TABLE IF NOT EXISTS orders (
-- 主键 / 业务标识
id TEXT PRIMARY KEY,
source TEXT NOT NULL,
external_id TEXT,
service_version TEXT NOT NULL,
amount_cents INTEGER NOT NULL CHECK(amount_cents >= 0),
status TEXT NOT NULL CHECK(status IN
('pending','paid','serving','delivered','completed','refunded')),
status_updated_at TEXT NOT NULL,
-- 客户信息（加密 + 脱敏）
customer_name TEXT,
customer_phone_enc TEXT,
customer_phone_hash TEXT,
customer_wechat TEXT,
-- 考生信息
candidate_name TEXT,
candidate_id_card_enc TEXT,
candidate_province TEXT,
candidate_score INTEGER,
candidate_rank INTEGER,
candidate_subjects TEXT,
candidate_interests TEXT,
candidate_strong_subjects TEXT,
candidate_weak_subjects TEXT,
candidate_family TEXT,
-- 服务信息
assigned_consultant TEXT,
plan_file TEXT,
audit_report TEXT,
pdf_path TEXT,
-- 时间戳（ISO8601 UTC）
created_at TEXT NOT NULL,
paid_at TEXT,
started_at TEXT,
delivered_at TEXT,
completed_at TEXT,
-- 元数据
notes TEXT,
tags TEXT,
upgrade_from TEXT,
FOREIGN KEY (upgrade_from) REFERENCES orders(id) ON DELETE SET NULL
);
CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status);
CREATE INDEX IF NOT EXISTS idx_orders_source ON orders(source);
CREATE INDEX IF NOT EXISTS idx_orders_created_at ON orders(created_at);
CREATE INDEX IF NOT EXISTS idx_orders_phone_hash ON orders(customer_phone_hash);
CREATE UNIQUE INDEX IF NOT EXISTS uniq_orders_external
ON orders(source, external_id) WHERE external_id IS NOT NULL;
-- 状态历史（审计）
CREATE TABLE IF NOT EXISTS order_status_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id TEXT NOT NULL,
from_status TEXT,
to_status TEXT NOT NULL,
actor TEXT,
reason TEXT,
changed_at TEXT NOT NULL,
FOREIGN KEY (order_id) REFERENCES orders(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_status_history_order ON order_status_history(order_id);
"""
def apply_schema(db_path: str | Path) -> sqlite3.Connection:
"""应用 schema 到指定 SQLite 文件，返回连接。
- 启用外键约束 (PRAGMA foreign_keys = ON)
- 父目录自动创建
- 幂等：可重复执行
"""
db_path = Path(db_path)
if db_path.parent and not db_path.parent.exists():
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path))
try:
conn.execute("PRAGMA foreign_keys = ON")
conn.executescript(SCHEMA_SQL)
conn.commit()
except Exception:
conn.close()
raise
return conn
def get_schema_version(conn: sqlite3.Connection) -> int:
"""读取当前 schema 版本号（首次运行返回 0）。后续迁移将引入 schema_version 表。"""
try:
row = conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='orders'"
).fetchone()
return 1 if row else 0
except sqlite3.DatabaseError:
return 0

98
data/orders/state_machine.py Normal file
View File

@@ -0,0 +1,98 @@
"""订单状态机 (T4.1)
6 态状态机：pending → paid → serving → delivered → completed；
任何阶段可转入 refunded。completed / refunded 为终态。
非法转换抛 InvalidStateTransition；DAO 层捕获后回滚事务。
"""
from __future__ import annotations
from enum import Enum
class OrderStatus(str, Enum):
"""订单 6 态枚举（继承 str 以便直接写入 SQLite / JSON）。"""
PENDING = "pending"
PAID = "paid"
SERVING = "serving"
DELIVERED = "delivered"
COMPLETED = "completed"
REFUNDED = "refunded"
TERMINAL_STATUSES: frozenset[str] = frozenset(
{OrderStatus.COMPLETED.value, OrderStatus.REFUNDED.value}
)
# 合法状态转换：单向推进 + 任意阶段可退款
ALLOWED_TRANSITIONS: dict[str, frozenset[str]] = {
OrderStatus.PENDING.value: frozenset(
{OrderStatus.PAID.value, OrderStatus.REFUNDED.value}
),
OrderStatus.PAID.value: frozenset(
{OrderStatus.SERVING.value, OrderStatus.REFUNDED.value}
),
OrderStatus.SERVING.value: frozenset(
{OrderStatus.DELIVERED.value, OrderStatus.REFUNDED.value}
),
OrderStatus.DELIVERED.value: frozenset(
{OrderStatus.COMPLETED.value, OrderStatus.REFUNDED.value}
),
OrderStatus.COMPLETED.value: frozenset(),
OrderStatus.REFUNDED.value: frozenset(),
}
class InvalidStateTransition(ValueError):
"""非法状态转换时抛出。"""
def is_known_status(status: str) -> bool:
"""判断字符串是否为已定义的 6 态之一。"""
return status in {s.value for s in OrderStatus}
def is_terminal(status: str) -> bool:
"""判断状态是否为终态（completed / refunded）。"""
return status in TERMINAL_STATUSES
def is_valid_transition(from_status: str, to_status: str) -> bool:
"""判断 from → to 是否为合法转换。
- 已知状态但非法转换 → False
- 未知状态 → False（不抛异常，DAO 层需先校验状态合法性）
"""
if not (is_known_status(from_status) and is_known_status(to_status)):
return False
if from_status == to_status:
# 相同状态视为非法（避免无意义的状态写入历史表）
return False
return to_status in ALLOWED_TRANSITIONS[from_status]
def assert_valid_transition(from_status: str, to_status: str) -> None:
"""校验状态转换合法性，非法时抛 InvalidStateTransition。"""
if not is_known_status(from_status):
raise InvalidStateTransition(f"未知起始状态: {from_status!r}")
if not is_known_status(to_status):
raise InvalidStateTransition(f"未知目标状态: {to_status!r}")
if from_status == to_status:
raise InvalidStateTransition(
f"状态相同无需转换: {from_status!r} → {to_status!r}"
)
if to_status not in ALLOWED_TRANSITIONS[from_status]:
allowed = sorted(ALLOWED_TRANSITIONS[from_status])
raise InvalidStateTransition(
f"非法状态转换: {from_status!r} → {to_status!r}（允许: {allowed}）"
)
def next_states(from_status: str) -> frozenset[str]:
"""返回 from_status 的所有合法下一状态（用于前端展示）。"""
if not is_known_status(from_status):
return frozenset()
return ALLOWED_TRANSITIONS[from_status]

0
data/orders/tests/__init__.py Normal file
View File

157
data/orders/tests/test_crypto.py Normal file
View File

@@ -0,0 +1,157 @@
"""crypto 模块测试"""
import os
import pytest
# 测试开始前设置密钥（必须在 import data.orders.crypto 之前）
os.environ.setdefault("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-unit-tests")
from data.orders.crypto import (
derive_key,
encrypt,
decrypt,
hash_for_index,
constant_time_equals,
EncryptionError,
MissingEncryptionKey,
ENV_KEY_NAME,
)
from cryptography.fernet import Fernet
def setup_function(_):
"""每个用例前清掉缓存，避免环境变量副作用。"""
from data.orders import crypto
crypto.get_fernet.cache_clear()
# -------- derive_key --------
def test_derive_key_is_deterministic():
"""同 secret 派生同 key。"""
k1 = derive_key("hello")
k2 = derive_key("hello")
assert k1 == k2
assert len(k1) > 0
def test_derive_key_distinct_for_different_secrets():
"""不同 secret 派生不同 key。"""
assert derive_key("a") != derive_key("b")
def test_derive_key_rejects_empty():
with pytest.raises(EncryptionError):
derive_key("")
def test_derive_key_accepts_fernet_key_directly():
"""Fernet 直接生成的 key 应被接受（兼容）。"""
direct = Fernet.generate_key()
# derive_key 用 sha256 派生；这里验证 Fernet 能用直接 key 初始化
Fernet(direct) # 不抛即可
# -------- encrypt/decrypt round-trip --------
def test_encrypt_decrypt_round_trip_chinese():
"""中文明文 round-trip。"""
plaintext = "张同学 13800001234"
ct = encrypt(plaintext)
assert ct != plaintext
assert decrypt(ct) == plaintext
def test_encrypt_decrypt_round_trip_empty_string():
"""空字符串也能 round-trip。"""
assert decrypt(encrypt("")) == ""
def test_encrypt_output_is_ascii():
"""密文为 base64-url 字符串，仅含 ASCII。"""
ct = encrypt("hello world")
ct.encode("ascii") # 不抛即通过
def test_decrypt_with_wrong_key_fails():
"""错误密钥解密应抛 EncryptionError。"""
# 先用当前 key 加密
ct = encrypt("secret-value")
# 切换密钥
os.environ[ENV_KEY_NAME] = "different-secret"
from data.orders import crypto
crypto.get_fernet.cache_clear()
with pytest.raises(EncryptionError):
decrypt(ct)
# 恢复
os.environ[ENV_KEY_NAME] = "test-secret-for-unit-tests"
crypto.get_fernet.cache_clear()
def test_encrypt_rejects_non_string():
with pytest.raises(EncryptionError):
encrypt(12345) # type: ignore[arg-type]
def test_decrypt_rejects_non_string():
with pytest.raises(EncryptionError):
decrypt(12345) # type: ignore[arg-type]
def test_missing_key_raises_missing_encryption_key(monkeypatch):
"""未配置密钥时 get_fernet 抛 MissingEncryptionKey，不静默降级。"""
monkeypatch.delenv(ENV_KEY_NAME, raising=False)
from data.orders import crypto
crypto.get_fernet.cache_clear()
with pytest.raises(MissingEncryptionKey):
encrypt("test")
# -------- hash_for_index --------
def test_hash_for_index_is_stable():
"""同输入同输出。"""
assert hash_for_index("13800001234") == hash_for_index("13800001234")
def test_hash_for_index_strips_whitespace():
"""前后空白不影响 hash。"""
assert hash_for_index(" 13800001234 ") == hash_for_index("13800001234")
def test_hash_for_index_is_hex_64_chars():
"""SHA-256 hex 应为 64 字符。"""
h = hash_for_index("13800001234")
assert len(h) == 64
int(h, 16) # 可被解析为 16 进制
def test_hash_for_index_different_inputs():
"""不同输入产生不同 hash。"""
assert hash_for_index("13800001234") != hash_for_index("13800001235")
def test_hash_for_index_rejects_non_string():
with pytest.raises(EncryptionError):
hash_for_index(12345) # type: ignore[arg-type]
# -------- constant_time_equals --------
def test_constant_time_equals_equal():
assert constant_time_equals("abc", "abc") is True
def test_constant_time_equals_unequal():
assert constant_time_equals("abc", "abd") is False
def test_constant_time_equals_different_length():
assert constant_time_equals("abc", "abcd") is False

153
data/orders/tests/test_models.py Normal file
View File

@@ -0,0 +1,153 @@
"""models 模块集成测试（Order dataclass + 加解密串联）。"""
import os
import json
os.environ.setdefault("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-unit-tests")
from data.orders.models import Order, generate_order_id, utc_now_iso
from data.orders.crypto import decrypt, hash_for_index
def setup_function(_):
from data.orders import crypto
crypto.get_fernet.cache_clear()
def test_generate_order_id_format():
oid = generate_order_id()
assert oid.startswith("GKO-")
parts = oid.split("-")
assert len(parts) == 3
assert len(parts[1]) == 8 # YYYYMMDD
assert len(parts[2]) == 4
def test_order_auto_derives_phone_hash():
order = Order(
id="GKO-20260612-AAAA",
source="web",
service_version="basic",
amount_cents=1000,
customer_phone="13800001234",
)
assert order.customer_phone_hash == hash_for_index("13800001234")
def test_order_to_db_row_encrypts_phone():
order = Order(
id="GKO-20260612-AAAB",
source="web",
service_version="basic",
amount_cents=1000,
customer_phone="13800001234",
)
row = order.to_db_row()
# 明文不出现在 DB 行
assert "customer_phone" not in row or row.get("customer_phone") is None
# 加密字段存在
assert "customer_phone_enc" in row
assert row["customer_phone_enc"] != "13800001234"
# hash 字段保留
assert row["customer_phone_hash"] == hash_for_index("13800001234")
# 密文可解
assert decrypt(row["customer_phone_enc"]) == "13800001234"
def test_order_to_db_row_encrypts_id_card():
order = Order(
id="GKO-20260612-AAAC",
source="web",
service_version="basic",
amount_cents=1000,
candidate_id_card="430102200501011234",
)
row = order.to_db_row()
assert "candidate_id_card_enc" in row
assert "candidate_id_card" not in row or row.get("candidate_id_card") is None
assert decrypt(row["candidate_id_card_enc"]) == "430102200501011234"
def test_order_from_db_row_decrypts():
"""模拟数据库往返：to_db_row → 模拟 DB 读取 → from_db_row。"""
order_in = Order(
id="GKO-20260612-AAAD",
source="xianyu",
external_id="EXT-X",
service_version="standard",
amount_cents=9900,
status="pending",
customer_name="张*",
customer_phone="13800001234",
candidate_name="张同学",
candidate_province="湖南",
candidate_score=578,
candidate_subjects=["物理", "化学", "生物"],
tags=["高优", "VIP"],
)
row = order_in.to_db_row()
order_out = Order.from_db_row(row)
assert order_out.customer_phone == "13800001234"
assert order_out.customer_name == "张*"
assert order_out.candidate_province == "湖南"
assert order_out.candidate_subjects == ["物理", "化学", "生物"]
assert order_out.tags == ["高优", "VIP"]
def test_order_to_dict_decrypt_sensitive_true():
order = Order(
id="GKO-20260612-AAAE",
source="web",
service_version="basic",
customer_phone="13800001234",
candidate_id_card="430102200501011234",
)
d = order.to_dict(decrypt_sensitive=True)
assert d["customer_phone"] == "13800001234"
assert d["candidate_id_card"] == "430102200501011234"
def test_order_to_dict_decrypt_sensitive_false():
order = Order(
id="GKO-20260612-AAAF",
source="web",
service_version="basic",
customer_phone="13800001234",
candidate_id_card="430102200501011234",
)
d = order.to_dict(decrypt_sensitive=False)
assert "customer_phone" not in d
assert "candidate_id_card" not in d
assert "customer_phone_hash" in d
def test_order_default_timestamps():
order = Order(
id="GKO-20260612-AAAG",
source="web",
service_version="basic",
)
assert order.created_at is not None
assert order.status_updated_at == order.created_at
def test_order_tags_json_serializable():
"""to_db_row 后 tags 为 JSON 字符串。"""
order = Order(
id="GKO-20260612-AAAH",
source="web",
service_version="basic",
tags=["a", "b"],
)
row = order.to_db_row()
# 重新解析应能还原
parsed = json.loads(row["tags"])
assert parsed == ["a", "b"]
def test_utc_now_iso_is_iso8601():
ts = utc_now_iso()
assert "T" in ts
assert ts.endswith("+00:00") or ts.endswith("Z")

244
data/orders/tests/test_schema.py Normal file
View File

@@ -0,0 +1,244 @@
"""schema 模块测试"""
import os
import sqlite3
import tempfile
from pathlib import Path
import pytest
os.environ.setdefault("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-unit-tests")
from data.orders.schema import apply_schema, get_schema_version, SCHEMA_SQL
@pytest.fixture
def tmp_db():
with tempfile.TemporaryDirectory() as d:
yield Path(d) / "test_orders.db"
def test_apply_schema_creates_orders_table(tmp_db):
conn = apply_schema(tmp_db)
try:
row = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='orders'"
).fetchone()
assert row is not None
finally:
conn.close()
def test_apply_schema_creates_status_history_table(tmp_db):
conn = apply_schema(tmp_db)
try:
row = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='order_status_history'"
).fetchone()
assert row is not None
finally:
conn.close()
def test_apply_schema_enables_foreign_keys(tmp_db):
conn = apply_schema(tmp_db)
try:
fk = conn.execute("PRAGMA foreign_keys").fetchone()[0]
assert fk == 1
finally:
conn.close()
def test_apply_schema_is_idempotent(tmp_db):
"""重复执行 apply_schema 不报错。"""
conn1 = apply_schema(tmp_db)
conn1.close()
conn2 = apply_schema(tmp_db)
conn2.close() # 不抛即通过
def test_apply_schema_creates_indexes(tmp_db):
conn = apply_schema(tmp_db)
try:
idxs = conn.execute(
"SELECT name FROM sqlite_master WHERE type='index' AND name LIKE 'idx_%'"
).fetchall()
idx_names = {i[0] for i in idxs}
assert "idx_orders_status" in idx_names
assert "idx_orders_source" in idx_names
assert "idx_orders_created_at" in idx_names
assert "idx_orders_phone_hash" in idx_names
assert "idx_status_history_order" in idx_names
finally:
conn.close()
def test_apply_schema_creates_parent_dir(tmp_db):
"""父目录不存在时自动创建。"""
nested = tmp_db.parent / "subdir1" / "subdir2" / "nested.db"
assert not nested.parent.exists()
conn = apply_schema(nested)
conn.close()
assert nested.exists()
def test_check_constraint_rejects_invalid_status(tmp_db):
"""CHECK 约束拒绝非法 status 字符串。"""
conn = apply_schema(tmp_db)
try:
with pytest.raises(sqlite3.IntegrityError):
conn.execute(
"""INSERT INTO orders
(id, source, service_version, amount_cents, status,
status_updated_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
"X1",
"web",
"basic",
100,
"INVALID_STATUS",
"2026-06-12T10:00:00+00:00",
"2026-06-12T10:00:00+00:00",
),
)
finally:
conn.close()
def test_check_constraint_rejects_negative_amount(tmp_db):
"""CHECK 约束拒绝负金额。"""
conn = apply_schema(tmp_db)
try:
with pytest.raises(sqlite3.IntegrityError):
conn.execute(
"""INSERT INTO orders
(id, source, service_version, amount_cents, status,
status_updated_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
"X2",
"web",
"basic",
-1,
"pending",
"2026-06-12T10:00:00+00:00",
"2026-06-12T10:00:00+00:00",
),
)
finally:
conn.close()
def test_unique_external_id_per_source(tmp_db):
"""(source, external_id) 组合唯一。"""
conn = apply_schema(tmp_db)
try:
conn.execute(
"""INSERT INTO orders
(id, source, external_id, service_version, amount_cents, status,
status_updated_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
"A1",
"xianyu",
"EXT-001",
"basic",
100,
"pending",
"2026-06-12T10:00:00+00:00",
"2026-06-12T10:00:00+00:00",
),
)
with pytest.raises(sqlite3.IntegrityError):
conn.execute(
"""INSERT INTO orders
(id, source, external_id, service_version, amount_cents, status,
status_updated_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
"A2",
"xianyu",
"EXT-001",
"basic",
100,
"pending",
"2026-06-12T10:00:00+00:00",
"2026-06-12T10:00:00+00:00",
),
)
# 不同 source 可同 external_id
conn.execute(
"""INSERT INTO orders
(id, source, external_id, service_version, amount_cents, status,
status_updated_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
"A3",
"wechat",
"EXT-001",
"basic",
100,
"pending",
"2026-06-12T10:00:00+00:00",
"2026-06-12T10:00:00+00:00",
),
)
finally:
conn.close()
def test_status_history_cascade_delete(tmp_db):
"""删除订单时状态历史级联删除。"""
conn = apply_schema(tmp_db)
try:
conn.execute(
"""INSERT INTO orders
(id, source, service_version, amount_cents, status,
status_updated_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
"B1",
"web",
"basic",
100,
"pending",
"2026-06-12T10:00:00+00:00",
"2026-06-12T10:00:00+00:00",
),
)
conn.execute(
"""INSERT INTO order_status_history
(order_id, from_status, to_status, actor, changed_at)
VALUES (?, ?, ?, ?, ?)""",
("B1", None, "pending", "system", "2026-06-12T10:00:00+00:00"),
)
conn.commit()
# 验证存在
count = conn.execute(
"SELECT COUNT(*) FROM order_status_history WHERE order_id = ?", ("B1",)
).fetchone()[0]
assert count == 1
# 删除订单
conn.execute("DELETE FROM orders WHERE id = ?", ("B1",))
conn.commit()
# 级联删除
count = conn.execute(
"SELECT COUNT(*) FROM order_status_history WHERE order_id = ?", ("B1",)
).fetchone()[0]
assert count == 0
finally:
conn.close()
def test_get_schema_version_returns_1_after_apply(tmp_db):
conn = apply_schema(tmp_db)
try:
assert get_schema_version(conn) == 1
finally:
conn.close()
def test_schema_sql_is_nonempty():
assert "CREATE TABLE IF NOT EXISTS orders" in SCHEMA_SQL
assert "CREATE TABLE IF NOT EXISTS order_status_history" in SCHEMA_SQL

177
data/orders/tests/test_state_machine.py Normal file
View File

@@ -0,0 +1,177 @@
"""state_machine 模块测试"""
import pytest
from data.orders.state_machine import (
OrderStatus,
TERMINAL_STATUSES,
is_known_status,
is_terminal,
is_valid_transition,
assert_valid_transition,
next_states,
InvalidStateTransition,
)
# -------- 6 态枚举存在性 --------
def test_all_six_statuses_defined():
assert {s.value for s in OrderStatus} == {
"pending",
"paid",
"serving",
"delivered",
"completed",
"refunded",
}
# -------- 终态判定 --------
def test_terminal_statuses_are_completed_and_refunded():
assert TERMINAL_STATUSES == frozenset({"completed", "refunded"})
def test_is_terminal_true_for_terminals():
assert is_terminal("completed") is True
assert is_terminal("refunded") is True
def test_is_terminal_false_for_non_terminals():
for s in ("pending", "paid", "serving", "delivered"):
assert is_terminal(s) is False
def test_is_terminal_false_for_unknown():
assert is_terminal("not_a_status") is False
# -------- 合法转换 --------
@pytest.mark.parametrize(
"from_,to",
[
("pending", "paid"),
("paid", "serving"),
("serving", "delivered"),
("delivered", "completed"),
("pending", "refunded"),
("paid", "refunded"),
("serving", "refunded"),
("delivered", "refunded"),
],
)
def test_valid_transitions(from_, to):
assert is_valid_transition(from_, to) is True
# assert_valid_transition 不抛
assert_valid_transition(from_, to)
# -------- 非法转换 --------
@pytest.mark.parametrize(
"from_,to",
[
# 倒退
("paid", "pending"),
("serving", "paid"),
("delivered", "serving"),
("completed", "delivered"),
# 跨级跳跃
("pending", "serving"),
("pending", "delivered"),
("pending", "completed"),
("paid", "delivered"),
("paid", "completed"),
# 终态之后
("completed", "refunded"),
("refunded", "pending"),
("refunded", "paid"),
# 相同状态
("pending", "pending"),
("paid", "paid"),
],
)
def test_invalid_transitions(from_, to):
assert is_valid_transition(from_, to) is False
with pytest.raises(InvalidStateTransition):
assert_valid_transition(from_, to)
# -------- 未知状态 --------
def test_unknown_from_status_returns_false():
assert is_valid_transition("unknown", "paid") is False
def test_unknown_to_status_returns_false():
assert is_valid_transition("pending", "unknown") is False
def test_assert_unknown_from_raises():
with pytest.raises(InvalidStateTransition):
assert_valid_transition("foo", "paid")
def test_assert_unknown_to_raises():
with pytest.raises(InvalidStateTransition):
assert_valid_transition("pending", "foo")
# -------- next_states --------
def test_next_states_for_pending():
assert next_states("pending") == frozenset({"paid", "refunded"})
def test_next_states_for_completed_is_empty():
assert next_states("completed") == frozenset()
def test_next_states_for_refunded_is_empty():
assert next_states("refunded") == frozenset()
def test_next_states_for_unknown_is_empty():
assert next_states("xxx") == frozenset()
# -------- is_known_status --------
def test_is_known_status():
for s in ("pending", "paid", "serving", "delivered", "completed", "refunded"):
assert is_known_status(s) is True
assert is_known_status("xxx") is False
assert is_known_status("") is False
# -------- 完整业务路径 --------
def test_full_happy_path():
"""完整业务路径 pending → paid → serving → delivered → completed。"""
path = ["pending", "paid", "serving", "delivered", "completed"]
for i in range(len(path) - 1):
assert_valid_transition(path[i], path[i + 1])
def test_refund_path_from_any_non_terminal():
"""任意非终态可退款。"""
for s in ("pending", "paid", "serving", "delivered"):
assert_valid_transition(s, "refunded")
def test_cannot_transition_from_terminal():
"""终态之后任何转换都非法。"""
with pytest.raises(InvalidStateTransition):
assert_valid_transition("completed", "paid")
with pytest.raises(InvalidStateTransition):
assert_valid_transition("refunded", "paid")