62 lines
2.6 KiB
Python
62 lines
2.6 KiB
Python
from pathlib import Path
|
|
import sqlite3
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_sqlite_session_migration_executes_and_survives_reopen(tmp_path):
|
|
db_path = tmp_path / "rustbox.sqlite"
|
|
migration = (ROOT / "migrations/sqlite/V047__sessions.sql").read_text()
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(migration)
|
|
conn.execute(
|
|
"""INSERT INTO sessions (
|
|
token, account_id, browser_id, email, password_ciphertext, user_hash,
|
|
imap_host, imap_port, imap_tls, imap_tls_verify,
|
|
smtp_host, smtp_port, smtp_tls, smtp_tls_verify,
|
|
timeout_secs, created_at_unix, last_accessed_unix, user_agent
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
"token-1", "account-1", "browser-1", "user@example.com", "ciphertext", "hash-1",
|
|
"imap.example.com", 993, 1, 1, "smtp.example.com", 587, 1, 1,
|
|
3600, 1000, 1000, "pytest",
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
reopened = sqlite3.connect(db_path)
|
|
row = reopened.execute(
|
|
"SELECT token, account_id, browser_id, password_ciphertext FROM sessions WHERE token = ?",
|
|
("token-1",),
|
|
).fetchone()
|
|
reopened.close()
|
|
assert row == ("token-1", "account-1", "browser-1", "ciphertext")
|
|
|
|
|
|
def test_sqlite_expiry_query_removes_only_expired_sessions(tmp_path):
|
|
db_path = tmp_path / "rustbox.sqlite"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript((ROOT / "migrations/sqlite/V047__sessions.sql").read_text())
|
|
base = (
|
|
"account", "browser", "user@example.com", "ciphertext", "hash",
|
|
"imap", 993, 1, 1, "smtp", 587, 1, 1, 100, 1000, "pytest",
|
|
)
|
|
for token, last_accessed in [("expired", 1000), ("live", 1950)]:
|
|
conn.execute(
|
|
"""INSERT INTO sessions (
|
|
token, account_id, browser_id, email, password_ciphertext, user_hash,
|
|
imap_host, imap_port, imap_tls, imap_tls_verify,
|
|
smtp_host, smtp_port, smtp_tls, smtp_tls_verify,
|
|
timeout_secs, created_at_unix, last_accessed_unix, user_agent
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(token, f"{base[0]}-{token}", *base[1:15], last_accessed, base[15]),
|
|
)
|
|
delete_sql = (ROOT / "src/queries/sqlite/queries.rs").read_text()
|
|
assert "last_accessed_unix + timeout_secs < ?" in delete_sql
|
|
conn.execute("DELETE FROM sessions WHERE last_accessed_unix + timeout_secs < ?", (2000,))
|
|
remaining = [row[0] for row in conn.execute("SELECT token FROM sessions ORDER BY token")]
|
|
conn.close()
|
|
assert remaining == ["live"]
|