Compare commits
3 Commits
525d38224c
...
e9f73c7562
| Author | SHA1 | Date | |
|---|---|---|---|
| e9f73c7562 | |||
| c30bd1c0c5 | |||
| 85e6e502e5 |
@@ -15,6 +15,7 @@ Architecture et composants clés :
|
|||||||
|
|
||||||
import os
|
import os
|
||||||
import tomllib
|
import tomllib
|
||||||
|
from collections.abc import Mapping
|
||||||
|
|
||||||
import sqlalchemy as sa
|
import sqlalchemy as sa
|
||||||
from flask import Flask
|
from flask import Flask
|
||||||
@@ -23,6 +24,15 @@ from flask_sqlalchemy import SQLAlchemy
|
|||||||
db = 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):
|
def _migrate_db(app):
|
||||||
"""Applique les migrations de schéma manquantes de manière incrémentale (sans Alembic).
|
"""Applique les migrations de schéma manquantes de manière incrémentale (sans Alembic).
|
||||||
|
|
||||||
@@ -36,6 +46,10 @@ def _migrate_db(app):
|
|||||||
"""
|
"""
|
||||||
import sqlite3
|
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")
|
db_path = os.path.join(app.instance_path, "worklog.db")
|
||||||
if not os.path.exists(db_path):
|
if not os.path.exists(db_path):
|
||||||
return # Nouvelle DB, create_all() s'en charge
|
return # Nouvelle DB, create_all() s'en charge
|
||||||
@@ -120,7 +134,12 @@ def _date_fr(d):
|
|||||||
return f"{jour} {d.day} {mois} {d.year}"
|
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.
|
"""Factory de création et de configuration de l'application Flask.
|
||||||
|
|
||||||
Cette fonction réalise les étapes suivantes :
|
Cette fonction réalise les étapes suivantes :
|
||||||
@@ -136,6 +155,11 @@ def create_app(config_path=None):
|
|||||||
Paramètres:
|
Paramètres:
|
||||||
config_path (str | None): Chemin optionnel vers le fichier de configuration TOML.
|
config_path (str | None): Chemin optionnel vers le fichier de configuration TOML.
|
||||||
Par défaut, cherche `config.toml` à la racine du projet.
|
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:
|
Retourne:
|
||||||
Flask: L'instance de l'application Flask configurée et prête à l'emploi.
|
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)
|
os.makedirs(app.instance_path, exist_ok=True)
|
||||||
|
|
||||||
app.config["SQLALCHEMY_DATABASE_URI"] = (
|
if database_uri is None:
|
||||||
f"sqlite:///{os.path.join(app.instance_path, 'worklog.db')}"
|
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["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||||
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev-secret-change-in-prod")
|
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:
|
else:
|
||||||
app.config["TOML"] = {}
|
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)
|
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["date_fr"] = _date_fr
|
||||||
app.jinja_env.filters["day_type_fr"] = _day_type_fr
|
app.jinja_env.filters["day_type_fr"] = _day_type_fr
|
||||||
|
|
||||||
|
|||||||
@@ -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.
|
- 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
|
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():
|
def get_vehicles():
|
||||||
"""Récupère l'ensemble des véhicules configurés dans le fichier TOML.
|
"""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", {})
|
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:
|
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.
|
"""Vérifie si un profil de trajet donné inclut une distance pour véhicule à moteur.
|
||||||
|
|
||||||
|
|||||||
@@ -37,6 +37,9 @@ class WorkEntry(db.Model):
|
|||||||
time_slots: so.Mapped[list["TimeSlot"]] = so.relationship(
|
time_slots: so.Mapped[list["TimeSlot"]] = so.relationship(
|
||||||
back_populates="entry", cascade="all, delete-orphan", order_by="TimeSlot.start_time"
|
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:
|
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)
|
end_time: so.Mapped[time] = so.mapped_column(sa.Time, nullable=False)
|
||||||
|
|
||||||
entry: so.Mapped["WorkEntry"] = so.relationship(back_populates="time_slots")
|
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):
|
class LeaveBalance(db.Model):
|
||||||
|
|||||||
@@ -37,6 +37,12 @@ distances = { moteur = 14, velo = 8 }
|
|||||||
name = "Vélo seul"
|
name = "Vélo seul"
|
||||||
distances = { velo = 24 }
|
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) ---
|
# --- Barème kilométrique voitures 2025 (revenus 2024) ---
|
||||||
# Source : https://www.service-public.gouv.fr/particuliers/actualites/A14686
|
# Source : https://www.service-public.gouv.fr/particuliers/actualites/A14686
|
||||||
# Majoration +20% pour véhicules électriques gérée dans travel_calc.py
|
# Majoration +20% pour véhicules électriques gérée dans travel_calc.py
|
||||||
|
|||||||
@@ -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).
|
- **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.
|
- **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).
|
- **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"]`.*
|
*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)
|
### SQLite / `instance/worklog.db` (Données dynamiques)
|
||||||
La base de données stocke l'activité saisie par l'utilisateur :
|
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).
|
- **`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).
|
- **`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).
|
- **`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.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ import pytest
|
|||||||
|
|
||||||
from app import create_app
|
from app import create_app
|
||||||
from app import db as _db
|
from app import db as _db
|
||||||
|
from tests.in_memory_db import IN_MEMORY_DATABASE_URI, in_memory_engine_options
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
@@ -48,6 +49,12 @@ distances = { moteur = 14, velo = 8 }
|
|||||||
name = "Vélo seul"
|
name = "Vélo seul"
|
||||||
distances = { velo = 24 }
|
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]]
|
[[bareme_kilometrique.2025.cv_5.tranches]]
|
||||||
km_max = 3000
|
km_max = 3000
|
||||||
taux = 0.548
|
taux = 0.548
|
||||||
@@ -66,9 +73,12 @@ forfait = 0
|
|||||||
encoding="utf-8",
|
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["TESTING"] = True
|
||||||
application.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
|
|
||||||
|
|
||||||
with application.app_context():
|
with application.app_context():
|
||||||
_db.create_all()
|
_db.create_all()
|
||||||
|
|||||||
20
tests/in_memory_db.py
Normal file
20
tests/in_memory_db.py
Normal 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
16
tests/test_app_factory.py
Normal 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
|
||||||
@@ -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):
|
def test_get_vehicles_returns_configured_vehicles(app):
|
||||||
with app.app_context():
|
with app.app_context():
|
||||||
from app.config_loader import get_vehicles
|
from app.config_loader import get_vehicles
|
||||||
@@ -59,3 +95,82 @@ def test_day_types_without_journey(app):
|
|||||||
types = day_types_without_journey()
|
types = day_types_without_journey()
|
||||||
assert "TT" in types
|
assert "TT" in types
|
||||||
assert "WORK" not 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
111
tests/test_models.py
Normal 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")
|
||||||
Reference in New Issue
Block a user