Compare commits

..
1 Commits
Author SHA1 Message Date
Laurent Chedanne b5c84635cf Code changes for sql-to-api branch 2026-05-29 10:40:34 +02:00
11 changed files with 82 additions and 1091 deletions

No files matched your search

+1 -3
View File
@@ -1,5 +1,3 @@
env/*
config.json
config.json.*
csv/*
logs/
config.json.orig
-28
View File
@@ -1,28 +0,0 @@
# Consignes de maintenance
- Mettre à jour `docs/crealead-caldav.md` lors de toute évolution impactant
`crealead-caldav.py`, notamment ses pré-requis, sa configuration, son
lancement, le traitement des événements, ses sorties CSV, sa journalisation
ou ses tests dans `tests/`.
- Garder les exemples de configuration, les commandes d'exécution et les cas
de test documentés cohérents avec le comportement du script.
- Chaque nouvelle ligne écrite dans `logs/crealead-caldav.log` doit commencer
par l'heure d'écriture en UTC au format ISO 8601 avec suffixe `Z`, entre
crochets. Le format par défaut est :
`[2026-10-15T12:34:56Z]: [info] - message`.
Cela s'applique aux sorties standard et d'erreur ainsi qu'aux lignes des
traces d'exception.
- Tout événement non exporté doit laisser une trace au format
`[2026-10-15T12:34:56Z]: [warning][event] - message`, sans préfixe `[info]`
supplémentaire. Préciser le calendrier concerné, le titre, le moment de
l'événement (début et fin disponibles) et la raison de son exclusion.
Ajouter le fichier CSV et l'UID pour faciliter son identification.
- Ne jamais ignorer silencieusement un événement, notamment en cas de
`SUMMARY` absent, de dates absentes ou invalides, de document iCalendar
illisible ou d'export CSV interrompu. Récupérer les métadonnées disponibles
dans les données brutes si nécessaire et indiquer `indisponible` pour celles
qui ne peuvent pas être retrouvées. Échapper les retours à la ligne des
métadonnées et du motif pour conserver un avertissement sur une seule ligne.
- Lors d'une évolution du traitement des événements ou de la journalisation,
vérifier dans les tests que les événements écartés sont bien tracés avec les
informations et les préfixes ci-dessus.
+46 -212
View File
@@ -17,143 +17,9 @@ import sys
import caldav
import json
import csv
from datetime import date, datetime, time, timezone
from icalendar import Calendar
import calendar
from datetime import date, datetime, timezone
import pytz
from urllib.parse import unquote
import re
import traceback
from contextlib import contextmanager, redirect_stdout, redirect_stderr
from pathlib import Path
def utc_timestamp():
return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
class TimestampedLog:
"""Prefix each log line, including lines written in several fragments."""
def __init__(self, log_file):
self.log_file = log_file
self.line_start = True
def write(self, value):
# Keep an event warning separate from an unfinished console message.
if value.startswith('[warning][event] - ') and not self.line_start:
self.log_file.write('\n')
self.line_start = True
for fragment in value.splitlines(keepends=True):
if self.line_start:
self.log_file.write('[{}]: '.format(utc_timestamp()))
if not fragment.startswith('[warning][event] - '):
self.log_file.write('[info] - ')
self.log_file.write(fragment)
self.line_start = fragment.endswith(('\n', '\r'))
return len(value)
def flush(self):
self.log_file.flush()
class TeeOutput:
"""Write to the original console stream and to the shared log file."""
def __init__(self, stream, log_file):
self.stream = stream
self.log_file = log_file
def write(self, value):
written = self.stream.write(value)
self.log_file.write(value)
self.log_file.flush()
return written
def flush(self):
self.stream.flush()
self.log_file.flush()
def __getattr__(self, name):
return getattr(self.stream, name)
@contextmanager
def log_output():
"""Append console output to logs/<script name>.log next to the script."""
script_path = Path(__file__).resolve()
log_path = script_path.parent / 'logs' / script_path.with_suffix('.log').name
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open('a', encoding='utf-8') as log_file:
log_writer = TimestampedLog(log_file)
with redirect_stdout(TeeOutput(sys.stdout, log_writer)), \
redirect_stderr(TeeOutput(sys.stderr, log_writer)):
try:
yield
except BaseException as error:
# The interpreter prints the traceback after the streams are restored.
# Save a copy here while the log is still open, without duplicating it
# in the console. SystemExit normally has no traceback.
if not log_writer.line_start:
log_writer.write('\n')
if isinstance(error, SystemExit):
if error.code is not None and not isinstance(error.code, int):
log_writer.write(str(error.code) + '\n')
else:
traceback.print_exc(file=log_writer)
raise
finally:
if not log_writer.line_start:
log_writer.write('\n')
def recover_event_contexts(data):
"""Recover identifying fields even when the iCalendar parser rejects data."""
if isinstance(data, bytes):
data = data.decode('utf-8', errors='replace')
unfolded = re.sub(r'\r?\n[ \t]', '', str(data))
blocks = re.split(r'(?im)^BEGIN:VEVENT\s*\r?$', unfolded)
contexts = []
for block in blocks[1:] or [unfolded]:
context = {}
depth = 0
for line in block.splitlines():
name, separator, value = line.partition(':')
name = name.split(';', 1)[0].upper()
if name == 'BEGIN':
depth += 1
elif name == 'END':
if depth == 0:
break
depth -= 1
elif depth == 0 and separator and name in ('SUMMARY', 'DTSTART', 'DTEND', 'UID'):
context.setdefault(name, value)
contexts.append(context)
return contexts
def log_skipped_event(calendar_find, component, reason, fallback=None):
"""Every rejected event must have an identifiable warning in the log."""
fallback = fallback or {}
def field(name):
value = component.get(name)
if value is None:
return str(fallback.get(name, 'indisponible'))
decoded = getattr(value, 'dt', None)
if isinstance(decoded, datetime):
return local_datetime(decoded).astimezone(timezone.utc).isoformat().replace('+00:00', 'Z')
if isinstance(decoded, date):
return decoded.isoformat()
return str(value)
calendar_name = getattr(calendar_find.get('calendar_server'), 'name', None)
calendar_name = calendar_name or calendar_find.get('calendar_name', 'indisponible')
# repr keeps newlines in event metadata from breaking the warning format.
print('[warning][event] - Calendrier={!r} ; CSV={!r} ; titre={!r} ; '
'début={!r} ; fin={!r} ; UID={!r} ; événement non exporté : {}'.format(
str(calendar_name), calendar_find['csv_name'], field('SUMMARY'),
field('DTSTART'), field('DTEND'), field('UID'),
str(reason).replace('\r', '\\r').replace('\n', '\\n')), file=sys.stderr)
class EventRow():
@@ -167,7 +33,6 @@ class EventRow():
self.categories = ''
self.location = ''
self.description = ''
self.event_context = {}
def set_title(self, title):
self.title = title
@@ -245,18 +110,21 @@ def link_calendars(config, calendars_server):
return calendars_find
def local_datetime(value):
"""Convert an iCalendar date or datetime to Europe/Paris."""
def split_data(data):
"""Return string splitted by ';' separator"""
return data.split(':')
def datetime_format(date_str):
"""Return date formated"""
try: # Nextcloud Format
date_formated = datetime.strptime(date_str, '%Y%m%dT%H%M%SZ')
except: # Zimbra format
date_formated = datetime.strptime(date_str, '%Y%m%d')
paris_tz = pytz.timezone('Europe/Paris')
if isinstance(value, datetime):
result = value
elif isinstance(value, date):
result = datetime.combine(value, time.min)
else:
raise ValueError("date absente ou invalide")
if result.tzinfo is None:
return paris_tz.localize(result)
return result.astimezone(paris_tz)
date_formated = date_formated.replace(tzinfo=timezone.utc).astimezone(paris_tz)
#date_formated = date_formated.astimezone(paris_tz)
return date_formated
def get_all_events(config, calendars_find):
@@ -273,66 +141,44 @@ def get_all_events(config, calendars_find):
event_rows = []
for event in events_fetched:
contexts = recover_event_contexts(event.data)
# Parse parameters (TZID, VALUE=DATE), folded lines and escaped text.
try:
event_calendar = Calendar.from_ical(event.data)
except (ValueError, TypeError) as error:
for context in contexts:
log_skipped_event(calendar_find, {},
'document iCalendar illisible ({})'.format(error), context)
continue
event_fields = list(map(split_data, event.data.splitlines()))
components = event_calendar.walk('VEVENT')
if not components:
log_skipped_event(calendar_find, {}, 'aucun composant VEVENT', contexts[0])
for index, component in enumerate(components):
context = contexts[index] if index < len(contexts) else {}
# Fix problem #2023022110004936: ignore events without a summary.
if 'SUMMARY' not in component:
log_skipped_event(calendar_find, component, 'propriété SUMMARY absente', context)
continue
try:
start = component.decoded('DTSTART')
end = component.decoded('DTEND', None)
if end is None and 'DURATION' in component:
end = start + component.decoded('DURATION')
start = local_datetime(start)
end = local_datetime(end)
except (KeyError, ValueError, TypeError) as error:
log_skipped_event(calendar_find, component,
'dates de début/fin absentes ou invalides ({})'.format(error), context)
# Fix problem #2023022110004936
has = False
for sub in event_fields:
if 'SUMMARY' in sub:
has = True
break
if not has:
continue
event_row = EventRow()
event_row.event_context = component
event_row.set_title(str(component['SUMMARY']))
event_row.set_start(start)
event_row.set_end(end)
categories = component.get('CATEGORIES', [])
if not isinstance(categories, list):
categories = [categories]
event_row.set_categories(','.join(
str(category) for group in categories for category in group.cats))
event_row.set_location(str(component.get('LOCATION', '')))
event_row.set_description(clean_description(str(component.get('DESCRIPTION', ''))))
for field in ['SUMMARY', 'DTSTART', 'DTEND', 'CATEGORIES', 'LOCATION', 'DESCRIPTION']:
for event_field in event_fields:
if field in event_field[0]:
if field == "SUMMARY":
event_row.set_title(event_field[1])
elif field == "DTSTART":
date = datetime_format(event_field[1])
event_row.set_start(date)
elif field == "DTEND":
date = datetime_format(event_field[1])
event_row.set_end(date)
elif field == "CATEGORIES":
event_row.set_categories(event_field[1])
elif field == "LOCATION":
event_row.set_location(event_field[1])
elif field == "DESCRIPTION":
event_row.set_description(event_field[1])
event_rows.append(event_row)
sorted_rows = sorted(event_rows, key=lambda row: (row.start_date, row.start_time))
sorted_rows = sorted(event_rows, key=lambda l: datetime.strptime(str(l.start_date)+'T'+str(l.start_time), "%Y-%m-%dT%H:%M:%S"))
calendars_to_write.append({'csv_name' : calendar_find['csv_name'],
'calendar' : calendar_find,
'line' : sorted_rows})
return calendars_to_write
def clean_description(value):
if value.startswith("text/html,"):
value = value[len("text/html,"):]
value = unquote(value)
value = re.sub(r'<[^>]+>', '', value)
return value.strip().strip('"')
def write_csv(filename, rows):
"""Write CSV file"""
@@ -344,7 +190,8 @@ def write_csv(filename, rows):
writer.writerow(row.get_csv_row())
def main():
if __name__ == "__main__":
# Load config
config = load_json_config("./config.json")
@@ -358,18 +205,5 @@ def main():
calendars_to_write = get_all_events(config, calendars_find)
# Write CSV
for index, calendar_to_write in enumerate(calendars_to_write):
try:
for calendar_to_write in calendars_to_write:
write_csv(calendar_to_write['csv_name'], calendar_to_write['line'])
except Exception as error:
# The current CSV may be incomplete; later calendars were not written.
for pending in calendars_to_write[index:]:
for row in pending['line']:
log_skipped_event(pending['calendar'], row.event_context,
'export CSV interrompu, écriture non garantie ({})'.format(error))
raise
if __name__ == "__main__":
with log_output():
main()
+35 -8
View File
@@ -139,10 +139,13 @@ def get_client_uuid_by_name(client_name, user_id):
query = ("select uuid from oc_timemanager_client where name=\""+client_name+"\" AND user_id=\""+user_id+"\"")
cursor.execute(query)
client_uuid = cursor.fetchone()[0]
row = cursor.fetchone()
save_db()
return client_uuid.decode('utf-8')
if not row:
return None
return row[0].decode('utf-8')
def get_project_uuid_by_name(project_name, client_uuid, user_id):
@@ -264,7 +267,6 @@ def create_project(name, client_uuid):
#print(json.dumps(response.json(), indent=1))
else:
print("Erreur de création project")
print(response)
print(response.reason)
def create_client(name):
@@ -315,12 +317,25 @@ def create_client(name):
# Vérification de la réponse
if response.status_code == 200:
pass
#print("Client créé avec succès.")
#print(json.dumps(response.json(), indent=1))
try:
resp = response.json()
# Tenter d'extraire l'UUID renvoyé par l'API (structure attendue)
client_uuid = None
if isinstance(resp, dict):
client_created = resp.get('data', {}).get('clients', {}).get('created', [])
if client_created and isinstance(client_created, list):
client_uuid = client_created[0].get('uuid')
if client_uuid:
return client_uuid
# si pas d'uuid dans la réponse, retourner None (le caller gérera)
return None
except Exception:
return None
else:
print("Erreur de création client")
print(response.reason)
return None
def insert_task_time(name, project_name, client_name, task_uuid, user_id, dstart, hstart, dend, hend):
@@ -423,11 +438,23 @@ for user_info in config['calendars']:
continue
# Description => Client
client_uuid = None
if not client_exist_by_name(client_name, user_id):
print("Création du client {}".format(client_name))
create_client(client_name)
created_uuid = create_client(client_name)
if created_uuid:
client_uuid = created_uuid
else:
# tentative de récupération depuis la base au cas où l'API a créé l'entrée
client_uuid = get_client_uuid_by_name(client_name, user_id)
if not client_uuid:
print("Erreur: impossible d'obtenir l'UUID du client '{}' après création, saut.".format(client_name))
continue
else:
client_uuid = get_client_uuid_by_name(client_name, user_id)
if not client_uuid:
print("Erreur: le client '{}' existe mais son UUID est introuvable, saut.".format(client_name))
continue
# Categories => Project
if not project_exist_by_name(project_name, client_uuid, user_id):
-235
View File
@@ -1,235 +0,0 @@
# Export CalDAV vers CSV
Le script [`crealead-caldav.py`](../crealead-caldav.py) récupère des événements
d'agendas accessibles en CalDAV, notamment Nextcloud, et produit un fichier CSV
par calendrier configuré. Ces fichiers peuvent ensuite être utilisés par
`csv-to-timemanager.py`.
## Pré-requis
- Python 3, avec `pip` et le module `venv`. L'environnement Docker du dépôt
utilise Python 3.9.2.
- Les dépendances de [`requirements.txt`](../requirements.txt), notamment
`caldav`, `icalendar` et `pytz`.
- Un serveur CalDAV accessible depuis la machine d'exécution et un compte
autorisé à lire les calendriers à exporter.
- Des droits de lecture sur `config.json` et d'écriture sur les fichiers CSV
ainsi que sur le dossier `logs/`, situé à côté du script.
- Bash pour utiliser le lanceur `exec-export-caldav.sh`.
Depuis la racine du dépôt, préparer l'environnement :
```bash
python3 -m venv env
source env/bin/activate
python -m pip install -r requirements.txt
```
L'environnement Docker fourni constitue une autre façon d'exécuter le script
et les tests. Sa construction est décrite dans
[`support/docker-dev/README.md`](../support/docker-dev/README.md).
## Usage
### Configuration minimale
Créer un fichier `config.json` à la racine du dépôt. Exemple pour exporter les
événements d'octobre 2026 :
```json
{
"caldav_url": "https://cloud.example.com/remote.php/dav/",
"username": "utilisateur",
"password": "mot-de-passe-application",
"year_start": 2026,
"month_start": 10,
"day_start": 1,
"year_end": 2026,
"month_end": 11,
"day_end": 1,
"calendars": [
{
"calendar_name": "Temps de travail",
"csv_name": "csv/utilisateur.csv"
}
]
}
```
| Champ | Rôle |
| --- | --- |
| `caldav_url` | URL du serveur CalDAV. |
| `username`, `password` | Identifiants du compte qui peut lire les calendriers. |
| `year_start`, `month_start`, `day_start` | Date de début de la recherche, à minuit. Les valeurs sont des nombres entiers. |
| `year_end`, `month_end`, `day_end` | Date de fin de la recherche, à minuit. Pour couvrir un mois, utiliser le premier jour du mois suivant. |
| `calendars` | Liste des calendriers à exporter. |
| `calendars[].calendar_name` | Texte contenu dans le nom du calendrier distant, avec respect de la casse. Il doit correspondre à un seul calendrier. |
| `calendars[].csv_name` | Chemin du fichier CSV à écrire. Un chemin relatif est résolu depuis le répertoire de lancement. |
Ajouter une entrée dans `calendars` pour chaque calendrier à exporter.
Les champs `db_*`, `user_name` et `api_password`, utilisés par d'autres scripts
du dépôt, ne sont pas nécessaires à cet export.
`config.json` est ignoré par Git. Remplacer les valeurs d'exemple par les
informations du serveur et du compte utilisés.
### Lancement
Lancer les commandes depuis la racine du dépôt : le script lit
`./config.json`, et le lanceur recherche `env/bin/activate` dans le répertoire
courant. Créer au préalable les dossiers de destination des CSV ; le script
ne les crée pas.
```bash
mkdir -p csv
bash ./exec-export-caldav.sh
```
Le lanceur active l'environnement `env/`, puis exécute le script. Il est aussi
possible de lancer directement Python :
```bash
env/bin/python crealead-caldav.py
```
Le script ne propose pas d'arguments en ligne de commande. La période et les
calendriers sont définis dans `config.json`.
### Fichiers produits et traitement des événements
Chaque CSV existant est remplacé lors de l'export. Les lignes sont triées par
date et heure de début. Le séparateur est `;` et les colonnes sont :
```text
Titre;Date debut;Heure debut;Date fin;Heure fin;Categories;Emplacement;Description
```
Les événements récurrents sont demandés au serveur avec `expand=True`.
Le lecteur iCalendar prend en charge les paramètres des propriétés, comme
`TZID` et `VALUE=DATE`, ainsi que les textes répartis sur plusieurs lignes.
Les heures sont converties vers `Europe/Paris`. Une heure sans fuseau est
interprétée dans ce fuseau ; une date seule est exportée à minuit.
Si `DTEND` est absent, `DURATION` peut fournir la date de fin.
Les événements sans propriété `SUMMARY` sont ignorés. Ceux dont les dates
de début ou de fin sont absentes ou invalides sont également ignorés. Chaque
exclusion laisse un avertissement détaillé dans le journal, y compris en
l'absence de `SUMMARY`. Un document iCalendar illisible ou une ressource sans
composant `VEVENT` est signalé puis ignoré. Les autres événements restent
exportés. Si un calendrier configuré est introuvable ou correspond à plusieurs
calendriers distants, le script affiche un message et s'arrête avant l'écriture
des CSV.
En cas d'erreur d'écriture d'un CSV, le script s'arrête avec une exception.
Il laisse un avertissement pour chaque événement du calendrier en cours,
dont le fichier peut être incomplet, et des calendriers restant à écrire.
Les fichiers déjà écrits avec succès sont conservés.
### Journalisation
Les sorties standard et d'erreur sont affichées dans le terminal et copiées
dans `logs/crealead-caldav.log`, à côté du script. Le dossier est créé
automatiquement. Le journal utilise UTF-8, conserve les exécutions précédentes
et enregistre les traces d'exception en cas de plantage pendant le traitement.
Il n'y a pas de rotation automatique du journal. `logs/` est ignoré par Git.
Chaque nouvelle ligne du journal est préfixée par l'heure d'écriture en UTC,
au format ISO 8601 `YYYY-MM-DDTHH:MM:SSZ`, puis par `[info]` :
```text
[2026-10-15T12:34:56Z]: [info] - csv/utilisateur.csv ...
```
Cela s'applique aussi aux messages écrits en plusieurs fragments et à chaque
ligne d'une trace d'exception. L'horodatage est ajouté dans le fichier de
journal, sans modifier les messages ordinaires du terminal.
Tout événement non exporté est signalé avec `[warning][event]`, à la place de
`[info]`. Le message indique le calendrier distant (ou son nom configuré),
le fichier CSV, le titre, le début, la fin, l'UID et la raison de l'exclusion :
```text
[2026-10-15T12:34:56Z]: [warning][event] - Calendrier='Temps de travail' ; CSV='csv/utilisateur.csv' ; titre='Réunion' ; début='2026-10-15T08:00:00Z' ; fin='indisponible' ; UID='reunion-1' ; événement non exporté : dates de début/fin absentes ou invalides (date absente ou invalide)
```
L'heure entre crochets est celle du message ; les champs `début` et `fin`
décrivent le moment de l'événement. Les dates et heures décodées sont affichées
en UTC, les dates de journée entière restent des dates seules. En cas de
document illisible ou de date invalide, le script récupère les champs bruts
disponibles. Une métadonnée absente ou irrécupérable vaut `indisponible`.
Les retours à la ligne dans les métadonnées et le motif sont échappés pour
garder un avertissement sur une seule ligne. Ces avertissements sont aussi
visibles dans le terminal.
## Tests
Les tests se trouvent dans
[`tests/test_crealead_caldav.py`](../tests/test_crealead_caldav.py) et utilisent
`unittest`, fourni avec Python. Aucune dépendance de test supplémentaire
n'est nécessaire.
### Cas couverts
Les 13 tests de `ExportEventsTests` vérifient :
- La lecture de `DTSTART` et `DTEND` avec `TZID`, leur conversion vers Paris
et le tri chronologique des lignes.
- La conversion des dates UTC en été et en hiver, ainsi que celle d'un autre
fuseau horaire.
- Les événements sur une journée entière (`VALUE=DATE`), les heures sans
fuseau et le calcul de la fin à partir de `DURATION`.
- L'exclusion des événements sans propriété `SUMMARY`, des dates manquantes
ou invalides et des documents iCalendar illisibles, sans bloquer les
événements valides, ainsi que la présence d'un avertissement contextualisé
dans le journal pour chaque événement écarté.
- La récupération des métadonnées de chaque événement dans un document
illisible, la conservation d'une date brute invalide et le signalement
d'une ressource sans composant `VEVENT`.
- La conservation des deux-points dans les textes, des paramètres de texte,
des catégories et des lignes repliées, ainsi que le nettoyage de la
description HTML encodée et le contenu du CSV écrit.
Les 9 tests de `LogOutputTests` vérifient :
- La création de `logs/` à côté du script, la copie des deux sorties et
l'écriture immédiate des messages, y compris les accents.
- L'ajout au journal existant et le choix du nom du journal d'après celui du
script, y compris lorsqu'une exécution se termine sans retour à la ligne.
- L'horodatage en UTC, le préfixe `[info]` de chaque ligne, les écritures en
plusieurs fragments, les messages sur plusieurs lignes et les lignes vides.
- Le préfixe `[warning][event]` sans `[info]`, les métadonnées de l'événement,
l'échappement des retours à la ligne et la séparation d'un avertissement
après un message ordinaire inachevé.
- La sauvegarde des traces d'exception et la restauration des sorties après
une erreur.
- La conservation du code de sortie d'un `SystemExit`, sans ajouter de trace
d'exception, et la sauvegarde de son éventuel message.
Le test de `ExportFailureTests` provoque une erreur d'écriture CSV. Il vérifie
qu'un avertissement est enregistré pour les événements du calendrier en
échec et des calendriers restant à écrire, et que le premier CSV déjà écrit
est conservé.
Ces tests utilisent le vrai lecteur `icalendar`, des événements synthétiques
et des dossiers temporaires pour les CSV et les journaux. Le module CalDAV
et les réponses du serveur sont simulés : aucun accès réseau, compte,
`config.json` ou calendrier réel n'est nécessaire. Ils vérifient que la
recherche demande `expand=True`, mais ne valident pas l'expansion des
récurrences par un serveur réel ni l'import dans TimeManager.
### Exécution
Depuis la racine du dépôt, avec les dépendances installées dans `env/` :
```bash
env/bin/python -B -m unittest discover -s tests -v
```
L'option `-B` évite de créer des fichiers de cache Python. La suite contient
actuellement 23 tests ; une exécution réussie se termine par `OK`.
Avec l'image Docker du dépôt déjà construite :
```bash
docker run --rm -v "$(pwd)":/app -w /app cal-to-time-manager:dev -B -m unittest discover -s tests -v
```
-2
View File
@@ -23,8 +23,6 @@ Chaque entrée doit contenir au minimum :
- `user_name` : login Nextcloud utilisé pour authentifier les requêtes,
- `api_password` : mot de passe ou mot de passe d'application Nextcloud.
> Il est important de générer la clé API en s'étant connecté avec l'identifiant et non l'email car le script user_name utilise le champ "user_name" pour faire des vérifications en bases de données avant d'ajouter des données.
Colonnes CSV utilisées
----------------------
-4
View File
@@ -1,4 +0,0 @@
#!/bin/bash
source env/bin/activate
python3 remove-users-data-timemanager.py $1
-116
View File
@@ -1,116 +0,0 @@
import csv
import requests
import json
import uuid
import re
import base64
import sys
import mysql.connector
from unidecode import unidecode
##########################################
# Config
##########################################
# Récupération de la config
with open('config.json', 'r') as f:
config = json.load(f)
mydb = mysql.connector.connect(
host=config['db_host'],
user=config['db_user'],
password=config['db_password'],
database=config['db_database']
)
cursor = mydb.cursor(buffered=True)
# Argument --yes pour auto-valider toutes les suppressions
valid_args = {'-y', '--yes'}
extra_args = [arg for arg in sys.argv[1:] if arg not in valid_args]
if extra_args:
print(f"Erreur : argument(s) non reconnu(s) : {' '.join(extra_args)}")
print(f"Usage : python {sys.argv[0]} [--yes | -y]")
sys.exit(1)
auto_confirm = bool(set(sys.argv[1:]) & valid_args)
##########################################
# Fonctions
##########################################
def check_users_exist(user_id):
"""Vérifie si l'utilisateur existe dans oc_users."""
cursor.execute("SELECT uid FROM oc_users WHERE uid = %s", (user_id,))
return cursor.fetchone() is not None
def remove_row_on_table_for_user(user_id, table):
"""Supprime toutes les entrées d'une table pour un utilisateur donné."""
query = f"DELETE FROM `{table}` WHERE user_id = %s"
cursor.execute(query, (user_id,))
affected = cursor.rowcount
mydb.commit()
print(f" [{table}] {affected} ligne(s) supprimée(s)")
##########################################
# Main
##########################################
tables = [
"oc_timemanager_client",
"oc_timemanager_commit",
"oc_timemanager_project",
"oc_timemanager_task",
"oc_timemanager_time"
]
# Listing de tous les utilisateurs configurés avant toute action
print("==========================================")
print("Utilisateurs configurés dans config.json :")
print("==========================================")
valid_users = []
for user_info in config['calendars']:
user_id = user_info.get('user_name', '')
if not user_id:
print(f" ⚠ [{user_id or '???'}] — infos de connexion incomplètes, ignoré")
continue
exists = check_users_exist(user_id)
status = "✔ existe en BDD" if exists else "✘ introuvable en BDD"
print(f" • {user_id} — {status}")
if exists:
valid_users.append(user_info)
print()
if not valid_users:
print("Aucun utilisateur valide à traiter. Fin du script.")
cursor.close()
mydb.close()
sys.exit(0)
if auto_confirm:
print("Mode auto-confirmation activé (--yes) : toutes les suppressions seront effectuées sans demande.\n")
# Traitement par utilisateur
for user_info in valid_users:
user_id = user_info['user_name']
print("==========================================")
print(f"User : {user_id}")
print("==========================================")
if auto_confirm:
confirmed = True
else:
reponse = input(f"\nSupprimer toutes les données TimeManager de '{user_id}' ? [o/N] : ").strip().lower()
confirmed = reponse in ('o', 'oui', 'y', 'yes')
if confirmed:
print("Suppression en cours...")
for table in tables:
remove_row_on_table_for_user(user_id, table)
print(f"✔ Données de '{user_id}' supprimées.\n")
else:
print(f"✘ Suppression annulée pour '{user_id}'.\n")
cursor.close()
mydb.close()
print("Terminé.")
-116
View File
@@ -1,116 +0,0 @@
import csv
import requests
import json
import uuid
import re
import base64
import sys
import mysql.connector
from unidecode import unidecode
##########################################
# Config
##########################################
# Récupération de la config
with open('config.json', 'r') as f:
config = json.load(f)
mydb = mysql.connector.connect(
host=config['db_host'],
user=config['db_user'],
password=config['db_password'],
database=config['db_database']
)
cursor = mydb.cursor(buffered=True)
# Argument --yes pour auto-valider toutes les suppressions
valid_args = {'-y', '--yes'}
extra_args = [arg for arg in sys.argv[1:] if arg not in valid_args]
if extra_args:
print(f"Erreur : argument(s) non reconnu(s) : {' '.join(extra_args)}")
print(f"Usage : python {sys.argv[0]} [--yes | -y]")
sys.exit(1)
auto_confirm = bool(set(sys.argv[1:]) & valid_args)
##########################################
# Fonctions
##########################################
def check_users_exist(user_id):
"""Vérifie si l'utilisateur existe dans oc_users."""
cursor.execute("SELECT uid FROM oc_users WHERE uid = %s", (user_id,))
return cursor.fetchone() is not None
def remove_row_on_table_for_user(user_id, table):
"""Supprime toutes les entrées d'une table pour un utilisateur donné."""
query = f"DELETE FROM `{table}` WHERE user_id = %s"
cursor.execute(query, (user_id,))
affected = cursor.rowcount
mydb.commit()
print(f" [{table}] {affected} ligne(s) supprimée(s)")
##########################################
# Main
##########################################
tables = [
"oc_timemanager_client",
"oc_timemanager_commit",
"oc_timemanager_project",
"oc_timemanager_task",
"oc_timemanager_time"
]
# Listing de tous les utilisateurs configurés avant toute action
print("==========================================")
print("Utilisateurs configurés dans config.json :")
print("==========================================")
valid_users = []
for user_info in config['calendars']:
user_id = user_info.get('user_name', '')
if not user_id:
print(f" ⚠ [{user_id or '???'}] — infos de connexion incomplètes, ignoré")
continue
exists = check_users_exist(user_id)
status = "✔ existe en BDD" if exists else "✘ introuvable en BDD"
print(f" • {user_id} — {status}")
if exists:
valid_users.append(user_info)
print()
if not valid_users:
print("Aucun utilisateur valide à traiter. Fin du script.")
cursor.close()
mydb.close()
sys.exit(0)
if auto_confirm:
print("Mode auto-confirmation activé (--yes) : toutes les suppressions seront effectuées sans demande.\n")
# Traitement par utilisateur
for user_info in valid_users:
user_id = user_info['user_name']
print("==========================================")
print(f"User : {user_id}")
print("==========================================")
if auto_confirm:
confirmed = True
else:
reponse = input(f"\nSupprimer toutes les données TimeManager de '{user_id}' ? [o/N] : ").strip().lower()
confirmed = reponse in ('o', 'oui', 'y', 'yes')
if confirmed:
print("Suppression en cours...")
for table in tables:
remove_row_on_table_for_user(user_id, table)
print(f"✔ Données de '{user_id}' supprimées.\n")
else:
print(f"✘ Suppression annulée pour '{user_id}'.\n")
cursor.close()
mydb.close()
print("Terminé.")
-3
View File
@@ -1,3 +0,0 @@
#!/bin/bash
ssh execcsv@crl-01.ovea.com
-364
View File
@@ -1,364 +0,0 @@
import contextlib
import csv
import importlib.util
import io
import re
import sys
from datetime import date, datetime, time, timezone
from pathlib import Path
import tempfile
from types import ModuleType, SimpleNamespace
import unittest
from unittest.mock import Mock, patch
# The regression tests use real iCalendar parsing, with no CalDAV connection.
spec = importlib.util.spec_from_file_location(
'crealead_caldav', Path(__file__).resolve().parents[1] / 'crealead-caldav.py')
exporter = importlib.util.module_from_spec(spec)
with patch.dict('sys.modules', {'caldav': ModuleType('caldav')}):
spec.loader.exec_module(exporter)
def event_data(properties):
return SimpleNamespace(data=(
'BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\n'
+ properties.replace('\n', '\r\n')
+ '\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n'))
class ExportEventsTests(unittest.TestCase):
def export(self, *events):
server = SimpleNamespace(name='Agenda de test', date_search=Mock(return_value=events))
config = dict(year_start=2026, month_start=10, day_start=1,
year_end=2026, month_end=11, day_end=1)
errors = io.StringIO()
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(errors):
with exporter.log_output():
result = exporter.get_all_events(config, [
{'csv_name': 'test.csv', 'calendar_server': server}])
self.log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
self.assertTrue(server.date_search.call_args.kwargs['expand'])
return result[0]['line'], errors.getvalue()
def test_timezone_parameters_and_chronological_sort(self):
rows, errors = self.export(
event_data('SUMMARY:Later\nDTSTART;TZID=Europe/Paris:20261015T140000\n'
'DTEND;TZID=Europe/Paris:20261015T150000'),
event_data('SUMMARY:Earlier\nDTSTART:20261015T080000Z\n'
'DTEND:20261015T090000Z'))
self.assertEqual(errors, '')
self.assertEqual([row.title for row in rows], ['Earlier', 'Later'])
self.assertEqual([row.start_time for row in rows], [time(10), time(14)])
self.assertEqual([row.end_time for row in rows], [time(11), time(15)])
def test_utc_conversion_in_winter(self):
rows, errors = self.export(event_data(
'SUMMARY:Winter\nDTSTART:20260115T080000Z\nDTEND:20260115T090000Z'))
self.assertEqual(errors, '')
self.assertEqual(rows[0].start_time, time(9))
self.assertEqual(rows[0].end_time, time(10))
def test_other_timezone_is_converted_to_paris(self):
rows, errors = self.export(event_data(
'SUMMARY:Call\nDTSTART;TZID=America/New_York:20261015T090000\n'
'DTEND;TZID=America/New_York:20261015T100000'))
self.assertEqual(errors, '')
self.assertEqual(rows[0].start_time, time(15))
self.assertEqual(rows[0].end_time, time(16))
def test_date_only_values_keep_their_dates_at_midnight(self):
rows, errors = self.export(event_data(
'SUMMARY:Day\nDTSTART;VALUE=DATE:20261015\nDTEND;VALUE=DATE:20261016'))
self.assertEqual(errors, '')
self.assertEqual(rows[0].start_date, date(2026, 10, 15))
self.assertEqual(rows[0].end_date, date(2026, 10, 16))
self.assertEqual(rows[0].start_time, time(0))
self.assertEqual(rows[0].end_time, time(0))
def test_floating_times_are_local(self):
rows, errors = self.export(event_data(
'SUMMARY:Local\nDTSTART:20261015T090000\nDTEND:20261015T100000'))
self.assertEqual(errors, '')
self.assertEqual(rows[0].start_time, time(9))
self.assertEqual(rows[0].end_time, time(10))
def test_duration_can_supply_end_time(self):
rows, errors = self.export(event_data(
'SUMMARY:Duration\nDTSTART:20261015T080000Z\nDURATION:PT1H30M'))
self.assertEqual(errors, '')
self.assertEqual(rows[0].end_time, time(11, 30))
def test_missing_or_invalid_dates_are_logged_and_skipped(self):
for dates in (
'DTEND:20261015T090000Z',
'DTSTART:20261015T080000Z',
'DTSTART:invalid\nDTEND:20261015T090000Z',
'DTSTART:20261015T080000Z\nDTEND:invalid',
):
with self.subTest(dates=dates):
rows, errors = self.export(
event_data('UID:bad-event\nSUMMARY:Invalid\n' + dates),
event_data('SUMMARY:Valid\nDTSTART:20261015T080000Z\n'
'DTEND:20261015T090000Z'))
self.assertEqual([row.title for row in rows], ['Valid'])
self.assertIn('bad-event', errors)
self.assertIn('test.csv', errors)
self.assertIn('événement non exporté', errors)
self.assertIn("Calendrier='Agenda de test'", self.log)
self.assertIn("titre='Invalid'", self.log)
self.assertIn('début=', self.log)
self.assertIn('fin=', self.log)
self.assertIn('dates de début/fin absentes ou invalides', self.log)
self.assertEqual(self.log.count('[warning][event] - '), 1)
def test_event_without_summary_is_still_skipped(self):
rows, errors = self.export(event_data(
'DTSTART:20261015T080000Z\nDTEND:20261015T090000Z'))
self.assertEqual(rows, [])
self.assertIn('propriété SUMMARY absente', errors)
self.assertIn("titre='indisponible'", self.log)
self.assertIn("début='2026-10-15T08:00:00Z'", self.log)
self.assertIn("Calendrier='Agenda de test'", self.log)
self.assertEqual(self.log.count('[warning][event] - '), 1)
def test_malformed_calendar_does_not_block_other_events(self):
rows, errors = self.export(
SimpleNamespace(data='not an iCalendar document'),
event_data('SUMMARY:Valid\nDTSTART:20261015T080000Z\n'
'DTEND:20261015T090000Z'))
self.assertEqual([row.title for row in rows], ['Valid'])
self.assertIn('document iCalendar illisible', errors)
self.assertIn("titre='indisponible'", self.log)
self.assertIn("début='indisponible'", self.log)
def test_malformed_document_recovers_each_event_context(self):
rows, errors = self.export(SimpleNamespace(data=(
'BEGIN:VEVENT\nUID:first\nSUMMARY;LANGUAGE=fr:Réunion:\n suite\n'
'DTSTART;TZID=Europe/Paris:20261015T090000\nEND:VEVENT\n'
'BEGIN:VEVENT\nUID:second\nSUMMARY:Autre\n'
'DTSTART:20261016T080000Z\nEND:VEVENT\n')))
# Multiple root components make Calendar.from_ical reject the document.
self.assertEqual(rows, [])
self.assertEqual(self.log.count('[warning][event] - '), 2)
self.assertIn("titre='Réunion:suite'", self.log)
self.assertIn("début='20261015T090000'", self.log)
self.assertIn("titre='Autre'", self.log)
self.assertIn("UID='second'", self.log)
self.assertIn('document iCalendar illisible', errors)
def test_resource_without_event_is_logged(self):
rows, errors = self.export(SimpleNamespace(data=(
'BEGIN:VCALENDAR\r\nVERSION:2.0\r\nEND:VCALENDAR\r\n')))
self.assertEqual(rows, [])
self.assertIn('aucun composant VEVENT', errors)
self.assertIn("Calendrier='Agenda de test'", self.log)
self.assertEqual(self.log.count('[warning][event] - '), 1)
def test_invalid_date_keeps_raw_value_in_warning(self):
rows, errors = self.export(event_data(
'SUMMARY:Invalid\nDTSTART:invalid\nDTEND:20261015T090000Z'))
self.assertEqual(rows, [])
self.assertIn("début='invalid'", self.log)
self.assertIn('dates de début/fin absentes ou invalides', errors)
def test_text_parameters_folding_colons_and_csv_output(self):
rows, errors = self.export(event_data(
'SUMMARY;LANGUAGE=fr:Projet: réunion\n'
'DTSTART:20261015T080000Z\nDTEND:20261015T090000Z\n'
'CATEGORIES:Projet A,Projet B\nLOCATION:Bureau: Montpellier\n'
'DESCRIPTION:text/html,%3Cb%3EClient%3C/b%3E:\n suite'))
self.assertEqual(errors, '')
with tempfile.TemporaryDirectory() as directory:
filename = Path(directory) / 'export.csv'
exporter.write_csv(filename, rows)
with filename.open(newline='') as csv_file:
exported = list(csv.reader(csv_file, delimiter=';'))
self.assertEqual(len(exported), 2)
self.assertEqual(exported[1], [
'Projet: réunion', '2026-10-15', '10:00:00', '2026-10-15',
'11:00:00', 'Projet A,Projet B', 'Bureau: Montpellier', 'Client:suite'])
class LogOutputTests(unittest.TestCase):
def messages(self, log):
messages = []
for line in log.splitlines():
match = re.fullmatch(r'\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\]: (.*)', line)
self.assertIsNotNone(match, line)
messages.append(match.group(1))
return messages
def test_timestamp_uses_utc(self):
with patch.object(exporter, 'datetime') as clock:
clock.now.return_value = datetime(2026, 10, 15, 12, 34, 56, tzinfo=timezone.utc)
self.assertEqual(exporter.utc_timestamp(), '2026-10-15T12:34:56Z')
clock.now.assert_called_once_with(timezone.utc)
def test_fragmented_and_multiline_writes_have_one_prefix_per_line(self):
console = io.StringIO()
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
contextlib.redirect_stdout(console):
with exporter.log_output():
print('CSV', '...', 2)
sys.stdout.write('Début')
sys.stdout.write(' de ligne\n\nAutre ligne\n')
messages = self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8'))
self.assertEqual(messages, [
'[info] - CSV ... 2', '[info] - Début de ligne', '[info] - ', '[info] - Autre ligne'])
self.assertEqual(console.getvalue(), 'CSV ... 2\nDébut de ligne\n\nAutre ligne\n')
def test_event_warning_has_own_level_context_and_escaped_newlines(self):
console, errors = io.StringIO(), io.StringIO()
component = exporter.Calendar.from_ical(event_data(
'UID:event-1\nSUMMARY:Réunion\\nSuite\n'
'DTSTART;TZID=Europe/Paris:20261015T100000\n'
'DTEND;TZID=Europe/Paris:20261015T110000').data).walk('VEVENT')[0]
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
contextlib.redirect_stdout(console), contextlib.redirect_stderr(errors):
with exporter.log_output():
sys.stdout.write('En cours')
exporter.log_skipped_event(
{'calendar_name': 'Calendrier configuré', 'csv_name': 'test.csv'},
component, 'motif\nsur deux lignes')
print('Suite')
messages = self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8'))
self.assertEqual(len(messages), 3)
self.assertEqual(messages[0], '[info] - En cours')
self.assertTrue(messages[1].startswith('[warning][event] - '))
self.assertNotIn('[info]', messages[1])
self.assertIn("Calendrier='Calendrier configuré'", messages[1])
self.assertIn("titre='Réunion\\nSuite'", messages[1])
self.assertIn("début='2026-10-15T08:00:00Z'", messages[1])
self.assertIn('motif\\nsur deux lignes', messages[1])
self.assertEqual(messages[2], '[info] - Suite')
self.assertEqual(console.getvalue(), 'En coursSuite\n')
self.assertTrue(errors.getvalue().startswith('[warning][event] - '))
def test_appended_runs_without_final_newline_remain_separate(self):
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
contextlib.redirect_stdout(io.StringIO()):
for message in ('Premier', 'Deuxième'):
with exporter.log_output():
sys.stdout.write(message)
self.assertEqual(
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
['[info] - Premier', '[info] - Deuxième'])
def test_system_exit_message_is_saved(self):
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')):
with self.assertRaises(SystemExit):
with exporter.log_output():
sys.exit('Configuration invalide')
self.assertEqual(
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
['[info] - Configuration invalide'])
def test_creates_logs_next_to_script_and_copies_both_streams(self):
stdout, stderr = io.StringIO(), io.StringIO()
with tempfile.TemporaryDirectory() as directory:
script_path = Path(directory) / 'crealead-caldav.py'
with patch.object(exporter, '__file__', str(script_path)), \
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
with exporter.log_output():
print('Export démarré')
print('Événement ignoré', file=sys.stderr)
sys.stdout.write('Terminé')
sys.stdout.flush()
self.assertEqual(
self.messages((Path(directory) / 'logs/crealead-caldav.log').read_text(encoding='utf-8')),
['[info] - Export démarré', '[info] - Événement ignoré', '[info] - Terminé'])
self.assertIs(sys.stdout, stdout)
self.assertIs(sys.stderr, stderr)
self.assertEqual(stdout.getvalue(), 'Export démarré\nTerminé')
self.assertEqual(stderr.getvalue(), 'Événement ignoré\n')
def test_appends_on_next_execution_and_uses_script_name(self):
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
contextlib.redirect_stdout(io.StringIO()):
for message in ('Premier export', 'Deuxième export'):
with exporter.log_output():
print(message)
self.assertEqual(
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
['[info] - Premier export', '[info] - Deuxième export'])
def test_exception_traceback_is_saved_and_streams_are_restored(self):
stdout, stderr = io.StringIO(), io.StringIO()
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
with self.assertRaisesRegex(ValueError, 'date invalide'):
with exporter.log_output():
print('Avant erreur')
raise ValueError('date invalide')
self.assertIs(sys.stdout, stdout)
self.assertIs(sys.stderr, stderr)
log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
messages = self.messages(log)
self.assertEqual(messages[0], '[info] - Avant erreur')
self.assertTrue(messages[1].startswith('[info] - Traceback'))
self.assertTrue(all(message.startswith('[info] - ') for message in messages))
self.assertIn('ValueError: date invalide', log)
self.assertEqual(stdout.getvalue(), 'Avant erreur\n')
self.assertEqual(stderr.getvalue(), '')
def test_system_exit_keeps_exit_code_without_logging_traceback(self):
with tempfile.TemporaryDirectory() as directory:
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit) as caught:
with exporter.log_output():
print('Connexion impossible', file=sys.stderr)
sys.exit(2)
self.assertEqual(caught.exception.code, 2)
self.assertEqual(
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
['[info] - Connexion impossible'])
class ExportFailureTests(unittest.TestCase):
def test_csv_failure_logs_events_in_current_and_pending_calendars(self):
config = dict(year_start=2026, month_start=10, day_start=1,
year_end=2026, month_end=11, day_end=1)
with tempfile.TemporaryDirectory() as directory:
paths = [Path(directory) / 'first.csv',
Path(directory) / 'missing/failed.csv',
Path(directory) / 'pending.csv']
calendars = [
{'csv_name': str(path), 'calendar_server': SimpleNamespace(
name='Agenda {}'.format(index), date_search=Mock(return_value=[event_data(
'SUMMARY:Event {}\nUID:uid-{}\nDTSTART:20261015T080000Z\n'
'DTEND:20261015T090000Z'.format(index, index))]))}
for index, path in enumerate(paths)
]
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
patch.object(exporter, 'load_json_config', return_value=config), \
patch.object(exporter, 'init_connection', return_value=[]), \
patch.object(exporter, 'link_calendars', return_value=calendars), \
contextlib.redirect_stdout(io.StringIO()), \
contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(OSError):
with exporter.log_output():
exporter.main()
log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
self.assertTrue(paths[0].is_file())
self.assertFalse(paths[2].exists())
warnings = [line for line in log.splitlines() if '[warning][event]' in line]
self.assertEqual(len(warnings), 2)
for index, warning in enumerate(warnings, start=1):
self.assertIn("Calendrier='Agenda {}'".format(index), warning)
self.assertIn("titre='Event {}'".format(index), warning)
self.assertIn("début='2026-10-15T08:00:00Z'", warning)
self.assertIn('export CSV interrompu', warning)
if __name__ == '__main__':
unittest.main()