Skip to content

Commit a87ae03

Browse files
committed
Add comprehensive test suite for timeline features and privacy guard
- Introduced conftest.py for test setup and path configuration. - Implemented CLI tests for timeline JSON output and environment variable handling. - Added privacy guard tests to scan for sensitive information in tracked files. - Created SQLite read client tests to ensure retry logic and error handling. - Developed timeline pretty formatter tests to validate output formatting. - Established timeline service tests to verify chronological event ordering and tag merging.
1 parent 6d8561b commit a87ae03

7 files changed

Lines changed: 1319 additions & 0 deletions

File tree

docs/schema.md

Lines changed: 623 additions & 0 deletions
Large diffs are not rendered by default.

tests/conftest.py

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
"""Test bootstrap."""
2+
3+
from __future__ import annotations
4+
5+
import sys
6+
from pathlib import Path
7+
8+
ROOT = Path(__file__).resolve().parents[1]
9+
SRC = ROOT / "src"
10+
if str(SRC) not in sys.path:
11+
sys.path.insert(0, str(SRC))

tests/test_cli.py

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
"""CLI tests."""
2+
3+
from __future__ import annotations
4+
5+
import json
6+
from datetime import date, datetime
7+
from zoneinfo import ZoneInfo
8+
9+
import rond_api.cli as cli
10+
from rond_api.cli import _resolve_tree_mode, main
11+
from rond_api.domain.timeline_types import TimelineResult, VisitEvent
12+
13+
14+
def test_cli_timeline_json_output(capsys, monkeypatch) -> None:
15+
tz = ZoneInfo("UTC")
16+
timeline = TimelineResult(
17+
query_date=date(2026, 1, 29),
18+
timezone="中国/上海",
19+
events=[
20+
VisitEvent(
21+
visit_id=1,
22+
location_name="示例地点A",
23+
category_name="示例分类",
24+
location_type=0,
25+
poi_category=None,
26+
tags=["测试标签"],
27+
arrival_at=datetime(2026, 1, 29, 9, 0, tzinfo=tz),
28+
departure_at=datetime(2026, 1, 29, 10, 0, tzinfo=tz),
29+
is_cross_day=False,
30+
)
31+
],
32+
)
33+
34+
monkeypatch.setattr(cli, "get_timeline", lambda **_: timeline)
35+
36+
exit_code = main(
37+
[
38+
"timeline",
39+
"--date",
40+
"2026-01-29",
41+
"--output",
42+
"json",
43+
]
44+
)
45+
46+
assert exit_code == 0
47+
output = capsys.readouterr().out
48+
payload = json.loads(output)
49+
assert payload["query_date"] == "2026-01-29"
50+
assert isinstance(payload["events"], list)
51+
52+
53+
def test_resolve_tree_mode_from_env(monkeypatch) -> None:
54+
monkeypatch.delenv("tree", raising=False)
55+
monkeypatch.delenv("TIMELINE_TREE", raising=False)
56+
monkeypatch.delenv("TREE", raising=False)
57+
assert _resolve_tree_mode(None) is False
58+
59+
monkeypatch.setenv("tree", "on")
60+
assert _resolve_tree_mode(None) is True
61+
62+
63+
def test_resolve_tree_mode_cli_overrides_env(monkeypatch) -> None:
64+
monkeypatch.setenv("tree", "on")
65+
assert _resolve_tree_mode(False) is False

