laurent
f360332626
Surveille inbox/picnic/ et inbox/leclerc/ avec watchdog.
Chaque nouveau fichier est importé automatiquement :
- succès/doublon → processed/{source}_{date}_{nom}
- erreur → failed/{nom} + failed/{nom}.log
Nouvelle commande CLI : python -m tickettracker.cli watch [--inbox] [--db]
22 tests ajoutés (test_watcher.py), tous passent.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
242 lines
8.4 KiB
Python
242 lines
8.4 KiB
Python
"""
|
|
Tests du watch folder (tickettracker/watcher.py).
|
|
|
|
Stratégie :
|
|
- Utilise tmp_path pour les dossiers inbox/processed/failed
|
|
- Mocke pipeline.import_receipt pour contrôler le résultat sans parser de vrais fichiers
|
|
- Teste _process_file directement (évite la dépendance à watchdog / inotify)
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from tickettracker.watcher import _process_file, ReceiptHandler
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def dirs(tmp_path):
|
|
"""Structure de dossiers inbox/picnic, inbox/leclerc, processed, failed."""
|
|
inbox = tmp_path / "inbox"
|
|
(inbox / "picnic").mkdir(parents=True)
|
|
(inbox / "leclerc").mkdir(parents=True)
|
|
processed = tmp_path / "processed"
|
|
processed.mkdir()
|
|
failed = tmp_path / "failed"
|
|
failed.mkdir()
|
|
return {
|
|
"inbox": inbox,
|
|
"processed": processed,
|
|
"failed": failed,
|
|
"tmp_path": tmp_path,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_file(dirs):
|
|
"""Crée un faux fichier HTML Picnic dans inbox/picnic/."""
|
|
f = dirs["inbox"] / "picnic" / "ticket_picnic.html"
|
|
f.write_text("<html>Picnic</html>", encoding="utf-8")
|
|
return f
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_leclerc(dirs):
|
|
"""Crée un faux fichier PDF Leclerc dans inbox/leclerc/."""
|
|
f = dirs["inbox"] / "leclerc" / "ticket_leclerc.pdf"
|
|
f.write_bytes(b"%PDF-1.4 fake")
|
|
return f
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests _process_file — import réussi
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_process_file_success_moves_to_processed(dirs, sample_file):
|
|
"""Import réussi : le fichier est déplacé dans processed/."""
|
|
with patch("tickettracker.watcher.pipeline.import_receipt", return_value=True):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
# Le fichier original ne doit plus être dans inbox/
|
|
assert not sample_file.exists()
|
|
# Un fichier doit être présent dans processed/
|
|
processed_files = list(dirs["processed"].iterdir())
|
|
assert len(processed_files) == 1
|
|
|
|
|
|
def test_process_file_success_naming_convention(dirs, sample_file):
|
|
"""Le fichier déplacé suit le pattern {source}_{date}_{nom_original}."""
|
|
with patch("tickettracker.watcher.pipeline.import_receipt", return_value=True):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
processed_files = list(dirs["processed"].iterdir())
|
|
name = processed_files[0].name
|
|
assert name.startswith("picnic_")
|
|
assert name.endswith("ticket_picnic.html")
|
|
|
|
|
|
def test_process_file_duplicate_moves_to_processed(dirs, sample_file):
|
|
"""Doublon (import_receipt retourne False) : fichier dans processed/ quand même."""
|
|
with patch("tickettracker.watcher.pipeline.import_receipt", return_value=False):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
assert not sample_file.exists()
|
|
processed_files = list(dirs["processed"].iterdir())
|
|
assert len(processed_files) == 1
|
|
|
|
|
|
def test_process_file_error_moves_to_failed(dirs, sample_file):
|
|
"""Erreur pendant l'import : le fichier est déplacé dans failed/."""
|
|
with patch(
|
|
"tickettracker.watcher.pipeline.import_receipt",
|
|
side_effect=ValueError("format invalide"),
|
|
):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
assert not sample_file.exists()
|
|
failed_files = [f for f in dirs["failed"].iterdir() if not f.name.endswith(".log")]
|
|
assert len(failed_files) == 1
|
|
|
|
|
|
def test_process_file_error_creates_log(dirs, sample_file):
|
|
"""Erreur : un fichier .log est créé dans failed/ avec le message d'erreur."""
|
|
with patch(
|
|
"tickettracker.watcher.pipeline.import_receipt",
|
|
side_effect=ValueError("format invalide"),
|
|
):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
log_files = [f for f in dirs["failed"].iterdir() if f.name.endswith(".log")]
|
|
assert len(log_files) == 1
|
|
log_content = log_files[0].read_text(encoding="utf-8")
|
|
assert "format invalide" in log_content
|
|
|
|
|
|
def test_process_file_nothing_in_failed_on_success(dirs, sample_file):
|
|
"""Import réussi : aucun fichier dans failed/."""
|
|
with patch("tickettracker.watcher.pipeline.import_receipt", return_value=True):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
assert list(dirs["failed"].iterdir()) == []
|
|
|
|
|
|
def test_process_file_leclerc_source(dirs, sample_leclerc):
|
|
"""Source leclerc : le fichier déplacé commence par 'leclerc_'."""
|
|
with patch("tickettracker.watcher.pipeline.import_receipt", return_value=True):
|
|
_process_file(
|
|
sample_leclerc, "leclerc",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
processed_files = list(dirs["processed"].iterdir())
|
|
assert processed_files[0].name.startswith("leclerc_")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests ReceiptHandler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_handler_detects_source_from_parent_folder(dirs):
|
|
"""ReceiptHandler détecte la source depuis le nom du sous-dossier."""
|
|
handler = ReceiptHandler(
|
|
db_path=dirs["tmp_path"] / "test.db",
|
|
inbox_path=dirs["inbox"],
|
|
)
|
|
# Le sous-dossier parent du fichier donne la source
|
|
assert handler.processed_dir == dirs["processed"]
|
|
assert handler.failed_dir == dirs["failed"]
|
|
|
|
|
|
def test_handler_ignores_unknown_subfolder(dirs):
|
|
"""Un fichier dans un sous-dossier inconnu (ni picnic ni leclerc) est ignoré."""
|
|
unknown_dir = dirs["inbox"] / "autre"
|
|
unknown_dir.mkdir()
|
|
f = unknown_dir / "fichier.txt"
|
|
f.write_text("test")
|
|
|
|
handler = ReceiptHandler(
|
|
db_path=dirs["tmp_path"] / "test.db",
|
|
inbox_path=dirs["inbox"],
|
|
)
|
|
|
|
# Simuler un événement de création de fichier
|
|
class FakeEvent:
|
|
is_directory = False
|
|
src_path = str(f)
|
|
|
|
with patch("tickettracker.watcher._process_file") as mock_process:
|
|
handler.on_created(FakeEvent())
|
|
mock_process.assert_not_called()
|
|
|
|
|
|
def test_handler_ignores_directory_events(dirs):
|
|
"""Un événement de création de répertoire est ignoré."""
|
|
handler = ReceiptHandler(
|
|
db_path=dirs["tmp_path"] / "test.db",
|
|
inbox_path=dirs["inbox"],
|
|
)
|
|
|
|
class FakeEvent:
|
|
is_directory = True
|
|
src_path = str(dirs["inbox"] / "picnic" / "subdir")
|
|
|
|
with patch("tickettracker.watcher._process_file") as mock_process:
|
|
handler.on_created(FakeEvent())
|
|
mock_process.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests création des dossiers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_process_file_creates_processed_dir_if_missing(dirs, sample_file):
|
|
"""_process_file crée processed/ s'il est absent."""
|
|
dirs["processed"].rmdir() # supprime le dossier
|
|
assert not dirs["processed"].exists()
|
|
with patch("tickettracker.watcher.pipeline.import_receipt", return_value=True):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
assert dirs["processed"].exists()
|
|
|
|
|
|
def test_process_file_creates_failed_dir_if_missing(dirs, sample_file):
|
|
"""_process_file crée failed/ s'il est absent."""
|
|
dirs["failed"].rmdir()
|
|
assert not dirs["failed"].exists()
|
|
with patch(
|
|
"tickettracker.watcher.pipeline.import_receipt",
|
|
side_effect=RuntimeError("boom"),
|
|
):
|
|
_process_file(
|
|
sample_file, "picnic",
|
|
dirs["tmp_path"] / "test.db",
|
|
dirs["processed"], dirs["failed"],
|
|
)
|
|
assert dirs["failed"].exists()
|