from django.test import TransactionTestCase from django.utils import timezone from datetime import timedelta from django.contrib.auth.models import User from interventions.models import Intervention, InterventionTimeLine from django.core.management import call_command from django.db import connection class MaterializedViewFirstResolutionTests(TransactionTestCase): def setUp(self): # Nettoyer les objets existants pour éviter les collisions sous TransactionTestCase InterventionTimeLine.objects.all().delete() Intervention.objects.all().delete() User.objects.all().delete() self.user = User.objects.create_user(username="test_reporting_user", password="password") def test_first_resolution_counts_pause(self): # Création d'une intervention intervention = Intervention.objects.create( title="Test First Resolution", status="finished" ) now = timezone.now() t_to_be_processed = now - timedelta(minutes=120) t_on_pause = now - timedelta(minutes=90) t_finished = now - timedelta(minutes=30) # Événements de timeline: to_be_processed -> on_pause -> finished InterventionTimeLine.objects.create( intervention=intervention, event_user=self.user, event_time=t_to_be_processed, event_type='status_change', from_status='in_preparation', to_status='to_be_processed' ) InterventionTimeLine.objects.create( intervention=intervention, event_user=self.user, event_time=t_on_pause, event_type='status_change', from_status='to_be_processed', to_status='on_pause' ) InterventionTimeLine.objects.create( intervention=intervention, event_user=self.user, event_time=t_finished, event_type='status_change', from_status='on_pause', to_status='finished' ) # Régénérer les vues de reporting call_command('refresh_reporting_views', force_recreate=True) # Vérifier le calcul dans la base de données with connection.cursor() as cursor: cursor.execute( "SELECT first_resolution_min, final_resolution_min FROM reporting.mv_interventions WHERE id = %s", [intervention.id] ) row = cursor.fetchone() self.assertIsNotNone(row, "L'intervention devrait être présente dans la vue matérialisée.") first_res, final_res = row # Temps de première réalisation : de t_to_be_processed à t_on_pause (30 minutes) self.assertEqual(first_res, 30) # Temps de réalisation final : de t_to_be_processed à t_finished (90 minutes) self.assertEqual(final_res, 90) from django.urls import reverse from common.models import UserConfig, Thematic, AppView from observations.models import Observation class ReportingObservationsTests(TransactionTestCase): def setUp(self): # Nettoyer Observation.objects.all().delete() User.objects.all().delete() Thematic.objects.all().delete() AppView.objects.all().delete() # Create user & config self.user = User.objects.create_user(username="reporter", password="password", is_superuser=True) self.user_config = UserConfig.objects.create(user=self.user, is_intern=True) # Add reporting access view self.reporting_view = AppView.objects.create(code="reporting", name_fr="Reporting", name_nl="Reporting", url_name="reporting:dashboard") self.user_config.accessible_views.add(self.reporting_view) # Create Thematic self.thematic = Thematic.objects.create(code="T1", name_fr="Thématique Test", name_nl="Test Thematiek") # Add thematic permission to user_config from common.models import UserThematics UserThematics.objects.create(user_config=self.user_config, thematic=self.thematic, can_view_interventions=True) self.observation = Observation.objects.create( code="OBS00001", description="Test Observation", latitude=50.85, longitude=4.35, thematic=self.thematic, status="to_process", observation_type="incident" ) self.client.login(username="reporter", password="password") def test_reporting_observations_list(self): url = reverse("zapi:reporting-observations") response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["count"], 1) self.assertEqual(response.json()["results"][0]["code"], "OBS00001") def test_reporting_observations_volume_timeseries(self): url = reverse("zapi:reporting-observations-volume-timeseries") response = self.client.get(url, {"granularity": "week", "group_by": "thematic"}) self.assertEqual(response.status_code, 200) self.assertTrue(len(response.json()["results"]) > 0) self.assertEqual(response.json()["results"][0]["count"], 1) def test_reporting_observations_map_locations(self): url = reverse("zapi:reporting-observations-map-locations") response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["type"], "FeatureCollection") self.assertEqual(len(response.json()["features"]), 1) self.assertEqual(response.json()["features"][0]["properties"]["code"], "OBS00001") class ReportingOperatorRoleTests(TransactionTestCase): def setUp(self): Observation.objects.all().delete() User.objects.all().delete() Thematic.objects.all().delete() AppView.objects.all().delete() from common.models import Role self.operator_role, _ = Role.objects.get_or_create(name="operator") self.user = User.objects.create_user(username="operator_user", password="password") self.user_config = UserConfig.objects.create(user=self.user, is_intern=True) self.user_config.roles.add(self.operator_role) self.reporting_view = AppView.objects.create(code="reporting", name_fr="Reporting", name_nl="Reporting", url_name="reporting:dashboard") self.user_config.accessible_views.add(self.reporting_view) self.thematic = Thematic.objects.create(code="T1", name_fr="Thématique Test", name_nl="Test Thematiek") self.observation = Observation.objects.create( code="OBS00001", description="Test Observation", latitude=50.85, longitude=4.35, thematic=self.thematic, status="to_process", observation_type="incident" ) self.client.login(username="operator_user", password="password") def test_operator_permission_clauses(self): from reporting.views import _get_permission_clause, _get_observations_permission_clause clause, params = _get_permission_clause(self.user) self.assertIsNone(clause) self.assertEqual(params, []) obs_clause, obs_params = _get_observations_permission_clause(self.user, table_prefix="o") self.assertIsNone(obs_clause) self.assertEqual(obs_params, []) def test_operator_contract_ids(self): from reporting.views import _get_reporting_contract_ids from contracts.models import Contract, Company from reporting.models import ReportingContractConfig company = Company.objects.create(name="Test Company") contract = Contract.objects.create( contract_number="C01", company=company, start_date=timezone.now().date(), end_date=timezone.now().date() + timedelta(days=365) ) ReportingContractConfig.objects.create(contract=contract, is_enabled=True) contracts = _get_reporting_contract_ids(self.user) self.assertEqual(contracts, [contract.id])