nextcloud: публичные RO-ссылки на документы аптек; аптеки больше не на Group Folders.
Нужен внешний (без входа в ACORE) read-only доступ к документам аптеки — для сторон вроде поставщиков/проверяющих. Для сотрудников внутренний доступ уже был (access.can_access_pharmacy_docs через виджет портала), это отдельный, внешний путь. - NextcloudSharingService — клиент OCS Files Sharing API (публичные ссылки, идемпотентно: сначала проверяет существующую шару на путь, не плодит дубли). - webdav_service.ensure_dir() — идемпотентный MKCOL (в отличие от mkdir(), который должен падать на дубликате — так и оставлен для UI). - PharmacyPublicLink — аптека ↔ URL + share_id (для последующего отзыва/ обновления). - sync_service.sync_pharmacy(): больше не использует Group Folders (аптечные сотрудники учёток Nextcloud не имеют — ACL по AD-группам тут не нужен, это была подготовка под LDAP, которая пока не горит). Теперь — обычная WebDAV-папка под сервисным аккаунтом + публичная RO-ссылка. Попутно чинит сегодняшний блокер: --pharmacies-only больше не зависит от сломанного (404) Group Folders API — клиенты в NextcloudSyncService стали ленивыми. - Отделы (sync_department) не тронуты — Group Folders/LDAP остаётся архитектурным заделом на будущее, не нужно прямо сейчас. Тесты: SharingServiceTests, SyncPharmacyPublicLinkTests (идемпотентность, обновление URL, обработка ошибок WebDAV, dry-run). 326 тестов в затронутых приложениях — зелёные. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
from django.contrib import admin
|
||||
from unfold.admin import ModelAdmin
|
||||
|
||||
from .models import DocumentAccessGrant, NextcloudSyncLog
|
||||
from .models import DocumentAccessGrant, NextcloudSyncLog, PharmacyPublicLink
|
||||
|
||||
|
||||
@admin.register(DocumentAccessGrant)
|
||||
@@ -25,3 +25,15 @@ class NextcloudSyncLogAdmin(ModelAdmin):
|
||||
|
||||
def has_add_permission(self, request):
|
||||
return False
|
||||
|
||||
|
||||
@admin.register(PharmacyPublicLink)
|
||||
class PharmacyPublicLinkAdmin(ModelAdmin):
|
||||
"""Только чтение — ссылки провижинит nextcloud_sync, руками тут не правят."""
|
||||
list_display = ['pharmacy', 'url', 'updated_at']
|
||||
search_fields = ['pharmacy__pharmacy_number', 'pharmacy__address', 'url']
|
||||
autocomplete_fields = ['pharmacy']
|
||||
readonly_fields = ['pharmacy', 'share_id', 'url', 'created_at', 'updated_at']
|
||||
|
||||
def has_add_permission(self, request):
|
||||
return False
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
# Generated by Django 6.0.3 on 2026-11-02 21:07
|
||||
|
||||
import django.db.models.deletion
|
||||
import uuid
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('locations', '0016_ukrainian_phone_field'),
|
||||
('nextcloud_connector', '0003_document_global_permissions'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='PharmacyPublicLink',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('uuid', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('updated_at', models.DateTimeField(auto_now=True)),
|
||||
('share_id', models.CharField(max_length=32, verbose_name='ID ссылки в Nextcloud')),
|
||||
('url', models.URLField(max_length=500, verbose_name='Публичная ссылка')),
|
||||
('pharmacy', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='nextcloud_public_link', to='locations.pharmacy', verbose_name='Аптека')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'Публичная ссылка на документы аптеки',
|
||||
'verbose_name_plural': 'Публичные ссылки на документы аптек',
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -145,3 +145,27 @@ class NextcloudSyncLog(BaseModel):
|
||||
|
||||
def __str__(self):
|
||||
return f'{self.get_entity_type_display()} {self.entity_ref} — {self.get_action_display()} — {self.get_status_display()}'
|
||||
|
||||
|
||||
class PharmacyPublicLink(BaseModel):
|
||||
"""Публичная RO-ссылка Nextcloud на папку документов аптеки.
|
||||
|
||||
Для доступа БЕЗ входа в портал (поставщики, проверяющие и т.п.) — не
|
||||
путать с виджетом портала (access.can_access_pharmacy_docs), который
|
||||
обслуживает залогиненных сотрудников своим ACL.
|
||||
share_id — id шары в Nextcloud, нужен для отзыва/обновления прав.
|
||||
"""
|
||||
pharmacy = models.OneToOneField(
|
||||
'locations.Pharmacy', on_delete=models.CASCADE,
|
||||
related_name='nextcloud_public_link',
|
||||
verbose_name=_('Аптека'),
|
||||
)
|
||||
share_id = models.CharField(_('ID ссылки в Nextcloud'), max_length=32)
|
||||
url = models.URLField(_('Публичная ссылка'), max_length=500)
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('Публичная ссылка на документы аптеки')
|
||||
verbose_name_plural = _('Публичные ссылки на документы аптек')
|
||||
|
||||
def __str__(self):
|
||||
return f'{self.pharmacy} → {self.url}'
|
||||
|
||||
@@ -1,21 +1,24 @@
|
||||
"""Провижининг Group Folders из сущностей портала.
|
||||
"""Провижининг из сущностей портала.
|
||||
|
||||
Портал — источник правды. Отделы (Directory type=department) → папки в
|
||||
Портал — источник правды. Отделы (Directory type=department) → Group Folders в
|
||||
`03_Отделы`, привязанные к AD-группе `nc-dept-<код>` (её ведёт AD-коннектор,
|
||||
Nextcloud читает через LDAP). Аптеки → папки в `02_Аптеки`, привязанные к
|
||||
сервисной группе `nc-svc`: аптечные сотрудники учёток Nextcloud не имеют,
|
||||
доступ к документам идёт через виджет портала под сервисным аккаунтом.
|
||||
Nextcloud читает через LDAP) — задел под будущую связку Nextcloud↔LDAP, пока не
|
||||
подключённую. Аптеки → обычные WebDAV-папки в `02_Аптеки` под сервисным
|
||||
аккаунтом (учёток Nextcloud у аптечных сотрудников нет: внутренний доступ —
|
||||
через виджет портала, access.can_access_pharmacy_docs; внешний, без входа в
|
||||
ACORE, — через публичную RO-ссылку, которую сюда же и провижинит сервис).
|
||||
"""
|
||||
import logging
|
||||
|
||||
from apps.nextcloud_connector.models import NextcloudSyncLog
|
||||
from apps.nextcloud_connector.nextcloud_service import NextcloudService
|
||||
from apps.nextcloud_connector.sharing_service import NextcloudSharingService
|
||||
from apps.nextcloud_connector.webdav_service import NextcloudWebDAVService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEPT_ROOT = '03_Отделы'
|
||||
PHARM_ROOT = '02_Аптеки'
|
||||
SVC_GROUP = 'nc-svc' # сервисная группа для доступа портала к папкам аптек
|
||||
|
||||
|
||||
def safe_component(name: str) -> str:
|
||||
@@ -69,14 +72,41 @@ def pharmacy_mountpoint(pharmacy) -> str:
|
||||
|
||||
|
||||
class NextcloudSyncService:
|
||||
"""Тонкая обёртка над NextcloudService с логированием в NextcloudSyncLog."""
|
||||
"""Тонкая обёртка над Nextcloud-клиентами с логированием в NextcloudSyncLog.
|
||||
|
||||
Клиенты (Group Folders / WebDAV / Sharing) создаются лениво — по одному на
|
||||
то, что реально используется. Это важно для --pharmacies-only: аптеки
|
||||
больше не трогают Group Folders API, так что массовый прогон только по
|
||||
аптекам не зависит от его доступности.
|
||||
"""
|
||||
|
||||
def __init__(self, dry_run: bool = False):
|
||||
self.dry_run = dry_run
|
||||
self.svc = None if dry_run else NextcloudService()
|
||||
self._folder_cache = None if dry_run else self.svc.list_folders()
|
||||
self._svc = None
|
||||
self._folder_cache = None
|
||||
self._webdav = None
|
||||
self._sharing = None
|
||||
|
||||
# --- отделы ---
|
||||
@property
|
||||
def svc(self) -> NextcloudService:
|
||||
if self._svc is None:
|
||||
self._svc = NextcloudService()
|
||||
self._folder_cache = self._svc.list_folders()
|
||||
return self._svc
|
||||
|
||||
@property
|
||||
def webdav(self) -> NextcloudWebDAVService:
|
||||
if self._webdav is None:
|
||||
self._webdav = NextcloudWebDAVService()
|
||||
return self._webdav
|
||||
|
||||
@property
|
||||
def sharing(self) -> NextcloudSharingService:
|
||||
if self._sharing is None:
|
||||
self._sharing = NextcloudSharingService()
|
||||
return self._sharing
|
||||
|
||||
# --- отделы (Group Folders — задел под LDAP, см. докстринг модуля) ---
|
||||
|
||||
def sync_department(self, directory) -> NextcloudSyncLog:
|
||||
mp = dept_mountpoint(directory)
|
||||
@@ -85,7 +115,8 @@ class NextcloudSyncService:
|
||||
return self._log('department', directory.code, mp, 'skip', 'dry_run',
|
||||
f'Папка «{mp}» ↔ группа {group} (RW)')
|
||||
try:
|
||||
_, created = self.svc.ensure_folder(
|
||||
svc = self.svc # инициализирует self._folder_cache
|
||||
_, created = svc.ensure_folder(
|
||||
mp, {group: NextcloudService.PERM_RW},
|
||||
folder_cache=self._folder_cache,
|
||||
)
|
||||
@@ -96,21 +127,27 @@ class NextcloudSyncService:
|
||||
logger.error(f'Nextcloud: отдел {directory.code} — {e}')
|
||||
return self._log('department', directory.code, mp, 'create', 'error', str(e))
|
||||
|
||||
# --- аптеки ---
|
||||
# --- аптеки (WebDAV-папка + публичная RO-ссылка) ---
|
||||
|
||||
def sync_pharmacy(self, pharmacy) -> NextcloudSyncLog:
|
||||
mp = pharmacy_mountpoint(pharmacy)
|
||||
ref = str(pharmacy.pharmacy_number)
|
||||
if self.dry_run:
|
||||
return self._log('pharmacy', ref, mp, 'skip', 'dry_run',
|
||||
f'Папка «{mp}» ↔ {SVC_GROUP} (RW, доступ через портал)')
|
||||
f'Папка «{mp}» (WebDAV) + публичная RO-ссылка')
|
||||
try:
|
||||
_, created = self.svc.ensure_folder(
|
||||
mp, {SVC_GROUP: NextcloudService.PERM_RW},
|
||||
folder_cache=self._folder_cache,
|
||||
created_dir = self.webdav.ensure_dir(mp)
|
||||
share, created_link = self.sharing.ensure_public_link(
|
||||
mp, permissions=NextcloudSharingService.PERM_RO,
|
||||
)
|
||||
action = 'create' if created else 'update'
|
||||
return self._log('pharmacy', ref, mp, action, 'success', f'{SVC_GROUP} (RW)')
|
||||
from apps.nextcloud_connector.models import PharmacyPublicLink
|
||||
PharmacyPublicLink.objects.update_or_create(
|
||||
pharmacy=pharmacy,
|
||||
defaults={'share_id': str(share['id']), 'url': share['url']},
|
||||
)
|
||||
action = 'create' if (created_dir or created_link) else 'update'
|
||||
return self._log('pharmacy', ref, mp, action, 'success',
|
||||
f'папка + публичная ссылка: {share["url"]}')
|
||||
except Exception as e:
|
||||
logger.error(f'Nextcloud: аптека {ref} — {e}')
|
||||
return self._log('pharmacy', ref, mp, 'create', 'error', str(e))
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
"""Клиент Nextcloud OCS Files Sharing API — публичные RO-ссылки на папки.
|
||||
|
||||
Нужен для доступа БЕЗ входа в портал (внешним сторонам — поставщики,
|
||||
проверяющие и т.п.). Для сотрудников аптек есть отдельный, внутренний путь —
|
||||
`access.can_access_pharmacy_docs()` через виджет портала.
|
||||
|
||||
Настройки — тот же сервисный аккаунт ConnectorConfig(nextcloud), что и у
|
||||
WebDAV/Group Folders клиентов.
|
||||
"""
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
|
||||
from apps.nextcloud_connector.nextcloud_service import (
|
||||
NextcloudConfigError, NextcloudServiceError,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class NextcloudSharingService:
|
||||
SHARE_TYPE_PUBLIC_LINK = 3
|
||||
|
||||
# Битовая маска прав (совпадает с Nextcloud permissions).
|
||||
PERM_READ = 1
|
||||
PERM_RO = PERM_READ # просмотр/скачивание, без редактирования — то, что нужно аптекам
|
||||
|
||||
def __init__(self):
|
||||
from core.models import ConnectorConfig
|
||||
cfg = ConnectorConfig.objects.filter(connector_type='nextcloud').first()
|
||||
|
||||
if not cfg:
|
||||
raise NextcloudConfigError(
|
||||
'Настройки коннектора Nextcloud не найдены. '
|
||||
'Добавьте запись в Core → Настройки коннекторов.'
|
||||
)
|
||||
if not cfg.is_enabled:
|
||||
raise NextcloudConfigError('Коннектор Nextcloud отключён.')
|
||||
secret = cfg.token or cfg.password
|
||||
if not cfg.host or not cfg.username or not secret:
|
||||
raise NextcloudConfigError(
|
||||
'Для коннектора Nextcloud нужны host, username и token (app password) '
|
||||
'сервисного аккаунта.'
|
||||
)
|
||||
|
||||
extra = cfg.extra_config or {}
|
||||
scheme = 'https' if cfg.use_ssl else 'http'
|
||||
port = f':{cfg.port}' if cfg.port else ''
|
||||
self.base = f'{scheme}://{cfg.host}{port}'
|
||||
self.shares_url = f'{self.base}/ocs/v2.php/apps/files_sharing/api/v1/shares'
|
||||
|
||||
verify = bool(extra.get('verify_ssl', True))
|
||||
self._client = httpx.Client(
|
||||
auth=(cfg.username, secret),
|
||||
verify=verify,
|
||||
timeout=extra.get('timeout', 30),
|
||||
headers={'OCS-APIRequest': 'true', 'Accept': 'application/json'},
|
||||
)
|
||||
|
||||
# --- низкоуровневый вызов OCS ---
|
||||
|
||||
def _req(self, method: str, url: str, params: dict | None = None, data: dict | None = None):
|
||||
try:
|
||||
resp = self._client.request(method, url, params=params or {}, data=data or {})
|
||||
except httpx.HTTPError as e:
|
||||
raise NextcloudServiceError(f'{method} {url}: {type(e).__name__}: {e}') from e
|
||||
|
||||
try:
|
||||
payload = resp.json()
|
||||
except ValueError as e:
|
||||
raise NextcloudServiceError(
|
||||
f'{method} {url}: ответ не JSON (HTTP {resp.status_code})'
|
||||
) from e
|
||||
|
||||
ocs = payload.get('ocs', {})
|
||||
meta = ocs.get('meta', {})
|
||||
if meta.get('status') != 'ok':
|
||||
raise NextcloudServiceError(
|
||||
f'{method} {url}: OCS {meta.get("statuscode")} {meta.get("message")}'
|
||||
)
|
||||
return ocs.get('data')
|
||||
|
||||
# --- Shares ---
|
||||
|
||||
def find_public_link(self, path: str) -> dict | None:
|
||||
"""Существующая публичная ссылка на путь (относительно корня сервис-аккаунта),
|
||||
либо None — если её ещё нет."""
|
||||
data = self._req('GET', self.shares_url, params={'path': path})
|
||||
items = data if isinstance(data, list) else ([data] if data else [])
|
||||
for item in items:
|
||||
if int(item.get('share_type', -1)) == self.SHARE_TYPE_PUBLIC_LINK:
|
||||
return item
|
||||
return None
|
||||
|
||||
def create_public_link(self, path: str, permissions: int = PERM_RO) -> dict:
|
||||
return self._req('POST', self.shares_url, data={
|
||||
'path': path,
|
||||
'shareType': self.SHARE_TYPE_PUBLIC_LINK,
|
||||
'permissions': permissions,
|
||||
})
|
||||
|
||||
def ensure_public_link(self, path: str, permissions: int = PERM_RO) -> tuple[dict, bool]:
|
||||
"""Гарантирует публичную ссылку на путь. Возвращает (share_data, created)."""
|
||||
existing = self.find_public_link(path)
|
||||
if existing is not None:
|
||||
return existing, False
|
||||
return self.create_public_link(path, permissions), True
|
||||
|
||||
def delete_share(self, share_id) -> None:
|
||||
"""Отозвать ссылку. 404 (уже удалена/не найдена) — не ошибка."""
|
||||
try:
|
||||
resp = self._client.delete(f'{self.shares_url}/{share_id}')
|
||||
except httpx.HTTPError as e:
|
||||
raise NextcloudServiceError(
|
||||
f'DELETE {self.shares_url}/{share_id}: {type(e).__name__}: {e}'
|
||||
) from e
|
||||
if resp.status_code == 200:
|
||||
return
|
||||
try:
|
||||
meta = resp.json().get('ocs', {}).get('meta', {})
|
||||
except ValueError:
|
||||
meta = {}
|
||||
if str(meta.get('statuscode')) == '404':
|
||||
return
|
||||
raise NextcloudServiceError(
|
||||
f'DELETE shares/{share_id}: HTTP {resp.status_code} {meta.get("message", "")}'
|
||||
)
|
||||
|
||||
def close(self):
|
||||
self._client.close()
|
||||
@@ -3,6 +3,7 @@ from django.test import TestCase
|
||||
from apps.nextcloud_connector.nextcloud_service import (
|
||||
NextcloudService, NextcloudConfigError,
|
||||
)
|
||||
from apps.nextcloud_connector.sharing_service import NextcloudSharingService
|
||||
from apps.nextcloud_connector.services import sync_service as ss
|
||||
|
||||
|
||||
@@ -86,6 +87,144 @@ class EnsureFolderTests(TestCase):
|
||||
self.assertIn(('perm', 7, 'nc-dept-hr', 15), svc.calls)
|
||||
|
||||
|
||||
class SharingServiceTests(TestCase):
|
||||
"""Идемпотентность публичных RO-ссылок без сети — _req замокан."""
|
||||
|
||||
def _service(self, existing_shares=None, created_share=None):
|
||||
svc = object.__new__(NextcloudSharingService) # без __init__ (без конфига/сети)
|
||||
svc.shares_url = 'https://nc.example/ocs/v2.php/apps/files_sharing/api/v1/shares'
|
||||
svc.calls = []
|
||||
existing_shares = existing_shares if existing_shares is not None else []
|
||||
|
||||
def fake_req(method, url, params=None, data=None):
|
||||
svc.calls.append((method, params, data))
|
||||
if method == 'GET':
|
||||
return existing_shares
|
||||
return created_share or {'id': 99, 'url': 'https://nc.example/s/newtoken'}
|
||||
|
||||
svc._req = fake_req
|
||||
return svc
|
||||
|
||||
def test_creates_when_no_existing_link(self):
|
||||
svc = self._service(existing_shares=[])
|
||||
share, created = svc.ensure_public_link('02_Аптеки/A-001')
|
||||
self.assertTrue(created)
|
||||
self.assertEqual(share['id'], 99)
|
||||
self.assertIn(('GET', {'path': '02_Аптеки/A-001'}, None), svc.calls)
|
||||
post_calls = [c for c in svc.calls if c[0] == 'POST']
|
||||
self.assertEqual(len(post_calls), 1)
|
||||
self.assertEqual(post_calls[0][2]['shareType'], NextcloudSharingService.SHARE_TYPE_PUBLIC_LINK)
|
||||
self.assertEqual(post_calls[0][2]['permissions'], NextcloudSharingService.PERM_RO)
|
||||
|
||||
def test_reuses_existing_public_link(self):
|
||||
existing = [{'id': 7, 'url': 'https://nc.example/s/old', 'share_type': 3}]
|
||||
svc = self._service(existing_shares=existing)
|
||||
share, created = svc.ensure_public_link('02_Аптеки/A-001')
|
||||
self.assertFalse(created)
|
||||
self.assertEqual(share['id'], 7)
|
||||
self.assertNotIn('POST', [c[0] for c in svc.calls])
|
||||
|
||||
def test_ignores_non_public_link_shares(self):
|
||||
# Шара пользователю (share_type=0) не публичная ссылка — не должна считаться найденной.
|
||||
existing = [{'id': 1, 'url': '', 'share_type': 0}]
|
||||
svc = self._service(existing_shares=existing)
|
||||
share, created = svc.ensure_public_link('02_Аптеки/A-001')
|
||||
self.assertTrue(created) # пришлось создать новую, т.к. публичной нет
|
||||
|
||||
|
||||
class SyncPharmacyPublicLinkTests(TestCase):
|
||||
"""sync_pharmacy: WebDAV-папка + публичная RO-ссылка, без Group Folders.
|
||||
|
||||
Клиенты подменены напрямую через lazy-свойства (_webdav/_sharing), сети нет.
|
||||
"""
|
||||
|
||||
@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.pharmacy = Pharmacy.objects.create(
|
||||
organization=org, location_region=region,
|
||||
pharmacy_number=1, location_type='pharmacy',
|
||||
)
|
||||
|
||||
def _service(self, *, dir_created=True, link_created=True,
|
||||
share=None, fail_webdav=False):
|
||||
from apps.nextcloud_connector.services.sync_service import NextcloudSyncService
|
||||
|
||||
svc = NextcloudSyncService(dry_run=False)
|
||||
|
||||
class _FakeWebDAV:
|
||||
def ensure_dir(self, relpath):
|
||||
if fail_webdav:
|
||||
raise RuntimeError('WebDAV недоступен')
|
||||
return dir_created
|
||||
|
||||
share_ = share or {'id': 42, 'url': 'https://nc.example/s/tok1'}
|
||||
|
||||
class _FakeSharing:
|
||||
def ensure_public_link(self, path, permissions=None):
|
||||
return share_, link_created
|
||||
|
||||
svc._webdav = _FakeWebDAV()
|
||||
svc._sharing = _FakeSharing()
|
||||
return svc
|
||||
|
||||
def test_creates_folder_and_link_and_stores_model(self):
|
||||
from apps.nextcloud_connector.models import PharmacyPublicLink
|
||||
|
||||
svc = self._service(dir_created=True, link_created=True)
|
||||
log = svc.sync_pharmacy(self.pharmacy)
|
||||
|
||||
self.assertEqual(log.status, 'success')
|
||||
self.assertEqual(log.action, 'create')
|
||||
link = PharmacyPublicLink.objects.get(pharmacy=self.pharmacy)
|
||||
self.assertEqual(link.share_id, '42')
|
||||
self.assertEqual(link.url, 'https://nc.example/s/tok1')
|
||||
|
||||
def test_second_run_is_idempotent_update(self):
|
||||
from apps.nextcloud_connector.models import PharmacyPublicLink
|
||||
|
||||
PharmacyPublicLink.objects.create(
|
||||
pharmacy=self.pharmacy, share_id='42', url='https://nc.example/s/tok1',
|
||||
)
|
||||
svc = self._service(dir_created=False, link_created=False)
|
||||
log = svc.sync_pharmacy(self.pharmacy)
|
||||
|
||||
self.assertEqual(log.status, 'success')
|
||||
self.assertEqual(log.action, 'update')
|
||||
self.assertEqual(PharmacyPublicLink.objects.filter(pharmacy=self.pharmacy).count(), 1)
|
||||
|
||||
def test_link_url_updates_if_changed_upstream(self):
|
||||
from apps.nextcloud_connector.models import PharmacyPublicLink
|
||||
|
||||
PharmacyPublicLink.objects.create(
|
||||
pharmacy=self.pharmacy, share_id='42', url='https://nc.example/s/stale',
|
||||
)
|
||||
new_share = {'id': 42, 'url': 'https://nc.example/s/fresh'}
|
||||
svc = self._service(dir_created=False, link_created=False, share=new_share)
|
||||
svc.sync_pharmacy(self.pharmacy)
|
||||
|
||||
link = PharmacyPublicLink.objects.get(pharmacy=self.pharmacy)
|
||||
self.assertEqual(link.url, 'https://nc.example/s/fresh')
|
||||
|
||||
def test_webdav_failure_logs_error_not_exception(self):
|
||||
svc = self._service(fail_webdav=True)
|
||||
log = svc.sync_pharmacy(self.pharmacy)
|
||||
self.assertEqual(log.status, 'error')
|
||||
self.assertIn('WebDAV недоступен', log.message)
|
||||
|
||||
def test_dry_run_touches_nothing(self):
|
||||
from apps.nextcloud_connector.services.sync_service import NextcloudSyncService
|
||||
from apps.nextcloud_connector.models import PharmacyPublicLink
|
||||
|
||||
svc = NextcloudSyncService(dry_run=True)
|
||||
log = svc.sync_pharmacy(self.pharmacy)
|
||||
self.assertEqual(log.status, 'dry_run')
|
||||
self.assertFalse(PharmacyPublicLink.objects.filter(pharmacy=self.pharmacy).exists())
|
||||
|
||||
|
||||
import datetime
|
||||
import json
|
||||
from django.test import Client, TestCase
|
||||
|
||||
@@ -199,6 +199,24 @@ class NextcloudWebDAVService:
|
||||
raise NextcloudServiceError(f'Уже существует: {relpath}')
|
||||
self._raise(resp, relpath)
|
||||
|
||||
def ensure_dir(self, relpath: str) -> bool:
|
||||
"""Идемпотентный вариант mkdir — не ошибка, если папка уже есть.
|
||||
|
||||
Для массового провижининга (nextcloud_sync), где повторный прогон на
|
||||
существующую сущность — норма, а не ошибка пользователя.
|
||||
Возвращает True, если папка создана заново.
|
||||
"""
|
||||
self.ensure_parent_collections(relpath)
|
||||
try:
|
||||
resp = self._client.request('MKCOL', self._url(relpath))
|
||||
except httpx.HTTPError as e:
|
||||
raise NextcloudServiceError(f'WebDAV MKCOL «{relpath}»: {e}') from e
|
||||
if resp.status_code == 201:
|
||||
return True
|
||||
if resp.status_code == 405:
|
||||
return False
|
||||
self._raise(resp, relpath)
|
||||
|
||||
def move(self, src: str, dst: str) -> None:
|
||||
"""Переименовать/переместить (WebDAV MOVE). Не перезаписывает существующее."""
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user