ShowDocs
搜索文章Ctrl K

外观

← 返回PostgreSQL列表
应用实践

PostgreSQL 完整项目：参数化库存预留 CLI · 完整测试、反例与执行计划至综合练习 ​

返回分篇目录

完整测试、反例与执行计划 ​

py
"""Real integration tests; unavailable database is an error, never a skip."""
from concurrent.futures import ThreadPoolExecutor
import threading
import unittest

import psycopg
from inventory import connect, initialize, reserve


class InventoryTest(unittest.TestCase):
    def setUp(self):
        self.conn = connect()
        initialize(self.conn)
        self.conn.execute("TRUNCATE reservations,products")
        self.conn.execute("INSERT INTO products(sku,name,stock,price) VALUES ('book','中文📚',5,12.30),('pen','笔',0,1)")

    def tearDown(self):
        self.conn.close()

    def test_parameterization_encoding_and_replay(self):
        key = "中文'; DROP TABLE products;--"
        self.assertFalse(reserve(self.conn, key, "book", 2)["replayed"])
        self.assertTrue(reserve(self.conn, key, "book", 2)["replayed"])
        self.assertEqual(self.conn.execute("SELECT name,stock FROM products WHERE sku='book'").fetchone(), ("中文📚", 3))
        with self.assertRaises(ValueError):
            reserve(self.conn, key, "book", 1)
        self.assertEqual(self.conn.execute("SELECT count(*) FROM products").fetchone()[0], 2)

    def test_failures_rollback_both_tables(self):
        for key, sku, qty in [("empty", "pen", 1), ("zero", "book", 0), ("minus", "book", -1)]:
            with self.assertRaises(ValueError):
                reserve(self.conn, key, sku, qty)
        with self.assertRaises(psycopg.errors.ForeignKeyViolation):
            reserve(self.conn, "missing", "none", 1)
        self.assertEqual(self.conn.execute("SELECT count(*) FROM reservations").fetchone()[0], 0)
        self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 5)

    def test_caller_transaction_survives_failure(self):
        with self.conn.transaction():
            self.conn.execute("UPDATE products SET note='caller' WHERE sku='book'")
            with self.assertRaises(ValueError):
                reserve(self.conn, "fail", "pen", 1)
            self.assertEqual(self.conn.execute("SELECT note FROM products WHERE sku='book'").fetchone()[0], "caller")
        self.assertEqual(self.conn.execute("SELECT note FROM products WHERE sku='book'").fetchone()[0], "caller")

    def test_outer_rollback_undoes_success(self):
        with self.assertRaises(RuntimeError):
            with self.conn.transaction():
                reserve(self.conn, "outer", "book", 1)
                raise RuntimeError("caller rollback")
        self.assertEqual(self.conn.execute("SELECT count(*) FROM reservations").fetchone()[0], 0)
        self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 5)

    def test_concurrent_stock_and_same_key(self):
        barrier = threading.Barrier(8)
        def attempt(index):
            with connect() as conn:
                barrier.wait(timeout=10)
                try:
                    reserve(conn, f"concurrent-{index}", "book", 1)
                    return True
                except ValueError:
                    return False
        with ThreadPoolExecutor(max_workers=8) as pool:
            self.assertEqual(sum(pool.map(attempt, range(8))), 5)
        self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 0)
        self.conn.execute("TRUNCATE reservations")
        self.conn.execute("UPDATE products SET stock=5 WHERE sku='book'")
        replay_barrier = threading.Barrier(8)
        def replay(_):
            with connect() as conn:
                replay_barrier.wait(timeout=10)
                return reserve(conn, "same", "book", 1)["replayed"]
        with ThreadPoolExecutor(max_workers=8) as pool:
            self.assertEqual(sum(pool.map(replay, range(8))), 7)
        self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 4)

    def test_constraints_and_null(self):
        for query in ["UPDATE products SET stock=-1", "UPDATE products SET name=NULL"]:
            with self.assertRaises(psycopg.IntegrityError):
                self.conn.execute(query)
        self.assertEqual(self.conn.execute("SELECT NULL=NULL, NULL IS NULL, count(note) FROM products").fetchone(), (None, True, 0))

    def test_isolation_and_lock_timeout(self):
        with connect() as other:
            with self.conn.transaction():
                self.conn.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
                self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 5)
                other.execute("UPDATE products SET stock=4 WHERE sku='book'")
                self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 5)
            with self.conn.transaction():
                self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 4)
                other.execute("UPDATE products SET stock=3 WHERE sku='book'")
                self.assertEqual(self.conn.execute("SELECT stock FROM products WHERE sku='book'").fetchone()[0], 3)
            with self.conn.transaction():
                self.conn.execute("SELECT * FROM products WHERE sku='book' FOR UPDATE")
                other.execute("SET lock_timeout='150ms'")
                with self.assertRaises(psycopg.errors.LockNotAvailable):
                    other.execute("UPDATE products SET stock=2 WHERE sku='book'")


if __name__ == "__main__":
    unittest.main(verbosity=2)

