Archived
snowy-yang
0f2e72db44
- core/repo.py：经由 core.db(aiosqlite) 读写 auth_system 的**同一份** site.db （settings/announcements/users 只读/ledger 只读），字段口径对齐 auth_system； 读并发、写串行、WAL 读写同时进行 - web/routers/site.py：/api/v1/site/announcements、/api/v1/site/stats（异步多表并发读） - web/app.py：lifespan 关闭时释放 aiohttp 连接池（含按 loop 隔离的 session） 写路径说明：迁移期写操作仍归 auth_system 的 _DB_LOCK 事务口径，避免两套写 路径互相踩踏；本层先只做读。 验证：pytest 22/22（新增 6 项含「异步读 + 同步写者并发 WAL」）； /api/v1/site/* 实测读到共享库（3 用户/1 管理员）；ruff/pyright 干净。
97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""异步仓储层测试：读写共享库、并发读、与 auth_system schema 兼容。"""
|
|
import asyncio
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
import core.db as db
|
|
import core.repo as repo
|
|
|
|
|
|
@pytest.fixture()
|
|
def site_db(tmp_path, monkeypatch):
|
|
"""建一个与 auth_system 口径一致的迷你库，指给 core.db。"""
|
|
path = tmp_path / 'site.db'
|
|
conn = sqlite3.connect(str(path))
|
|
conn.executescript("""
|
|
CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT UNIQUE, role TEXT DEFAULT 'user');
|
|
CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT DEFAULT '');
|
|
CREATE TABLE announcements (id INTEGER PRIMARY KEY, title TEXT DEFAULT '', content TEXT DEFAULT '',
|
|
popup_mode TEXT DEFAULT 'off', updated_at TEXT DEFAULT '');
|
|
CREATE TABLE billing_ledger (task_id TEXT PRIMARY KEY, user_id INTEGER, status TEXT DEFAULT 'open');
|
|
INSERT INTO users (username, role) VALUES ('alice','user'), ('bob','user'), ('qfbsh','admin');
|
|
INSERT INTO settings (key, value) VALUES ('server_nickname','云村测试');
|
|
INSERT INTO announcements (id, title, content) VALUES (1, '公告', '正文');
|
|
INSERT INTO billing_ledger (task_id, user_id, status) VALUES ('t1', 1, 'open');
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
monkeypatch.setattr(db, 'DATA_DIR', tmp_path)
|
|
monkeypatch.setattr(db, 'DB_PATH', path)
|
|
return path
|
|
|
|
|
|
async def test_get_setting(site_db):
|
|
assert await repo.get_setting('server_nickname') == '云村测试'
|
|
assert await repo.get_setting('missing', 'fallback') == 'fallback'
|
|
|
|
|
|
async def test_set_setting_upsert(site_db):
|
|
await repo.set_setting('a', '1')
|
|
assert await repo.get_setting('a') == '1'
|
|
await repo.set_setting('a', '2')
|
|
assert await repo.get_setting('a') == '2'
|
|
|
|
|
|
async def test_counts(site_db):
|
|
assert await repo.count_users() == 3
|
|
assert await repo.count_users('admin') == 1
|
|
assert await repo.ledger_open_count() == 1
|
|
|
|
|
|
async def test_list_announcements(site_db):
|
|
rows = await repo.list_announcements()
|
|
assert rows and rows[0]['title'] == '公告'
|
|
|
|
|
|
async def test_concurrent_reads(site_db):
|
|
"""并发读同一库不互相阻塞、结果一致。"""
|
|
results = await asyncio.gather(*[repo.count_users() for _ in range(20)])
|
|
assert results == [3] * 20
|
|
|
|
|
|
async def test_read_write_concurrent_with_sync_writer(site_db):
|
|
"""异步读与『另一个同步写者』（模拟遗留栈写库）可同时进行（WAL）。"""
|
|
stop = asyncio.Event()
|
|
errors = []
|
|
|
|
def sync_writer():
|
|
# 模拟 auth_system 的同步写：独立连接持续写 settings
|
|
conn = sqlite3.connect(str(site_db), timeout=15)
|
|
conn.execute('PRAGMA journal_mode=WAL')
|
|
try:
|
|
for i in range(80):
|
|
conn.execute("INSERT INTO settings (key, value) VALUES (?, 'x') "
|
|
"ON CONFLICT(key) DO UPDATE SET value='x'", (f'k{i}',))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
stop.set()
|
|
|
|
async def reader():
|
|
while not stop.is_set():
|
|
try:
|
|
await repo.get_setting('server_nickname')
|
|
except Exception as e:
|
|
errors.append(e)
|
|
await asyncio.sleep(0)
|
|
|
|
# sync_writer 放线程池，reader 留在事件循环
|
|
loop = asyncio.get_running_loop()
|
|
writer_task = loop.run_in_executor(None, sync_writer)
|
|
await asyncio.gather(writer_task, *[reader() for _ in range(3)])
|
|
|
|
assert not errors, f'读期间异常: {errors[:3]}'
|
|
assert await repo.get_setting('k79') == 'x'
|