5 Commits

Author SHA1 Message Date
a27f282b69 feat(presence): ajoute le service métier de présence et ses tests
Étape 3 validée du service métier de présence : implémentation de
app/business/presence_service.py et couverture par
tests/test_presence_service.py.

Co-authored-by: OpenAI/GPT-5.6-Luna-Pro <vibecoder@antoineve.me>
2026-08-13 18:54:33 +02:00
e9f73c7562 feat: persister les événements de présence et activer les clés étrangères SQLite
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>
2026-08-13 17:36:26 +02:00
c30bd1c0c5 fix(tests): isolate SQLite test databases
Co-authored-by: OpenAI/GPT-5.6-Terra <vibecoder@antoineve.me>
2026-08-13 17:25:48 +02:00
85e6e502e5 feat(config): ajouter la configuration Home Assistant (étape 1)
Ajoute la section [home_assistant] au TOML et sa validation au chargement
via get_home_assistant_config(). La section est facultative pour préserver
la compatibilité avec les configurations existantes, mais si elle est
présente elle doit être complète et référencer un type de journée, un
trajet et un véhicule à moteur existants, sous peine de refuser le
démarrage de l'application. La configuration validée est exposée dans
app.config['HOME_ASSISTANT'].

Co-authored-by: OpenAI/GPT-5.6-Luna-Pro <vibecoder@antoineve.me>
2026-08-13 16:13:18 +02:00
525d38224c docs: planifier l'API REST Home Assistant
Co-authored-by: OpenAI/GPT-5.6-Luna-Pro <vibecoder@antoineve.me>
2026-08-13 15:57:32 +02:00
13 changed files with 1120 additions and 6 deletions

View File

