62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from src import session_store
|
|
from src.session_store import SessionStore, SessionStoreError
|
|
|
|
|
|
def test_revocation_is_durable_and_scoped_to_one_session(tmp_path):
|
|
now = [1_000.0]
|
|
database = tmp_path / "sessions.sqlite3"
|
|
first = SessionStore(database, clock=lambda: now[0])
|
|
first.activate("first-session-secret", 2_000)
|
|
first.activate("second-session-secret", 2_000)
|
|
|
|
reconstructed = SessionStore(database, clock=lambda: now[0])
|
|
reconstructed.revoke("first-session-secret")
|
|
|
|
assert reconstructed.is_active("first-session-secret", 2_000) is False
|
|
assert reconstructed.is_active("second-session-secret", 2_000) is True
|
|
assert b"first-session-secret" not in database.read_bytes()
|
|
assert b"second-session-secret" not in database.read_bytes()
|
|
|
|
|
|
def test_validation_does_not_create_a_missing_registry(tmp_path):
|
|
database = tmp_path / "sessions.sqlite3"
|
|
store = SessionStore(database, clock=lambda: 1_000.0)
|
|
|
|
with pytest.raises(SessionStoreError):
|
|
store.is_active("unknown-session", 2_000)
|
|
|
|
assert database.exists() is False
|
|
|
|
|
|
def test_validation_executes_only_a_read_query(tmp_path, monkeypatch):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
store.activate("active-session", 2_000)
|
|
statements = []
|
|
connect = sqlite3.connect
|
|
|
|
def traced_connect(*args, **kwargs):
|
|
connection = connect(*args, **kwargs)
|
|
connection.set_trace_callback(statements.append)
|
|
return connection
|
|
|
|
monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect)
|
|
|
|
assert store.is_active("active-session", 2_000) is True
|
|
assert [statement.split()[0].upper() for statement in statements] == ["SELECT"]
|
|
|
|
|
|
def test_expired_sessions_are_rejected_without_writing_during_validation(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("expiring-session", 1_001)
|
|
|
|
now[0] = 1_001.0
|
|
|
|
assert store.is_active("expiring-session", 1_001) is False
|
|
with sqlite3.connect(store.path) as connection:
|
|
assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (1,)
|