From 75dd0bd50e42e504064aa95f9404964c45932452 Mon Sep 17 00:00:00 2001 From: kdeterme Date: Mon, 7 Sep 2026 11:17:00 +0200 Subject: [PATCH] refactor: move output file resolution to model method with log-based fallback and added unit tests --- loko/common/admin_scheduled_tasks.py | 17 +---- .../commands/run_scheduled_commands.py | 2 +- loko/common/models_scheduled_tasks.py | 45 +++++++++++ loko/common/tests.py | 75 +++++++++++++++++++ 4 files changed, 122 insertions(+), 17 deletions(-) diff --git a/loko/common/admin_scheduled_tasks.py b/loko/common/admin_scheduled_tasks.py index 9844049..9b382b7 100644 --- a/loko/common/admin_scheduled_tasks.py +++ b/loko/common/admin_scheduled_tasks.py @@ -180,22 +180,7 @@ class ScheduledCommandAdmin(admin.ModelAdmin): subject = task.email_subject or f"[{prod_name}] Tâche planifiée réussie: {task.name}" # Vérifier si un fichier a été généré - output_file = task.command_args.get('output', '') if isinstance(task.command_args, dict) else '' - - # Substituer les variables dans le nom du fichier - if output_file and '%(year)s' in output_file: - from datetime import date - from dateutil.relativedelta import relativedelta - today = date.today() - prev_month = today.replace(day=1) - relativedelta(months=1) - output_file = output_file.replace('%(year)s', str(prev_month.year)) - output_file = output_file.replace('%(month)s', f'{prev_month.month:02d}') - output_file = output_file.replace('%(day)s', f'{today.day:02d}') - output_file = output_file.replace('%(date)s', today.strftime('%Y-%m-%d')) - - # Convertir en chemin absolu - if not os.path.isabs(output_file): - output_file = os.path.join(settings.BASE_DIR, output_file) + output_file = task.resolve_output_file(log=log) body = f"""Bonjour, diff --git a/loko/common/management/commands/run_scheduled_commands.py b/loko/common/management/commands/run_scheduled_commands.py index 5ecef93..f2c9968 100644 --- a/loko/common/management/commands/run_scheduled_commands.py +++ b/loko/common/management/commands/run_scheduled_commands.py @@ -140,7 +140,7 @@ class Command(BaseCommand): subject = task.email_subject or f"[{prod_name}] Tâche planifiée réussie: {task.name}" # Vérifier si un fichier a été généré - output_file = task.command_args.get('output', '') + output_file = task.resolve_output_file(log=log) body = f""" Bonjour, diff --git a/loko/common/models_scheduled_tasks.py b/loko/common/models_scheduled_tasks.py index 573e55a..fc7c78e 100644 --- a/loko/common/models_scheduled_tasks.py +++ b/loko/common/models_scheduled_tasks.py @@ -7,6 +7,10 @@ from django.conf import settings from django.utils import timezone from django.core.validators import MinValueValidator, MaxValueValidator import json +import os +import re +from datetime import date +from dateutil.relativedelta import relativedelta class ScheduledCommand(models.Model): @@ -216,6 +220,47 @@ class ScheduledCommand(models.Model): except: return str(self.command_args) + def resolve_output_file(self, log=None): + """ + Résout le chemin du fichier généré par la commande (s'il existe). + Gère les variables temporelles (%(year)s, %(month)s, %(day)s, %(date)s) + et convertit les chemins relatifs par rapport à settings.BASE_DIR. + Inspecte également log.output en repli si le fichier n'est pas trouvé directement. + """ + output_file = '' + if isinstance(self.command_args, dict): + output_file = self.command_args.get('output', '') + + if output_file: + today = date.today() + # Par cohérence avec les exports (ex: export_interventions_xlsx), + # pour le mois et l'année, on utilise le mois précédent par défaut + prev_month = today.replace(day=1) - relativedelta(months=1) + + output_file = str(output_file) + output_file = output_file.replace('%(year)s', str(prev_month.year)) + output_file = output_file.replace('%(month)s', f'{prev_month.month:02d}') + output_file = output_file.replace('%(day)s', f'{today.day:02d}') + output_file = output_file.replace('%(date)s', today.strftime('%Y-%m-%d')) + + if not os.path.isabs(output_file): + output_file = os.path.join(settings.BASE_DIR, output_file) + + if os.path.exists(output_file): + return output_file + + # Repli : analyse de log.output pour trouver un fichier généré + if log and getattr(log, 'output', None): + match = re.search(r'(?:Fichier généré|Fichier de sortie)\s*:\s*(.+)', log.output) + if match: + candidate = match.group(1).strip() + if not os.path.isabs(candidate): + candidate = os.path.join(settings.BASE_DIR, candidate) + if os.path.exists(candidate): + return candidate + + return output_file if (output_file and os.path.exists(output_file)) else None + class CommandExecutionLog(models.Model): """ diff --git a/loko/common/tests.py b/loko/common/tests.py index 3343e6c..357d5ad 100644 --- a/loko/common/tests.py +++ b/loko/common/tests.py @@ -3,6 +3,7 @@ from django.contrib.messages import get_messages from django.test import TestCase from django.urls import reverse from django.conf import settings +from django.utils import timezone from django.utils.translation import gettext as _ import json @@ -317,6 +318,80 @@ class UserConfigCopyTests(TestCase): self.assertTrue(target_ut.can_inspect) +class ScheduledCommandResolveOutputFileTests(TestCase): + def test_resolve_output_file_with_placeholders(self): + import os + from unittest.mock import patch + from datetime import date + from dateutil.relativedelta import relativedelta + from common.models_scheduled_tasks import ScheduledCommand + + cmd = ScheduledCommand( + name="Export test", + command_name="export_interventions_xlsx", + command_args={"output": "exports/interventions_export_%(year)s_%(month)s.xlsx"} + ) + + today = date.today() + prev_month = today.replace(day=1) - relativedelta(months=1) + expected_filename = f"interventions_export_{prev_month.year}_{prev_month.month:02d}.xlsx" + expected_path = os.path.join(settings.BASE_DIR, "exports", expected_filename) + + with patch("os.path.exists", side_effect=lambda p: p == expected_path): + resolved = cmd.resolve_output_file() + self.assertEqual(resolved, expected_path) + + def test_resolve_output_file_fallback_from_log_output(self): + import os + from unittest.mock import patch, MagicMock + from common.models_scheduled_tasks import ScheduledCommand + + cmd = ScheduledCommand( + name="Export test", + command_name="export_interventions_xlsx", + command_args={"output": "unknown_file.xlsx"} + ) + + log = MagicMock() + log.output = "Démarrage...\n✓ Fichier généré: /tmp/custom_generated.xlsx\n" + + with patch("os.path.exists", side_effect=lambda p: p == "/tmp/custom_generated.xlsx"): + resolved = cmd.resolve_output_file(log=log) + self.assertEqual(resolved, "/tmp/custom_generated.xlsx") + + def test_send_success_email_attaches_resolved_file(self): + import os + from unittest.mock import patch, MagicMock + from common.models_scheduled_tasks import ScheduledCommand + from common.management.commands.run_scheduled_commands import Command as RunScheduledCommand + + cmd = ScheduledCommand.objects.create( + name="Export mensuel", + command_name="export_interventions_xlsx", + command_args={"output": "exports/interventions_export_%(year)s_%(month)s.xlsx"}, + send_email=True, + email_recipients="test@example.com" + ) + + log = MagicMock() + log.started_at = timezone.now() + log.finished_at = timezone.now() + log.duration_seconds = 1.0 + log.output = "Succès" + + fake_path = os.path.join(settings.BASE_DIR, "exports", "test.xlsx") + + with patch.object(ScheduledCommand, "resolve_output_file", return_value=fake_path), \ + patch("os.path.exists", return_value=True), \ + patch("django.core.mail.EmailMessage.attach_file") as mock_attach, \ + patch("django.core.mail.EmailMessage.send") as mock_send: + runner = RunScheduledCommand() + runner.send_success_email(cmd, log) + + mock_attach.assert_called_once_with(fake_path) + mock_send.assert_called_once() + +