diff --git a/app_new/api/routes.py b/app_new/api/routes.py index c54b8d0..85c4413 100644 --- a/app_new/api/routes.py +++ b/app_new/api/routes.py @@ -5,7 +5,7 @@ Endpoints publics avec authentification par clé API. from flask import Blueprint, request, jsonify, current_app from flask_login import login_required, current_user from functools import wraps -import hmac +from werkzeug.security import check_password_hash from app_new.extensions import db api_bp = Blueprint('api_v1', __name__, url_prefix='/api/v1') @@ -19,12 +19,8 @@ def api_key_required(f): api_key = request.headers.get('X-API-Key') if api_key: from app_new.core.models.settings import AppSettings - settings = AppSettings.query.first() - if ( - settings - and settings.api_key - and hmac.compare_digest(api_key, settings.api_key) - ): + api_key_hash = AppSettings.get('api_key_hash') + if api_key_hash and check_password_hash(api_key_hash, api_key): return f(*args, **kwargs) return jsonify({'error': 'Clé API invalide'}), 401 # Sinon, fallback sur session login @@ -47,7 +43,7 @@ def status(): 'interventions_en_cours': Intervention.query.filter_by(status='en_cours').count(), 'interventions_en_attente': Intervention.query.filter_by(status='en_attente').count(), 'equipments_total': Equipment.query.count(), - 'equipments_panne': Equipment.query.filter_by(status='panne').count(), + 'equipments_panne': Equipment.query.filter(Equipment.status.in_(['en_panne', 'hors_service', 'hs'])).count(), 'rooms_total': Room.query.count(), 'buildings_total': Building.query.count(), }) @@ -59,8 +55,8 @@ def list_interventions(): """Liste des interventions avec pagination.""" from app_new.core.models.maintenance import Intervention - page = int(request.args.get('page', 1)) - per_page = min(int(request.args.get('per_page', 20)), 100) + page = max(request.args.get('page', 1, type=int) or 1, 1) + per_page = min(max(request.args.get('per_page', 20, type=int) or 20, 1), 100) status = request.args.get('status') query = Intervention.query @@ -120,8 +116,8 @@ def list_equipments(): """Liste des équipements avec pagination.""" from app_new.core.models.equipment import Equipment - page = int(request.args.get('page', 1)) - per_page = min(int(request.args.get('per_page', 20)), 100) + page = max(request.args.get('page', 1, type=int) or 1, 1) + per_page = min(max(request.args.get('per_page', 20, type=int) or 20, 1), 100) status = request.args.get('status') query = Equipment.query diff --git a/app_new/core/authorization.py b/app_new/core/authorization.py index f366862..cc16711 100644 --- a/app_new/core/authorization.py +++ b/app_new/core/authorization.py @@ -121,6 +121,12 @@ def required_permission(endpoint, method): if endpoint in {"interventions.create", "interventions.create_for_group"}: return "intervention.create" return "intervention.manage" if mutating else "intervention.view" + if blueprint == "documents": + if "intervention" in endpoint: + return "intervention.manage" if mutating else "intervention.view" + if "equipment" in endpoint: + return "patrimoine.manage" if mutating else "patrimoine.view" + return "system.admin" if blueprint in STOCK_BLUEPRINTS: return "stock.manage" if mutating else "stock.view" if blueprint in CONTRACT_BLUEPRINTS: diff --git a/app_new/core/models/equipment.py b/app_new/core/models/equipment.py index 8743190..4c294d4 100644 --- a/app_new/core/models/equipment.py +++ b/app_new/core/models/equipment.py @@ -243,10 +243,9 @@ class Equipment(db.Model): return self.room def create_scheduled_tasks_from_lot(self, commit=True): - """Crée les tâches planifiées à partir des tâches du lot héritées.""" + """Crée les échéances via le moteur unique de maintenance.""" from .maintenance import LotTask - from .planning import ScheduledTask - from datetime import date, timedelta + from ..services.maintenance_engine import generate_due_tasks # Utiliser le lot effectif (hérité du parent si non défini) effective_lot_id = self.effective_lot_id @@ -255,37 +254,9 @@ class Equipment(db.Model): return [] created_tasks = [] - - # Récupérer les tâches du lot - lot_tasks = LotTask.query.filter_by(lot_id=effective_lot_id).all() + lot_tasks = LotTask.query.filter_by(lot_id=effective_lot_id, is_active=True).all() for lt in lot_tasks: - # Vérifier si une tâche planifiée existe déjà - existing = ScheduledTask.query.filter_by( - lot_task_id=lt.id, - equipment_id=self.id - ).first() - - if not existing: - # Calculer la première date planifiée - scheduled_date = date.today() - if lt.jours_entre_interventions: - scheduled_date = date.today() + timedelta(days=lt.jours_entre_interventions) - - # Créer la tâche planifiée - scheduled_task = ScheduledTask( - lot_task_id=lt.id, - equipment_id=self.id, - room_id=self.room_id or (self.effective_room.id if self.effective_room else None), - scheduled_date=scheduled_date, - estimated_duration=lt.duree_minutes, - status='planned' - ) - db.session.add(scheduled_task) - created_tasks.append(scheduled_task) - - if created_tasks and commit: - db.session.commit() - + created_tasks.extend(generate_due_tasks(lt, equipment_ids={self.id}, commit=commit)) return created_tasks def __repr__(self): diff --git a/app_new/core/routes/admin.py b/app_new/core/routes/admin.py index 2ea1a9a..f01a0e4 100644 --- a/app_new/core/routes/admin.py +++ b/app_new/core/routes/admin.py @@ -3,6 +3,8 @@ Core Routes - Administration GMAO Collège """ from flask import Blueprint, render_template, redirect, url_for, request, flash +import secrets +from werkzeug.security import generate_password_hash from flask_login import login_required, current_user from ...extensions import db from ..models.user import User, Staff @@ -285,6 +287,30 @@ def settings_openrouter(): return jsonify({'success': True, 'message': 'Clé API OpenRouter sauvegardée'}) +@admin_bp.route('/settings/api-key', methods=['POST']) +@login_required +@admin_required +def generate_api_key(): + """Génère une clé API affichée une seule fois et stockée hachée.""" + from ..models.settings import AppSettings + raw_key = f"gmao_{secrets.token_urlsafe(32)}" + AppSettings.set( + 'api_key_hash', generate_password_hash(raw_key), + description='Empreinte de la clé API REST GMAO', + ) + return render_template('admin/api_key_created.html', api_key=raw_key) + + +@admin_bp.route('/settings/api-key/revoke', methods=['POST']) +@login_required +@admin_required +def revoke_api_key(): + from ..models.settings import AppSettings + AppSettings.set('api_key_hash', '', description='Clé API REST révoquée') + flash('Clé API révoquée.', 'success') + return redirect(url_for('admin.settings')) + + @admin_bp.route('/setup-wizard') @login_required @admin_required diff --git a/app_new/core/services/maintenance_engine.py b/app_new/core/services/maintenance_engine.py index 3536ffb..26387e8 100644 --- a/app_new/core/services/maintenance_engine.py +++ b/app_new/core/services/maintenance_engine.py @@ -3,7 +3,7 @@ from datetime import date, timedelta from app_new.extensions import db from app_new.core.models.planning import Meter, ScheduledTask -from app_new.core.models.maintenance import LotTask +from app_new.core.models.maintenance import Intervention, LotTask from app_new.core.services.planning_service import PlanningService @@ -27,7 +27,41 @@ def _working_day(candidate): raise ValueError("Aucune journée travaillée disponible dans les 12 prochains mois.") -def generate_due_tasks(task: LotTask, event=None, today=None): +def _last_completion(task, equipment): + dates = [] + scheduled = ScheduledTask.query.filter( + ScheduledTask.lot_task_id == task.id, + ScheduledTask.equipment_id == equipment.id, + ScheduledTask.status.in_(['completed', 'done']), + ).order_by(ScheduledTask.completed_at.desc(), ScheduledTask.scheduled_date.desc()).first() + if scheduled: + if scheduled.completed_at: + dates.append(scheduled.completed_at.date()) + elif scheduled.scheduled_date: + dates.append(scheduled.scheduled_date) + intervention = Intervention.query.filter( + Intervention.lot_task_id == task.id, + Intervention.equipment_id == equipment.id, + Intervention.status.in_(['terminee', 'cloturee']), + Intervention.is_deleted.is_(False), + ).order_by(Intervention.completed_at.desc(), Intervention.completed_date.desc()).first() + if intervention: + if intervention.completed_at: + dates.append(intervention.completed_at.date()) + elif intervention.completed_date: + dates.append(intervention.completed_date) + elif intervention.scheduled_date: + dates.append(intervention.scheduled_date) + return max(dates) if dates else None + + +def _calendar_due(task, equipment, today): + last = _last_completion(task, equipment) + interval = task.jours_entre_interventions or 0 + return last + timedelta(days=interval) if last and interval else today + + +def generate_due_tasks(task: LotTask, event=None, today=None, equipment_ids=None, commit=True): """Génère les échéances sans doublon et retourne les objets créés.""" today = today or date.today() if not task.is_active: @@ -41,18 +75,54 @@ def generate_due_tasks(task: LotTask, event=None, today=None): return [] created = [] for equipment in _targets(task.lot): + if equipment_ids is not None and equipment.id not in set(equipment_ids): + continue if task.trigger_type == "meter": meter = Meter.query.filter_by(equipment_id=equipment.id, is_active=True).order_by(Meter.current_value.desc()).first() - if not meter or task.meter_threshold is None or meter.current_value < task.meter_threshold: + base_value = meter.last_maintenance_value if meter else None + used_since_maintenance = (meter.current_value - (base_value or meter.initial_value or 0)) if meter else 0 + if not meter or task.meter_threshold is None or used_since_maintenance < task.meter_threshold: continue - existing = ScheduledTask.query.filter_by(lot_task_id=task.id, equipment_id=equipment.id, status="planned").first() + existing = ScheduledTask.query.filter( + ScheduledTask.lot_task_id == task.id, + ScheduledTask.equipment_id == equipment.id, + ScheduledTask.status.in_(['planned', 'in_progress']), + ).first() if existing: continue - due = _working_day(today + timedelta(days=max(task.advance_days or 0, 0))) + due = _calendar_due(task, equipment, today) if task.trigger_type == 'calendar' else today + if task.trigger_type == 'calendar' and due > today + timedelta(days=max(task.advance_days or 0, 0)): + continue + if task.trigger_type == 'season': + completed_this_season = ScheduledTask.query.filter( + ScheduledTask.lot_task_id == task.id, + ScheduledTask.equipment_id == equipment.id, + ScheduledTask.status.in_(['completed', 'done']), + db.extract('year', ScheduledTask.scheduled_date) == today.year, + ).first() + if completed_this_season: + continue + due = _working_day(max(due, today)) item = ScheduledTask(lot_task_id=task.id, equipment_id=equipment.id, room_id=equipment.effective_room.id if equipment.effective_room else None, scheduled_date=due, estimated_duration=task.effective_duration(), status="planned") db.session.add(item) created.append(item) - db.session.commit() + if commit: + db.session.commit() + elif created: + db.session.flush() return created + + +def reconcile_planned_tasks(): + """Déplace les échéances ouvertes placées sur un jour non travaillé.""" + changed = [] + tasks = ScheduledTask.query.filter(ScheduledTask.status.in_(['planned', 'in_progress'])).all() + for item in tasks: + if item.scheduled_date and not PlanningService.get_working_hours(item.scheduled_date, user_id=item.assigned_to_id): + item.scheduled_date = _working_day(item.scheduled_date + timedelta(days=1)) + changed.append(item) + if changed: + db.session.commit() + return changed diff --git a/app_new/core/services/planning_service.py b/app_new/core/services/planning_service.py index a79e1fc..390fbfd 100644 --- a/app_new/core/services/planning_service.py +++ b/app_new/core/services/planning_service.py @@ -7,7 +7,8 @@ from typing import List, Dict, Optional, Tuple from app_new.extensions import db from app_new.core.models.planning import ( ScheduledTask, AdminTask, WorkSchedule, CollegeClosure, - ClosureSchedule, ClosureWorkDay, TechnicianAvailability, PlanningDay, PlanningItem + ClosureSchedule, ClosureWorkDay, TechnicianAvailability, PersonalLeave, + Training, TrainingParticipant, PlanningDay, PlanningItem ) from app_new.core.models.maintenance import Intervention from app_new.core.models.equipment import Equipment @@ -107,7 +108,32 @@ class PlanningService: PRIORITY_FORMATION = 1 @staticmethod - def get_working_hours(d: date) -> Optional[Tuple[time, time, Optional[time], Optional[time]]]: + def maintenance_unavailability(d: date, user_id=None): + """Retourne le motif qui interdit toute maintenance ce jour-là.""" + leave_query = PersonalLeave.query.filter( + PersonalLeave.start_date <= d, + PersonalLeave.end_date >= d, + ) + if user_id is not None: + leave_query = leave_query.filter(PersonalLeave.user_id == user_id) + leave = leave_query.first() + if leave: + return f"Absence : {leave.leave_type or 'indisponible'}" + + training_query = Training.query.join(TrainingParticipant).filter( + Training.start_date <= d, + Training.end_date >= d, + TrainingParticipant.is_confirmed.is_(True), + ) + if user_id is not None: + training_query = training_query.filter(TrainingParticipant.user_id == user_id) + training = training_query.first() + if training: + return f"Formation : {training.name}" + return None + + @staticmethod + def get_working_hours(d: date, user_id=None) -> Optional[Tuple[time, time, Optional[time], Optional[time]]]: """ Retourne les horaires de travail pour une date donnée. Retourne (start, end, lunch_start, lunch_end) ou None si non travaillé. @@ -116,6 +142,9 @@ class PlanningService: is_holiday, holiday_name = is_public_holiday(d) if is_holiday: return None + + if PlanningService.maintenance_unavailability(d, user_id): + return None # Vérifier si c'est un jour de vacances closure = CollegeClosure.query.filter( diff --git a/app_new/documents/routes.py b/app_new/documents/routes.py index 91c79ba..a08cc10 100644 --- a/app_new/documents/routes.py +++ b/app_new/documents/routes.py @@ -7,6 +7,7 @@ from flask_login import login_required, current_user from werkzeug.utils import secure_filename from datetime import datetime, timezone import os +from uuid import uuid4 from app_new.extensions import db from ..core.models.maintenance import InterventionDocument @@ -22,6 +23,8 @@ def allowed_file(filename): @login_required def upload_intervention(intervention_id): """Upload un document pour une intervention.""" + from ..core.models.maintenance import Intervention + Intervention.query.get_or_404(intervention_id) if 'file' not in request.files: flash('Aucun fichier sélectionné.', 'error') return redirect(url_for('interventions.detail', id=intervention_id)) @@ -34,8 +37,7 @@ def upload_intervention(intervention_id): if file and allowed_file(file.filename): filename = secure_filename(file.filename) - timestamp = datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S') - unique_filename = f"{timestamp}_{filename}" + unique_filename = f"{uuid4().hex}_{filename}" upload_dir = os.path.join(current_app.config.get('UPLOAD_FOLDER', 'uploads'), 'interventions') os.makedirs(upload_dir, exist_ok=True) @@ -64,7 +66,8 @@ def upload_intervention(intervention_id): @login_required def upload_equipment(equipment_id): """Upload un document pour un équipement.""" - from ..core.models.equipment import EquipmentDocument + from ..core.models.equipment import Equipment, EquipmentDocument + Equipment.query.get_or_404(equipment_id) if 'file' not in request.files: flash('Aucun fichier sélectionné.', 'error') @@ -78,8 +81,7 @@ def upload_equipment(equipment_id): if file and allowed_file(file.filename): filename = secure_filename(file.filename) - timestamp = datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S') - unique_filename = f"{timestamp}_{filename}" + unique_filename = f"{uuid4().hex}_{filename}" upload_dir = os.path.join(current_app.config.get('UPLOAD_FOLDER', 'uploads'), 'equipments') os.makedirs(upload_dir, exist_ok=True) diff --git a/app_new/interventions/crud.py b/app_new/interventions/crud.py index 528f9c4..4fda0c2 100644 --- a/app_new/interventions/crud.py +++ b/app_new/interventions/crud.py @@ -470,6 +470,15 @@ def change_status(id): old_status = intervention.status new_status = request.form.get('status') comment = request.form.get('comment', '') + + if new_status == 'en_cours': + from ..core.services.planning_service import PlanningService + reason = PlanningService.maintenance_unavailability( + datetime.now(timezone.utc).date(), user_id=current_user.id + ) + if reason: + flash(f'Impossible de démarrer une maintenance aujourd’hui : {reason}.', 'danger') + return redirect(url_for('interventions.detail', id=intervention.id)) if new_status != old_status: intervention.status = new_status @@ -484,7 +493,9 @@ def change_status(id): db.session.add(status_change) if new_status == 'terminee': - intervention.completed_date = datetime.now(timezone.utc).date() + now = datetime.now(timezone.utc) + intervention.completed_date = now.date() + intervention.completed_at = now db.session.commit() flash('Statut modifié.', 'success') @@ -572,4 +583,3 @@ def postpone(id): flash(f"Intervention reportée au {new_date.strftime('%d/%m/%Y')}.", 'success') return redirect(url_for('interventions.detail', id=id)) - diff --git a/app_new/interventions/planning.py b/app_new/interventions/planning.py index 5d29351..12cd0cb 100644 --- a/app_new/interventions/planning.py +++ b/app_new/interventions/planning.py @@ -29,11 +29,13 @@ def planning(): @planning_bp.route('//delete', methods=['POST']) @login_required def delete(id): - """Supprimer définitivement une intervention.""" + """Archiver une intervention sans détruire son historique.""" intervention = Intervention.query.get_or_404(id) - db.session.delete(intervention) + intervention.is_deleted = True + intervention.deleted_at = datetime.utcnow() + intervention.deleted_by_id = current_user.id db.session.commit() - flash('Intervention supprimée définitivement.', 'success') + flash('Intervention placée dans la corbeille. Son historique est conservé.', 'success') return redirect(url_for('interventions.index')) @@ -51,4 +53,4 @@ def restore(id): intervention.deleted_by_id = None db.session.commit() flash('Intervention restaurée.', 'success') - return redirect(url_for('interventions.detail', id=intervention.id)) \ No newline at end of file + return redirect(url_for('interventions.detail', id=intervention.id)) diff --git a/app_new/planning/schedules.py b/app_new/planning/schedules.py index fc3a0ac..10a4628 100644 --- a/app_new/planning/schedules.py +++ b/app_new/planning/schedules.py @@ -250,6 +250,15 @@ def scheduled(): return render_template('planning/scheduled.html', tasks=tasks, status=status) +@planning_bp.route('/scheduled/reconcile', methods=['POST']) +@login_required +def reconcile_scheduled(): + from app_new.core.services.maintenance_engine import reconcile_planned_tasks + changed = reconcile_planned_tasks() + flash(f'{len(changed)} échéance(s) déplacée(s) vers un jour travaillé.', 'success' if changed else 'info') + return redirect(url_for('planning.scheduled')) + + @planning_bp.route('/scheduled/') @login_required def scheduled_detail(id): diff --git a/app_new/planning/templates/planning/scheduled.html b/app_new/planning/templates/planning/scheduled.html index a2bf3d4..f3a51c5 100644 --- a/app_new/planning/templates/planning/scheduled.html +++ b/app_new/planning/templates/planning/scheduled.html @@ -2,6 +2,7 @@ {% block title %}Tâches Planifiées — GMAO Collège{% endblock %} {% block content %} +

