18113637e9
Adds a UtcDateTime TypeDecorator (models.py) that enforces aware-UTC on write and stamps tzinfo=UTC on read. Replaces 26 DateTime column declarations. Removes 5 production sites that defensively stripped tzinfo to match SQLite's lossy round-trip. auth.py now coerces legacy session cookies upward (stamp UTC on parsed naive marker) instead of stripping live aware markers downward. The change is Python-side only: UtcDateTime.impl = DateTime, so DDL and emitted SQL are unchanged. No Alembic migration needed. Adds 2 unit tests in test_models.py pinning the decorator's contract independently of the column declarations. The three deliberately-naive test_timeago.py fixtures (lines 67, 73, 113) remain naive on purpose -- they exercise _ensure_utc's normalize-up path at the public filter boundary, which stays as belt-and-braces defense. See docs/superpowers/specs/2026-05-16-tz-aware-datetime-design.md. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from l4d2web.db import init_db, session_scope
|
|
from l4d2web.models import (
|
|
Blueprint,
|
|
Server,
|
|
ServerLiveState,
|
|
ServerPlayerSession,
|
|
SteamUserProfile,
|
|
User,
|
|
UtcDateTime,
|
|
now_utc as now_utc_aware,
|
|
)
|
|
|
|
|
|
def test_create_user_and_blueprint(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'app.db'}")
|
|
|
|
init_db()
|
|
|
|
with session_scope() as session:
|
|
user = User(username="alice", password_digest="digest", admin=False)
|
|
session.add(user)
|
|
session.flush()
|
|
|
|
blueprint = Blueprint(user_id=user.id, name="default", arguments="[]", config="[]")
|
|
session.add(blueprint)
|
|
session.flush()
|
|
|
|
assert user.id is not None
|
|
assert blueprint.id is not None
|
|
|
|
|
|
def test_user_has_password_changed_at_default(tmp_path, monkeypatch):
|
|
from datetime import UTC, datetime
|
|
from l4d2web.app import create_app
|
|
from l4d2web.auth import hash_password
|
|
|
|
db_url = f"sqlite:///{tmp_path/'pw.db'}"
|
|
monkeypatch.setenv("DATABASE_URL", db_url)
|
|
create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
|
|
init_db()
|
|
|
|
before = datetime.now(UTC)
|
|
with session_scope() as db:
|
|
db.add(User(username="alice", password_digest=hash_password("secret")))
|
|
with session_scope() as db:
|
|
user = db.query(User).filter_by(username="alice").one()
|
|
|
|
assert user.password_changed_at is not None
|
|
assert user.password_changed_at >= before
|
|
|
|
|
|
def test_server_has_rcon_password_column(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'t.db'}")
|
|
init_db()
|
|
with session_scope() as db:
|
|
u = User(username="u", password_digest="x")
|
|
db.add(u); db.flush()
|
|
bp = Blueprint(user_id=u.id, name="bp", arguments="[]", config="[]")
|
|
db.add(bp); db.flush()
|
|
s = Server(
|
|
user_id=u.id, blueprint_id=bp.id, name="s", port=27500,
|
|
rcon_password="abc",
|
|
)
|
|
db.add(s); db.flush()
|
|
assert db.scalar(select(Server.rcon_password).where(Server.id == s.id)) == "abc"
|
|
|
|
|
|
def test_server_live_state_table_columns(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'t.db'}")
|
|
init_db()
|
|
with session_scope() as db:
|
|
u = User(username="u", password_digest="x"); db.add(u); db.flush()
|
|
bp = Blueprint(user_id=u.id, name="bp", arguments="[]", config="[]"); db.add(bp); db.flush()
|
|
s = Server(user_id=u.id, blueprint_id=bp.id, name="s", port=27501, rcon_password="x")
|
|
db.add(s); db.flush()
|
|
row = ServerLiveState(
|
|
server_id=s.id, started_at=now_utc_aware(), last_seen_at=now_utc_aware(),
|
|
players=2, max_players=4, bots=0, map="c1m1_hotel", hibernating=False,
|
|
)
|
|
db.add(row); db.flush()
|
|
|
|
|
|
def test_server_player_session_table_columns(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'t.db'}")
|
|
init_db()
|
|
with session_scope() as db:
|
|
u = User(username="u", password_digest="x"); db.add(u); db.flush()
|
|
bp = Blueprint(user_id=u.id, name="bp", arguments="[]", config="[]"); db.add(bp); db.flush()
|
|
s = Server(user_id=u.id, blueprint_id=bp.id, name="s", port=27502, rcon_password="x")
|
|
db.add(s); db.flush()
|
|
row = ServerPlayerSession(
|
|
server_id=s.id, steam_id_64="76561197960828710",
|
|
joined_at=now_utc_aware(), left_at=None,
|
|
name_at_join="Crone", min_ping=42, max_ping=185,
|
|
)
|
|
db.add(row); db.flush()
|
|
|
|
|
|
def test_steam_user_profile_table_columns(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'t.db'}")
|
|
init_db()
|
|
with session_scope() as db:
|
|
row = SteamUserProfile(
|
|
steam_id_64="76561197960828710",
|
|
persona_name="MrCool42",
|
|
avatar_url="https://avatars.cloudflare.steamstatic.com/abc_medium.jpg",
|
|
fetched_at=now_utc_aware(),
|
|
)
|
|
db.add(row); db.flush()
|
|
|
|
|
|
def test_utc_datetime_rejects_naive_bind() -> None:
|
|
with pytest.raises(TypeError, match="naive"):
|
|
UtcDateTime().process_bind_param(datetime(2026, 5, 16, 12, 0), None)
|
|
|
|
|
|
def test_utc_datetime_stamps_utc_on_read() -> None:
|
|
naive = datetime(2026, 5, 16, 12, 0)
|
|
result = UtcDateTime().process_result_value(naive, None)
|
|
assert result == datetime(2026, 5, 16, 12, 0, tzinfo=UTC)
|
|
assert result.tzinfo == UTC
|