Antoine Van Elstraete
e9f73c7562
Ajoute le modèle WorkplacePresenceEvent pour stocker les événements de présence reçus de Home Assistant, avec clé d'idempotence unique, lien vers une journée et, facultativement, une plage horaire. Active les contraintes de clés étrangères sur chaque connexion SQLite et documente le schéma dans l'onboarding. Couvre le tout par des tests de modèle et de factory. Co-authored-by: OpenAI/GPT-5.6-Luna-Pro <vibecoder@antoineve.me>
112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
from datetime import date, datetime, time
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app import db
|
|
from app.models import TimeSlot, WorkEntry, WorkplacePresenceEvent
|
|
|
|
|
|
def make_entry() -> WorkEntry:
|
|
return WorkEntry(date=date(2026, 8, 13), day_type="WORK")
|
|
|
|
|
|
def make_event(entry: WorkEntry, key: str = "ha-arrival-1") -> WorkplacePresenceEvent:
|
|
return WorkplacePresenceEvent(
|
|
idempotency_key=key,
|
|
event_type="arrival",
|
|
received_at=datetime(2026, 8, 13, 6, 23, 10),
|
|
occurred_at=datetime(2026, 8, 13, 8, 23, 10),
|
|
local_date=date(2026, 8, 13),
|
|
entry=entry,
|
|
)
|
|
|
|
|
|
def test_presence_event_creation_and_relations(app):
|
|
with app.app_context():
|
|
entry = make_entry()
|
|
slot = TimeSlot(start_time=time(8), end_time=time(12), entry=entry)
|
|
event = make_event(entry)
|
|
event.time_slot = slot
|
|
db.session.add(entry)
|
|
db.session.commit()
|
|
|
|
assert event.entry is entry
|
|
assert event in entry.presence_events
|
|
assert event.time_slot is slot
|
|
assert event in slot.presence_events
|
|
assert event.processed_at is None
|
|
|
|
|
|
def test_idempotency_key_is_unique(app):
|
|
with app.app_context():
|
|
entry = make_entry()
|
|
db.session.add_all([entry, make_event(entry), make_event(entry, "ha-arrival-1")])
|
|
with pytest.raises(IntegrityError):
|
|
db.session.commit()
|
|
db.session.rollback()
|
|
|
|
|
|
def test_event_type_check_constraint(app):
|
|
with app.app_context():
|
|
entry = make_entry()
|
|
event = make_event(entry)
|
|
event.event_type = "unknown"
|
|
db.session.add_all([entry, event])
|
|
with pytest.raises(IntegrityError):
|
|
db.session.commit()
|
|
db.session.rollback()
|
|
|
|
|
|
def test_time_slot_link_is_nullable_and_set_null_on_slot_delete(app):
|
|
with app.app_context():
|
|
entry = make_entry()
|
|
slot = TimeSlot(start_time=time(8), end_time=time(12), entry=entry)
|
|
event = make_event(entry)
|
|
event.time_slot = slot
|
|
db.session.add(entry)
|
|
db.session.commit()
|
|
|
|
db.session.delete(slot)
|
|
db.session.commit()
|
|
|
|
assert db.session.get(WorkplacePresenceEvent, event.id).time_slot_id is None
|
|
|
|
|
|
def test_events_cascade_when_work_entry_is_deleted(app):
|
|
with app.app_context():
|
|
entry = make_entry()
|
|
db.session.add(make_event(entry))
|
|
db.session.commit()
|
|
event_id = entry.presence_events[0].id
|
|
|
|
db.session.delete(entry)
|
|
db.session.commit()
|
|
|
|
assert db.session.get(WorkplacePresenceEvent, event_id) is None
|
|
|
|
|
|
def test_create_all_adds_presence_table_without_losing_existing_entries(app):
|
|
with app.app_context():
|
|
db.session.add(make_entry())
|
|
db.session.commit()
|
|
db.session.execute(sa.text("DROP TABLE workplace_presence_events"))
|
|
db.session.commit()
|
|
|
|
db.create_all()
|
|
|
|
assert db.session.scalar(sa.select(sa.func.count()).select_from(WorkEntry)) == 1
|
|
assert sa.inspect(db.engine).has_table("workplace_presence_events")
|
|
|
|
|
|
def test_create_all_creates_all_tables_on_empty_sqlite_database(app):
|
|
with app.app_context():
|
|
db.drop_all()
|
|
db.create_all()
|
|
|
|
inspector = sa.inspect(db.engine)
|
|
assert inspector.has_table("work_entries")
|
|
assert inspector.has_table("time_slots")
|
|
assert inspector.has_table("workplace_presence_events")
|