PowerShell执行 examples/postgres/.venv/Scripts/python.exe -X utf8 -m unittest discover -s examples/postgres -p test_inventory.py -v；Bash换成.venv/bin/python。需要真实服务，服务不可达是错误而非skip。7项测试分别覆盖参数化与中文/幂等、失败回滚、调用者事务保留、外层回滚、两类并发竞争、CHECK/NOT NULL/NULL、隔离快照与锁超时。多线程使用各自连接，不跨线程共享事务。

回滚RED验证器仅在内存生成“移除事务上下文”的错误实现，不修改正式源码或测试。使用上面安装了 Psycopg 的虚拟环境解释器运行 examples/postgres/.venv/Scripts/python.exe -X utf8 examples/postgres/red_check.py（Bash 用 examples/postgres/.venv/bin/python examples/postgres/red_check.py） 预期 退出1 并指出 reservations数量1而非0；正常测试随后应7项通过退出0。这是行为反例，不是把未连接数据库的报错当RED。文件全文如下：

py
"""Mutation check: missing transaction must fail an existing atomicity test.

Runs a throwaway module in memory; does not alter the real implementation or tests.
Expected exit: 1. Run the normal tests afterwards for GREEN evidence.
"""
import contextlib
from pathlib import Path
import sys
import types
import unittest

path = Path(__file__).with_name("inventory.py")
mutant = types.ModuleType("inventory")
mutant.__file__ = str(path)
mutant.contextlib = contextlib
source = path.read_text(encoding="utf-8").replace("with conn.transaction():", "with contextlib.nullcontext():")
exec(compile(source, str(path), "exec"), mutant.__dict__)
sys.modules["inventory"] = mutant
from test_inventory import InventoryTest

suite = unittest.TestSuite([InventoryTest("test_failures_rollback_both_tables")])
result = unittest.TextTestRunner(verbosity=2).run(suite)
raise SystemExit(0 if result.wasSuccessful() else 1)

查询与计划验证器全文如下。历史 Windows 及 Linux 运行记录获得 Index Only Scan，但具体时延、估算和buffers依环境变化，正文不硬编码性能门槛。当前 verify 检查实际返回行及顺序，完整输出计划供观察，不把优化器必须选择某类索引节点设为成功条件。请勿使用 python -O 跳过脚本中的 assert。

py
"""Run only against postgres_lab. Evidence is JSON; every failing assertion exits nonzero."""
import json
from inventory import connect, initialize, reserve


with connect() as conn:
    initialize(conn)
    conn.execute("TRUNCATE reservations,products")
    conn.execute("INSERT INTO products(sku,name,stock,price) VALUES ('book','中文📚',5,12.30),('pen','笔',0,1)")
    reserve(conn, "seed", "book", 2)
    # A temporary relation isolates the EXPLAIN workload from application data.
    conn.execute("CREATE TEMP TABLE plan_lab AS SELECT n AS id,n%100 AS bucket FROM generate_series(1,50000) n")
    conn.execute("CREATE INDEX plan_lab_bucket_id ON plan_lab(bucket,id)")
    conn.execute("ANALYZE plan_lab")
    plan = conn.execute("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) SELECT * FROM plan_lab WHERE bucket=42 ORDER BY id LIMIT 10").fetchone()[0]
    # Planner choices vary; verify the query result, report the plan for study.
    rows = conn.execute("SELECT * FROM plan_lab WHERE bucket=42 ORDER BY id LIMIT 10").fetchall()
    assert rows == [(42 + 100 * n, 42) for n in range(10)], rows
    joined = conn.execute("SELECT p.sku,count(r.request_id),coalesce(sum(r.quantity),0) FROM products p LEFT JOIN reservations r USING(sku) GROUP BY p.sku ORDER BY p.sku").fetchall()
    assert joined == [("book", 1, 2), ("pen", 0, 0)], joined
    page = conn.execute("SELECT sku FROM products WHERE sku>%s ORDER BY sku LIMIT 10", ("book",)).fetchall()
    assert page == [("pen",)], page
    print(json.dumps({"server": conn.execute("SELECT version()").fetchone()[0],
                      "encoding": conn.execute("SHOW server_encoding").fetchone()[0],
                      "join": joined, "page": page, "plan": plan}, ensure_ascii=False, default=str, indent=2))

verify预期输出 PostgreSQL18.6、UTF8、两行JOIN结果、下一页pen及JSON计划。临时表只存在当前连接，不进入业务备份。测试结束再运行verify得到一致的演示数据，然后执行备份恢复。

综合练习 ​

给CLI添加“查询某request_id结果”的只读命令：参数仍绑定；不存在返回明确错误；读取中文及带引号键保持原值；重复查询不能修改库存。先写测试验证不存在和合法情况，再实现查询。提示：复用 argparse 子命令分支和 fetchone()，把“没查到”与SQL异常分开；先记录库存，再查两次并比较，不额外创建HTTP服务。参考实现思路见复习，现有完整项目不依赖这项扩展才能运行。

继续阅读