@@ -15,6 +15,7 @@ Architecture et composants clés :
import os
import tomllib
from collections.abc import Mapping
import sqlalchemy as sa
from flask import Flask
@@ -23,6 +24,15 @@ from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
def _enable_sqlite_foreign_keys(dbapi_connection, connection_record):
"""Active les contraintes de clés étrangères sur chaque connexion SQLite."""
cursor = dbapi_connection.cursor()
try:
cursor.execute("PRAGMA foreign_keys=ON")
finally:
cursor.close()
def _migrate_db(app):
"""Applique les migrations de schéma manquantes de manière incrémentale (sans Alembic).
@@ -36,6 +46,10 @@ def _migrate_db(app):
"""
import sqlite3
with app.app_context():
if db.engine.url.database in (None, ":memory:"):
return # Une base mémoire est initialisée par create_all().
db_path = os.path.join(app.instance_path, "worklog.db")
if not os.path.exists(db_path):
return # Nouvelle DB, create_all() s'en charge
@@ -120,7 +134,12 @@ def _date_fr(d):
return f"{jour} {d.day} {mois} {d.year}"
def create_app(config_path=None):
def create_app(
config_path: str | None = None,
*,
database_uri: str | None = None,
engine_options: Mapping[str, object] | None = None,
) -> Flask:
"""Factory de création et de configuration de l'application Flask.
Cette fonction réalise les étapes suivantes :
@@ -136,6 +155,11 @@ def create_app(config_path=None):
Paramètres:
config_path (str | None): Chemin optionnel vers le fichier de configuration TOML.
Par défaut, cherche `config.toml` à la racine du projet.
database_uri (str | None): URI SQLAlchemy à utiliser à la place de la base SQLite
de l'instance. Cette option est appliquée avant l'initialisation
de Flask-SQLAlchemy.
engine_options (Mapping[str, object] | None): Options SQLAlchemy appliquées avant
l'initialisation de Flask-SQLAlchemy.
Retourne:
Flask: L'instance de l'application Flask configurée et prête à l'emploi.
@@ -144,9 +168,11 @@ def create_app(config_path=None):
os.makedirs(app.instance_path, exist_ok=True)
app.config["SQLALCHEMY_DATABASE_URI"] = (
f"sqlite:///{os.path.join(app.instance_path, 'worklog.db')}"
)
if database_uri is None:
database_uri = f"sqlite:///{os.path.join(app.instance_path, 'worklog.db')}"
app.config["SQLALCHEMY_DATABASE_URI"] = database_uri
if engine_options is not None:
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = engine_options
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev-secret-change-in-prod")
@@ -159,7 +185,15 @@ def create_app(config_path=None):
else:
app.config["TOML"] = {}
from app.config_loader import get_home_assistant_config
with app.app_context():
app.config["HOME_ASSISTANT"] = get_home_assistant_config()
db.init_app(app)
with app.app_context():
if db.engine.dialect.name == "sqlite":
sa.event.listen(db.engine, "connect", _enable_sqlite_foreign_keys)
app.jinja_env.filters["date_fr"] = _date_fr
app.jinja_env.filters["day_type_fr"] = _day_type_fr

View File

@@ -0,0 +1,227 @@
"""Enregistrement métier des événements de présence Home Assistant."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, date, datetime, time
from typing import Any, Callable, Literal, Mapping
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.models import TimeSlot, WorkEntry, WorkplacePresenceEvent
EventType = Literal["arrival", "departure"]
SlotState = Literal["open", "closed"]
class PresenceServiceError(ValueError):
"""Erreur métier prévisible lors de l'enregistrement d'une présence."""
class InvalidPresenceEventError(PresenceServiceError):
"""Les données de l'événement ne respectent pas le contrat métier."""
class IdempotencyConflictError(PresenceServiceError):
"""La clé est déjà utilisée par un événement différent."""
class ArrivalAlreadyOpenError(PresenceServiceError):
"""Une arrivée est déjà ouverte, quelle que soit sa journée."""
class DepartureWithoutArrivalError(PresenceServiceError):
"""Aucune arrivée ouverte ne peut être fermée."""
@dataclass(frozen=True)
class PresenceResult:
"""Résultat sérialisable par la future route API."""
event_id: int
entry_id: int
time_slot_id: int | None
replayed: bool
slot_state: SlotState
@property
def event_created(self) -> bool:
return not self.replayed
def _parse_occurred_at(value: str, timezone: ZoneInfo) -> tuple[datetime, datetime, date, time]:
try:
parsed = datetime.fromisoformat(value)
except (TypeError, ValueError) as exc:
raise InvalidPresenceEventError("occurred_at doit être un ISO 8601 valide") from exc
if parsed.tzinfo is None or parsed.utcoffset() is None:
raise InvalidPresenceEventError("occurred_at doit comporter un offset explicite")
local = parsed.astimezone(timezone)
wall_time = local.replace(tzinfo=None)
return wall_time, local, local.date(), local.time()
def _received_at(value: datetime | None, clock: Callable[[], datetime] | None) -> datetime:
received = value if value is not None else (clock() if clock else datetime.now(UTC))
if received.tzinfo is None or received.utcoffset() is None:
return received
return received.astimezone(UTC).replace(tzinfo=None)
def _validate_config(config: Mapping[str, Any]) -> tuple[ZoneInfo, str, str, str]:
try:
timezone_name = config["timezone"]
defaults = (
config["default_day_type"],
config["default_journey_profile_id"],
config["default_motor_vehicle_id"],
)
timezone = ZoneInfo(timezone_name)
except (KeyError, TypeError, ZoneInfoNotFoundError, ValueError) as exc:
raise InvalidPresenceEventError("Configuration Home Assistant invalide") from exc
if not isinstance(timezone_name, str) or not all(isinstance(item, str) for item in defaults):
raise InvalidPresenceEventError("Configuration Home Assistant invalide")
return timezone, defaults[0], defaults[1], defaults[2]
def _result(event: WorkplacePresenceEvent, replayed: bool) -> PresenceResult:
return PresenceResult(
event_id=event.id,
entry_id=event.entry_id,
time_slot_id=event.time_slot_id,
replayed=replayed,
slot_state="closed" if event.time_slot_id is not None else "open",
)
def record_presence_event(
session: Session,
config: Mapping[str, Any],
event_type: str,
occurred_at: str,
idempotency_key: str,
*,
received_at: datetime | None = None,
clock: Callable[[], datetime] | None = None,
) -> PresenceResult:
"""Enregistre une arrivée ou un départ sans valider la transaction SQLAlchemy."""
if not hasattr(session, "in_transaction"):
session = session()
if event_type not in ("arrival", "departure"):
raise InvalidPresenceEventError("event_type doit valoir arrival ou departure")
if not isinstance(idempotency_key, str) or not idempotency_key or len(idempotency_key) > 255:
raise InvalidPresenceEventError(
"idempotency_key doit être non vide et limitée à 255 caractères"
)
timezone, day_type, journey_id, vehicle_id = _validate_config(config)
occurred_wall, occurred_local, local_date, wall_time = _parse_occurred_at(occurred_at, timezone)
received_wall = _received_at(received_at, clock)
# Le SELECT démarre explicitement la transaction racine. Aucun contexte ne
# valide cette transaction : la route appelante garde la décision finale.
session.execute(select(1))
try:
if session:
existing = session.scalar(
select(WorkplacePresenceEvent).where(
WorkplacePresenceEvent.idempotency_key == idempotency_key
)
)
if existing is not None:
if existing.event_type != event_type or existing.occurred_at != occurred_wall:
raise IdempotencyConflictError("La clé d'idempotence est déjà utilisée")
return _result(existing, replayed=True)
if event_type == "arrival":
# Une seule arrivée peut être ouverte dans toute l'application :
# le prochain départ doit toujours avoir un rattachement unique.
open_arrival = session.scalar(
select(WorkplacePresenceEvent).where(
WorkplacePresenceEvent.event_type == "arrival",
WorkplacePresenceEvent.time_slot_id.is_(None),
)
)
if open_arrival is not None:
raise ArrivalAlreadyOpenError("Une arrivée est déjà ouverte")
entry = session.scalar(select(WorkEntry).where(WorkEntry.date == local_date))
if entry is None:
entry = WorkEntry(
date=local_date,
day_type=day_type,
journey_profile_id=journey_id,
motor_vehicle_id=vehicle_id,
)
session.add(entry)
session.flush()
event = WorkplacePresenceEvent(
idempotency_key=idempotency_key,
event_type="arrival",
received_at=received_wall,
occurred_at=occurred_wall,
local_date=local_date,
entry=entry,
)
session.add(event)
session.flush()
return _result(event, replayed=False)
open_arrivals = session.scalars(
select(WorkplacePresenceEvent)
.where(
WorkplacePresenceEvent.event_type == "arrival",
WorkplacePresenceEvent.time_slot_id.is_(None),
)
.order_by(WorkplacePresenceEvent.occurred_at, WorkplacePresenceEvent.id)
).all()
if not open_arrivals:
raise DepartureWithoutArrivalError("Aucune arrivée ouverte")
if len(open_arrivals) > 1:
raise ArrivalAlreadyOpenError("Plusieurs arrivées sont ouvertes")
arrival = open_arrivals[0]
arrival_local = arrival.occurred_at.replace(tzinfo=timezone)
if occurred_local <= arrival_local:
raise InvalidPresenceEventError(
"L'instant du départ doit être postérieur à celui de l'arrivée"
)
slot = TimeSlot(
entry_id=arrival.entry_id, start_time=arrival.occurred_at.time(), end_time=wall_time
)
session.add(slot)
departure = WorkplacePresenceEvent(
idempotency_key=idempotency_key,
event_type="departure",
received_at=received_wall,
occurred_at=occurred_wall,
local_date=local_date,
entry_id=arrival.entry_id,
time_slot=slot,
processed_at=received_wall,
)
session.add(departure)
session.flush()
arrival.time_slot = slot
arrival.processed_at = received_wall
session.flush()
return _result(departure, replayed=False)
except IntegrityError as exc:
# Une autre requête peut avoir gagné la clé entre le contrôle et le flush.
session.rollback()
existing = session.scalar(
select(WorkplacePresenceEvent).where(
WorkplacePresenceEvent.idempotency_key == idempotency_key
)
)
if (
existing is not None
and existing.event_type == event_type
and existing.occurred_at == occurred_wall
):
return _result(existing, replayed=True)
raise IdempotencyConflictError("Conflit d'unicité lors de l'enregistrement") from exc
record_home_assistant_event = record_presence_event

