46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
"""Offline SQLite check for the production contact SQL; no WeChat database is read."""
|
|
|
|
import re
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
source = (
|
|
Path(__file__).resolve().parents[2] / "node-agent/WxAgent.Core/WechatContactQuery.cs"
|
|
).read_text()
|
|
match = re.search(r'public const string Sql = """\s*(.*?)\s*""";', source, re.S)
|
|
assert match is not None
|
|
sql = match.group(1)
|
|
|
|
with sqlite3.connect(":memory:") as connection:
|
|
connection.execute(
|
|
"CREATE TABLE contact(username TEXT PRIMARY KEY, nick_name TEXT, remark TEXT, small_head_url TEXT)"
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO contact VALUES (?, ?, ?, NULL)",
|
|
[(f"wxid_{i:04}", "same display name", None) for i in range(501)]
|
|
+ [("room@chatroom", "group", None), ("literal", "O'Brien_%", None)],
|
|
)
|
|
|
|
def query(limit=201, offset=0, contains=None, groups=None):
|
|
return connection.execute(
|
|
sql, {"limit": limit, "offset": offset, "contains": contains, "groups": groups}
|
|
).fetchall()
|
|
|
|
first = query()
|
|
assert len(first) == 201
|
|
second = query(offset=200)
|
|
third = query(offset=400)
|
|
ids = [row[0] for row in first[:200] + second[:200] + third]
|
|
assert len(ids) == 503 and len(set(ids)) == 503 and ids == sorted(ids)
|
|
assert [row[0] for row in query(contains="_%")] == ["literal"]
|
|
assert query(contains="'; DROP TABLE contact;--") == []
|
|
assert [row[0] for row in query(groups=1)] == ["room@chatroom"]
|
|
assert len(query(limit=10000, groups=0)) == 502
|
|
assert len(query(limit=10000, contains="same display name")) == 501
|
|
assert query(offset=503) == []
|
|
assert connection.execute("SELECT count(*) FROM contact").fetchone()[0] == 503
|
|
|
|
print(
|
|
"contact SQL: pagination, duplicate names, literal filtering and group filtering passed"
|
|
)
|