import io from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase from django.urls import reverse from django.utils.translation import gettext as _ from openpyxl import load_workbook from assets.models import AssetCategory from common.models import Role, Thematic, UserConfig, UserThematics from contracts.models import Company, CompanyTeam, CompanyMember from .models import Observation class ObservationPermissionTests(TestCase): def setUp(self): User = get_user_model() # Thematics self.thematic_a = Thematic.objects.create( code='THEME-A', name_fr='Thématique A', name_nl='Thema A', ) self.thematic_b = Thematic.objects.create( code='THEME-B', name_fr='Thématique B', name_nl='Thema B', ) # Users self.internal_user = User.objects.create_user('internal', password='pass') self.admin_user = User.objects.create_user('admin', password='pass') self.external_user = User.objects.create_user('external', password='pass') self.colleague_user = User.objects.create_user('colleague', password='pass') self.other_company_user = User.objects.create_user('otherco', password='pass') self.unprivileged_user = User.objects.create_user('noaccess', password='pass') # Configurations self.internal_config = UserConfig.objects.create(user=self.internal_user, is_intern=True) self.admin_config = UserConfig.objects.create(user=self.admin_user, is_intern=True) self.external_config = UserConfig.objects.create(user=self.external_user, is_intern=False) self.unprivileged_config = UserConfig.objects.create(user=self.unprivileged_user, is_intern=True) self.admin_role, _ = Role.objects.get_or_create(name='admin') self.admin_config.roles.add(self.admin_role) UserThematics.objects.create( user_config=self.internal_config, thematic=self.thematic_a, can_view_interventions=True, ) UserThematics.objects.create( user_config=self.external_config, thematic=self.thematic_a, can_view_interventions=True, ) # User without thematic access intentionally left blank # Companies for external users self.company = Company.objects.create(name='Ext Company') self.company_team = CompanyTeam.objects.create(name='Ext Team', company=self.company) self.other_company = Company.objects.create(name='Other Company') self.other_team = CompanyTeam.objects.create(name='Other Team', company=self.other_company) m1 = CompanyMember.objects.create(user=self.external_user, name='External Member') m1.teams.add(self.company_team) m2 = CompanyMember.objects.create(user=self.colleague_user, name='Colleague Member') m2.teams.add(self.company_team) m3 = CompanyMember.objects.create(user=self.other_company_user, name='Other Member') m3.teams.add(self.other_team) # Observations for tests self.category_a = AssetCategory.objects.create( thematic=self.thematic_a, code='CAT-A', name_fr='Catégorie A', name_nl='Categorie A', ) self.category_b = AssetCategory.objects.create( thematic=self.thematic_b, code='CAT-B', name_fr='Catégorie B', name_nl='Categorie B', ) self.allowed_observation = Observation.objects.create( description='Allowed internal', latitude=50.0, longitude=4.0, created_by=self.internal_user, thematic=self.thematic_a, ) self.blocked_observation = Observation.objects.create( description='Blocked internal', latitude=50.1, longitude=4.1, created_by=self.internal_user, thematic=self.thematic_b, ) self.external_own = Observation.objects.create( description='External own', latitude=51.0, longitude=4.5, created_by=self.external_user, thematic=self.thematic_a, ) self.external_colleague = Observation.objects.create( description='External colleague', latitude=51.1, longitude=4.6, created_by=self.colleague_user, thematic=self.thematic_a, ) self.external_other_company = Observation.objects.create( description='External other company', latitude=51.2, longitude=4.7, created_by=self.other_company_user, thematic=self.thematic_a, ) self.to_redirect_observation = Observation.objects.create( description='To redirect observation', latitude=52.0, longitude=4.8, created_by=self.internal_user, thematic=self.thematic_a, status='to_redirect', ) def test_geojson_limits_results_to_user_thematics(self): self.client.login(username='internal', password='pass') response = self.client.get(reverse('observations:observations_geojson')) self.assertEqual(response.status_code, 200) data = response.json() feature_ids = {feature['id'] for feature in data['features']} self.assertIn(self.allowed_observation.id, feature_ids) self.assertNotIn(self.blocked_observation.id, feature_ids) def test_external_user_only_sees_company_observations(self): self.client.login(username='external', password='pass') response = self.client.get(reverse('observations:observations_geojson')) self.assertEqual(response.status_code, 200) data = response.json() feature_ids = {feature['id'] for feature in data['features']} self.assertIn(self.external_own.id, feature_ids) self.assertIn(self.external_colleague.id, feature_ids) self.assertNotIn(self.external_other_company.id, feature_ids) def test_detail_view_forbidden_without_permissions(self): self.client.login(username='noaccess', password='pass') response = self.client.get( reverse('observations:observation_detail', args=[self.allowed_observation.id]) ) self.assertEqual(response.status_code, 403) def test_to_redirect_observation_can_be_modified_by_author_or_admin(self): edit_url = reverse('observations:observation_edit', args=[self.to_redirect_observation.id]) detail_url = reverse('observations:observation_detail', args=[self.to_redirect_observation.id]) mobile_detail_url = reverse('observations:observation_detail_mobile', args=[self.to_redirect_observation.id]) self.client.login(username='internal', password='pass') detail_response = self.client.get(detail_url) self.assertEqual(detail_response.status_code, 200) self.assertContains(detail_response, edit_url) mobile_response = self.client.get(mobile_detail_url) self.assertEqual(mobile_response.status_code, 200) self.assertContains(mobile_response, edit_url) edit_response = self.client.get(edit_url) self.assertEqual(edit_response.status_code, 200) self.client.logout() self.client.login(username='admin', password='pass') admin_detail_response = self.client.get(detail_url) self.assertEqual(admin_detail_response.status_code, 200) self.assertContains(admin_detail_response, edit_url) admin_edit_response = self.client.get(edit_url) self.assertEqual(admin_edit_response.status_code, 200) def test_detail_view_shows_company_next_to_shortname(self): self.client.login(username='external', password='pass') detail_url = reverse('observations:observation_detail', args=[self.external_own.id]) mobile_detail_url = reverse('observations:observation_detail_mobile', args=[self.external_own.id]) # Test desktop detail_response = self.client.get(detail_url) self.assertEqual(detail_response.status_code, 200) self.assertContains(detail_response, "(Ext Company)") # Test mobile mobile_response = self.client.get(mobile_detail_url) self.assertEqual(mobile_response.status_code, 200) self.assertContains(mobile_response, "(Ext Company)") def _build_creation_payload(self, *, thematic_id, category_id, description): return { 'description': description, 'latitude': '50.5', 'longitude': '4.5', 'observation_type': 'inventory', 'thematic': str(thematic_id), 'category': str(category_id), 'address': 'Test address', } def test_observation_add_rejects_disallowed_thematic(self): self.client.login(username='internal', password='pass') photo = SimpleUploadedFile('photo.jpg', b'filecontent', content_type='image/jpeg') payload = self._build_creation_payload( thematic_id=self.thematic_b.id, category_id=self.category_b.id, description='Disallowed thematic', ) payload['photos'] = photo response = self.client.post( reverse('observations:observation_add'), payload, ) self.assertEqual(response.status_code, 200) form = response.context['obs_form'] self.assertIn('thematic', form.errors) self.assertIn("Vous n'avez pas accès à cette thématique.", form.errors['thematic']) self.assertFalse( Observation.objects.filter(description='Disallowed thematic', created_by=self.internal_user).exists() ) def test_observation_add_allows_accessible_thematic(self): self.client.login(username='internal', password='pass') photo = SimpleUploadedFile('photo2.jpg', b'filecontent', content_type='image/jpeg') payload = self._build_creation_payload( thematic_id=self.thematic_a.id, category_id=self.category_a.id, description='Allowed thematic creation', ) payload['photos'] = photo response = self.client.post( reverse('observations:observation_add'), payload, ) self.assertEqual(response.status_code, 302) self.assertTrue( Observation.objects.filter(description='Allowed thematic creation', created_by=self.internal_user).exists() ) def test_geojson_includes_symptom(self): from interventions.models import Symptom symptom = Symptom.objects.create( name_fr='Symptome A', name_nl='Symptoom A', thematic=self.thematic_a, ) self.allowed_observation.symptom = symptom self.allowed_observation.save() self.client.login(username='internal', password='pass') response = self.client.get(reverse('observations:observations_geojson')) self.assertEqual(response.status_code, 200) data = response.json() features = data['features'] allowed_feature = next(f for f in features if f['id'] == self.allowed_observation.id) self.assertEqual(allowed_feature['properties']['symptom'], 'Symptome A') def test_export_excel_includes_symptom(self): from interventions.models import Symptom symptom = Symptom.objects.create( name_fr='Symptome A', name_nl='Symptoom A', thematic=self.thematic_a, ) self.internal_user.first_name = 'Alice' self.internal_user.last_name = 'Dupont' self.internal_user.save(update_fields=['first_name', 'last_name']) self.allowed_observation.symptom = symptom self.allowed_observation.address = '80 Boulevard Industriel / Industrielaan (Anderlecht)' self.allowed_observation.save(update_fields=['symptom', 'address']) self.client.login(username='internal', password='pass') response = self.client.get(reverse('observations:observations_export_excel')) self.assertEqual(response.status_code, 200) self.assertEqual( response['Content-Type'], "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) wb = load_workbook(filename=io.BytesIO(response.content)) ws = wb.active headers = [cell.value for cell in ws[1]] symptom_header = str(_('Symptôme')) author_header = str(_('Auteur')) street_header = str(_('Rue')) commune_header = str(_('Commune')) code_header = str(_('Code')) self.assertIn(symptom_header, headers) self.assertIn(author_header, headers) self.assertIn(street_header, headers) self.assertIn(commune_header, headers) code_col = headers.index(code_header) + 1 target_row = None for row_idx in range(2, ws.max_row + 1): if ws.cell(row=row_idx, column=code_col).value == self.allowed_observation.code: target_row = row_idx break self.assertIsNotNone(target_row) symptom_col = headers.index(symptom_header) + 1 author_col = headers.index(author_header) + 1 street_col = headers.index(street_header) + 1 commune_col = headers.index(commune_header) + 1 self.assertEqual(ws.cell(row=target_row, column=symptom_col).value, 'Symptome A') self.assertEqual(ws.cell(row=target_row, column=author_col).value, 'Alice D.') self.assertEqual(ws.cell(row=target_row, column=street_col).value, 'Boulevard Industriel / Industrielaan') self.assertEqual(ws.cell(row=target_row, column=commune_col).value, 'Anderlecht') def test_auto_create_intervention_on_to_process(self): from interventions.models import Symptom, Intervention, InterventionTimeLine from contracts.models import Contract, Company from django.utils import timezone company = Company.objects.create(name='Provider Co') contract = Contract.objects.create( contract_number='C-123', company=company, start_date=timezone.now().date(), end_date=timezone.now().date() + timezone.timedelta(days=365) ) symptom = Symptom.objects.create( name_fr='Symptome Auto', name_nl='Symptoom Auto', thematic=self.thematic_a, auto_create_intervention=True, auto_create_intervention_status='to_be_planned', contract=contract, provider=company, priority='1', ) obs = Observation.objects.create( description='Auto creation test description', latitude=50.5, longitude=4.5, address='123 Rue de la Paix', created_by=self.internal_user, thematic=self.thematic_a, category=self.category_a, symptom=symptom, status='to_process', ) obs.refresh_from_db() self.assertEqual(obs.status, 'in_progress') self.assertIsNotNone(obs.intervention) # Check intervention was created with correct fields intervention = obs.intervention self.assertEqual(intervention.title, 'Symptome Auto - 123 Rue de la Paix') self.assertEqual(intervention.status, 'to_be_planned') self.assertEqual(intervention.description, 'Auto creation test description') self.assertEqual(intervention.priority, '1') self.assertEqual(intervention.contract, contract) self.assertEqual(intervention.assigned_provider, company) self.assertEqual(intervention.asset_category, self.category_a) self.assertEqual(intervention.thematic, self.thematic_a) self.assertEqual(intervention.lat, 50.5) self.assertEqual(intervention.lon, 4.5) self.assertIsNotNone(intervention.geom) # Check timeline events timeline_exists = InterventionTimeLine.objects.filter( intervention=intervention, event_type='creation', to_status='to_be_planned' ).exists() self.assertTrue(timeline_exists) # Check comment exists comment_exists = obs.comments.filter( comment__contains=intervention.code ).exists() self.assertTrue(comment_exists) def test_no_auto_create_on_in_preparation(self): from interventions.models import Symptom, Intervention symptom = Symptom.objects.create( name_fr='Symptome Auto 2', name_nl='Symptoom Auto 2', thematic=self.thematic_a, auto_create_intervention=True, ) obs = Observation.objects.create( description='Draft observation', latitude=50.5, longitude=4.5, created_by=self.internal_user, thematic=self.thematic_a, symptom=symptom, status='in_preparation', ) obs.refresh_from_db() self.assertIsNone(obs.intervention) self.assertEqual(obs.status, 'in_preparation') obs.status = 'to_process' obs.save() obs.refresh_from_db() self.assertEqual(obs.status, 'in_progress') self.assertIsNotNone(obs.intervention) def test_photo_copy_auto_created_intervention(self): from interventions.models import Symptom, Intervention, InterventionDocument from .models import ObservationPhoto from PIL import Image symptom = Symptom.objects.create( name_fr='Symptome Auto 3', name_nl='Symptoom Auto 3', thematic=self.thematic_a, auto_create_intervention=True, ) obs = Observation.objects.create( description='Photo test observation', latitude=50.5, longitude=4.5, created_by=self.internal_user, thematic=self.thematic_a, symptom=symptom, status='to_process', ) obs.refresh_from_db() self.assertIsNotNone(obs.intervention) intervention = obs.intervention file_obj = io.BytesIO() image = Image.new('RGB', (1, 1), color='red') image.save(file_obj, 'JPEG') file_obj.seek(0) photo_file = SimpleUploadedFile('test_photo.jpg', file_obj.read(), content_type='image/jpeg') photo = ObservationPhoto.objects.create( observation=obs, photo=photo_file, ) docs = InterventionDocument.objects.filter(intervention=intervention) self.assertEqual(docs.count(), 1) doc = docs.first() self.assertEqual(doc.uploaded_by, self.internal_user) def test_canceled_intervention_alert_on_observation_detail(self): from interventions.models import Intervention intervention = Intervention.objects.create( code="INT-TEST-CANCEL", title="Canceled intervention test", status="canceled", cancellation_type="duplicate", thematic=self.thematic_a, lat=50.5, lon=4.5, ) obs = Observation.objects.create( description="Observation with canceled intervention", latitude=50.5, longitude=4.5, created_by=self.internal_user, thematic=self.thematic_a, status="to_process", intervention=intervention, ) self.client.login(username="admin", password="pass") detail_url = reverse("observations:observation_detail", args=[obs.id]) response = self.client.get(detail_url) self.assertEqual(response.status_code, 200) self.assertContains(response, "Une intervention liée à cette observation a été annulée") self.assertContains(response, "INT-TEST-CANCEL") def test_no_auto_intervention_loop_on_cancellation(self): from contracts.models import Contract, Company from interventions.models import Symptom, Intervention from interventions.views.workflow import attempt_status_update from django.utils import timezone company = Company.objects.create(name='Provider Co Loop') contract = Contract.objects.create( contract_number='C-LOOP', company=company, start_date=timezone.now().date(), end_date=timezone.now().date() + timezone.timedelta(days=365) ) symptom = Symptom.objects.create( name_fr='Symptome Auto Loop Test', name_nl='Symptoom Auto Loop Test', thematic=self.thematic_a, auto_create_intervention=True, contract=contract, provider=company, ) # 1. Create observation in 'to_process' -> auto-creates intervention obs = Observation.objects.create( description='Loop test observation', latitude=50.5, longitude=4.5, created_by=self.internal_user, thematic=self.thematic_a, symptom=symptom, status='to_process', ) obs.refresh_from_db() self.assertIsNotNone(obs.intervention) self.assertEqual(obs.status, 'in_progress') first_intervention = obs.intervention # 2. Cancel the intervention success, msg, _ = attempt_status_update( request=None, intervention=first_intervention, user=self.admin_user, new_status='canceled', cancellation_type='other', cancellation_reason='Test cancellation reason', ) self.assertTrue(success, msg) # 3. Observation status is reset to 'to_process' obs.refresh_from_db() self.assertEqual(obs.status, 'to_process') # 4. Save observation again (or unlink and save) - MUST NOT create a new intervention obs.save() obs.refresh_from_db() self.assertEqual(obs.status, 'to_process') # Unlink intervention and save observation again obs.intervention = None obs.save() obs.refresh_from_db() # Total interventions with source_ref equal to obs.code or obs.pk must still be 1 possible_refs = [str(obs.pk), obs.code, f"OBS{obs.pk:05d}"] total_interventions = Intervention.objects.filter(source_ref__in=possible_refs).count() self.assertEqual(total_interventions, 1) self.assertEqual(obs.status, 'to_process')