Tâches Planifiées

@@ -67,4 +68,4 @@ Aucune tâche planifiée pour le moment.
{% endif %} -{% endblock %} \ No newline at end of file +{% endblock %} diff --git a/app_new/scheduler/engine.py b/app_new/scheduler/engine.py index 6a3e47d..2045baa 100644 --- a/app_new/scheduler/engine.py +++ b/app_new/scheduler/engine.py @@ -25,62 +25,15 @@ def get_working_hours(check_date, user_id=1): Prend en compte les vacances (CollegeClosure) et les horaires normaux (WorkSchedule). Retourne (start_time, end_time, lunch_start, lunch_end) ou None si non travaille. """ - # Verifier si c'est un jour de vacances - closure = CollegeClosure.query.filter( - CollegeClosure.start_date <= check_date, - CollegeClosure.end_date >= check_date - ).first() - - if closure: - exceptional_day = ClosureWorkDay.query.filter_by( - closure_id=closure.id, work_date=check_date - ).first() - if exceptional_day: - return (exceptional_day.start_time, exceptional_day.end_time, - exceptional_day.lunch_start, exceptional_day.lunch_end) - if closure.work_hours_type == 'none': - return None # Ferme - # Horaires reduits pendant les vacances - from ..core.models.planning import ClosureSchedule - cs = ClosureSchedule.query.filter_by( - closure_id=closure.id, - day_of_week=check_date.weekday(), - is_active=True - ).first() - if cs: - return (cs.start_time, cs.end_time, cs.lunch_start, cs.lunch_end) - return None - - # Horaires normaux - ws = WorkSchedule.query.filter_by( - day_of_week=check_date.weekday(), - is_active=True - ).first() - if ws: - return (ws.start_time, ws.end_time, ws.lunch_start, ws.lunch_end) - return None + from ..core.services.planning_service import PlanningService + return PlanningService.get_working_hours(check_date, user_id=user_id) def is_user_available(check_date, user_id=1): """Verifie si l'utilisateur est disponible (pas en conge, pas en formation).""" - # Conge personnel - leave = PersonalLeave.query.filter( - PersonalLeave.user_id == user_id, - PersonalLeave.start_date <= check_date, - PersonalLeave.end_date >= check_date - ).first() - if leave: - return False, f"Conge: {leave.leave_type or 'N/A'}" - - # Formation - training = Training.query.filter( - Training.start_date <= check_date, - Training.end_date >= check_date - ).first() - if training: - return False, f"Formation: {training.name}" - - return True, None + from ..core.services.planning_service import PlanningService + reason = PlanningService.maintenance_unavailability(check_date, user_id=user_id) + return (reason is None, reason) def get_due_tasks(weeks_ahead=4): diff --git a/app_new/templates/admin/api_key_created.html b/app_new/templates/admin/api_key_created.html new file mode 100644 index 0000000..50db0b0 --- /dev/null +++ b/app_new/templates/admin/api_key_created.html @@ -0,0 +1,10 @@ +{% extends "base.html" %} +{% block title %}Nouvelle clé API — GMAO{% endblock %} +{% block content %} +
Copiez cette clé maintenant
+

