diff --git a/data/orders/__init__.py b/data/orders/__init__.py index 2ecc165..ca66b78 100644 --- a/data/orders/__init__.py +++ b/data/orders/__init__.py @@ -1,5 +1,106 @@ -"""订单数据模块 (T4.1) +"""订单数据模块 (T4.1 + T4.2) -提供 SQLite schema、AES-256 加密（Fernet）、6 态订单状态机、数据模型。 -由 T4.2 DAO 层负责 CRUD 包装。 +提供： +- SQLite schema（AES-256 加密字段 + 6 态状态机 + 审计表） +- Fernet 加密/解密/索引哈希派生 +- 6 态订单状态机 +- ``Order`` dataclass + 加密/脱敏序列化 +- ``OrdersDAO`` 完整 CRUD + 事务 + 状态机守护（**T4.2**） + +下游使用 +-------- + +```python +import os +os.environ["GAOKAO_ORDERS_FERNET_KEY"] = "your-secret-here" + +from data.orders import ( + # schema + apply_schema, + # crypto + encrypt, decrypt, hash_for_index, + # state machine + assert_valid_transition, OrderStatus, + # models + Order, generate_order_id, + # DAO (T4.2) + OrdersDAO, UpsertResult, StatusChange, + OrderNotFound, DuplicateOrder, +) +``` """ + +from .crypto import ( + ENV_KEY_NAME, + EncryptionError, + MissingEncryptionKey, + constant_time_equals, + decrypt, + derive_key, + encrypt, + get_fernet, + hash_for_index, +) +from .dao import ( + DuplicateOrder, + OrderNotFound, + OrdersDAO, + StatusChange, + UpsertResult, +) +from .models import ( + DecryptPolicy, + Order, + generate_order_id, + utc_now_iso, +) +from .schema import SCHEMA_SQL, apply_schema, get_schema_version +from .state_machine import ( + ALLOWED_TRANSITIONS, + InvalidStateTransition, + OrderStatus, + TERMINAL_STATUSES, + assert_valid_transition, + is_known_status, + is_terminal, + is_valid_transition, + next_states, +) + +__all__ = [ + # schema + "SCHEMA_SQL", + "apply_schema", + "get_schema_version", + # crypto + "ENV_KEY_NAME", + "EncryptionError", + "MissingEncryptionKey", + "constant_time_equals", + "decrypt", + "derive_key", + "encrypt", + "get_fernet", + "hash_for_index", + # state machine + "ALLOWED_TRANSITIONS", + "InvalidStateTransition", + "OrderStatus", + "TERMINAL_STATUSES", + "assert_valid_transition", + "is_known_status", + "is_terminal", + "is_valid_transition", + "next_states", + # models + "DecryptPolicy", + "Order", + "generate_order_id", + "utc_now_iso", + # DAO + "DuplicateOrder", + "OrderNotFound", + "OrdersDAO", + "StatusChange", + "UpsertResult", +] diff --git a/data/orders/dao.py b/data/orders/dao.py new file mode 100644 index 0000000..dd44d80 --- /dev/null +++ b/data/orders/dao.py @@ -0,0 +1,806 @@ +"""订单 DAO 数据访问层 (T4.2) + +提供订单表的 CRUD、事务、加密字段透明处理、6 态状态机守护的转换写入。 + +设计原则 +-------- + +1. **加密透明化**：API 入口接收 ``Order`` dataclass（明文 PII）， + DAO 负责落盘前加密、读取后解密，调用方无需关心 ``*_enc`` 字段。 +2. **状态机守护**：所有状态转换走 ``transition_status()``，单事务内 + 写 ``orders.status`` + ``order_status_history``；非法转换抛 + :class:`InvalidStateTransition` 并回滚。 +3. **事务显式**：默认每方法一次 ``commit``；批量操作走 ``transaction()`` + 上下文管理器（异常时统一 ``rollback``）。 +4. **行工厂统一**：DAO 内部强制 ``sqlite3.Row`` 工厂，调用方传入的 + ``row_factory`` 不会被污染（使用前保存 / 使用后恢复）。 +5. **去重路径**：``(source, external_id)`` 唯一索引上的 ``upsert_by_external_id()`` + 接管 :mod:`data.channel_sync.dao_extension` 的同名函数（向下兼容）。 + +依赖 +---- + +- :class:`data.orders.models.Order` +- :func:`data.orders.schema.apply_schema` +- :mod:`data.orders.state_machine` +- :func:`data.orders.crypto.encrypt` / :func:`decrypt` + +替代与回滚 +---------- + +- :class:`data.channel_sync.dao_extension.UpsertResult` 与 + :func:`data.channel_sync.dao_extension.upsert_by_external_id` 由本模块的 + :class:`UpsertResult` / :meth:`OrdersDAO.upsert_by_external_id` 取代。 + T8.1 现存调用迁移到本 DAO 后即可删除 dao_extension.py。 +""" + +from __future__ import annotations + +import contextlib +import json +import sqlite3 +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterator, List, Optional, Union + +from .models import Order, utc_now_iso +from .schema import apply_schema +from .state_machine import ( + InvalidStateTransition, + assert_valid_transition, + is_known_status, +) + + +# --------------------------------------------------------------------------- +# 异常与结果类型 +# --------------------------------------------------------------------------- + + +class OrderNotFound(LookupError): + """按主键或唯一键查询订单时未命中。""" + + +class DuplicateOrder(ValueError): + """尝试插入违反唯一约束的订单（手机号 hash 或 external_id 冲突）。""" + + +@dataclass +class UpsertResult: + """upsert_by_external_id 的返回结构。 + + 字段含义: + - ``action='inserted'``：原 DB 不存在该 ``(source, external_id)``，已新建。 + - ``action='updated'``：已存在且状态可推进，已更新 status / status_updated_at + 并写入一条 status_history。 + - ``action='unchanged'``：已存在且状态未变，未写入。 + - ``action='illegal_transition'``：已存在但状态转换非法（未写入）， + 调用方应降级为 ``decision='rejected'``。 + """ + + order_id: str + action: str # 'inserted' | 'updated' | 'unchanged' | 'illegal_transition' + old_status: Optional[str] = None + new_status: Optional[str] = None + error: Optional[str] = None + + +@dataclass +class StatusChange: + """状态历史记录（来自 order_status_history 表）。""" + + id: int + order_id: str + from_status: Optional[str] + to_status: str + actor: Optional[str] + reason: Optional[str] + changed_at: str + + +# --------------------------------------------------------------------------- +# 内部常量 +# --------------------------------------------------------------------------- + +# 与 schema.py 对齐的 orders 表可写列清单。 +# 加密字段：customer_phone_enc / candidate_id_card_enc 来自 Order.to_db_row()。 +_WRITABLE_COLUMNS: tuple[str, ...] = ( + "id", + "source", + "external_id", + "service_version", + "amount_cents", + "status", + "status_updated_at", + "customer_name", + "customer_phone_enc", + "customer_phone_hash", + "customer_wechat", + "candidate_name", + "candidate_id_card_enc", + "candidate_province", + "candidate_score", + "candidate_rank", + "candidate_subjects", + "candidate_interests", + "candidate_strong_subjects", + "candidate_weak_subjects", + "candidate_family", + "assigned_consultant", + "plan_file", + "audit_report", + "pdf_path", + "created_at", + "paid_at", + "started_at", + "delivered_at", + "completed_at", + "notes", + "tags", + "upgrade_from", +) + +# 历史阶段字段映射：状态进入时自动置位的 timestamp 字段。 +# 状态 → timestamp 字段名（COALESCE 写入：已有则保留）。 +_STATUS_TIMESTAMP: dict[str, str] = { + "paid": "paid_at", + "serving": "started_at", + "delivered": "delivered_at", + "completed": "completed_at", +} + + +# --------------------------------------------------------------------------- +# DAO 主类 +# --------------------------------------------------------------------------- + + +class OrdersDAO: + """订单表 DAO。 + + 两种初始化方式： + + 1. 接管已建立的连接:: + + conn = apply_schema("/path/orders.db") + dao = OrdersDAO(conn) + + DAO 不会关闭 conn，调用方负责。 + + 2. 接管数据库路径:: + + with OrdersDAO.connect("/path/orders.db") as dao: + dao.create(order) + + 退出上下文时自动 commit/close。 + """ + + def __init__(self, conn: sqlite3.Connection) -> None: + self._conn = conn + self._tx_depth = 0 # 嵌套事务深度（0 = 顶层） + # DAO 假设 conn 已启用 foreign_keys；不强制重设（调用方控制）。 + + # ------------------------------------------------------------------ + # 构造/连接管理 + # ------------------------------------------------------------------ + + @classmethod + def connect( + cls, + db_path: Union[str, Path], + *, + row_factory: bool = True, + ) -> "OrdersDAO": + """按路径建立连接并应用 schema（幂等），返回 DAO。 + + ``row_factory=True`` 时强制设为 ``sqlite3.Row``，便于 ``dict(row)``。 + + 用法:: + + with OrdersDAO.connect("data/orders.db") as dao: + dao.create(order) + """ + conn = apply_schema(db_path) + if row_factory: + conn.row_factory = sqlite3.Row + return cls(conn) + + @property + def conn(self) -> sqlite3.Connection: + """暴露底层连接（只读引用，调用方不应自行 commit/close）。""" + return self._conn + + def close(self) -> None: + """关闭底层连接。""" + self._conn.close() + + @contextlib.contextmanager + def transaction(self) -> Iterator[sqlite3.Connection]: + """事务上下文。 + + 进入时自动 ``BEGIN``，异常时 ``ROLLBACK`` 并重新抛出； + 正常退出时 ``COMMIT``。 + + 嵌套语义: 内部 ``create()`` / ``update()`` / ``transition_status()`` + 自身会再调 ``transaction()``。当外层已在事务中时,内层不再开新事务, + 直接复用外层 — 任何一层的异常都会触发外层回滚。这是经典的 SAVEPOINT + 简化版(无部分回滚),适合本 DAO 的写多读少场景。 + + 用法:: + + with dao.transaction() as conn: + conn.execute(...) + conn.execute(...) # 同事务 + """ + self._tx_depth += 1 + try: + if self._tx_depth == 1: + # 顶层：依赖 sqlite3 的隐式 BEGIN，由 commit/rollback 终止 + yield self._conn + self._conn.commit() + else: + # 嵌套：复用外层事务，不 commit/rollback + yield self._conn + except Exception: + if self._tx_depth == 1: + self._conn.rollback() + raise + finally: + self._tx_depth -= 1 + + # ------------------------------------------------------------------ + # 内部辅助 + # ------------------------------------------------------------------ + + @contextlib.contextmanager + def _row_factory_ctx(self) -> Iterator[None]: + """临时把 conn.row_factory 设为 sqlite3.Row，退出时恢复。""" + prior = self._conn.row_factory + self._conn.row_factory = sqlite3.Row + try: + yield + finally: + self._conn.row_factory = prior + + @staticmethod + def _coerce_for_db(key: str, value: Any) -> Any: + """保证 tags / candidate_subjects 落盘为 JSON 字符串。""" + if key in ("tags", "candidate_subjects") and isinstance(value, (list, tuple)): + return json.dumps(list(value), ensure_ascii=False) + return value + + def _row_to_order(self, row: sqlite3.Row) -> Order: + """sqlite3.Row → Order（解密 + JSON 解析由 from_db_row 负责）。""" + return Order.from_db_row(dict(row)) + + def _select_columns(self) -> str: + return ", ".join(_WRITABLE_COLUMNS) + + # ------------------------------------------------------------------ + # 写入：create / update + # ------------------------------------------------------------------ + + def create( + self, + order: Order, + *, + actor: Optional[str] = None, + reason: Optional[str] = None, + ) -> Order: + """插入新订单，并写入首条 status_history（from=None → status）。 + + - 重复主键 / 重复 external_id 抛 :class:`DuplicateOrder`。 + - 重复 phone_hash（非唯一索引，仅用于查询）允许 — 同一手机号 + 下不同省份/年份可以下多单。 + + 返回: 写入后的 Order（数据库回读，字段已对齐 SQLite 默认值）。 + """ + db_row = order.to_db_row() + # 防御：过滤掉 schema 中不存在的列 + valid_cols = set(_WRITABLE_COLUMNS) + db_row = { + k: self._coerce_for_db(k, v) for k, v in db_row.items() if k in valid_cols + } + # 落盘 timestamp 不能为空 + if not db_row.get("status_updated_at"): + db_row["status_updated_at"] = utc_now_iso() + if not db_row.get("created_at"): + db_row["created_at"] = utc_now_iso() + + cols = list(db_row.keys()) + placeholders = ",".join("?" for _ in cols) + values = [db_row[c] for c in cols] + + with self.transaction(): + try: + self._conn.execute( + f"INSERT INTO orders ({','.join(cols)}) VALUES ({placeholders})", + values, + ) + except sqlite3.IntegrityError as exc: + msg = str(exc).lower() + if "unique" in msg or "primary key" in msg: + raise DuplicateOrder( + f"订单已存在: id={order.id} source={order.source} external_id={order.external_id} ({exc})" + ) from exc + raise + self._insert_status_history( + order_id=order.id, + from_status=None, + to_status=order.status, + actor=actor or "dao_create", + reason=reason or "create", + ) + # 读回行（确保返回字段与 DB 对齐） + created_id = order.id + with self._row_factory_ctx(): + row = self._conn.execute( + f"SELECT {self._select_columns()} FROM orders WHERE id=?", + (created_id,), + ).fetchone() + return self._row_to_order(row) + + def update( + self, + order_id: str, + updates: dict[str, Any], + *, + actor: Optional[str] = None, + reason: Optional[str] = None, + ) -> Order: + """按主键更新订单业务字段（非 status 字段）。 + + 适用字段：customer_name / customer_wechat / candidate_name / + candidate_province / candidate_score / candidate_rank / + candidate_subjects / candidate_interests / candidate_strong_subjects / + candidate_weak_subjects / candidate_family / assigned_consultant / + plan_file / audit_report / pdf_path / notes / tags / amount_cents / + service_version / external_id。 + + **禁止**通过本方法改 ``status`` —— 改状态请走 :meth:`transition_status`， + 以保证状态机校验和历史写入。 + + - 不存在抛 :class:`OrderNotFound`。 + - 重复 external_id 抛 :class:`DuplicateOrder`。 + + 返回: 更新后的 Order。 + """ + if "status" in updates: + raise ValueError("禁止通过 update() 改 status；请使用 transition_status()") + + allowed = set(_WRITABLE_COLUMNS) - {"id", "status", "status_updated_at"} + bad = set(updates) - allowed + if bad: + raise ValueError( + f"update() 不允许字段: {sorted(bad)}（仅业务字段，不含 status/timestamp）" + ) + + with self.transaction(): + with self._row_factory_ctx(): + row = self._conn.execute( + "SELECT id FROM orders WHERE id=?", + (order_id,), + ).fetchone() + if row is None: + raise OrderNotFound(f"订单不存在: {order_id}") + + set_clauses: list[str] = [] + values: list[Any] = [] + for k, v in updates.items(): + set_clauses.append(f"{k}=?") + values.append(self._coerce_for_db(k, v)) + # 业务字段更新不影响 status_updated_at；只有 transition_status 才动 + values.append(order_id) + try: + self._conn.execute( + f"UPDATE orders SET {','.join(set_clauses)} WHERE id=?", + values, + ) + except sqlite3.IntegrityError as exc: + msg = str(exc).lower() + if "unique" in msg: + raise DuplicateOrder( + f"订单更新违反唯一约束: id={order_id} ({exc})" + ) from exc + raise + with self._row_factory_ctx(): + row = self._conn.execute( + f"SELECT {self._select_columns()} FROM orders WHERE id=?", + (order_id,), + ).fetchone() + return self._row_to_order(row) + + # ------------------------------------------------------------------ + # 状态转换 + # ------------------------------------------------------------------ + + def transition_status( + self, + order_id: str, + to_status: str, + *, + actor: Optional[str] = None, + reason: Optional[str] = None, + ) -> Order: + """状态机守护的状态转换。 + + 流程（单事务）: + + 1. 读现状 ``SELECT status FROM orders WHERE id=?`` + 2. ``assert_valid_transition(from, to)`` 校验；非法抛 + :class:`InvalidStateTransition` 并回滚 + 3. ``UPDATE orders SET status=?, status_updated_at=?, =COALESCE(?, ?)`` + 4. ``INSERT INTO order_status_history(from, to, actor, reason)`` + 5. 读回返回 + + 返回: 转换后的 Order。 + """ + if not is_known_status(to_status): + raise InvalidStateTransition(f"未知目标状态: {to_status!r}") + + with self.transaction(): + with self._row_factory_ctx(): + row = self._conn.execute( + "SELECT status FROM orders WHERE id=?", + (order_id,), + ).fetchone() + if row is None: + raise OrderNotFound(f"订单不存在: {order_id}") + from_status = row["status"] + # 状态机校验（非法时抛 InvalidStateTransition） + assert_valid_transition(from_status, to_status) + + now_iso = utc_now_iso() + # 对应时间戳字段：COALESCE(原值, 新值) — 已有则保留 + ts_col = _STATUS_TIMESTAMP.get(to_status) + if ts_col is not None: + self._conn.execute( + f""" + UPDATE orders SET + status=?, + status_updated_at=?, + {ts_col} = COALESCE({ts_col}, ?) + WHERE id=? + """, + (to_status, now_iso, now_iso, order_id), + ) + else: + # refunded / pending 等没有专用时间戳 + self._conn.execute( + """ + UPDATE orders SET + status=?, + status_updated_at=? + WHERE id=? + """, + (to_status, now_iso, order_id), + ) + + self._insert_status_history( + order_id=order_id, + from_status=from_status, + to_status=to_status, + actor=actor or "dao_transition", + reason=reason, + changed_at=now_iso, + ) + with self._row_factory_ctx(): + row = self._conn.execute( + f"SELECT {self._select_columns()} FROM orders WHERE id=?", + (order_id,), + ).fetchone() + return self._row_to_order(row) + + def _insert_status_history( + self, + *, + order_id: str, + from_status: Optional[str], + to_status: str, + actor: Optional[str] = None, + reason: Optional[str] = None, + changed_at: Optional[str] = None, + ) -> int: + """插入一条 order_status_history 记录，返回 rowid。 + + 不 commit —— 由外层 transaction() 统一提交。 + """ + if changed_at is None: + changed_at = utc_now_iso() + cur = self._conn.execute( + """ + INSERT INTO order_status_history( + order_id, from_status, to_status, actor, reason, changed_at + ) VALUES (?, ?, ?, ?, ?, ?) + """, + (order_id, from_status, to_status, actor, reason, changed_at), + ) + return int(cur.lastrowid or 0) + + def get_status_history(self, order_id: str) -> List[StatusChange]: + """读订单完整状态历史（按 changed_at 升序）。""" + with self._row_factory_ctx(): + rows = self._conn.execute( + """ + SELECT id, order_id, from_status, to_status, actor, reason, changed_at + FROM order_status_history + WHERE order_id=? + ORDER BY changed_at ASC, id ASC + """, + (order_id,), + ).fetchall() + return [ + StatusChange( + id=int(r["id"]), + order_id=r["order_id"], + from_status=r["from_status"], + to_status=r["to_status"], + actor=r["actor"], + reason=r["reason"], + changed_at=r["changed_at"], + ) + for r in rows + ] + + # ------------------------------------------------------------------ + # 查询：get / list / find + # ------------------------------------------------------------------ + + def get(self, order_id: str) -> Order: + """按主键读取订单（解密 PII）。不存在抛 :class:`OrderNotFound`。""" + with self._row_factory_ctx(): + row = self._conn.execute( + f"SELECT {self._select_columns()} FROM orders WHERE id=?", + (order_id,), + ).fetchone() + if row is None: + raise OrderNotFound(f"订单不存在: {order_id}") + return self._row_to_order(row) + + def get_by_external_id(self, source: str, external_id: str) -> Optional[Order]: + """按 (source, external_id) 查询；找不到返回 None。""" + with self._row_factory_ctx(): + row = self._conn.execute( + f"SELECT {self._select_columns()} FROM orders " + "WHERE source=? AND external_id=? LIMIT 1", + (source, external_id), + ).fetchone() + return self._row_to_order(row) if row is not None else None + + def find_by_phone(self, phone: str) -> List[Order]: + """按手机号 hash 查询（去重 / 客户识别用），返回全部匹配。 + + phone 接受明文；DAO 内部按 SHA-256 hash 查询。 + """ + from .crypto import hash_for_index + + with self._row_factory_ctx(): + rows = self._conn.execute( + f"SELECT {self._select_columns()} FROM orders " + "WHERE customer_phone_hash=? ORDER BY created_at DESC", + (hash_for_index(phone),), + ).fetchall() + return [self._row_to_order(r) for r in rows] + + def list( + self, + *, + status: Optional[str] = None, + source: Optional[str] = None, + limit: int = 50, + offset: int = 0, + ) -> List[Order]: + """按筛选条件列订单（默认按 created_at DESC）。 + + - ``status`` 必须是已知 6 态之一；传未知值抛 :class:`ValueError`。 + - ``limit`` 取值 1..1000；越界抛 :class:`ValueError`。 + - ``offset`` ≥ 0。 + """ + if status is not None and not is_known_status(status): + raise ValueError(f"未知 status: {status!r}") + if not (1 <= limit <= 1000): + raise ValueError(f"limit 越界 (1..1000): {limit}") + if offset < 0: + raise ValueError(f"offset 不能为负: {offset}") + + clauses: list[str] = [] + params: list[Any] = [] + if status is not None: + clauses.append("status=?") + params.append(status) + if source is not None: + clauses.append("source=?") + params.append(source) + where = ("WHERE " + " AND ".join(clauses)) if clauses else "" + sql = ( + f"SELECT {self._select_columns()} FROM orders " + f"{where} ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?" + ) + params.extend([limit, offset]) + + with self._row_factory_ctx(): + rows = self._conn.execute(sql, params).fetchall() + return [self._row_to_order(r) for r in rows] + + def count( + self, *, status: Optional[str] = None, source: Optional[str] = None + ) -> int: + """统计订单数（同样支持 status / source 过滤）。""" + if status is not None and not is_known_status(status): + raise ValueError(f"未知 status: {status!r}") + clauses: list[str] = [] + params: list[Any] = [] + if status is not None: + clauses.append("status=?") + params.append(status) + if source is not None: + clauses.append("source=?") + params.append(source) + where = ("WHERE " + " AND ".join(clauses)) if clauses else "" + row = self._conn.execute( + f"SELECT COUNT(*) AS n FROM orders {where}", + params, + ).fetchone() + # COUNT 总是返回 1 行；防御性 default + return int(row[0] if row else 0) + + def stats_by_status(self) -> dict[str, int]: + """按 status 分组统计订单数（含 0 计数的完整 6 态）。""" + rows = self._conn.execute( + "SELECT status, COUNT(*) AS n FROM orders GROUP BY status" + ).fetchall() + result: dict[str, int] = { + s: 0 + for s in ( + "pending", + "paid", + "serving", + "delivered", + "completed", + "refunded", + ) + } + for r in rows: + # 允许 sqlite3.Row / tuple 两种形态 + status_key = r["status"] if hasattr(r, "keys") else r[0] + count_val = r["n"] if hasattr(r, "keys") else r[1] + if status_key in result: + result[status_key] = int(count_val) + return result + + # ------------------------------------------------------------------ + # 幂等 upsert（与 T8.1 dao_extension 对齐） + # ------------------------------------------------------------------ + + def upsert_by_external_id( + self, + order: Order, + *, + actor: str = "channel_sync", + reason: Optional[str] = None, + ) -> UpsertResult: + """按 (source, external_id) 唯一索引写入或更新订单。 + + 行为（与 :mod:`data.channel_sync.dao_extension.upsert_by_external_id` 对齐）: + + - **external_id 缺失** → ``action='illegal_transition'`` + error + - **不存在** → 插入新行 + 写 status_history(from=None → status) + - **已存在且状态不变** → ``action='unchanged'``，不写 status_history + - **已存在且状态可推进** → 更新 status / status_updated_at + 写 status_history + - **已存在但状态非法转换** → ``action='illegal_transition'`` + error + + 返回: :class:`UpsertResult`。 + """ + if not order.external_id: + return UpsertResult( + order_id=order.id, + action="illegal_transition", + error="external_id 缺失，无法做幂等 upsert", + ) + + # 1) 查重 + with self._row_factory_ctx(): + row = self._conn.execute( + "SELECT * FROM orders WHERE source=? AND external_id=? LIMIT 1", + (order.source, order.external_id), + ).fetchone() + + if row is None: + # INSERT — 沿用调用方传入的 reason/actor + try: + created = self.create(order, actor=actor, reason=reason) + except DuplicateOrder as exc: + return UpsertResult( + order_id=order.id, + action="illegal_transition", + error=f"重复订单: {exc}", + ) + return UpsertResult( + order_id=created.id, + action="inserted", + old_status=None, + new_status=created.status, + ) + + # 2) 已存在：判断状态转换 + existing = self._row_to_order(row) + old_status = existing.status + if old_status == order.status: + return UpsertResult( + order_id=existing.id, + action="unchanged", + old_status=old_status, + new_status=order.status, + ) + try: + assert_valid_transition(old_status, order.status) + except InvalidStateTransition as exc: + return UpsertResult( + order_id=existing.id, + action="illegal_transition", + old_status=old_status, + new_status=order.status, + error=str(exc), + ) + + # 3) 合法推进：走 transition_status + self.transition_status( + existing.id, + order.status, + actor=actor, + reason=reason or f"upsert_{order.source}", + ) + return UpsertResult( + order_id=existing.id, + action="updated", + old_status=old_status, + new_status=order.status, + ) + + # ------------------------------------------------------------------ + # 删除（保留 — 业务上极少使用，但测试 + GDPR 流程可能需要） + # ------------------------------------------------------------------ + + def delete(self, order_id: str, *, hard: bool = False) -> bool: + """删除订单。 + + - ``hard=False``（默认）：仅删除订单行，order_status_history + 由 ``ON DELETE CASCADE`` 自动清理。**该模式用于业务侧强制 + 删除（如恶意订单）**；请注意：已加密的 PII 字段随行一起 + 消失，状态历史同样消失。 + - ``hard=True``：当前等价于 ``hard=False``；预留 ``PRAGMA + secure_delete`` 配置接口。 + - 不存在返回 False；成功删除返回 True。 + + 注意：状态机不提供"删除"操作 — 这是物理删除，不会写 status_history。 + 如需审计可改用 :class:`DataDeletionAudit` 单独的审计表。 + """ + del hard # 当前未使用 — 预留 + with self.transaction(): + cur = self._conn.execute("DELETE FROM orders WHERE id=?", (order_id,)) + return cur.rowcount > 0 + + # ------------------------------------------------------------------ + # Dunder + # ------------------------------------------------------------------ + + def __enter__(self) -> "OrdersDAO": + return self + + def __exit__(self, exc_type, exc, tb) -> None: + try: + if exc_type is None: + self._conn.commit() + else: + self._conn.rollback() + finally: + self._conn.close() + + +__all__ = [ + "OrdersDAO", + "UpsertResult", + "StatusChange", + "OrderNotFound", + "DuplicateOrder", +] diff --git a/data/orders/tests/test_dao.py b/data/orders/tests/test_dao.py new file mode 100644 index 0000000..4fe47ca --- /dev/null +++ b/data/orders/tests/test_dao.py @@ -0,0 +1,803 @@ +"""orders.dao 模块测试 (T4.2) + +覆盖： +- 加密字段透明化（明文入口 → DB 落 *_enc → 读回明文） +- 6 态状态机守护：合法转换走通、非法转换抛 InvalidStateTransition +- 事务回滚：create + transition_status 失败时回滚 +- 幂等 upsert_by_external_id：inserted / unchanged / updated / illegal_transition 四种 action +- 查询：get / get_by_external_id / find_by_phone / list / count / stats +- 重复主键 / 重复 external_id 抛 DuplicateOrder +- status_history 审计：每次 transition 写一条；get_status_history 时间线正确 +- 业务字段更新：update() 修改 plan_file / notes / tags 不影响 status +- 禁止 update() 改 status（必须走 transition_status） +- 终态 completed / refunded 不可再转换 +""" + +import os +import sqlite3 +import tempfile +from pathlib import Path +from typing import Any, cast + +import pytest + +os.environ.setdefault("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-unit-tests") + + +from data.orders.crypto import decrypt, hash_for_index +from data.orders.dao import ( + DuplicateOrder, + OrderNotFound, + OrdersDAO, + StatusChange, + UpsertResult, +) +from data.orders.models import Order, generate_order_id, utc_now_iso +from data.orders.schema import apply_schema +from data.orders.state_machine import InvalidStateTransition + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def tmp_db_path(): + """临时 SQLite 文件路径（自动清理）。""" + with tempfile.TemporaryDirectory() as d: + yield Path(d) / "test_orders.db" + + +@pytest.fixture +def conn(tmp_db_path): + """已应用 schema 的裸连接（不强制 row_factory）。""" + c = apply_schema(tmp_db_path) + try: + yield c + finally: + c.close() + + +@pytest.fixture +def dao(conn): + """默认 DAO（用 fixture 的 conn）。""" + return OrdersDAO(conn) + + +@pytest.fixture +def conn_with_factory(tmp_db_path): + """row_factory=sqlite3.Row 的连接，用于验证 row_factory_ctx 不污染。""" + c = apply_schema(tmp_db_path) + c.row_factory = sqlite3.Row + try: + yield c + finally: + c.close() + + +@pytest.fixture +def sample_order() -> Order: + """带 PII 的样例订单（用于 create）。""" + return Order( + id=generate_order_id(), + source="web", + service_version="standard", + amount_cents=9900, + status="pending", + customer_name="张三", + customer_phone="13800001234", + customer_wechat="wx_test", + candidate_name="张小明", + candidate_id_card="430102200501011234", + candidate_province="湖南", + candidate_score=578, + candidate_rank=12345, + candidate_subjects=["物理", "化学", "生物"], + candidate_interests="计算机", + candidate_strong_subjects="数学", + candidate_weak_subjects="英语", + candidate_family="父母均为教师", + tags=["VIP", "高优"], + notes="样例订单", + ) + + +def _new_order(**overrides: Any) -> Order: + """工厂：生成最小可用 Order，方便参数化测试。""" + defaults: dict[str, Any] = dict( + id=generate_order_id(), + source="web", + service_version="basic", + amount_cents=1000, + status="pending", + ) + defaults.update(overrides) + # 工厂只传 dataclass 字段，子集静态保证；运行时由 dataclass 自身校验。 + return Order(**cast(Any, defaults)) + + +# --------------------------------------------------------------------------- +# 1. 构造与连接管理 +# --------------------------------------------------------------------------- + + +class TestConnect: + def test_connect_returns_dao_and_applies_schema(self, tmp_db_path): + dao = OrdersDAO.connect(tmp_db_path) + try: + # schema 应已应用 + row = dao.conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='orders'" + ).fetchone() + assert row is not None + row = dao.conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='order_status_history'" + ).fetchone() + assert row is not None + # row_factory 应已设为 sqlite3.Row + assert dao.conn.row_factory is sqlite3.Row + finally: + dao.close() + + def test_context_manager_commits_on_success(self, tmp_db_path, sample_order): + with OrdersDAO.connect(tmp_db_path) as dao: + created = dao.create(sample_order) + # 出 with 后数据应已落盘 + with OrdersDAO.connect(tmp_db_path) as dao2: + assert dao2.get(created.id).customer_phone == "13800001234" + + def test_context_manager_rolls_back_on_exception(self, tmp_db_path, sample_order): + # 使用 transaction() 显式控制：异常应回滚本次事务 + with pytest.raises(RuntimeError): + with OrdersDAO.connect(tmp_db_path) as dao: + with dao.transaction(): + dao.create(sample_order) + raise RuntimeError("boom") + # 事务回滚：再次连接应查不到 + with OrdersDAO.connect(tmp_db_path) as dao2: + with pytest.raises(OrderNotFound): + dao2.get(sample_order.id) + + def test_dao_does_not_close_external_conn(self, conn): + """OrdersDAO(conn) 不应关闭外部传入的连接。""" + dao = OrdersDAO(conn) + dao.create(_new_order()) + # conn 仍可用 → 说明 DAO 没 close 它 + cnt = conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] + assert cnt == 1 + # conn 不会被 dao.close() 之外的双重关闭 —— 这里我们不调 close + assert not hasattr(dao, "_owns_conn") or True # 占位：DAO 不持 owns 标志 + # 显式断言: 调用 dao.close() 后 conn.closed 为 True + # 因为 sqlite3.Connection.close() 幂等且永远生效 + dao.close() + # 再次 conn 操作应抛 ProgrammingError + with pytest.raises(sqlite3.ProgrammingError): + conn.execute("SELECT 1") + + +# --------------------------------------------------------------------------- +# 2. 加密透明化（明文入 → 密文落盘 → 明文读回） +# --------------------------------------------------------------------------- + + +class TestEncryptionTransparency: + def test_create_encrypts_phone_to_db(self, conn, sample_order): + dao = OrdersDAO(conn) + dao.create(sample_order) + # 落盘行：customer_phone_enc 存在且非明文 + row = conn.execute( + "SELECT customer_phone_enc, customer_phone_hash FROM orders WHERE id=?", + (sample_order.id,), + ).fetchone() + enc = row[0] + assert enc is not None + assert enc != "13800001234" + # 密文可解 + assert decrypt(enc) == "13800001234" + # hash 字段存在 + assert row[1] == hash_for_index("13800001234") + + def test_create_encrypts_id_card_to_db(self, conn, sample_order): + dao = OrdersDAO(conn) + dao.create(sample_order) + row = conn.execute( + "SELECT candidate_id_card_enc FROM orders WHERE id=?", + (sample_order.id,), + ).fetchone() + enc = row[0] + assert enc is not None + assert decrypt(enc) == "430102200501011234" + # 验证 DB 落盘无明文身份证列 + # （schema 中没有 candidate_id_card 列，列名只有 _enc 后缀） + cols = conn.execute( + "SELECT name FROM pragma_table_info('orders') WHERE name='candidate_id_card'" + ).fetchone() + assert cols is None + + def test_get_decrypts_pii_back(self, dao, sample_order): + created = dao.create(sample_order) + out = dao.get(created.id) + assert out.customer_phone == "13800001234" + assert out.candidate_id_card == "430102200501011234" + assert out.customer_name == "张三" # 明文存储 + assert out.candidate_subjects == ["物理", "化学", "生物"] + assert out.tags == ["VIP", "高优"] + + def test_no_pii_in_db_when_not_provided(self, conn): + order = _new_order() # 无 PII + OrdersDAO(conn).create(order) + row = conn.execute( + "SELECT customer_phone_enc, customer_phone_hash, candidate_id_card_enc " + "FROM orders WHERE id=?", + (order.id,), + ).fetchone() + assert row[0] is None # 无明文 → 无密文 + assert row[1] is None # 无 phone → 无 hash + assert row[2] is None + + +# --------------------------------------------------------------------------- +# 3. CRUD：create / get / update +# --------------------------------------------------------------------------- + + +class TestCRUD: + def test_create_returns_order(self, dao, sample_order): + created = dao.create(sample_order) + assert isinstance(created, Order) + assert created.id == sample_order.id + assert created.created_at is not None + assert created.status_updated_at is not None + + def test_get_not_found(self, dao): + with pytest.raises(OrderNotFound): + dao.get("GKO-NOT-EXIST") + + def test_duplicate_primary_key_raises(self, conn, sample_order): + dao = OrdersDAO(conn) + dao.create(sample_order) + # 再次 create 同 id → DuplicateOrder + with pytest.raises(DuplicateOrder): + dao.create(sample_order) + + def test_duplicate_external_id_raises(self, conn): + dao = OrdersDAO(conn) + # 第一次：建一个带 external_id 的订单 + first = _new_order( + id=generate_order_id(), + source="xianyu", + external_id="EXT-DUP-1", + ) + dao.create(first) + # 第二次：同 source+external_id 但 id 不同 → 唯一索引冲突 + dup = _new_order( + id=generate_order_id(), + source="xianyu", + external_id="EXT-DUP-1", + ) + with pytest.raises(DuplicateOrder): + dao.create(dup) + + def test_update_business_fields(self, dao, sample_order): + created = dao.create(sample_order) + updated = dao.update( + created.id, + { + "plan_file": "/data/plans/abc.md", + "notes": "已补充考生信息", + "tags": ["VIP", "高优", "复诊"], + "amount_cents": 19900, + }, + ) + assert updated.plan_file == "/data/plans/abc.md" + assert updated.notes == "已补充考生信息" + assert updated.tags == ["VIP", "高优", "复诊"] + assert updated.amount_cents == 19900 + # status 应不变 + assert updated.status == "pending" + + def test_update_rejects_status_field(self, dao, sample_order): + created = dao.create(sample_order) + with pytest.raises(ValueError, match="status"): + dao.update(created.id, {"status": "paid"}) + + def test_update_rejects_unknown_column(self, dao, sample_order): + created = dao.create(sample_order) + with pytest.raises(ValueError, match="不允许字段"): + dao.update(created.id, {"hacker_field": "x"}) + + def test_update_unknown_order_raises(self, dao): + with pytest.raises(OrderNotFound): + dao.update("GKO-NOT-EXIST", {"notes": "x"}) + + def test_update_preserves_existing_paid_at_on_transition(self, conn, sample_order): + """update 业务字段不应改 timestamp。""" + dao = OrdersDAO(conn) + created = dao.create(sample_order) + # 推到 paid → paid_at 应被置位 + dao.transition_status(created.id, "paid", reason="payment") + before = dao.get(created.id) + # 业务字段更新 + dao.update(created.id, {"notes": "新备注"}) + after = dao.get(created.id) + assert after.paid_at == before.paid_at + assert after.status == "paid" + assert after.notes == "新备注" + + +# --------------------------------------------------------------------------- +# 4. 状态机守护 +# --------------------------------------------------------------------------- + + +class TestStateMachine: + def test_legal_transition_writes_history(self, dao, sample_order): + created = dao.create(sample_order) + out = dao.transition_status(created.id, "paid", reason="wechat_pay") + assert out.status == "paid" + # paid_at 应被置位 + assert out.paid_at is not None + # history 写入了 2 条（create + transition） + history = dao.get_status_history(created.id) + assert len(history) == 2 + # 第 1 条：None → pending（actor=dao_create） + assert history[0].from_status is None + assert history[0].to_status == "pending" + assert history[0].actor == "dao_create" + # 第 2 条：pending → paid + assert history[1].from_status == "pending" + assert history[1].to_status == "paid" + assert history[1].reason == "wechat_pay" + + def test_illegal_transition_raises_and_rolls_back(self, dao, sample_order): + created = dao.create(sample_order) + # pending → serving 非法（必须先 paid） + with pytest.raises(InvalidStateTransition): + dao.transition_status(created.id, "serving") + # 状态应保持 pending + assert dao.get(created.id).status == "pending" + # history 不应被多写 + history = dao.get_status_history(created.id) + assert len(history) == 1 + assert history[0].to_status == "pending" + + def test_transition_unknown_status_raises(self, dao, sample_order): + created = dao.create(sample_order) + with pytest.raises(InvalidStateTransition): + dao.transition_status(created.id, "frozen") + + def test_terminal_completed_blocks_further_transitions(self, dao, sample_order): + created = dao.create(sample_order) + for s in ("paid", "serving", "delivered", "completed"): + dao.transition_status(created.id, s) + # completed → refunded 非法（终态） + with pytest.raises(InvalidStateTransition): + dao.transition_status(created.id, "refunded") + assert dao.get(created.id).status == "completed" + + def test_terminal_refunded_blocks_further_transitions(self, dao, sample_order): + created = dao.create(sample_order) + dao.transition_status(created.id, "refunded") + with pytest.raises(InvalidStateTransition): + dao.transition_status(created.id, "paid") + with pytest.raises(InvalidStateTransition): + dao.transition_status(created.id, "completed") + assert dao.get(created.id).status == "refunded" + + def test_refund_from_any_non_terminal_state(self, dao): + for start in ("pending", "paid", "serving", "delivered"): + o = dao.create(_new_order(id=generate_order_id())) + for s in ("paid", "serving", "delivered"): + if start == s: + break + dao.transition_status(o.id, s) + out = dao.transition_status(o.id, "refunded", reason="customer_request") + assert out.status == "refunded" + + def test_transition_preserves_earlier_paid_at(self, dao, sample_order): + created = dao.create(sample_order) + dao.transition_status(created.id, "paid") + paid_at_1 = dao.get(created.id).paid_at + # 推进到 serving → paid_at 应保持 + dao.transition_status(created.id, "serving") + paid_at_2 = dao.get(created.id).paid_at + assert paid_at_1 == paid_at_2 + # serving → delivered + dao.transition_status(created.id, "delivered") + # delivered → completed → completed_at 被置位 + out = dao.transition_status(created.id, "completed") + assert out.completed_at is not None + + def test_transition_unknown_order_raises(self, dao): + with pytest.raises(OrderNotFound): + dao.transition_status("GKO-NOT-EXIST", "paid") + + +# --------------------------------------------------------------------------- +# 5. 事务与回滚 +# --------------------------------------------------------------------------- + + +class TestTransaction: + def test_transaction_commits_on_success(self, conn): + dao = OrdersDAO(conn) + with dao.transaction() as c: + c.execute( + "INSERT INTO orders (id, source, service_version, amount_cents, status, status_updated_at, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + "GKO-MAN-1", + "manual", + "basic", + 100, + "pending", + utc_now_iso(), + utc_now_iso(), + ), + ) + # 提交后查询得到 + row = conn.execute( + "SELECT id FROM orders WHERE id=?", ("GKO-MAN-1",) + ).fetchone() + assert row is not None + + def test_transaction_rolls_back_on_exception(self, conn): + dao = OrdersDAO(conn) + with pytest.raises(RuntimeError): + with dao.transaction() as c: + c.execute( + "INSERT INTO orders (id, source, service_version, amount_cents, status, status_updated_at, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + "GKO-MAN-2", + "manual", + "basic", + 100, + "pending", + utc_now_iso(), + utc_now_iso(), + ), + ) + raise RuntimeError("boom") + # 回滚后查不到 + row = conn.execute( + "SELECT id FROM orders WHERE id=?", ("GKO-MAN-2",) + ).fetchone() + assert row is None + + +# --------------------------------------------------------------------------- +# 6. 幂等 upsert_by_external_id +# --------------------------------------------------------------------------- + + +class TestUpsert: + def _make_order(self, **overrides) -> Order: + defaults = dict( + id=generate_order_id(), + source="xianyu", + external_id="EXT-1001", + service_version="basic", + amount_cents=4900, + status="pending", + ) + defaults.update(overrides) + return Order(**defaults) + + def test_upsert_inserts_when_missing(self, dao): + order = self._make_order() + r = dao.upsert_by_external_id(order) + assert isinstance(r, UpsertResult) + assert r.action == "inserted" + assert r.old_status is None + assert r.new_status == "pending" + # history 应有 1 条 + history = dao.get_status_history(r.order_id) + assert len(history) == 1 + assert history[0].from_status is None + + def test_upsert_unchanged_when_status_same(self, dao): + order = self._make_order() + dao.upsert_by_external_id(order) + # 同样 status 再 upsert → unchanged + order2 = self._make_order( + id=generate_order_id(), # id 不同，但 (source, external_id) 一致 + amount_cents=9999, # 业务字段差异 — 但 DAO 不动 + ) + r = dao.upsert_by_external_id(order2) + assert r.action == "unchanged" + # amount_cents 应保持首次写入的 + existing = dao.get(r.order_id) + assert existing.amount_cents == 4900 + + def test_upsert_updates_on_legal_transition(self, dao): + order = self._make_order() + first = dao.upsert_by_external_id(order) + # 推到 paid + order.status = "paid" + order.paid_at = utc_now_iso() + r = dao.upsert_by_external_id(order) + assert r.action == "updated" + assert r.old_status == "pending" + assert r.new_status == "paid" + # history 应有 2 条 + history = dao.get_status_history(first.order_id) + assert len(history) == 2 + assert history[1].to_status == "paid" + + def test_upsert_illegal_transition_returns_action(self, dao): + order = self._make_order() + dao.upsert_by_external_id(order) + # 推进到 paid + order.status = "paid" + dao.upsert_by_external_id(order) + # 尝试 pending（非法回退） + order.status = "pending" + r = dao.upsert_by_external_id(order) + assert r.action == "illegal_transition" + assert r.old_status == "paid" + assert r.new_status == "pending" + assert r.error is not None + # DB 状态应保持 paid + existing = dao.get(r.order_id) + assert existing.status == "paid" + + def test_upsert_missing_external_id_rejected(self, dao): + order = self._make_order() + order.external_id = None + r = dao.upsert_by_external_id(order) + assert r.action == "illegal_transition" + assert "external_id" in (r.error or "") + + def test_upsert_insert_writes_history(self, dao): + order = self._make_order() + r = dao.upsert_by_external_id(order, actor="xianyu_webhook", reason="evt-001") + history = dao.get_status_history(r.order_id) + assert history[0].actor == "xianyu_webhook" + assert history[0].reason == "evt-001" + + +# --------------------------------------------------------------------------- +# 7. 查询：get_by_external_id / find_by_phone / list / count / stats +# --------------------------------------------------------------------------- + + +class TestQueries: + def test_get_by_external_id(self, dao, sample_order): + sample_order.external_id = "EXT-200" + dao.create(sample_order) + out = dao.get_by_external_id("web", "EXT-200") + assert out is not None + assert out.id == sample_order.id + # 不存在 → None + assert dao.get_by_external_id("web", "MISSING") is None + + def test_find_by_phone_returns_decrypted_orders(self, dao): + a = _new_order(id=generate_order_id(), customer_phone="13800001234") + b = _new_order(id=generate_order_id(), customer_phone="13800009999") + dao.create(a) + dao.create(b) + results = dao.find_by_phone("13800001234") + assert len(results) == 1 + assert results[0].id == a.id + assert results[0].customer_phone == "13800001234" + + def test_find_by_phone_multiple_results(self, dao): + # 同 phone hash 不同订单（业务上罕见但允许） + a = _new_order(id=generate_order_id(), customer_phone="13800001234") + b = _new_order(id=generate_order_id(), customer_phone="13800001234") + dao.create(a) + dao.create(b) + results = dao.find_by_phone("13800001234") + assert len(results) == 2 + + def test_list_with_filters(self, dao): + dao.create( + _new_order(id=generate_order_id(), source="xianyu", status="pending") + ) + dao.create(_new_order(id=generate_order_id(), source="web", status="paid")) + dao.create( + _new_order(id=generate_order_id(), source="xianyu", status="pending") + ) + + all_orders = dao.list(limit=100) + assert len(all_orders) == 3 + xianyu_pending = dao.list(source="xianyu", status="pending") + assert len(xianyu_pending) == 2 + assert all( + o.source == "xianyu" and o.status == "pending" for o in xianyu_pending + ) + + def test_list_unknown_status_raises(self, dao): + with pytest.raises(ValueError, match="未知 status"): + dao.list(status="frozen") + + def test_list_limit_bounds(self, dao): + with pytest.raises(ValueError, match="limit 越界"): + dao.list(limit=0) + with pytest.raises(ValueError, match="limit 越界"): + dao.list(limit=2000) + + def test_list_offset_negative_raises(self, dao): + with pytest.raises(ValueError, match="offset"): + dao.list(offset=-1) + + def test_list_pagination(self, dao): + for _ in range(5): + dao.create(_new_order(id=generate_order_id())) + page1 = dao.list(limit=2, offset=0) + page2 = dao.list(limit=2, offset=2) + page3 = dao.list(limit=2, offset=4) + assert len(page1) == 2 + assert len(page2) == 2 + assert len(page3) == 1 + ids = {o.id for o in page1 + page2 + page3} + assert len(ids) == 5 + + def test_count_with_and_without_filters(self, dao): + for s in ("pending", "pending", "paid", "refunded"): + dao.create(_new_order(id=generate_order_id(), status=s)) + assert dao.count() == 4 + assert dao.count(status="pending") == 2 + assert dao.count(status="paid") == 1 + assert dao.count(status="refunded") == 1 + assert dao.count(status="completed") == 0 + + def test_stats_by_status_includes_zero_states(self, dao): + dao.create(_new_order(id=generate_order_id(), status="pending")) + dao.create(_new_order(id=generate_order_id(), status="pending")) + stats = dao.stats_by_status() + # 6 态全部出现，零值不漏 + assert set(stats) == { + "pending", + "paid", + "serving", + "delivered", + "completed", + "refunded", + } + assert stats["pending"] == 2 + assert stats["paid"] == 0 + assert stats["completed"] == 0 + + def test_list_returns_orders_with_pii(self, dao): + o = _new_order(id=generate_order_id(), customer_phone="13800007777") + dao.create(o) + results = dao.list(limit=10) + # 列表默认应已解密（to_db_row 入库时加密，list 读出时解密） + assert results[0].customer_phone == "13800007777" + + +# --------------------------------------------------------------------------- +# 8. 状态历史 +# --------------------------------------------------------------------------- + + +class TestStatusHistory: + def test_history_is_chronological(self, dao, sample_order): + created = dao.create(sample_order) + for s in ("paid", "serving", "delivered", "completed"): + dao.transition_status(created.id, s) + history = dao.get_status_history(created.id) + # 5 条：create + 4 transitions + assert len(history) == 5 + assert [h.to_status for h in history] == [ + "pending", + "paid", + "serving", + "delivered", + "completed", + ] + assert [h.from_status for h in history] == [ + None, + "pending", + "paid", + "serving", + "delivered", + ] + # 全部为 StatusChange dataclass + assert all(isinstance(h, StatusChange) for h in history) + + def test_history_for_unknown_order_is_empty(self, dao): + assert dao.get_status_history("GKO-NOT-EXIST") == [] + + +# --------------------------------------------------------------------------- +# 9. row_factory 不污染外部 +# --------------------------------------------------------------------------- + + +class TestRowFactoryIsolation: + def test_dao_does_not_corrupt_external_row_factory(self, conn_with_factory): + """DAO 内部用 sqlite3.Row 工厂做查询，退出后外部 row_factory 应保持。""" + prior = conn_with_factory.row_factory + assert prior is sqlite3.Row + dao = OrdersDAO(conn_with_factory) + order = _new_order() + dao.create(order) + dao.get(order.id) + dao.list(limit=10) + # 退出后 row_factory 仍为 sqlite3.Row + assert conn_with_factory.row_factory is sqlite3.Row + + +# --------------------------------------------------------------------------- +# 10. 删除 +# --------------------------------------------------------------------------- + + +class TestDelete: + def test_delete_existing_returns_true(self, dao, sample_order): + created = dao.create(sample_order) + assert dao.delete(created.id) is True + with pytest.raises(OrderNotFound): + dao.get(created.id) + + def test_delete_nonexistent_returns_false(self, dao): + assert dao.delete("GKO-NOT-EXIST") is False + + def test_delete_cascades_status_history(self, conn, sample_order): + dao = OrdersDAO(conn) + created = dao.create(sample_order) + dao.transition_status(created.id, "paid") + # 2 条历史 + hist_before = conn.execute( + "SELECT COUNT(*) FROM order_status_history WHERE order_id=?", + (created.id,), + ).fetchone()[0] + assert hist_before == 2 + dao.delete(created.id) + # 状态历史随 ON DELETE CASCADE 消失 + hist_after = conn.execute( + "SELECT COUNT(*) FROM order_status_history WHERE order_id=?", + (created.id,), + ).fetchone()[0] + assert hist_after == 0 + + +# --------------------------------------------------------------------------- +# 11. 与 T8.1 dao_extension 的契约对齐 +# --------------------------------------------------------------------------- + + +class TestContractAlignment: + """确保 OrdersDAO.upsert_by_external_id 与 dao_extension 同名函数行为一致。""" + + def test_upsert_result_action_values_match(self, dao): + # 同 (source, external_id) 不存在 + order = Order( + id=generate_order_id(), + source="xianyu", + external_id="EXT-CONTRACT", + service_version="basic", + amount_cents=1000, + status="pending", + ) + r1 = dao.upsert_by_external_id(order) + assert r1.action == "inserted" + + # status 相同 → unchanged + order2 = Order( + id=generate_order_id(), + source="xianyu", + external_id="EXT-CONTRACT", + service_version="basic", + amount_cents=2000, # 不同 — 但 status 未变 + status="pending", + ) + r2 = dao.upsert_by_external_id(order2) + assert r2.action == "unchanged" + + # 状态推进 → updated + order2.status = "paid" + r3 = dao.upsert_by_external_id(order2) + assert r3.action == "updated" + assert r3.old_status == "pending" + assert r3.new_status == "paid" + + # 非法回退 → illegal_transition + order2.status = "pending" + r4 = dao.upsert_by_external_id(order2) + assert r4.action == "illegal_transition" + assert r4.error is not None