55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
"""引擎工厂（utils/db.py）单测（B7 评审 P1-1）：单例复用 + dispose 真实释放。
|
|
|
|
create_engine 与缓存字典均 monkeypatch 替换，不触网；验证 dispose_engines
|
|
对每个缓存 Engine 显式调用 dispose() 并清空缓存（不再是仅 cache_clear）。
|
|
"""
|
|
|
|
from app.utils import db
|
|
|
|
|
|
class SpyEngine:
|
|
def __init__(self):
|
|
self.dispose_calls = 0
|
|
|
|
def dispose(self, close: bool = True) -> None:
|
|
self.dispose_calls += 1
|
|
|
|
|
|
def _patch(monkeypatch):
|
|
created_urls = []
|
|
|
|
def fake_create_engine(url, **kwargs):
|
|
created_urls.append(url)
|
|
return SpyEngine()
|
|
|
|
monkeypatch.setattr(db, "create_engine", fake_create_engine)
|
|
monkeypatch.setattr(db, "_engines", {})
|
|
return created_urls
|
|
|
|
|
|
def test_get_engine_caches_one_engine_per_database(monkeypatch):
|
|
created = _patch(monkeypatch)
|
|
e1 = db.get_engine("db_a")
|
|
e2 = db.get_engine("db_a")
|
|
e3 = db.get_engine("db_b")
|
|
assert e1 is e2
|
|
assert e1 is not e3
|
|
assert len(created) == 2
|
|
|
|
|
|
def test_dispose_engines_calls_dispose_and_clears_cache(monkeypatch):
|
|
_patch(monkeypatch)
|
|
e = db.get_engine("db_a")
|
|
db.dispose_engines()
|
|
assert e.dispose_calls == 1
|
|
assert db._engines == {}
|
|
# 缓存已清：重建新实例而非复用已 dispose 的旧实例
|
|
fresh = db.get_engine("db_a")
|
|
assert fresh is not e
|
|
assert fresh.dispose_calls == 0
|
|
|
|
|
|
def test_dispose_engines_on_empty_cache_is_noop(monkeypatch):
|
|
_patch(monkeypatch)
|
|
db.dispose_engines() # 不抛异常即可
|