ACORE/core/tests_location_access.py

from datetime import date
from unittest.mock import patch

from django.contrib.auth.models import Group, Permission
from django.test import TestCase, RequestFactory
from django.urls import reverse

from apps.employees.models import Employee
from apps.locations.models import Pharmacy, GroupLocationAccess, PharmacyCardVisibility
from core.models import Directory, DirectoryType
from core.location_access import accessible_pharmacies, accessible_locations, has_all_locations
from core.test_utils import login_2fa


class LocationAccessTests(TestCase):
    @staticmethod
    def directory(kind, code):
        dtype, _ = DirectoryType.objects.get_or_create(code=kind, defaults={'name': kind})
        return Directory.objects.get_or_create(directory_type=dtype, code=code, defaults={'name': code})[0]

    @classmethod
    def setUpTestData(cls):
        directory = cls.directory
        org, region = directory('organization', 'scope-org'), directory('region', 'scope-region')
        cls.own = Pharmacy.objects.create(organization=org, location_region=region, pharmacy_number=701, location_type='pharmacy')
        cls.assigned = Pharmacy.objects.create(organization=org, location_region=region, pharmacy_number=702, location_type='pharmacy')
        cls.foreign = Pharmacy.objects.create(organization=org, location_region=region, pharmacy_number=703, location_type='pharmacy')
        cls.user = Employee.objects.create(username='scoped', email='scope@test.local', hire_date=date(2026, 1, 1),
            user_location_type='pharmacy', user_pharmacy=cls.own, is_staff=True,
            ad_sync_enabled=False, mattermost_sync_enabled=False)
        cls.group = Group.objects.create(name='Scoped operators')
        cls.group.permissions.add(*Permission.objects.filter(codename__in=[
            'access_pharmacies', 'access_utilities', 'access_rro', 'access_it_manager', 'access_internet_billing', 'access_reports']))
        cls.user.groups.add(cls.group)
        # QuerySet update avoids unrelated provisioning hooks/position validation.
        Pharmacy.objects.filter(pk=cls.assigned.pk).update(curator=cls.user)

    def setUp(self):
        self.user = Employee.objects.get(pk=self.user.pk)
        login_2fa(self.client, self.user)

    def request(self, path='/'):
        req = RequestFactory().get(path)
        req.user = self.user
        return req

    def test_staff_is_not_global_and_curator_gets_assigned(self):
        self.assertFalse(has_all_locations(self.user))
        self.assertSetEqual(set(accessible_pharmacies(self.user).values_list('pk', flat=True)), {self.own.pk, self.assigned.pk})
        self.assertSetEqual(set(accessible_locations(self.user).values_list('pk', flat=True)), {self.own.pk, self.assigned.pk})

    def test_head_and_responsible_manager_assignments(self):
        for field in ('head_pharmacist', 'responsible_manager'):
            Pharmacy.objects.filter(pk=self.foreign.pk).update(**{field: self.user})
            self.assertTrue(accessible_pharmacies(self.user).filter(pk=self.foreign.pk).exists())
            Pharmacy.objects.filter(pk=self.foreign.pk).update(**{field: None})
            self.assertFalse(accessible_pharmacies(self.user).filter(pk=self.foreign.pk).exists())

    def test_group_assignments_are_additive_and_revocable(self):
        scope = GroupLocationAccess.objects.create(group=self.group)
        scope.locations.add(self.foreign)
        self.assertTrue(accessible_pharmacies(self.user).filter(pk=self.foreign.pk).exists())
        scope.locations.clear()
        self.assertFalse(accessible_pharmacies(self.user).filter(pk=self.foreign.pk).exists())

    def test_global_permission_is_explicit(self):
        self.user.user_permissions.add(Permission.objects.get(content_type__app_label='locations', codename='view_all_locations'))
        self.user = Employee.objects.get(pk=self.user.pk)
        self.assertTrue(has_all_locations(self.user))
        self.assertEqual(accessible_pharmacies(self.user).count(), 3)

    def test_no_assignment_and_inactive_fail_closed(self):
        Employee.objects.filter(pk=self.user.pk).update(user_pharmacy=None)
        Pharmacy.objects.filter(curator=self.user).update(curator=None)
        user = Employee.objects.get(pk=self.user.pk)
        self.assertFalse(accessible_pharmacies(user).exists())
        user.is_superuser = True
        user.is_active = False
        self.assertFalse(has_all_locations(user))
        self.assertFalse(accessible_locations(user).exists())

    def test_preview_does_not_inherit_admin_locations(self):
        self.user.is_superuser = True
        self.user._access_preview_perms = set()
        self.assertFalse(accessible_locations(self.user).exists())

    def test_lists_and_direct_urls_use_same_scope(self):
        for htmx in ('', 'true'):
            response = self.client.get('/pharmacies/', HTTP_HX_REQUEST=htmx)
            self.assertEqual(response.status_code, 200)
            self.assertNotContains(response, f'/pharmacies/{self.foreign.pk}/')
            # Дислокация — открытый справочник сети: все аптеки видны.
            dislo = self.client.get('/dislocation/', HTTP_HX_REQUEST=htmx)
            self.assertEqual(dislo.status_code, 200)
            self.assertContains(dislo, f'#{self.foreign.pharmacy_number}')
            self.assertEqual(self.client.get(f'/pharmacies/{self.own.pk}/', HTTP_HX_REQUEST=htmx).status_code, 200)
            self.assertEqual(self.client.get(f'/pharmacies/{self.foreign.pk}/', HTTP_HX_REQUEST=htmx).status_code, 404)
        self.assertEqual(self.client.post(reverse('pharmacy_update', args=[self.foreign.pk]), {}).status_code, 404)

    def test_documents_block_foreign_pharmacy_before_webdav(self):
        with patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav:
            response = self.client.get(f'/pharmacies/{self.foreign.pk}/documents/')
        self.assertEqual(response.status_code, 404)
        dav.assert_not_called()

    def test_onlyoffice_token_rechecks_assignment(self):
        from apps.nextcloud_connector.views import _resolve_token_relpath
        payload = {'kind': 'pharmacy', 'pharmacy_id': self.assigned.pk, 'uid': self.user.pk, 'name': 'test.docx'}
        self.assertIsNotNone(_resolve_token_relpath(payload))
        Pharmacy.objects.filter(pk=self.assigned.pk).update(curator=None)
        self.assertIsNone(_resolve_token_relpath(payload))

    def test_utilities_json_and_post_are_scoped(self):
        from apps.utilities.models import UtilityMeterConfig, UtilityReading
        from apps.utilities.views import _utilities_context
        utility = self.directory('utility_type', 'scope-electricity')
        for pharmacy in (self.own, self.foreign):
            UtilityMeterConfig.objects.create(pharmacy=pharmacy, utility_type=utility)
        context = _utilities_context(self.request())
        self.assertNotIn(f'{self.foreign.pk}:', context['meter_configs_json'])
        response = self.client.post('/accounting/?tab=utilities', {
            'pharmacy': self.foreign.pk, 'utility_type': utility.pk,
            'reading_date': '2026-09-21', 'value': '10',
        })
        self.assertEqual(response.status_code, 404)
        self.assertFalse(UtilityReading.objects.exists())

    def test_power_scope_covers_list_counts_and_relocation(self):
        from apps.power_assets.models import PowerAsset
        from apps.power_assets.views import _power_assets_context
        kind, status = self.directory('power_asset_type', 'generator'), self.directory('power_asset_status', 'active')
        own = PowerAsset.objects.create(location=self.own, asset_type=kind, status=status, model_name='Own')
        other = PowerAsset.objects.create(location=self.foreign, asset_type=kind, status=status, model_name='Secret')
        context = _power_assets_context(self.request())
        self.assertEqual(context['power_counts']['total'], 1)
        self.assertEqual(list(context['power_assets_page'].object_list), [own])
        response = self.client.post(reverse('power_asset_update', args=[own.pk]), {
            'location_id': self.foreign.pk, 'asset_type': kind.pk, 'model_name': 'Moved',
        })
        self.assertEqual(response.status_code, 404)
        own.refresh_from_db()
        self.assertEqual(own.location_id, self.own.pk)
        self.assertEqual(self.client.post(reverse('power_asset_update', args=[other.pk]), {}).status_code, 404)

    def test_it_foreign_reference_rejected_before_write(self):
        from apps.it_manager.models import WorkLog
        response = self.client.post(reverse('it_worklog_create'), {
            'location_id': self.foreign.pk, 'description': 'Intrusion', 'work_type': 'other',
        })
        self.assertEqual(response.status_code, 404)
        self.assertFalse(WorkLog.objects.exists())

    def test_network_report_export_denied(self):
        """Экспорт живого отчёта — только руководителю отдела, не scoped pharmacy-user."""
        with patch('apps.reports.views.build_monthly_report') as builder:
            response = self.client.get('/reports/it/monthly/2026-09.docx')
        self.assertEqual(response.status_code, 403)
        builder.assert_not_called()

    def test_department_snapshot_visible_foreign_hidden(self):
        """Опубликованный снимок своего отдела виден; чужого отдела — нет."""
        from apps.reports.documents import user_can_view_report_document
        from apps.reports.models import ReportDocument
        dept = self.directory('department', 'scope-dept')
        other = self.directory('department', 'other-dept')
        self.user.department = dept
        self.user.save(update_fields=['department'])
        own = ReportDocument.objects.create(
            department=dept, title='Own', filename='own.docx',
            kind=ReportDocument.KIND_SNAPSHOT, status=ReportDocument.STATUS_PUBLISHED,
            created_by=self.user,
        )
        foreign = ReportDocument.objects.create(
            department=other, title='Foreign', filename='foreign.docx',
            kind=ReportDocument.KIND_SNAPSHOT, status=ReportDocument.STATUS_PUBLISHED,
            created_by=self.user,
        )
        self.assertTrue(user_can_view_report_document(self.user, own))
        self.assertFalse(user_can_view_report_document(self.user, foreign))

    def test_assignment_edit_requires_separate_permission(self):
        from apps.locations.forms import PortalPharmacyEditForm
        form = PortalPharmacyEditForm(instance=self.own, user=self.user)
        for field in ('curator', 'responsible_manager', 'head_pharmacist'):
            self.assertNotIn(field, form.fields)