test_s2_api.py
2.75 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
"""Test S2: Conversations & Messages API — unit tests for local_db CRUD functions"""
import json
import os
import sys
import time
import sqlite3
# Setup test DB (absolute path to avoid path issues)
TEST_DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_s2.db")
if os.path.exists(TEST_DB):
try:
os.remove(TEST_DB)
except:
pass
import config
config.LOCAL_DB = TEST_DB
import local_db
local_db.init_db()
def test_tables_exist():
conn = local_db.get_conn()
tables = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name IN ('conversations','messages')"
).fetchall()
conn.close()
names = [t[0] for t in tables]
assert 'conversations' in names, "conversations table missing"
assert 'messages' in names, "messages table missing"
def test_conv_create_and_list():
cid = local_db.conv_create("测试会话")
assert cid > 0
convs = local_db.conv_list()
assert len(convs) > 0
assert convs[0]['id'] == cid
assert convs[0]['title'] == "测试会话"
def test_msg_add_and_list():
cid = local_db.conv_create("消息测试")
mid = local_db.msg_add(cid, "user", "你好")
assert mid > 0
mid2 = local_db.msg_add(cid, "assistant", "你好！")
assert mid2 > 0
msgs = local_db.msg_list(cid)
assert len(msgs) == 2
assert msgs[0]['role'] == 'user'
assert msgs[1]['role'] == 'assistant'
def test_conv_delete_cascade():
cid = local_db.conv_create("删除测试")
local_db.msg_add(cid, "user", "测试")
local_db.conv_delete(cid)
convs = local_db.conv_list()
assert cid not in [c['id'] for c in convs]
msgs = local_db.msg_list(cid)
assert len(msgs) == 0
def test_conv_update_time():
cid = local_db.conv_create("时间测试")
conv1 = [c for c in local_db.conv_list() if c['id'] == cid][0]
time.sleep(1.1)
local_db.msg_add(cid, "user", "新消息")
conv2 = [c for c in local_db.conv_list() if c['id'] == cid][0]
assert conv2['updated_at'] > conv1['updated_at']
if __name__ == "__main__":
results = []
for name, fn in [
("tables_exist", test_tables_exist),
("conv_create_and_list", test_conv_create_and_list),
("msg_add_and_list", test_msg_add_and_list),
("conv_delete_cascade", test_conv_delete_cascade),
("conv_update_time", test_conv_update_time),
]:
try:
fn()
results.append((name, "PASS"))
except Exception as e:
results.append((name, f"FAIL: {e}"))
print("\n=== S2 UNIT TESTS ===")
for name, status in results:
print(f" {name}: {status}")
try:
os.remove(TEST_DB)
except:
pass
failed = [r for r in results if r[1].startswith("FAIL")]
if failed:
sys.exit(1)