tests/test_privacy_guard.py

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
"""Repository privacy guard tests."""
2+
3+
from __future__ import annotations
4+
5+
import re
6+
import subprocess
7+
from pathlib import Path
8+
9+
ROOT = Path(__file__).resolve().parents[1]
10+
11+
# 仅扫描版本库已跟踪文件,避免本地未提交临时文件干扰检查。
12+
SKIP_SUFFIXES = {
13+
".png",
14+
".jpg",
15+
".jpeg",
16+
".gif",
17+
".webp",
18+
".ico",
19+
".pdf",
20+
".sqlite",
21+
".sqlite3",
22+
".db",
23+
}
24+
SKIP_FILES = {"uv.lock"}
25+
26+
SENSITIVE_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
27+
(
28+
"email",
29+
re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"),
30+
),
31+
(
32+
"cn_mobile",
33+
re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"),
34+
),
35+
(
36+
"cn_id_card",
37+
re.compile(r"(?<!\d)\d{17}[\dXx](?!\d)"),
38+
),
39+
(
40+
"private_key_block",
41+
re.compile(r"BEGIN (?:RSA|OPENSSH|EC) PRIVATE KEY"),
42+
),
43+
(
44+
"aws_access_key",
45+
re.compile(r"AKIA[0-9A-Z]{16}"),
46+
),
47+
(
48+
"google_api_key",
49+
re.compile(r"AIza[0-9A-Za-z\-_]{35}"),
50+
),
51+
(
52+
"precise_coordinate_pair",
53+
re.compile(r"(?<!\d)-?\d{1,3}\.\d{4,}\s*,\s*-?\d{1,3}\.\d{4,}(?!\d)"),
54+
),
55+
(
56+
"city_level_timezone",
57+
re.compile(r"\b(?:Asia|Europe|America|Australia|Africa)/[A-Za-z_]+\b"),
58+
),
59+
(
60+
"street_level_address",
61+
re.compile(r"[\u4e00-\u9fffA-Za-z0-9]{0,24}(?:路|街|道|巷|弄)\d{1,5}号"),
62+
),
63+
)
64+
65+
SENSITIVE_CONTEXT_PATTERN = re.compile(
66+
r"(location_name|from_location_name|to_location_name|raw_name|raw_thoroughfare|"
67+
r"tags|note|remark|address|locality|sublocality)",
68+
re.IGNORECASE,
69+
)
70+
QUOTED_LITERAL_PATTERN = re.compile(r"""["']([^"'\n]{2,40})["']""")
71+
HAS_CJK_PATTERN = re.compile(r"[\u4e00-\u9fff]")
72+
COMMON_CN_SURNAMES = (
73+
"赵钱孙李周吴郑王冯陈褚卫蒋沈韩杨朱秦尤许何吕施张孔曹严华金魏陶姜戚谢"
74+
"邹喻柏水窦章云苏潘葛奚范彭郎鲁韦昌马苗凤花方俞任袁柳酆鲍史唐费廉岑薛"
75+
"雷贺倪汤殷罗毕郝邬安常乐于时傅皮卞齐康伍余元卜顾孟平黄和穆萧尹姚邵湛"
76+
"汪祁毛禹狄米贝明臧计伏成戴谈宋茅庞熊纪舒屈项祝董梁杜阮蓝闵席季麻强贾"
77+
"路娄危江童颜郭梅盛林刁钟徐邱骆高夏蔡田樊胡凌霍虞万支柯昝管卢莫经房裘"
78+
"缪干解应宗丁宣贲邓郁单杭洪包诸左石崔吉钮龚程嵇邢滑裴陆荣翁荀羊於惠甄"
79+
"曲家封芮羿储靳汲邴糜松井段富巫乌焦巴弓牧隗山谷车侯宓蓬全郗班仰秋仲伊"
80+
"宫宁仇栾暴甘厉戎祖武符刘景詹束龙叶司韶郜黎蓟薄印宿白怀蒲台从鄂索咸籍"
81+
"赖卓蔺屠蒙池乔阴胥能苍双闻莘党翟谭贡劳逄姬申扶堵冉宰郦雍却璩桑桂濮牛"
82+
"寿通边扈燕冀浦尚农温别庄晏柴瞿阎充慕连茹习宦艾鱼容向古易慎戈廖庾终暨"
83+
"居衡步都耿满弘国文寇广禄阙东欧殳沃利蔚越夔隆师巩厍聂晁勾敖融冷辛阚那"
84+
"简饶空曾毋沙乜养鞠须丰巢关蒯相查后荆红游竺权逯盖益桓公"
85+
)
86+
PERSON_NAME_PATTERN = re.compile(rf"^[{COMMON_CN_SURNAMES}][\u4e00-\u9fff]{{1,2}}$")
87+
LOCATION_SUFFIX_PATTERN = re.compile(
88+
r"[\u4e00-\u9fff]{2,24}(?:省|市|区|县|镇|乡|村|路|街|道|巷|弄|号|"
89+
r"花园|小区|社区|广场|大厦|中心|公寓|机场|车站|地铁站|高铁站)$"
90+
)
91+
SAFE_LITERAL_HINTS = (
92+
"示例",
93+
"未知",
94+
"测试",
95+
"中国",
96+
"中文",
97+
"某某",
98+
)
99+
100+
101+
def test_repository_contains_no_sensitive_literals() -> None:
102+
findings: list[str] = []
103+
for file_path in _iter_tracked_text_files():
104+
text = _read_text(file_path)
105+
if text is None:
106+
continue
107+
108+
for label, pattern in SENSITIVE_PATTERNS:
109+
for match in pattern.finditer(text):
110+
findings.append(
111+
f"{_rel(file_path)} | {label} | {match.group(0)[:32]}"
112+
)
113+
114+
findings.extend(_scan_suspect_chinese_literals(file_path, text))
115+
116+
assert not findings, "Privacy guard failed:\n" + "\n".join(sorted(findings))
117+
118+
119+
def _iter_tracked_text_files() -> list[Path]:
120+
result = subprocess.run(
121+
["git", "ls-files"],
122+
cwd=ROOT,
123+
check=True,
124+
capture_output=True,
125+
text=True,
126+
)
127+
128+
files: list[Path] = []
129+
for line in result.stdout.splitlines():
130+
rel_path = line.strip()
131+
if not rel_path:
132+
continue
133+
if rel_path in SKIP_FILES:
134+
continue
135+
path = ROOT / rel_path
136+
if path.suffix.lower() in SKIP_SUFFIXES:
137+
continue
138+
files.append(path)
139+
return files
140+
141+
142+
def _read_text(path: Path) -> str | None:
143+
try:
144+
return path.read_text(encoding="utf-8")
145+
except UnicodeDecodeError:
146+
return None
147+
148+
149+
def _rel(path: Path) -> str:
150+
return str(path.relative_to(ROOT))
151+
152+
153+
def _scan_suspect_chinese_literals(file_path: Path, text: str) -> list[str]:
154+
findings: list[str] = []
155+
for line_no, line in enumerate(text.splitlines(), start=1):
156+
if not SENSITIVE_CONTEXT_PATTERN.search(line):
157+
continue
158+
159+
for match in QUOTED_LITERAL_PATTERN.finditer(line):
160+
literal = match.group(1).strip()
161+
if not literal:
162+
continue
163+
if not HAS_CJK_PATTERN.search(literal):
164+
continue
165+
if any(hint in literal for hint in SAFE_LITERAL_HINTS):
166+
continue
167+
if _is_suspect_cn_literal(literal):
168+
findings.append(
169+
f"{_rel(file_path)}:{line_no} | suspect_cn_literal | {literal[:32]}"
170+
)
171+
172+
return findings
173+
174+
175+
def _is_suspect_cn_literal(literal: str) -> bool:
176+
if PERSON_NAME_PATTERN.fullmatch(literal):
177+
return True
178+
if LOCATION_SUFFIX_PATTERN.search(literal):
179+
return True
180+
return False

