818 lines
35 KiB
Python
818 lines
35 KiB
Python
from django.test import TestCase
|
|
|
|
from apps.nextcloud_connector.nextcloud_service import (
|
|
NextcloudService, NextcloudConfigError,
|
|
)
|
|
from apps.nextcloud_connector.services import sync_service as ss
|
|
|
|
|
|
class _FakeDept:
|
|
def __init__(self, code, name):
|
|
self.code, self.name = code, name
|
|
self.parent = None
|
|
self.pk = None
|
|
|
|
|
|
class _FakePharmacy:
|
|
def __init__(self, number, address=''):
|
|
self.pharmacy_number, self.address = number, address
|
|
|
|
|
|
class MountpointTests(TestCase):
|
|
def test_department_mountpoint_and_group(self):
|
|
d = _FakeDept('hr', 'HR')
|
|
self.assertEqual(ss.dept_mountpoint(d), '03_Отделы/HR')
|
|
self.assertEqual(ss.dept_group_name(d), 'nc-dept-hr')
|
|
|
|
def test_department_mountpoint_hierarchical(self):
|
|
parent = _FakeDept('admin', 'Административный')
|
|
parent.pk = 1
|
|
child = _FakeDept('legal', 'Юридическая служба')
|
|
child.pk = 2
|
|
child.parent = parent
|
|
self.assertEqual(
|
|
ss.dept_mountpoint(child),
|
|
'03_Отделы/Административный/Юридическая служба',
|
|
)
|
|
|
|
def test_pharmacy_mountpoint_zero_padded_with_address(self):
|
|
p = _FakePharmacy(17, 'Одесса, ул. Дерибасовская, 1')
|
|
self.assertEqual(ss.pharmacy_mountpoint(p),
|
|
'02_Аптеки/A-017 · Одесса, ул. Дерибасовская, 1')
|
|
|
|
def test_pharmacy_mountpoint_without_address(self):
|
|
self.assertEqual(ss.pharmacy_mountpoint(_FakePharmacy(3)), '02_Аптеки/A-003')
|
|
|
|
|
|
class ConfigGateTests(TestCase):
|
|
def test_service_raises_without_config(self):
|
|
# ConnectorConfig(nextcloud) не заведён — сервис должен упасть предсказуемо
|
|
with self.assertRaises(NextcloudConfigError):
|
|
NextcloudService()
|
|
|
|
|
|
class EnsureFolderTests(TestCase):
|
|
"""Идемпотентность ensure_folder без сети — низкоуровневые вызовы замоканы."""
|
|
|
|
def _service(self):
|
|
svc = object.__new__(NextcloudService) # без __init__ (не лезем в конфиг/сеть)
|
|
svc.calls = []
|
|
svc.create_folder = lambda mp: (svc.calls.append(('create', mp)) or 42)
|
|
svc.add_group = lambda fid, g: svc.calls.append(('add_group', fid, g))
|
|
svc.set_group_permissions = lambda fid, g, p: svc.calls.append(('perm', fid, g, p))
|
|
svc.set_quota = lambda fid, q: svc.calls.append(('quota', fid, q))
|
|
return svc
|
|
|
|
def test_creates_when_absent(self):
|
|
svc = self._service()
|
|
cache = {}
|
|
fid, created = svc.ensure_folder('03_Отделы/HR',
|
|
{'nc-dept-hr': NextcloudService.PERM_RW},
|
|
folder_cache=cache)
|
|
self.assertTrue(created)
|
|
self.assertEqual(fid, 42)
|
|
self.assertIn(('create', '03_Отделы/HR'), svc.calls)
|
|
self.assertIn(('perm', 42, 'nc-dept-hr', 15), svc.calls)
|
|
self.assertEqual(cache['03_Отделы/HR'], 42) # кэш обновлён
|
|
|
|
def test_skips_create_when_present(self):
|
|
svc = self._service()
|
|
fid, created = svc.ensure_folder('03_Отделы/HR',
|
|
{'nc-dept-hr': NextcloudService.PERM_RW},
|
|
folder_cache={'03_Отделы/HR': 7})
|
|
self.assertFalse(created)
|
|
self.assertEqual(fid, 7)
|
|
self.assertNotIn(('create', '03_Отделы/HR'), svc.calls)
|
|
self.assertIn(('perm', 7, 'nc-dept-hr', 15), svc.calls)
|
|
|
|
|
|
import datetime
|
|
import json
|
|
from django.test import Client, TestCase
|
|
from unittest.mock import patch
|
|
|
|
from core.models import ConnectorConfig, Directory, DirectoryType
|
|
from apps.nextcloud_connector.onlyoffice import (
|
|
can_edit_file, decode_file_token, document_key, encode_ds_jwt,
|
|
encode_file_token, is_allowed_ds_download_url, is_office_file,
|
|
is_previewable_file, office_content_type,
|
|
)
|
|
|
|
from apps.locations.models import Pharmacy
|
|
from apps.employees.models import Employee
|
|
from apps.nextcloud_connector.access import (
|
|
can_access_pharmacy_docs, can_edit_pharmacy_docs, safe_filename,
|
|
)
|
|
from core.test_utils import login_2fa
|
|
|
|
|
|
class SafeFilenameTests(TestCase):
|
|
def test_accepts_plain_names(self):
|
|
for n in ('licence.pdf', 'Договор 2026.docx', 'фото-1.jpg'):
|
|
self.assertEqual(safe_filename(n), n)
|
|
|
|
def test_rejects_traversal_and_separators(self):
|
|
for n in ('../secret', 'a/b.txt', 'a\\b.txt', '..', '.', '', None,
|
|
'.hidden', 'x\x00y'):
|
|
self.assertIsNone(safe_filename(n), f'должно отклонить: {n!r}')
|
|
|
|
|
|
class AccessControlTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
org_type = DirectoryType.objects.create(code='organization', name='Орг')
|
|
region_type = DirectoryType.objects.create(code='region', name='Регион')
|
|
region = Directory.objects.create(directory_type=region_type, code='od', name='Одесса')
|
|
org = Directory.objects.create(directory_type=org_type, code='akc', name='АКС')
|
|
cls.ph_a = Pharmacy.objects.create(organization=org, location_region=region,
|
|
pharmacy_number=17, location_type='pharmacy')
|
|
cls.ph_b = Pharmacy.objects.create(organization=org, location_region=region,
|
|
pharmacy_number=18, location_type='pharmacy')
|
|
common = dict(mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1))
|
|
cls.emp_a = Employee.objects.create_user(
|
|
username='ph_a_emp', user_location_type='pharmacy', user_pharmacy=cls.ph_a, **common)
|
|
cls.emp_b = Employee.objects.create_user(
|
|
username='ph_b_emp', user_location_type='pharmacy', user_pharmacy=cls.ph_b, **common)
|
|
cls.staff = Employee.objects.create_user(username='office', is_staff=True, **common)
|
|
|
|
def test_pharmacy_employee_sees_only_own(self):
|
|
self.assertTrue(can_access_pharmacy_docs(self.emp_a, self.ph_a))
|
|
self.assertFalse(can_access_pharmacy_docs(self.emp_a, self.ph_b))
|
|
|
|
def test_pharmacy_employee_view_only_by_default(self):
|
|
# show_documents=own_or_staff, edit_pharmacy=staff
|
|
self.assertTrue(can_access_pharmacy_docs(self.emp_a, self.ph_a))
|
|
self.assertFalse(can_edit_pharmacy_docs(self.emp_a, self.ph_a))
|
|
|
|
def test_staff_needs_explicit_network_access(self):
|
|
from core.test_utils import grant_all_locations
|
|
self.assertFalse(can_access_pharmacy_docs(self.staff, self.ph_a))
|
|
grant_all_locations(self.staff)
|
|
self.assertTrue(can_access_pharmacy_docs(self.staff, self.ph_a))
|
|
self.assertTrue(can_access_pharmacy_docs(self.staff, self.ph_b))
|
|
self.assertTrue(can_edit_pharmacy_docs(self.staff, self.ph_a))
|
|
|
|
|
|
class DocumentsViewTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
org_type = DirectoryType.objects.create(code='organization', name='Орг')
|
|
region_type = DirectoryType.objects.create(code='region', name='Регион')
|
|
region = Directory.objects.create(directory_type=region_type, code='od', name='Одесса')
|
|
org = Directory.objects.create(directory_type=org_type, code='akc', name='АКС')
|
|
cls.ph_a = Pharmacy.objects.create(organization=org, location_region=region,
|
|
pharmacy_number=17, location_type='pharmacy')
|
|
cls.ph_b = Pharmacy.objects.create(organization=org, location_region=region,
|
|
pharmacy_number=18, location_type='pharmacy')
|
|
common = dict(mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1))
|
|
cls.emp_a = Employee.objects.create_user(
|
|
username='ph_a_emp', user_location_type='pharmacy', user_pharmacy=cls.ph_a, **common)
|
|
|
|
def setUp(self):
|
|
self.client = Client()
|
|
|
|
def test_own_pharmacy_ok_but_storage_not_configured(self):
|
|
self.client.force_login(self.emp_a)
|
|
r = self.client.get(f'/pharmacies/{self.ph_a.pk}/documents/')
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, 'не настроен') # коннектор не сконфигурирован в тестах
|
|
|
|
def test_other_pharmacy_forbidden(self):
|
|
self.client.force_login(self.emp_a)
|
|
r = self.client.get(f'/pharmacies/{self.ph_b.pk}/documents/')
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_anonymous_redirected(self):
|
|
r = self.client.get(f'/pharmacies/{self.ph_a.pk}/documents/')
|
|
self.assertEqual(r.status_code, 302)
|
|
|
|
def test_download_rejects_bad_filename(self):
|
|
self.client.force_login(self.emp_a)
|
|
r = self.client.get(f'/pharmacies/{self.ph_a.pk}/documents/download/?name=../etc/passwd')
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
def test_download_other_pharmacy_forbidden(self):
|
|
self.client.force_login(self.emp_a)
|
|
r = self.client.get(f'/pharmacies/{self.ph_b.pk}/documents/download/?name=x.pdf')
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_upload_forbidden_without_edit(self):
|
|
self.client.force_login(self.emp_a)
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
r = self.client.post(
|
|
f'/pharmacies/{self.ph_a.pk}/documents/upload/',
|
|
{'file': SimpleUploadedFile('a.txt', b'x')},
|
|
)
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
|
|
from apps.nextcloud_connector.access import (
|
|
accessible_sections, get_section, safe_relpath,
|
|
)
|
|
|
|
|
|
class SafeRelpathTests(TestCase):
|
|
def test_empty_is_root(self):
|
|
self.assertEqual(safe_relpath(''), '')
|
|
self.assertEqual(safe_relpath(None), '')
|
|
|
|
def test_valid_nested(self):
|
|
self.assertEqual(safe_relpath('Приказы/2026'), 'Приказы/2026')
|
|
self.assertEqual(safe_relpath('/Приказы/'), 'Приказы')
|
|
|
|
def test_rejects_traversal(self):
|
|
for p in ('../etc', 'a/../b', 'a/./b', 'a//b', 'a/.hidden'):
|
|
self.assertIsNone(safe_relpath(p), f'должно отклонить: {p!r}')
|
|
|
|
|
|
class SectionsTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
common = dict(mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1))
|
|
cls.plain = Employee.objects.create_user(username='plain', **common)
|
|
cls.staff = Employee.objects.create_user(username='st', is_staff=True, **common)
|
|
cls.root = Employee.objects.create_user(
|
|
username='root', is_staff=True, is_superuser=True, **common)
|
|
|
|
def _keys(self, user):
|
|
return {s['key'] for s in accessible_sections(user)}
|
|
|
|
def test_plain_user_sections(self):
|
|
# без отдела — вкладки company/org/training без «Мой отдел»
|
|
self.assertEqual(self._keys(self.plain), {'company', 'org', 'training'})
|
|
|
|
def test_mine_tab_when_user_has_department(self):
|
|
from core.models import Directory, DirectoryType
|
|
dt = DirectoryType.objects.create(code='department', name='Відділ')
|
|
dept = Directory.objects.create(directory_type=dt, code='it', name='IT')
|
|
self.plain.department = dept
|
|
self.plain.save(update_fields=['department'])
|
|
keys = accessible_sections(self.plain)
|
|
self.assertEqual(keys[0]['key'], 'mine')
|
|
self.assertIn('mine', self._keys(self.plain))
|
|
|
|
def test_company_writable_only_for_staff(self):
|
|
company_plain = get_section(self.plain, 'company')
|
|
company_staff = get_section(self.staff, 'company')
|
|
self.assertFalse(company_plain['writable'])
|
|
self.assertTrue(company_staff['writable'])
|
|
|
|
def test_superuser_gets_hr(self):
|
|
self.assertIn('hr', self._keys(self.root))
|
|
self.assertNotIn('hr', self._keys(self.staff))
|
|
|
|
|
|
class DocumentsBrowserViewTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
from django.contrib.auth.models import Group
|
|
cls.user = Employee.objects.create_user(
|
|
username='browser', mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1))
|
|
# Любая группа = доступ к открытым разделам (в т.ч. /documents/).
|
|
cls.user.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
|
|
def setUp(self):
|
|
self.client = Client()
|
|
|
|
def test_browse_valid_section_ok(self):
|
|
self.client.force_login(self.user)
|
|
r = self.client.get('/documents/browse/?section=company')
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, 'не настроен')
|
|
|
|
def test_browse_inaccessible_section_forbidden(self):
|
|
self.client.force_login(self.user)
|
|
r = self.client.get('/documents/browse/?section=hr') # не суперюзер
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_browse_traversal_path_bad_request(self):
|
|
self.client.force_login(self.user)
|
|
r = self.client.get('/documents/browse/?section=company&path=../04_Кадровые дела')
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
def test_upload_to_readonly_section_forbidden(self):
|
|
self.client.force_login(self.user) # обычный юзер: company только чтение
|
|
r = self.client.post('/documents/upload/', {'section': 'company', 'path': ''})
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_anonymous_redirected(self):
|
|
r = self.client.get('/documents/browse/?section=company')
|
|
self.assertEqual(r.status_code, 302)
|
|
|
|
def test_mkdir_rename_delete_readonly_forbidden(self):
|
|
self.client.force_login(self.user)
|
|
for url in ('/documents/mkdir/', '/documents/rename/', '/documents/delete/'):
|
|
r = self.client.post(url, {'section': 'company', 'path': '', 'name': 'x'})
|
|
self.assertEqual(r.status_code, 403, url)
|
|
|
|
def test_staff_mkdir_rename_delete(self):
|
|
from unittest.mock import patch
|
|
from django.contrib.auth.models import Group
|
|
staff = Employee.objects.create_user(
|
|
username='docs_staff', is_staff=True,
|
|
mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1),
|
|
)
|
|
staff.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
login_2fa(self.client, staff)
|
|
dav = patch('apps.nextcloud_connector.views.NextcloudWebDAVService').start()
|
|
self.addCleanup(patch.stopall)
|
|
inst = dav.return_value
|
|
inst.list_dir.return_value = []
|
|
r = self.client.post('/documents/mkdir/', {'section': 'company', 'path': '', 'name': 'Папка'})
|
|
self.assertEqual(r.status_code, 200)
|
|
inst.mkdir.assert_called_once()
|
|
r = self.client.post(
|
|
'/documents/rename/',
|
|
{'section': 'company', 'path': '', 'name': 'a.txt', 'new_name': 'b.txt'},
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
inst.move.assert_called_once()
|
|
r = self.client.post('/documents/delete/', {'section': 'company', 'path': '', 'name': 'b.txt'})
|
|
self.assertEqual(r.status_code, 200)
|
|
inst.delete.assert_called_once()
|
|
r = self.client.post('/documents/mkdir/', {'section': 'company', 'path': '', 'name': '../x'})
|
|
self.assertContains(r, 'Некорректное имя', status_code=200)
|
|
inst.mkdir.assert_called_once() # второй раз не вызван
|
|
r = self.client.post(
|
|
'/documents/create/',
|
|
{'section': 'company', 'path': '', 'type': 'docx', 'name': 'Протокол'},
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
inst.upload.assert_called_once()
|
|
uploaded_name = inst.upload.call_args[0][0]
|
|
self.assertTrue(str(uploaded_name).endswith('Протокол.docx'))
|
|
|
|
def test_rename_via_hx_prompt_header(self):
|
|
from unittest.mock import patch
|
|
from django.contrib.auth.models import Group
|
|
staff = Employee.objects.create_user(
|
|
username='docs_prompt', is_staff=True,
|
|
mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1),
|
|
)
|
|
staff.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
login_2fa(self.client, staff)
|
|
with patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav:
|
|
dav.return_value.list_dir.return_value = []
|
|
r = self.client.post(
|
|
'/documents/rename/',
|
|
{'section': 'company', 'path': '', 'name': 'old.pdf'},
|
|
headers={'HX-Prompt': 'new.pdf'},
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
dav.return_value.move.assert_called_once()
|
|
|
|
|
|
from apps.nextcloud_connector.doc_acl import (
|
|
grant_owner_edit, resolve_access, user_can_create_in,
|
|
)
|
|
from apps.nextcloud_connector.services.sync_service import (
|
|
safe_component, dept_mountpoint,
|
|
)
|
|
|
|
|
|
class DocumentOwnerAclTests(TestCase):
|
|
"""Автор — edit; колеги — view; керівник відділу — edit."""
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
dtype = DirectoryType.objects.create(code='department', name='Відділ')
|
|
cls.dept = Directory.objects.create(directory_type=dtype, code='it', name='IT')
|
|
common = dict(
|
|
mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1),
|
|
user_location_type='office',
|
|
department=cls.dept,
|
|
)
|
|
cls.head = Employee.objects.create_user(username='acl_head', **common)
|
|
cls.author = Employee.objects.create_user(
|
|
username='acl_author', manager=cls.head, **common,
|
|
)
|
|
cls.peer = Employee.objects.create_user(
|
|
username='acl_peer', manager=cls.head, **common,
|
|
)
|
|
|
|
def test_owner_gets_edit_peer_view_head_edit(self):
|
|
grant_owner_edit(
|
|
self.author,
|
|
section_key='org',
|
|
path='doc.docx',
|
|
department_id=self.dept.pk,
|
|
is_folder=False,
|
|
)
|
|
owner = resolve_access(
|
|
self.author, section_key='org', path='doc.docx',
|
|
department_id=self.dept.pk,
|
|
)
|
|
peer = resolve_access(
|
|
self.peer, section_key='org', path='doc.docx',
|
|
department_id=self.dept.pk,
|
|
)
|
|
head = resolve_access(
|
|
self.head, section_key='org', path='doc.docx',
|
|
department_id=self.dept.pk,
|
|
)
|
|
self.assertTrue(owner.can_edit)
|
|
self.assertTrue(peer.can_view)
|
|
self.assertFalse(peer.can_edit)
|
|
self.assertTrue(head.can_edit)
|
|
|
|
def test_own_dept_can_create_without_folder_edit(self):
|
|
self.assertTrue(user_can_create_in(
|
|
self.author, section_key='org', path='', department_id=self.dept.pk,
|
|
))
|
|
folder = resolve_access(
|
|
self.author, section_key='org', path='', department_id=self.dept.pk,
|
|
)
|
|
self.assertTrue(folder.can_view)
|
|
self.assertFalse(folder.can_edit)
|
|
|
|
def test_upload_creates_owner_grant(self):
|
|
from django.contrib.auth.models import Group
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from unittest.mock import patch
|
|
self.author.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
client = Client()
|
|
client.force_login(self.author)
|
|
with patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav:
|
|
dav.return_value.list_dir.return_value = []
|
|
r = client.post(
|
|
'/documents/upload/',
|
|
{
|
|
'section': 'mine',
|
|
'path': '',
|
|
'file': SimpleUploadedFile('mine.docx', b'PK'),
|
|
},
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
from apps.nextcloud_connector.models import DocumentAccessGrant
|
|
g = DocumentAccessGrant.objects.filter(
|
|
employee=self.author, section_key='org', path='mine.docx',
|
|
department_id=self.dept.pk, is_active=True,
|
|
).first()
|
|
self.assertIsNotNone(g)
|
|
self.assertEqual(g.level, 'edit')
|
|
|
|
|
|
class PathHardeningTests(TestCase):
|
|
def test_safe_component_strips_separators(self):
|
|
self.assertEqual(safe_component('a/b'), 'a-b')
|
|
self.assertEqual(safe_component('a\\b'), 'a-b')
|
|
self.assertEqual(safe_component('..'), '_')
|
|
self.assertNotIn('/', safe_component('../secret'))
|
|
|
|
def test_dept_mountpoint_no_traversal(self):
|
|
class _D:
|
|
name = '../04_Кадровые дела'
|
|
parent = None
|
|
pk = 1
|
|
mp = dept_mountpoint(_D())
|
|
self.assertTrue(mp.startswith('03_Отделы/'))
|
|
self.assertNotIn('/../', mp)
|
|
self.assertNotIn('/04_', mp) # имя схлопнуто в один сегмент
|
|
|
|
|
|
class ExchangeWritableTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
common = dict(mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1))
|
|
cls.plain = Employee.objects.create_user(username='ex_plain', **common)
|
|
cls.staff = Employee.objects.create_user(username='ex_staff', is_staff=True, **common)
|
|
|
|
def test_training_writable_only_for_staff(self):
|
|
self.assertFalse(get_section(self.plain, 'training')['writable'])
|
|
self.assertTrue(get_section(self.staff, 'training')['writable'])
|
|
|
|
|
|
class OnlyOfficeHelpersTests(TestCase):
|
|
def test_office_extensions(self):
|
|
self.assertTrue(is_office_file('Договор.docx'))
|
|
self.assertTrue(is_office_file('table.XLSX'))
|
|
self.assertFalse(is_office_file('scan.pdf'))
|
|
self.assertFalse(is_office_file('photo.jpg'))
|
|
self.assertTrue(can_edit_file('a.docx'))
|
|
self.assertFalse(can_edit_file('a.pdf'))
|
|
self.assertTrue(is_previewable_file('scan.pdf'))
|
|
self.assertTrue(is_previewable_file('photo.JPG'))
|
|
self.assertFalse(is_previewable_file('a.docx'))
|
|
|
|
def test_file_token_roundtrip(self):
|
|
payload = {'kind': 'section', 'section': 'company', 'path': '', 'name': 'a.docx', 'uid': 1}
|
|
token = encode_file_token(payload)
|
|
self.assertEqual(decode_file_token(token)['name'], 'a.docx')
|
|
|
|
def test_document_key_changes_with_mtime(self):
|
|
path = '01_Компания/a.docx'
|
|
k1 = document_key(path, size=10, modified='Wed, 01 Jan 2026 00:00:00 GMT')
|
|
k2 = document_key(path, size=11, modified='Thu, 02 Jan 2026 00:00:00 GMT')
|
|
self.assertNotEqual(k1, k2)
|
|
self.assertEqual(document_key(path, size=10, modified='Wed, 01 Jan 2026 00:00:00 GMT'), k1)
|
|
|
|
|
|
class OnlyOfficeViewTests(TestCase):
|
|
JWT_SECRET = 'test-jwt-secret-32-bytes-long!!'
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
from django.contrib.auth.models import Group
|
|
cls.user = Employee.objects.create_user(
|
|
username='oo_user', mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1), is_staff=True,
|
|
)
|
|
cls.user.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
ConnectorConfig.objects.create(
|
|
connector_type='onlyoffice',
|
|
is_enabled=True,
|
|
host='docs.test.local',
|
|
use_ssl=True,
|
|
token=cls.JWT_SECRET,
|
|
extra_config={'callback_base': 'https://portal.test.local'},
|
|
)
|
|
|
|
def setUp(self):
|
|
self.client = Client()
|
|
login_2fa(self.client, self.user)
|
|
|
|
def _file_token(self):
|
|
return encode_file_token({
|
|
'kind': 'section',
|
|
'section': 'company',
|
|
'path': '',
|
|
'name': 'a.docx',
|
|
'uid': self.user.pk,
|
|
'mode': 'edit',
|
|
})
|
|
|
|
def test_editor_rejects_non_office(self):
|
|
r = self.client.get('/documents/edit/?section=company&name=photo.jpg')
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
def test_editor_page_for_docx(self):
|
|
r = self.client.get('/documents/edit/?section=company&name=Договор.docx')
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, 'DocsAPI')
|
|
self.assertContains(r, 'docs.test.local')
|
|
|
|
def test_content_rejects_garbage_token(self):
|
|
anon = Client()
|
|
jwt_hdr = f'Bearer {encode_ds_jwt({"payload": {"url": "https://portal.test.local/x"}}, self.JWT_SECRET)}'
|
|
r = anon.get(
|
|
'/documents/onlyoffice/content/not-a-token/',
|
|
HTTP_AUTHORIZATION=jwt_hdr,
|
|
)
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_content_rejects_missing_jwt(self):
|
|
anon = Client()
|
|
r = anon.get(f'/documents/onlyoffice/content/{self._file_token()}/')
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
def test_callback_rejects_unsigned_body(self):
|
|
anon = Client()
|
|
with patch('apps.nextcloud_connector.views.fetch_saved_file') as fetch:
|
|
r = anon.post(
|
|
f'/documents/onlyoffice/callback/{self._file_token()}/',
|
|
data=json.dumps({
|
|
'status': 2,
|
|
'url': 'https://evil.example/payload',
|
|
}),
|
|
content_type='application/json',
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json(), {'error': 1})
|
|
fetch.assert_not_called()
|
|
|
|
def test_callback_rejects_wrong_jwt_secret(self):
|
|
anon = Client()
|
|
bad = encode_ds_jwt(
|
|
{'status': 2, 'url': 'https://docs.test.local/cache/a.docx'},
|
|
'wrong-secret-32-bytes-long!!!!!',
|
|
)
|
|
with patch('apps.nextcloud_connector.views.fetch_saved_file') as fetch:
|
|
r = anon.post(
|
|
f'/documents/onlyoffice/callback/{self._file_token()}/',
|
|
data=json.dumps({'status': 2, 'url': 'https://docs.test.local/cache/a.docx'}),
|
|
content_type='application/json',
|
|
HTTP_AUTHORIZATION=f'Bearer {bad}',
|
|
)
|
|
self.assertEqual(r.json(), {'error': 1})
|
|
fetch.assert_not_called()
|
|
|
|
def test_callback_save_uploads_to_nextcloud(self):
|
|
ds_jwt = encode_ds_jwt(
|
|
{'status': 2, 'url': 'https://docs.test.local/cache/a.docx'},
|
|
self.JWT_SECRET,
|
|
)
|
|
anon = Client()
|
|
with patch('apps.nextcloud_connector.views.fetch_saved_file', return_value=b'DOCX') as fetch, \
|
|
patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav_cls:
|
|
dav_cls.return_value.upload.return_value = None
|
|
r = anon.post(
|
|
f'/documents/onlyoffice/callback/{self._file_token()}/',
|
|
data=json.dumps({'status': 2, 'url': 'https://evil.example/ignore'}),
|
|
content_type='application/json',
|
|
HTTP_AUTHORIZATION=f'Bearer {ds_jwt}',
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.json(), {'error': 0})
|
|
fetch.assert_called_once()
|
|
self.assertEqual(fetch.call_args[0][0], 'https://docs.test.local/cache/a.docx')
|
|
dav_cls.return_value.upload.assert_called_once()
|
|
|
|
def test_ssrf_url_not_from_ds_host(self):
|
|
settings = {
|
|
'public_url': 'https://docs.test.local',
|
|
'download_hosts': [],
|
|
}
|
|
self.assertFalse(is_allowed_ds_download_url('https://evil.example/x', settings))
|
|
self.assertTrue(is_allowed_ds_download_url('https://docs.test.local/cache/a.docx', settings))
|
|
|
|
def test_content_accepts_jwt_query_param(self):
|
|
ds_jwt = encode_ds_jwt({'url': 'https://portal.test.local/x'}, self.JWT_SECRET)
|
|
anon = Client()
|
|
with patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav:
|
|
dav.return_value.download.return_value = (b'PKDOCX', 'application/octet-stream')
|
|
r = anon.get(
|
|
f'/documents/onlyoffice/content/{self._file_token()}/',
|
|
{'token': ds_jwt},
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.content, b'PKDOCX')
|
|
self.assertIn(
|
|
'officedocument.wordprocessingml.document',
|
|
r['Content-Type'],
|
|
)
|
|
|
|
|
|
class DocumentPreviewTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
from django.contrib.auth.models import Group
|
|
org_type = DirectoryType.objects.create(code='organization', name='Орг')
|
|
region_type = DirectoryType.objects.create(code='region', name='Регион')
|
|
region = Directory.objects.create(directory_type=region_type, code='od', name='Одесса')
|
|
org = Directory.objects.create(directory_type=org_type, code='akc', name='АКС')
|
|
cls.ph = Pharmacy.objects.create(
|
|
organization=org, location_region=region,
|
|
pharmacy_number=33, location_type='pharmacy',
|
|
)
|
|
common = dict(
|
|
mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1),
|
|
)
|
|
cls.emp = Employee.objects.create_user(
|
|
username='prev_emp',
|
|
user_location_type='pharmacy',
|
|
user_pharmacy=cls.ph,
|
|
**common,
|
|
)
|
|
cls.staff = Employee.objects.create_user(
|
|
username='prev_staff', is_staff=True, **common,
|
|
)
|
|
# раздел /documents/ требует любую группу (perm=None)
|
|
cls.staff.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
|
|
def setUp(self):
|
|
self.client = Client()
|
|
|
|
def test_pharmacy_preview_page_for_pdf(self):
|
|
self.client.force_login(self.emp)
|
|
r = self.client.get(
|
|
f'/pharmacies/{self.ph.pk}/documents/preview/?name=scan.pdf',
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, 'doc-preview-frame')
|
|
self.assertContains(r, 'inline=1')
|
|
|
|
def test_pharmacy_preview_page_for_image(self):
|
|
self.client.force_login(self.emp)
|
|
r = self.client.get(
|
|
f'/pharmacies/{self.ph.pk}/documents/preview/?name=фото.png',
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, 'doc-preview-image')
|
|
|
|
def test_preview_rejects_office_file(self):
|
|
self.client.force_login(self.emp)
|
|
r = self.client.get(
|
|
f'/pharmacies/{self.ph.pk}/documents/preview/?name=a.docx',
|
|
)
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
def test_inline_download_serves_pdf(self):
|
|
self.client.force_login(self.emp)
|
|
with patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav:
|
|
dav.return_value.download.return_value = (b'%PDF-1.4', 'application/octet-stream')
|
|
r = self.client.get(
|
|
f'/pharmacies/{self.ph.pk}/documents/download/?name=scan.pdf&inline=1',
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r['Content-Type'], 'application/pdf')
|
|
self.assertIn('inline', r['Content-Disposition'])
|
|
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
|
|
|
|
def test_section_preview_for_staff(self):
|
|
login_2fa(self.client, self.staff)
|
|
r = self.client.get('/documents/preview/?section=company&name=scan.pdf')
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, 'doc-preview-frame')
|
|
|
|
|
|
class PharmacyOnlyOfficeAclTests(TestCase):
|
|
JWT_SECRET = 'test-jwt-secret-32-bytes-long!!'
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
from django.contrib.auth.models import Group
|
|
from core.test_utils import grant_all_locations
|
|
org_type = DirectoryType.objects.create(code='organization', name='Орг')
|
|
region_type = DirectoryType.objects.create(code='region', name='Регион')
|
|
region = Directory.objects.create(directory_type=region_type, code='od', name='Одесса')
|
|
org = Directory.objects.create(directory_type=org_type, code='akc', name='АКС')
|
|
cls.ph = Pharmacy.objects.create(
|
|
organization=org, location_region=region,
|
|
pharmacy_number=21, location_type='pharmacy',
|
|
)
|
|
common = dict(
|
|
mattermost_sync_enabled=False, ad_sync_enabled=False,
|
|
hire_date=datetime.date(2020, 1, 1),
|
|
)
|
|
cls.viewer = Employee.objects.create_user(
|
|
username='ph_viewer', user_location_type='pharmacy',
|
|
user_pharmacy=cls.ph, **common,
|
|
)
|
|
cls.editor = Employee.objects.create_user(
|
|
username='ph_editor', is_staff=True, **common,
|
|
)
|
|
grant_all_locations(cls.editor)
|
|
cls.viewer.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
cls.editor.groups.add(Group.objects.get_or_create(name='Сотрудники (тест)')[0])
|
|
ConnectorConfig.objects.create(
|
|
connector_type='onlyoffice',
|
|
is_enabled=True,
|
|
host='docs.test.local',
|
|
use_ssl=True,
|
|
token=cls.JWT_SECRET,
|
|
extra_config={'callback_base': 'https://portal.test.local'},
|
|
)
|
|
|
|
def setUp(self):
|
|
self.client = Client()
|
|
|
|
def test_viewer_opens_readonly(self):
|
|
login_2fa(self.client, self.viewer)
|
|
with patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav:
|
|
dav.return_value.stat.return_value = {'size': 10, 'modified': None}
|
|
r = self.client.get(
|
|
f'/documents/edit/?pharmacy={self.ph.pk}&name=Договор.docx',
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, '"mode": "view"')
|
|
|
|
def test_editor_opens_writable(self):
|
|
login_2fa(self.client, self.editor)
|
|
with patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav:
|
|
dav.return_value.stat.return_value = {'size': 10, 'modified': None}
|
|
r = self.client.get(
|
|
f'/documents/edit/?pharmacy={self.ph.pk}&name=Договор.docx',
|
|
)
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertContains(r, '"mode": "edit"')
|
|
|
|
def test_callback_rejects_save_for_viewer(self):
|
|
token = encode_file_token({
|
|
'kind': 'pharmacy',
|
|
'pharmacy_id': self.ph.pk,
|
|
'name': 'a.docx',
|
|
'uid': self.viewer.pk,
|
|
'mode': 'view',
|
|
})
|
|
ds_jwt = encode_ds_jwt(
|
|
{'status': 2, 'url': 'https://docs.test.local/cache/a.docx'},
|
|
self.JWT_SECRET,
|
|
)
|
|
with patch('apps.nextcloud_connector.views.fetch_saved_file') as fetch, \
|
|
patch('apps.nextcloud_connector.views.NextcloudWebDAVService') as dav_cls:
|
|
r = Client().post(
|
|
f'/documents/onlyoffice/callback/{token}/',
|
|
data=json.dumps({'status': 2, 'url': 'https://docs.test.local/cache/a.docx'}),
|
|
content_type='application/json',
|
|
HTTP_AUTHORIZATION=f'Bearer {ds_jwt}',
|
|
)
|
|
self.assertEqual(r.json(), {'error': 1})
|
|
fetch.assert_not_called()
|
|
dav_cls.return_value.upload.assert_not_called()
|
|
|
|
|
|
class BlankOfficeCreateTests(TestCase):
|
|
def test_office_mime(self):
|
|
self.assertIn('wordprocessingml', office_content_type('a.docx'))
|
|
self.assertIn('spreadsheetml', office_content_type('a.xlsx'))
|
|
|