Elle ne sera plus affichée après avoir quitté cette page.

+
+

Envoyez-la dans l’en-tête HTTP X-API-Key. Ne la placez jamais dans Git.

+ J’ai sauvegardé la clé +
+{% endblock %} diff --git a/app_new/templates/admin/settings.html b/app_new/templates/admin/settings.html index 2c3b2c8..db6a0a5 100644 --- a/app_new/templates/admin/settings.html +++ b/app_new/templates/admin/settings.html @@ -14,6 +14,13 @@
+
+
API REST GMAO
+

La clé donne un accès en lecture à l’API. Elle n’est jamais conservée en clair et ne sera affichée qu’une fois.

+
+
+
+
@@ -190,4 +197,4 @@ function testApiKey() { }); } -{% endblock %} \ No newline at end of file +{% endblock %} diff --git a/app_new/training/routes.py b/app_new/training/routes.py index f17145a..dc4c68f 100644 --- a/app_new/training/routes.py +++ b/app_new/training/routes.py @@ -36,6 +36,13 @@ def create(): location=request.form.get('location'), ) db.session.add(training) + db.session.flush() + db.session.add(TrainingParticipant( + training_id=training.id, + user_id=current_user.id, + is_confirmed=True, + notes='Créateur de la formation', + )) db.session.commit() flash('Formation créée avec succès.', 'success') return redirect(url_for('trainings.index')) diff --git a/docker/backup.py b/docker/backup.py index 345d1c4..dba27c4 100644 --- a/docker/backup.py +++ b/docker/backup.py @@ -7,6 +7,7 @@ import os import sys import subprocess import gzip +import tarfile from datetime import datetime from pathlib import Path @@ -15,6 +16,8 @@ DUMP_DIR.mkdir(parents=True, exist_ok=True) TIMESTAMP = datetime.now().strftime("%Y%m%d_%H%M%S") DUMP_FILE = DUMP_DIR / f"gmao_db_{TIMESTAMP}.sql.gz" +BUNDLE_FILE = DUMP_DIR / f"gmao_backup_{TIMESTAMP}.tar.gz" +UPLOAD_DIR = Path(os.environ.get("UPLOAD_FOLDER", "/app/app_new/uploads")) DB_HOST = os.environ.get("DB_HOST", "mariadb") DB_PORT = os.environ.get("DB_PORT", "3306") @@ -34,6 +37,8 @@ try: database=DB_NAME, charset="utf8mb4" ) cursor = conn.cursor() + cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ") + cursor.execute("START TRANSACTION WITH CONSISTENT SNAPSHOT") # Obtenir la liste des tables cursor.execute("SHOW TABLES") @@ -71,16 +76,38 @@ try: f.write("SET FOREIGN_KEY_CHECKS=1;\n") + conn.rollback() conn.close() + + # Une archive de reprise contient la base cohérente et les documents. + with tarfile.open(BUNDLE_FILE, "w:gz") as archive: + archive.add(DUMP_FILE, arcname=f"database/{DUMP_FILE.name}") + if UPLOAD_DIR.exists(): + archive.add(UPLOAD_DIR, arcname="uploads", recursive=True) + version_file = Path("/app/VERSION") + if version_file.exists(): + archive.add(version_file, arcname="VERSION") + + # Validation minimale immédiate : l'archive doit être lisible et contenir + # un dump SQL non vide. Une restauration complète reste à tester hors ligne. + with tarfile.open(BUNDLE_FILE, "r:gz") as archive: + sql_members = [m for m in archive.getmembers() if m.name.startswith("database/")] + if not sql_members or sql_members[0].size <= 0: + raise RuntimeError("Archive de sauvegarde invalide : dump SQL absent") # Rotation : garder les 7 derniers dumps = sorted(DUMP_DIR.glob("gmao_db_*.sql.gz"), key=lambda p: p.stat().st_mtime, reverse=True) for old in dumps[7:]: old.unlink() + + bundles = sorted(DUMP_DIR.glob("gmao_backup_*.tar.gz"), key=lambda p: p.stat().st_mtime, reverse=True) + for old in bundles[7:]: + old.unlink() size = DUMP_FILE.stat().st_size / 1024 - print(f"{datetime.now()}: Dump cree: {DUMP_FILE} ({size:.0f} KB, {len(tables)} tables)") + bundle_size = BUNDLE_FILE.stat().st_size / 1024 + print(f"{datetime.now()}: Sauvegarde validee: {BUNDLE_FILE} ({bundle_size:.0f} KB, {len(tables)} tables)") except Exception as e: print(f"{datetime.now()}: ERREUR dump: {e}", file=sys.stderr) - sys.exit(1) \ No newline at end of file + sys.exit(1) diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 8929a08..eff9bbc 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -8,7 +8,8 @@ export PYTHONPATH=/home/gmao/.local/lib/python3.13/site-packages:$PYTHONPATH # Si on est root, corriger les permissions des volumes montes if [ "$(id -u)" = "0" ]; then echo "Mise a jour des permissions des volumes..." - chown -R 1000:1000 /app/app_new /app/data /app/migrations /app/gmao_watchdog.py /app/ent_watchdog.py /app/pronote_watchdog.py /app/run_app_new.py /app/watchdog_dnd.py /var/log/supervisor /run/supervisor 2>/dev/null || true + mkdir -p /backups + chown -R 1000:1000 /app/app_new /app/data /app/migrations /app/gmao_watchdog.py /app/ent_watchdog.py /app/pronote_watchdog.py /app/run_app_new.py /app/watchdog_dnd.py /backups /var/log/supervisor /run/supervisor 2>/dev/null || true chmod -R u+rX /app 2>/dev/null || true fi diff --git a/tests/integration/test_lot2_reliability.py b/tests/integration/test_lot2_reliability.py new file mode 100644 index 0000000..4162344 --- /dev/null +++ b/tests/integration/test_lot2_reliability.py @@ -0,0 +1,37 @@ +import re + +from app_new.extensions import db + + +def test_api_key_is_hashed_and_authenticates(authenticated_client, app): + response = authenticated_client.post('/admin/settings/api-key') + assert response.status_code == 200 + match = re.search(rb'value="(gmao_[^"]+)"', response.data) + assert match + raw_key = match.group(1).decode() + + from app_new.core.models.settings import AppSettings + with app.app_context(): + stored = AppSettings.get('api_key_hash') + assert stored + assert raw_key not in stored + + anonymous = app.test_client() + assert anonymous.get('/api/v1/status', headers={'X-API-Key': raw_key}).status_code == 200 + assert anonymous.get('/api/v1/status', headers={'X-API-Key': 'incorrecte'}).status_code == 401 + + +def test_api_pagination_is_bounded(authenticated_client): + response = authenticated_client.get('/api/v1/equipments?page=-4&per_page=10000') + assert response.status_code == 200 + assert response.get_json()['page'] == 1 + assert response.get_json()['per_page'] == 100 + + +def test_document_permissions_follow_parent_domain(app): + from app_new.core.authorization import required_permission + with app.test_request_context('/'): + assert required_permission('documents.download_intervention_document', 'GET') == 'intervention.view' + assert required_permission('documents.delete_intervention_document', 'POST') == 'intervention.manage' + assert required_permission('documents.download_equipment_document', 'GET') == 'patrimoine.view' + assert required_permission('documents.upload_equipment', 'POST') == 'patrimoine.manage' diff --git a/tests/integration/test_maintenance_engine.py b/tests/integration/test_maintenance_engine.py index fb945ef..b5bcd71 100644 --- a/tests/integration/test_maintenance_engine.py +++ b/tests/integration/test_maintenance_engine.py @@ -5,9 +5,12 @@ from app_new.extensions import db from app_new.core.models.college import Building, Room from app_new.core.models.equipment import Equipment, EquipmentCategory from app_new.core.models.maintenance import Lot, LotTask -from app_new.core.models.planning import CollegeClosure, ClosureWorkDay, ScheduledTask +from app_new.core.models.planning import ( + CollegeClosure, ClosureWorkDay, Meter, ScheduledTask, Training, TrainingParticipant, +) from app_new.core.services.maintenance_engine import generate_due_tasks from app_new.scheduler.engine import get_working_hours +from app_new.core.models.user import User def test_exceptional_vacation_day_has_its_specific_hours(app): @@ -36,3 +39,52 @@ def test_lot_task_engine_uses_individual_units_and_prevents_duplicates(app): assert generate_due_tasks(task, event="gel") == [] scheduled = ScheduledTask.query.filter_by(lot_task_id=task.id).one() assert scheduled.equipment_id == unit.id + + +def test_calendar_task_waits_for_interval_after_completion(app): + with app.app_context(): + suffix = uuid4().hex[:7] + building = Building(name=f"B cal {suffix}"); db.session.add(building); db.session.flush() + room = Room(name="Local cal", building_id=building.id) + category = EquipmentCategory(name=f"Cat cal {suffix}") + lot = Lot(name=f"Lot cal {suffix}", category=category) + equipment = Equipment(name=f"Eq cal {suffix}", lot=lot, category=category, room=room) + task = LotTask(lot=lot, tache="Contrôle périodique", trigger_type="calendar", + jours_entre_interventions=30, advance_days=7, is_active=True) + db.session.add_all([room, category, lot, equipment, task]); db.session.flush() + completed = ScheduledTask(lot_task_id=task.id, equipment_id=equipment.id, + scheduled_date=date.today(), status='completed') + db.session.add(completed); db.session.commit() + assert generate_due_tasks(task, today=date.today()) == [] + created = generate_due_tasks(task, today=date.today() + timedelta(days=23)) + assert len(created) == 1 + + +def test_meter_trigger_uses_delta_since_last_maintenance(app): + with app.app_context(): + suffix = uuid4().hex[:7] + building = Building(name=f"B meter {suffix}"); db.session.add(building); db.session.flush() + room = Room(name="Local meter", building_id=building.id) + category = EquipmentCategory(name=f"Cat meter {suffix}") + lot = Lot(name=f"Lot meter {suffix}", category=category) + equipment = Equipment(name=f"Eq meter {suffix}", lot=lot, category=category, room=room) + task = LotTask(lot=lot, tache="Compteur", trigger_type="meter", meter_threshold=100, is_active=True) + db.session.add_all([room, category, lot, equipment, task]); db.session.flush() + meter = Meter(equipment_id=equipment.id, name='Copies', current_value=1050, + initial_value=0, last_maintenance_value=1000, is_active=True) + db.session.add(meter); db.session.commit() + assert generate_due_tasks(task) == [] + meter.current_value = 1100; db.session.commit() + assert len(generate_due_tasks(task)) == 1 + + +def test_confirmed_training_blocks_maintenance_day(app): + with app.app_context(): + suffix = uuid4().hex[:7] + user = User(username=f'tech_{suffix}', email=f'{suffix}@test.local', full_name='Tech', role='technicien') + user.set_password('test-password-strong') + training = Training(name='Formation bloquante', start_date=date.today(), end_date=date.today()) + db.session.add_all([user, training]); db.session.flush() + db.session.add(TrainingParticipant(training_id=training.id, user_id=user.id, is_confirmed=True)) + db.session.commit() + assert get_working_hours(date.today(), user_id=user.id) is None