connectors: команда check_connectors — health-check Mattermost/Nextcloud/SILO.
Первый шаг подключения внешних сервисов: проверка доступности + аутентификации, а не просто наличия настроек. - mattermost — GET users/me ботом provisioning; - nextcloud — WebDAV PROPFIND корня files-хранилища; - s3 (silo) — list_objects_v2 на media-бакете (таймаут 5с, без ретраев). Отключённый коннектор → SKIP; включённый недоступный → FAIL + ненулевой код возврата (для мониторинга/CI). Переиспользует существующих клиентов (MattermostService, NextcloudWebDAVService, django-storages S3-настройки). Тесты: core/tests_check_connectors.py. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,121 @@
|
||||
"""Health-check подключений ACORE к внешним сервисам.
|
||||
|
||||
Проверяет доступность + аутентификацию (а не просто наличие настроек):
|
||||
mattermost — GET /api/v4/users/me ботом provisioning
|
||||
nextcloud — WebDAV PROPFIND корня files-хранилища сервисного аккаунта
|
||||
s3 (silo) — list_objects_v2 на media-бакете
|
||||
|
||||
Отключённый/ненастроенный коннектор → SKIP (не ошибка). Любой включённый,
|
||||
но недоступный → FAIL и ненулевой код возврата — удобно для мониторинга/CI.
|
||||
|
||||
Использование:
|
||||
manage.py check_connectors # все три
|
||||
manage.py check_connectors mattermost # только выбранные
|
||||
"""
|
||||
from django.conf import settings
|
||||
from django.core.management.base import BaseCommand, CommandError
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = 'Проверяет соединение и аутентификацию с Mattermost, Nextcloud и SILO (S3).'
|
||||
|
||||
CHECKS = ('mattermost', 'nextcloud', 's3')
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument(
|
||||
'services', nargs='*', choices=self.CHECKS,
|
||||
help='Какие сервисы проверять (по умолчанию все три).',
|
||||
)
|
||||
|
||||
def handle(self, *args, **opts):
|
||||
targets = opts['services'] or list(self.CHECKS)
|
||||
failed = []
|
||||
for name in targets:
|
||||
status, detail = getattr(self, f'_check_{name}')()
|
||||
self._report(name, status, detail)
|
||||
if status == 'FAIL':
|
||||
failed.append(name)
|
||||
if failed:
|
||||
raise CommandError(
|
||||
f'Недоступны включённые коннекторы: {", ".join(failed)}.'
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Вывод
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _report(self, name, status, detail):
|
||||
label = f'{name:<11}'
|
||||
if status == 'OK':
|
||||
mark = self.style.SUCCESS('OK ')
|
||||
elif status == 'SKIP':
|
||||
mark = self.style.WARNING('SKIP')
|
||||
else:
|
||||
mark = self.style.ERROR('FAIL')
|
||||
self.stdout.write(f' {mark} {label} {detail}')
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Проверки
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _check_mattermost(self):
|
||||
from core.models import ConnectorConfig
|
||||
if ConnectorConfig.get_for('mattermost') is None:
|
||||
return 'SKIP', 'коннектор отключён или не настроен'
|
||||
try:
|
||||
from apps.mattermost_connector.mattermost_service import MattermostService
|
||||
svc = MattermostService(role='provisioning')
|
||||
me = svc.driver.users.get_user('me')
|
||||
return 'OK', f"бот @{me.get('username', '?')} (роль {svc.bot.role})"
|
||||
except Exception as e: # noqa: BLE001 — диагностика: любая ошибка = FAIL
|
||||
return 'FAIL', f'{type(e).__name__}: {e}'
|
||||
|
||||
def _check_nextcloud(self):
|
||||
from core.models import ConnectorConfig
|
||||
if ConnectorConfig.get_for('nextcloud') is None:
|
||||
return 'SKIP', 'коннектор отключён или не настроен'
|
||||
svc = None
|
||||
try:
|
||||
from apps.nextcloud_connector.webdav_service import NextcloudWebDAVService
|
||||
svc = NextcloudWebDAVService()
|
||||
entries = svc.list_dir('')
|
||||
return 'OK', f'WebDAV доступен, объектов в корне: {len(entries)}'
|
||||
except Exception as e: # noqa: BLE001
|
||||
return 'FAIL', f'{type(e).__name__}: {e}'
|
||||
finally:
|
||||
if svc is not None:
|
||||
try:
|
||||
svc.close()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
def _check_s3(self):
|
||||
if not getattr(settings, 'USE_S3_MEDIA', False):
|
||||
return 'SKIP', 'USE_S3_MEDIA=False (локальный MEDIA_ROOT)'
|
||||
try:
|
||||
import boto3
|
||||
from botocore.config import Config
|
||||
|
||||
client = boto3.client(
|
||||
's3',
|
||||
endpoint_url=settings.S3_ENDPOINT,
|
||||
aws_access_key_id=settings.S3_ACCESS_KEY,
|
||||
aws_secret_access_key=settings.S3_SECRET_KEY,
|
||||
region_name=settings.S3_REGION,
|
||||
config=Config(
|
||||
signature_version='s3v4',
|
||||
s3={'addressing_style': 'path'},
|
||||
connect_timeout=5,
|
||||
read_timeout=5,
|
||||
retries={'max_attempts': 1},
|
||||
),
|
||||
)
|
||||
resp = client.list_objects_v2(
|
||||
Bucket=settings.S3_MEDIA_BUCKET, MaxKeys=1,
|
||||
)
|
||||
return 'OK', (
|
||||
f'бакет «{settings.S3_MEDIA_BUCKET}» доступен '
|
||||
f'(объектов ≥ {resp.get("KeyCount", 0)})'
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
return 'FAIL', f'{type(e).__name__}: {e}'
|
||||
@@ -0,0 +1,40 @@
|
||||
from io import StringIO
|
||||
from unittest import mock
|
||||
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.test import TestCase, override_settings
|
||||
|
||||
|
||||
class CheckConnectorsTests(TestCase):
|
||||
"""health-check команда check_connectors: отключённые сервисы — SKIP (не
|
||||
ошибка), включённый недоступный — FAIL + ненулевой выход (CommandError)."""
|
||||
|
||||
@override_settings(USE_S3_MEDIA=False)
|
||||
def test_all_disabled_is_skip_not_error(self):
|
||||
# Нет ConnectorConfig для mattermost/nextcloud + S3 выключен → все SKIP.
|
||||
out = StringIO()
|
||||
call_command('check_connectors', stdout=out)
|
||||
text = out.getvalue()
|
||||
self.assertEqual(text.count('SKIP'), 3)
|
||||
self.assertNotIn('FAIL', text)
|
||||
|
||||
@override_settings(
|
||||
USE_S3_MEDIA=True,
|
||||
S3_ENDPOINT='http://silo.invalid:9000',
|
||||
S3_ACCESS_KEY='k', S3_SECRET_KEY='s',
|
||||
S3_MEDIA_BUCKET='acore-media', S3_REGION='us-east-1',
|
||||
)
|
||||
def test_enabled_but_unreachable_s3_fails(self):
|
||||
out, err = StringIO(), StringIO()
|
||||
with self.assertRaises(CommandError):
|
||||
call_command('check_connectors', 's3', stdout=out, stderr=err)
|
||||
self.assertIn('FAIL', out.getvalue())
|
||||
|
||||
def test_only_selected_service_checked(self):
|
||||
out = StringIO()
|
||||
call_command('check_connectors', 'nextcloud', stdout=out)
|
||||
text = out.getvalue()
|
||||
self.assertIn('nextcloud', text)
|
||||
self.assertNotIn('mattermost', text)
|
||||
self.assertNotIn('s3', text)
|
||||
Reference in New Issue
Block a user