View File

@@ -21,8 +21,28 @@ Contrat TOML :
- Certains types de journées (Télétravail, Maladie, Congé, RTT, Férié) n'impliquent aucun déplacement physique.
"""
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from flask import current_app
_HOME_ASSISTANT_REQUIRED_KEYS = {
"timezone",
"default_day_type",
"default_journey_profile_id",
"default_motor_vehicle_id",
}
_VALID_DAY_TYPES = {
"WORK",
"TT",
"GARDE",
"ASTREINTE",
"FORMATION",
"RTT",
"CONGE",
"MALADE",
"FERIE",
}
def get_vehicles():
"""Récupère l'ensemble des véhicules configurés dans le fichier TOML.
@@ -58,6 +78,66 @@ def get_journeys():
return current_app.config.get("TOML", {}).get("journeys", {})
def get_home_assistant_config() -> dict[str, str] | None:
"""Retourne la configuration Home Assistant, après validation.
L'absence de section désactive la future intégration et reste compatible avec
les anciens fichiers TOML. Une section présente doit en revanche être
complète et cohérente avec les véhicules, trajets et types de journées
connus de l'application.
"""
config = current_app.config.get("TOML", {}).get("home_assistant")
if config is None:
return None
if not isinstance(config, dict):
raise ValueError("Configuration [home_assistant] invalide : la section doit être une table")
missing = _HOME_ASSISTANT_REQUIRED_KEYS - config.keys()
if missing:
missing_keys = ", ".join(sorted(missing))
raise ValueError(
f"Configuration [home_assistant] incomplète : clé(s) manquante(s) {missing_keys}"
)
if any(
not isinstance(config[key], str) or not config[key] for key in _HOME_ASSISTANT_REQUIRED_KEYS
):
raise ValueError(
"Configuration [home_assistant] invalide : toutes les valeurs doivent être des chaînes non vides"
)
timezone = config["timezone"]
try:
ZoneInfo(timezone)
except (ZoneInfoNotFoundError, ValueError) as exc:
raise ValueError(
f"Configuration [home_assistant] invalide : fuseau horaire inconnu {timezone!r}"
) from exc
day_type = config["default_day_type"]
if day_type not in _VALID_DAY_TYPES:
raise ValueError(
f"Configuration [home_assistant] invalide : type de journée inconnu {day_type!r}"
)
journey_id = config["default_journey_profile_id"]
if journey_id not in get_journeys():
raise ValueError(f"Configuration [home_assistant] invalide : trajet inconnu {journey_id!r}")
vehicle_id = config["default_motor_vehicle_id"]
vehicle = get_vehicles().get(vehicle_id)
if vehicle is None:
raise ValueError(
f"Configuration [home_assistant] invalide : véhicule inconnu {vehicle_id!r}"
)
if vehicle.get("type") != "moteur":
raise ValueError(
f"Configuration [home_assistant] invalide : le véhicule {vehicle_id!r} n'est pas un véhicule moteur"
)
return {key: config[key] for key in _HOME_ASSISTANT_REQUIRED_KEYS}
def journey_has_motor(journey_profile_id: str | None) -> bool:
"""Vérifie si un profil de trajet donné inclut une distance pour véhicule à moteur.

View File

@@ -37,6 +37,9 @@ class WorkEntry(db.Model):
time_slots: so.Mapped[list["TimeSlot"]] = so.relationship(
back_populates="entry", cascade="all, delete-orphan", order_by="TimeSlot.start_time"
)
presence_events: so.Mapped[list["WorkplacePresenceEvent"]] = so.relationship(
back_populates="entry", cascade="all, delete-orphan"
)
def total_minutes(self) -> int:
"""
@@ -86,6 +89,50 @@ class TimeSlot(db.Model):
end_time: so.Mapped[time] = so.mapped_column(sa.Time, nullable=False)
entry: so.Mapped["WorkEntry"] = so.relationship(back_populates="time_slots")
presence_events: so.Mapped[list["WorkplacePresenceEvent"]] = so.relationship(
back_populates="time_slot", passive_deletes=True
)
class WorkplacePresenceEvent(db.Model):
"""Événement de présence reçu de Home Assistant.
``received_at`` est un instant normalisé en UTC, stocké naïf selon la convention
actuelle de l'application. À l'inverse, ``occurred_at`` est l'heure murale naïve
dans ``Europe/Paris`` et ``local_date`` est le jour local dérivé de cette heure.
Cette distinction est volontaire : elle sera utilisée par le service métier futur
pour rattacher les arrivées et départs aux journées, notamment autour de minuit.
"""
__tablename__ = "workplace_presence_events"
__table_args__ = (
sa.CheckConstraint("event_type IN ('arrival', 'departure')", name="ck_presence_event_type"),
sa.Index("ix_presence_events_local_date", "local_date"),
sa.Index("ix_presence_events_entry_id", "entry_id"),
)
id: so.Mapped[int] = so.mapped_column(primary_key=True)
idempotency_key: so.Mapped[str] = so.mapped_column(sa.String(255), unique=True, nullable=False)
event_type: so.Mapped[str] = so.mapped_column(sa.String(9), nullable=False)
received_at: so.Mapped[datetime] = so.mapped_column(sa.DateTime, nullable=False)
# Heure locale Europe/Paris, sans fuseau : ne pas la traiter comme un instant UTC.
occurred_at: so.Mapped[datetime] = so.mapped_column(sa.DateTime, nullable=False)
local_date: so.Mapped[date] = so.mapped_column(sa.Date, nullable=False)
entry_id: so.Mapped[int] = so.mapped_column(
sa.ForeignKey("work_entries.id", ondelete="CASCADE"), nullable=False
)
time_slot_id: so.Mapped[int | None] = so.mapped_column(
sa.ForeignKey("time_slots.id", ondelete="SET NULL"), nullable=True
)
created_at: so.Mapped[datetime] = so.mapped_column(
sa.DateTime, default=lambda: datetime.now(UTC), nullable=False
)
processed_at: so.Mapped[datetime | None] = so.mapped_column(sa.DateTime, nullable=True)
entry: so.Mapped["WorkEntry"] = so.relationship(back_populates="presence_events")
time_slot: so.Mapped["TimeSlot | None"] = so.relationship(
back_populates="presence_events", passive_deletes=True
)
class LeaveBalance(db.Model):

