200 lines
7.9 KiB
Python
200 lines
7.9 KiB
Python
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])
|
|
|
|
|