tests/test_sqlite_client.py

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
"""SQLite read client tests."""
2+
3+
from __future__ import annotations
4+
5+
import sqlite3
6+
from pathlib import Path
7+
8+
import pytest
9+
10+
from rond_api.db.sqlite_client import DatabaseReadError, SQLiteReadClient
11+
12+
13+
def test_sqlite_read_client_retries_when_locked(
14+
monkeypatch: pytest.MonkeyPatch,
15+
tmp_path: Path,
16+
) -> None:
17+
db_path = tmp_path / "retry.db"
18+
_init_demo_database(db_path)
19+
20+
calls = {"count": 0}
21+
original_execute_once = SQLiteReadClient._execute_once
22+
23+
def flaky_execute_once(self: SQLiteReadClient, sql: str, params: tuple[object, ...]):
24+
if calls["count"] == 0:
25+
calls["count"] += 1
26+
raise sqlite3.OperationalError("database is locked")
27+
return original_execute_once(self, sql, params)
28+
29+
monkeypatch.setattr(SQLiteReadClient, "_execute_once", flaky_execute_once)
30+
monkeypatch.setattr("rond_api.db.sqlite_client.time.sleep", lambda _seconds: None)
31+
32+
client = SQLiteReadClient(db_path=db_path, max_retries=3, retry_backoff_seconds=0.0)
33+
rows = client.execute_query("SELECT value FROM demo LIMIT 1;")
34+
assert rows[0]["value"] == "ok"
35+
assert calls["count"] == 1
36+
37+
38+
def test_sqlite_read_client_raises_after_non_retryable_error(
39+
monkeypatch: pytest.MonkeyPatch,
40+
tmp_path: Path,
41+
) -> None:
42+
db_path = tmp_path / "error.db"
43+
_init_demo_database(db_path)
44+
45+
def failing_execute_once(self: SQLiteReadClient, sql: str, params: tuple[object, ...]):
46+
raise sqlite3.OperationalError("no such table: missing")
47+
48+
monkeypatch.setattr(SQLiteReadClient, "_execute_once", failing_execute_once)
49+
client = SQLiteReadClient(db_path=db_path, max_retries=3, retry_backoff_seconds=0.0)
50+
51+
with pytest.raises(DatabaseReadError):
52+
client.execute_query("SELECT * FROM missing;")
53+
54+
55+
def _init_demo_database(path: Path) -> None:
56+
with sqlite3.connect(path) as connection:
57+
connection.execute("CREATE TABLE demo (value TEXT);")
58+
connection.execute("INSERT INTO demo (value) VALUES ('ok');")
59+
connection.commit()

0 commit comments

Comments
 (0)