View File

@@ -37,6 +37,12 @@ distances = { moteur = 14, velo = 8 }
name = "Vélo seul"
distances = { velo = 24 }
[home_assistant]
timezone = "Europe/Paris"
default_day_type = "WORK"
default_journey_profile_id = "moteur_seul"
default_motor_vehicle_id = "citadine"
# --- Barème kilométrique voitures 2025 (revenus 2024) ---
# Source : https://www.service-public.gouv.fr/particuliers/actualites/A14686
# Majoration +20% pour véhicules électriques gérée dans travel_calc.py

View File

@@ -101,14 +101,33 @@ Ce fichier contient les paramètres qui ne changent pas fréquemment et qui déf
- **Véhicules (`[vehicles.*]`)** : Nom, type (moteur ou velo), carburant (electric, diesel, essence, none), émissions de CO₂ par km, et puissance fiscale (CV).
- **Trajets (`[journeys.*]`)** : Profils de trajets prédéfinis (ex: "moteur_seul", "moteur_velo") avec les distances associées par type de moyen de transport.
- **Barème kilométrique (`[bareme_kilometrique.YYYY.*]`)** : Les tranches fiscales officielles de remboursement par année et par puissance fiscale (CV).
- **Home Assistant (`[home_assistant]`)** : Les valeurs par défaut et le fuseau utilisés par la future API de présence :
```toml
[home_assistant]
timezone = "Europe/Paris"
default_day_type = "WORK"
default_journey_profile_id = "moteur_seul"
default_motor_vehicle_id = "citadine"
```
Cette section est facultative pour préserver le fonctionnement de l'interface Web sur les configurations existantes. Si elle est présente, elle doit être complète et référencer un type de journée, un trajet et un véhicule à moteur existants ; l'application refuse alors de démarrer en cas d'erreur. La configuration est accessible via `get_home_assistant_config()` dans `app/config_loader.py`.
*Note : Ces données sont chargées en mémoire au démarrage de l'application dans `app.config["TOML"]`.*
Le secret de la future API ne doit pas être ajouté au fichier TOML. Il proviendra de la variable d'environnement `WORKLOG_API_TOKEN` lorsqu'elle sera implémentée ; cette étape ne le stocke ni ne le gère.
### SQLite / `instance/worklog.db` (Données dynamiques)
La base de données stocke l'activité saisie par l'utilisateur :
- **`work_entries`** : Une ligne par jour saisi (date, type de jour, ID du trajet, ID du véhicule, commentaire, timestamps).
- **`time_slots`** : Les plages horaires travaillées associées à une journée (heure de début, heure de fin, ID de l'entrée).
- **`leave_balance`** : Les quotas annuels de congés et de RTT (année, total congés, total RTT).
- **`workplace_presence_events`** : Les événements reçus de Home Assistant, avec une clé d'idempotence unique et leurs liens vers une journée et, facultativement, une plage horaire.
Les événements de présence stockent `received_at` comme un timestamp UTC naïf,
conformément à la convention existante des métadonnées. `occurred_at` est différent :
il représente une heure locale Europe/Paris naïve, et `local_date` est le jour local
qui en est dérivé. Une arrivée non encore rattachée à une plage est identifiée par
`event_type = "arrival"` et `time_slot_id IS NULL`; la signification de
`processed_at` sera précisée par le service métier de l'étape suivante.
---

View File

@@ -0,0 +1,266 @@
# Plan d'implémentation : API REST Home Assistant
## Objectif
Ajouter une API REST privée permettant à Home Assistant d'enregistrer automatiquement
les arrivées et les départs dans la zone `lieu de travail`, via `rest_command` et un
tracker `person`.
L'API doit créer automatiquement une journée de type `WORK`, utiliser par défaut le
profil de trajet `moteur_seul` et le véhicule `citadine` (Twingo ZE dans le fichier
de configuration), tout en laissant la modification complète de la journée dans
l'interface Web.
Le fuseau métier est `Europe/Paris`. Plusieurs plages horaires dans une même journée
doivent être supportées, comme dans l'interface existante.
## Décisions fonctionnelles
- L'arrivée crée la journée si elle n'existe pas, avec `day_type = "WORK"`.
- Le profil de trajet par défaut est `moteur_seul`.
- Le véhicule par défaut est `citadine`.
- Les journées `FORMATION` et `GARDE` restent compatibles avec le lieu de travail ;
l'API ne bloque donc pas ces types lorsqu'une journée existe déjà.
- Plusieurs couples arrivée/départ sont autorisés dans la même journée.
- Une arrivée ouverte ne doit pas être stockée dans `TimeSlot`, car le modèle actuel
exige un `end_time` et `WorkEntry.total_minutes()` suppose une plage complète.
- Une table d'événements de présence conserve les arrivées ouvertes et les événements
traités ; un `TimeSlot` est créé uniquement lorsqu'un départ complète une arrivée.
- Les utilisateurs continuent à corriger les journées et le mode de transport depuis
l'interface Web existante.
- L'import CSV direct en base n'est pas modifié dans cette livraison.
## Contrat HTTP cible
### Endpoint
```text
POST /api/v1/workplace-presence
```
### Requête
```json
{
"event": "arrival",
"occurred_at": "2026-08-13T08:23:10+02:00",
"idempotency_key": "person.telephone.arrival.20260813T082310+0200"
}
```
`event` vaut `arrival` ou `departure`. `occurred_at` est un timestamp ISO 8601
avec offset obligatoire. La date et l'heure enregistrées dans le modèle sont
calculées dans `Europe/Paris`.
La clé d'idempotence est obligatoire et doit être limitée en taille. Elle est
également acceptée dans `X-Idempotency-Key`; l'implémentation doit refuser une
valeur absente ou contradictoire entre l'en-tête et le JSON.
### Réponses
- `201 Created` pour une arrivée qui crée un nouvel événement.
- `200 OK` pour un départ qui complète une plage ou pour une requête idempotente
déjà traitée.
- `400` ou `422` pour un JSON ou une valeur métier invalide.
- `401` pour un token Bearer absent ou invalide.
- `409` pour un départ sans arrivée ouverte ou une transition incohérente.
- `415` si le contenu n'est pas `application/json`.
- `413` si le corps dépasse la limite configurée.
- `429` si la limite de débit est dépassée.
Toutes les erreurs sont des objets JSON homogènes, sans traceback ni détail interne.
Les réponses API indiquent `Cache-Control: no-store`.
## Étapes d'implémentation
### 1. Formaliser la configuration
Ajouter une section dédiée dans `config.toml` :
```toml
[home_assistant]
timezone = "Europe/Paris"
default_day_type = "WORK"
default_journey_profile_id = "moteur_seul"
default_motor_vehicle_id = "citadine"
```
Étendre `app/config_loader.py` avec un accès validé à cette configuration. Vérifier
au démarrage que le type de journée, le profil et le véhicule existent et que le
véhicule par défaut est bien un véhicule à moteur. Documenter que le token API ne
doit pas être placé dans TOML, mais fourni par `WORKLOG_API_TOKEN`.
### 2. Ajouter la persistance des événements de présence
Créer un modèle, par exemple `WorkplacePresenceEvent`, contenant au minimum :
- un identifiant primaire ;
- une clé d'idempotence unique ;
- le type (`arrival` ou `departure`) ;
- le timestamp reçu et le timestamp converti dans `Europe/Paris` ;
- la date locale ;
- l'identifiant de `WorkEntry` ;
- l'identifiant de `TimeSlot` lorsque le départ a complété une plage ;
- les dates de création et de traitement.
Le modèle doit permettre de retrouver une arrivée ouverte pour une date et de
conserver la réponse logique d'une requête rejouée. Ajouter les index nécessaires
sur la date locale, l'entrée et la clé unique.
Adapter `_migrate_db` dans `app/__init__.py` pour créer la nouvelle table dans les
installations existantes, en vérifiant d'abord la présence de la base et des tables.
Ajouter aussi la couverture de la base vide. Respecter la contrainte du projet :
il n'y a pas d'Alembic et les changements de schéma sont manuels.
### 3. Factoriser le service métier
Créer un service sans dépendance Flask, dans `app/business/`, chargé de :
- convertir et valider un timestamp ISO 8601 ;
- déterminer la date et l'heure locales dans `Europe/Paris` ;
- créer ou retrouver une `WorkEntry` avec les valeurs par défaut ;
- ouvrir une présence à l'arrivée ;
- retrouver l'arrivée ouverte la plus ancienne ou la plus récente selon la règle
retenue et la fermer au départ ;
- créer un `TimeSlot` complet pour chaque couple ;
- accepter plusieurs plages dans la même journée ;
- refuser un départ sans arrivée ouverte ;
- appliquer l'idempotence dans la même transaction SQLAlchemy.
La règle de rattachement doit être explicite pour les événements autour de minuit.
La date locale de l'arrivée ouvre la plage ; le départ doit être rattaché à cette
arrivée ouverte, même si son heure locale est le lendemain. Vérifier que cette
plage reste compatible avec le calcul existant du passage de minuit.
Réutiliser autant que possible les validations communes avec `entries.py`. Ne pas
faire dépendre le service des messages Flash, des redirections ou des templates.
### 4. Implémenter le blueprint API
Créer `app/routes/api.py` ou `app/routes/api/` et enregistrer le blueprint dans
la factory Flask sous `/api/v1`.
Implémenter :
- validation stricte du content type et du JSON ;
- authentification `Authorization: Bearer ...` ;
- comparaison du secret en temps constant ;
- contrôle de la clé d'idempotence ;
- appel du service métier ;
- sérialisation JSON stable ;
- traduction des erreurs métier en statuts HTTP ;
- gestion générique des erreurs inattendues sans fuite d'informations.
Limiter l'API à `POST` pour cette première version. Ajouter `405` pour les autres
méthodes et ne pas activer CORS, qui n'est pas nécessaire pour Home Assistant.
Configurer une taille maximale de requête adaptée à ce payload et prévoir un
rate limiting simple. Si aucune dépendance n'est souhaitable, une protection
minimale par token et fenêtre temporelle peut être implémentée ; sinon sélectionner
une dépendance légère et maintenue après vérification des contraintes de production.
### 5. Préserver le comportement Web
Vérifier que l'interface existante continue à afficher et modifier les `TimeSlot`
complets créés par l'API. Une journée créée par une arrivée doit être éditable
même avant le départ, avec zéro plage complète à ce stade.
Vérifier notamment que l'édition Web peut :
- changer `WORK` en `FORMATION` ou `GARDE` ;
- modifier le profil de trajet ;
- modifier le véhicule, notamment abandonner la Twingo par défaut ;
- ajouter, supprimer ou corriger plusieurs plages.
### 6. Documenter l'API séparément
Créer `docs/api.md` avec :
- le but et le périmètre de l'API ;
- l'URL, le contrat JSON et l'authentification ;
- les exemples de réponses `200`, `201`, `401`, `409` et erreurs de validation ;
- la sémantique d'idempotence ;
- la gestion du fuseau `Europe/Paris` ;
- le comportement autour de minuit ;
- les règles de déploiement HTTPS/HAProxy ;
- les commandes `curl` de test, sans secret en clair dans la documentation.
Ne jamais écrire de valeur réelle de `WORKLOG_API_TOKEN` dans ce fichier.
### 7. Documenter Home Assistant
Créer `docs/home-assistant.md` ou une section dédiée dans `docs/api.md` contenant
un exemple complet avec le tracker `person` :
- stockage du token dans `secrets.yaml` ;
- définition de `rest_command` en JSON ;
- automatisation d'arrivée lorsque `person.<nom>` passe à `lieu_de_travail` ;
- automatisation de départ lorsque l'état quitte cette zone ;
- génération d'une clé d'idempotence déterministe ;
- contrôle de `response_variable` et traitement des statuts non `200/201` ;
- avertissement sur les traces et l'accès administrateur Home Assistant aux secrets.
L'exemple doit conserver `verify_ssl: true` et utiliser l'en-tête Bearer.
### 8. Tester et vérifier
Ajouter des tests de routes et de service couvrant au minimum :
- arrivée authentifiée créant une journée `WORK` avec `moteur_seul` et `citadine` ;
- départ complétant la première plage ;
- deuxième arrivée et deuxième départ le même jour ;
- journée `FORMATION` ou `GARDE` existante ;
- départ sans arrivée ouverte ;
- timestamp avec offset et conversion `Europe/Paris` ;
- passage de minuit ;
- clé rejouée sans duplication ;
- clé réutilisée avec un contenu différent ;
- token absent ou invalide ;
- content type, JSON, champs et longueurs invalides ;
- limite de taille et rate limiting ;
- absence de secret ou de traceback dans les réponses ;
- non-régression des tests HTML existants.
Exécuter ensuite :
```bash
.venv/bin/python -m pytest
.venv/bin/ruff check .
.venv/bin/ruff format --check .
```
Tester manuellement depuis Home Assistant avec `rest_command` et vérifier la
réponse dans les traces d'automatisation, puis tester une répétition du même
événement.
## Hors périmètre et TODO ultérieure
### Import CSV via l'API
Ne pas modifier `scripts/import_csv.py` dans cette livraison. Prévoir une TODO
ultérieure pour faire passer l'import CSV par l'API plutôt que par des écritures
directes en base.
Cette évolution nécessitera probablement de nouveaux endpoints ou un contrat
d'import en lot, par exemple :
```text
POST /api/v1/work-entries
POST /api/v1/work-entries/bulk
```
Elle devra définir les règles de transaction, le comportement en cas de doublon,
les erreurs par ligne, l'idempotence d'un lot et une authentification adaptée à un
client local. Elle devra aussi réutiliser le service métier commun introduit par
la présente API.
## Sources de référence
- Home Assistant, `rest_command` : https://www.home-assistant.io/integrations/rest_command/
- Home Assistant, secrets : https://www.home-assistant.io/docs/configuration/secrets/
- OWASP REST Security Cheat Sheet : https://cheatsheetseries.owasp.org/cheatsheets/REST_Security_Cheat_Sheet.html
- OWASP API Security Top 10 : https://owasp.org/API-Security/editions/2023/en/0x11-t10/
- Flask, Web Security Considerations : https://flask.palletsprojects.com/en/stable/web-security/
La recherche a été réalisée avec des moyens Web alternatifs ; FireCrawl local
n'était pas disponible au moment de la préparation du plan.

View File

@@ -2,6 +2,7 @@ import pytest
from app import create_app
from app import db as _db
from tests.in_memory_db import IN_MEMORY_DATABASE_URI, in_memory_engine_options
@pytest.fixture
@@ -48,6 +49,12 @@ distances = { moteur = 14, velo = 8 }
name = "Vélo seul"
distances = { velo = 24 }
[home_assistant]
timezone = "Europe/Paris"
default_day_type = "WORK"
default_journey_profile_id = "moteur_seul"
default_motor_vehicle_id = "citadine"
[[bareme_kilometrique.2025.cv_5.tranches]]
km_max = 3000
taux = 0.548
@@ -66,9 +73,12 @@ forfait = 0
encoding="utf-8",
)
application = create_app(config_path=str(config_path))
application = create_app(
config_path=str(config_path),
database_uri=IN_MEMORY_DATABASE_URI,
engine_options=in_memory_engine_options(),
)
application.config["TESTING"] = True
application.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
with application.app_context():
_db.create_all()

20
tests/in_memory_db.py Normal file
View File

@@ -0,0 +1,20 @@
"""Configuration SQLite mémoire partagée entre fixtures et helpers de tests.
Ce module est volontairement indépendant de ``conftest`` afin d'être
importable de manière portable, y compris sous ``pytest --import-mode=importlib``
(où ``conftest`` n'est pas importable comme module ordinaire). Il centralise la
configuration mémoire partagée entre la fixture ``app`` et les helpers de tests
qui appellent directement ``create_app(...)``, évitant qu'une factory de test
initialise accidentellement ``instance/worklog.db``.
"""
from sqlalchemy.pool import StaticPool
IN_MEMORY_DATABASE_URI = "sqlite:///:memory:"
def in_memory_engine_options() -> dict[str, object]:
return {
"poolclass": StaticPool,
"connect_args": {"check_same_thread": False},
}

16
tests/test_app_factory.py Normal file
View File

@@ -0,0 +1,16 @@
import sqlalchemy as sa
from sqlalchemy.pool import StaticPool
from app import db
def test_app_fixture_uses_one_in_memory_database_connection(app):
with app.app_context():
assert str(db.engine.url) == "sqlite:///:memory:"
assert isinstance(db.engine.pool, StaticPool)
assert sa.inspect(db.engine).has_table("work_entries")
def test_sqlite_foreign_keys_are_enabled(app):
with app.app_context():
assert db.session.scalar(sa.text("PRAGMA foreign_keys")) == 1

View File

@@ -1,3 +1,39 @@
import pytest
from app import create_app
from tests.in_memory_db import IN_MEMORY_DATABASE_URI, in_memory_engine_options
_MINIMAL_CONFIG = """
[vehicles.citadine]
name = "Citadine"
type = "moteur"
[vehicles.velo]
name = "Vélo"
type = "velo"
[journeys.moteur_seul]
name = "Moteur seul"
distances = {{ moteur = 1 }}
[home_assistant]
timezone = "{timezone}"
default_day_type = "{day_type}"
default_journey_profile_id = "{journey_id}"
default_motor_vehicle_id = "{vehicle_id}"
"""
def _create_app_with_home_assistant_config(tmp_path, **values):
config_path = tmp_path / "config.toml"
config_path.write_text(_MINIMAL_CONFIG.format(**values), encoding="utf-8")
return create_app(
config_path=str(config_path),
database_uri=IN_MEMORY_DATABASE_URI,
engine_options=in_memory_engine_options(),
)
def test_get_vehicles_returns_configured_vehicles(app):
with app.app_context():
from app.config_loader import get_vehicles
@@ -59,3 +95,82 @@ def test_day_types_without_journey(app):
types = day_types_without_journey()
assert "TT" in types
assert "WORK" not in types
def test_get_home_assistant_config_returns_validated_defaults(app):
with app.app_context():
from app.config_loader import get_home_assistant_config
assert get_home_assistant_config() == {
"timezone": "Europe/Paris",
"default_day_type": "WORK",
"default_journey_profile_id": "moteur_seul",
"default_motor_vehicle_id": "citadine",
}
assert app.config["HOME_ASSISTANT"]["default_motor_vehicle_id"] == "citadine"
def test_home_assistant_section_absent_is_allowed(app):
with app.app_context():
from app.config_loader import get_home_assistant_config
app.config["TOML"].pop("home_assistant")
assert get_home_assistant_config() is None
@pytest.mark.parametrize(
("field", "value", "message"),
[
("timezone", "Mars/NoSuchPlace", "fuseau horaire"),
("day_type", "UNKNOWN", "type de journée"),
("journey_id", "unknown_journey", "trajet inconnu"),
("vehicle_id", "unknown_vehicle", "véhicule inconnu"),
],
)
def test_invalid_home_assistant_config_prevents_startup(tmp_path, field, value, message):
values = {
"timezone": "Europe/Paris",
"day_type": "WORK",
"journey_id": "moteur_seul",
"vehicle_id": "citadine",
}
values[field] = value
with pytest.raises(ValueError, match=message):
_create_app_with_home_assistant_config(tmp_path, **values)
def test_home_assistant_default_vehicle_must_be_motor_vehicle(tmp_path):
values = {
"timezone": "Europe/Paris",
"day_type": "WORK",
"journey_id": "moteur_seul",
"vehicle_id": "velo",
}
with pytest.raises(ValueError, match="véhicule moteur"):
_create_app_with_home_assistant_config(tmp_path, **values)
def test_incomplete_home_assistant_config_prevents_startup(tmp_path):
config_path = tmp_path / "config.toml"
config_path.write_text(
"""
[vehicles.citadine]
type = "moteur"
[journeys.moteur_seul]
distances = { moteur = 1 }
[home_assistant]
timezone = "Europe/Paris"
""",
encoding="utf-8",
)
with pytest.raises(ValueError, match="clé.*manquante"):
create_app(
config_path=str(config_path),
database_uri=IN_MEMORY_DATABASE_URI,
engine_options=in_memory_engine_options(),
)

111
tests/test_models.py Normal file
View File

@@ -0,0 +1,111 @@
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")

View File

@@ -0,0 +1,163 @@
from datetime import UTC, date, datetime
import pytest
from app import db
from app.business.presence_service import (
ArrivalAlreadyOpenError,
DepartureWithoutArrivalError,
IdempotencyConflictError,
InvalidPresenceEventError,
record_presence_event,
)
from app.models import TimeSlot, WorkEntry, WorkplacePresenceEvent
CONFIG = {
"timezone": "Europe/Paris",
"default_day_type": "WORK",
"default_journey_profile_id": "moteur_seul",
"default_motor_vehicle_id": "citadine",
}
RECEIVED = datetime(2026, 8, 13, 7, 0, tzinfo=UTC)
def call(session, event_type, occurred_at, key, **kwargs):
return record_presence_event(session, CONFIG, event_type, occurred_at, key, **kwargs)
def test_arrival_creates_entry_with_defaults_without_commit(app):
with app.app_context():
result = call(
db.session, "arrival", "2026-08-13T08:23:10+02:00", "a-1", received_at=RECEIVED
)
entry = db.session.get(WorkEntry, result.entry_id)
assert result.replayed is False
assert result.slot_state == "open"
assert (entry.day_type, entry.journey_profile_id, entry.motor_vehicle_id) == (
"WORK",
"moteur_seul",
"citadine",
)
assert db.session.query(WorkplacePresenceEvent).count() == 1
db.session.rollback()
assert db.session.query(WorkplacePresenceEvent).count() == 0
def test_departure_completes_both_events(app):
with app.app_context():
arrival = call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "a-1")
departure = call(db.session, "departure", "2026-08-13T17:00:00+02:00", "d-1")
events = db.session.scalars(
db.select(WorkplacePresenceEvent).order_by(WorkplacePresenceEvent.id)
).all()
assert departure.time_slot_id is not None
assert departure.entry_id == arrival.entry_id
assert all(event.time_slot_id == departure.time_slot_id for event in events)
assert all(event.processed_at is not None for event in events)
assert db.session.get(TimeSlot, departure.time_slot_id).start_time.hour == 8
def test_second_pair_same_day_is_supported(app):
with app.app_context():
first = call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "a-1")
call(db.session, "departure", "2026-08-13T12:00:00+02:00", "d-1")
call(db.session, "arrival", "2026-08-13T13:00:00+02:00", "a-2")
second = call(db.session, "departure", "2026-08-13T17:00:00+02:00", "d-2")
assert second.entry_id == first.entry_id
assert db.session.query(TimeSlot).count() == 2
def test_arrival_is_global_across_days_and_does_not_create_entry(app):
with app.app_context():
call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "a-1")
with pytest.raises(ArrivalAlreadyOpenError):
call(db.session, "arrival", "2026-08-14T08:00:00+02:00", "a-2")
assert db.session.query(WorkEntry).count() == 1
assert db.session.query(WorkplacePresenceEvent).count() == 1
assert db.session.query(TimeSlot).count() == 0
@pytest.mark.parametrize(
"departure_at",
["2026-08-13T08:00:00+02:00", "2026-08-13T07:59:59+02:00"],
ids=["equal", "before"],
)
def test_invalid_departure_does_not_create_slot_or_event(app, departure_at):
with app.app_context():
call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "a-1")
with pytest.raises(InvalidPresenceEventError):
call(db.session, "departure", departure_at, "d-1")
assert db.session.query(TimeSlot).count() == 0
assert db.session.query(WorkplacePresenceEvent).count() == 1
assert (
db.session.query(WorkplacePresenceEvent).filter_by(event_type="departure").count() == 0
)
@pytest.mark.parametrize("day_type", ["FORMATION", "GARDE"])
def test_existing_special_day_is_not_modified(app, day_type):
with app.app_context():
entry = WorkEntry(
date=date(2026, 8, 13),
day_type=day_type,
journey_profile_id="other-journey",
motor_vehicle_id="other-car",
)
db.session.add(entry)
db.session.flush()
result = call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "a-1")
db.session.refresh(entry)
assert result.entry_id == entry.id
assert (entry.day_type, entry.journey_profile_id, entry.motor_vehicle_id) == (
day_type,
"other-journey",
"other-car",
)
def test_invalid_transitions_and_payloads(app):
with app.app_context():
with pytest.raises(ArrivalAlreadyOpenError):
call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "a-1")
call(db.session, "arrival", "2026-08-13T09:00:00+02:00", "a-2")
db.session.rollback()
with pytest.raises(DepartureWithoutArrivalError):
call(db.session, "departure", "2026-08-13T09:00:00+02:00", "d-1")
for event_type, occurred_at, key in (
("other", "2026-08-13T09:00:00+02:00", "x"),
("arrival", "2026-08-13T09:00:00", "x"),
("arrival", "not-a-date", "x"),
("arrival", "2026-08-13T09:00:00+02:00", ""),
):
with pytest.raises(InvalidPresenceEventError):
call(db.session, event_type, occurred_at, key)
def test_departure_after_midnight_is_accepted_as_a_later_instant(app):
with app.app_context():
arrival = call(db.session, "arrival", "2026-08-13T23:30:00+00:00", "a-1")
departure = call(db.session, "departure", "2026-08-14T00:30:00+00:00", "d-1")
event = db.session.get(WorkplacePresenceEvent, arrival.event_id)
slot = db.session.get(TimeSlot, departure.time_slot_id)
assert event.occurred_at == datetime(2026, 8, 14, 1, 30)
assert event.local_date == date(2026, 8, 14)
assert (slot.start_time.hour, slot.end_time.hour) == (1, 2)
def test_replay_and_key_conflict(app):
with app.app_context():
first = call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "same")
replay = call(db.session, "arrival", "2026-08-13T08:00:00+02:00", "same")
assert replay.replayed is True
assert replay.event_id == first.event_id
with pytest.raises(IdempotencyConflictError):
call(db.session, "arrival", "2026-08-13T08:01:00+02:00", "same")