"""Tests for exchange admin configuration.""" from types import SimpleNamespace from unittest.mock import patch from apps.exchange.admin import ExchangeConnectionAdmin from apps.exchange.forms import ExchangeConnectionAdminForm from apps.exchange.models import ExchangeConnection from apps.exchange.services import ExchangePeriodicTaskService from django.contrib.admin.sites import AdminSite from django.contrib.messages.storage.fallback import FallbackStorage from django.test import RequestFactory, TestCase from django_celery_beat.models import IntervalSchedule, PeriodicTask from tests.apps.exchange.factories import ExchangeConnectionFactory from tests.apps.user.factories import UserFactory _DB_PASSWORD = "secret" # noqa: S105 class ExchangeAdminTest(TestCase): def setUp(self): self.site = AdminSite() self.admin = ExchangeConnectionAdmin(ExchangeConnection, self.site) self.factory = RequestFactory() self.user = UserFactory.create_superuser() def _request(self, path: str): request = self.factory.get(path) request.user = self.user request.session = {} request._messages = FallbackStorage(request) return request def _post_request(self, path: str, data: dict): request = self.factory.post(path, data=data) request.user = self.user request.session = {} request._messages = FallbackStorage(request) return request def test_exchange_admin_has_custom_routes(self): route_names = [route.name for route in self.admin.get_urls()] self.assertIn("exchange_exchangeconnection_test_connection", route_names) self.assertIn("exchange_exchangeconnection_copy_data", route_names) self.assertIn("exchange_exchangeconnection_periodic_tasks", route_names) self.assertIn("exchange_exchangeconnection_periodic_task_change", route_names) def test_exchange_admin_changelist_renders_custom_buttons(self): response = self.admin.changelist_view( self._request("/admin/exchange/exchangeconnection/") ) response.render() content = response.content.decode("utf-8") self.assertEqual(response.status_code, 200) self.assertIn("Проверить новое подключение", content) self.assertIn("Запустить обмен", content) self.assertIn("Настроить периодический обмен", content) self.assertIn("Добавить подключение обмена", content) def test_exchange_connection_admin_form_keeps_existing_password_when_blank(self): connection = ExchangeConnectionFactory(password=_DB_PASSWORD) form = ExchangeConnectionAdminForm( data={ "server": connection.server, "port": connection.port, "username": connection.username, "password": "", "database_name": connection.database_name, "schema_name": connection.schema_name, }, instance=connection, ) self.assertTrue(form.is_valid(), form.errors) saved_connection = form.save() self.assertEqual(saved_connection.get_decrypted_password(), _DB_PASSWORD) @patch("apps.exchange.admin.ExchangeConnectionService.validate_saved_connection") def test_validate_selected_connections_calls_service(self, validate_mock): connection = ExchangeConnectionFactory() self.admin.validate_selected_connections( self._request("/admin/exchange/exchangeconnection/"), ExchangeConnection.objects.filter(id=connection.id), ) validate_mock.assert_called_once_with(connection) @patch("apps.exchange.admin.ExchangeConnectionService.activate_connection") def test_activate_selected_connection_calls_service(self, activate_mock): connection = ExchangeConnectionFactory() self.admin.activate_selected_connection( self._request("/admin/exchange/exchangeconnection/"), ExchangeConnection.objects.filter(id=connection.id), ) activate_mock.assert_called_once_with(connection) @patch("apps.exchange.admin.ExchangeConnectionService.test_connection_payload") def test_test_connection_view_success_redirects_after_validation( self, test_connection_mock, ): test_connection_mock.return_value = { "status": "success", "message": "Подключение проверено.", } request = self._post_request( "/admin/exchange/exchangeconnection/test-connection/", { "server": "127.0.0.1", "port": 5432, "username": "postgres", "password": _DB_PASSWORD, "database_name": "target_db", "schema_name": "public", }, ) response = self.admin.test_connection_view(request) self.assertEqual(response.status_code, 302) test_connection_mock.assert_called_once_with( server="127.0.0.1", port=5432, username="postgres", password=_DB_PASSWORD, database_name="target_db", schema_name="public", ) @patch("apps.exchange.admin.BackgroundJobService.create_job") @patch("apps.exchange.admin.copy_parsers_data_async") def test_copy_data_view_enqueues_background_job(self, task_mock, create_job_mock): active_connection = ExchangeConnectionFactory(is_active=True) task_mock.delay.return_value = SimpleNamespace(id="task-123") request = self._post_request( "/admin/exchange/exchangeconnection/copy-data/", { "mode": "all", "truncate_before_copy": "on", }, ) response = self.admin.copy_data_view(request) self.assertEqual(response.status_code, 302) task_mock.delay.assert_called_once_with( connection_id=active_connection.id, payload={ "mode": "all", "table": None, "tables": None, "truncate_before_copy": True, }, requested_by_id=self.user.id, ) create_job_mock.assert_called_once() def test_periodic_tasks_view_lists_existing_tasks(self): interval = IntervalSchedule.objects.create(every=2, period="hours") PeriodicTask.objects.create( name="exchange-copy-hourly", task=ExchangePeriodicTaskService.TASK_NAME, interval=interval, kwargs='{"payload": {"mode": "all", "truncate_before_copy": true}}', ) response = self.admin.periodic_tasks_view( self._request("/admin/exchange/exchangeconnection/periodic-tasks/") ) response.render() content = response.content.decode("utf-8") self.assertEqual(response.status_code, 200) self.assertIn("exchange-copy-hourly", content) self.assertIn("Каждые 2 hours", content) @patch("apps.exchange.admin.ExchangePeriodicTaskService.create_periodic_task") def test_periodic_tasks_view_creates_task(self, create_task_mock): request = self._post_request( "/admin/exchange/exchangeconnection/periodic-tasks/", { "name": "exchange-copy-nightly", "description": "Nightly sync", "enabled": "on", "mode": "selected", "tables": ["parsers_manufacturer"], "truncate_before_copy": "on", "notify_on_error": "on", "schedule_type": "daily", "crontab_minute": 15, "crontab_hour": 3, }, ) response = self.admin.periodic_tasks_view(request) self.assertEqual(response.status_code, 302) create_task_mock.assert_called_once_with( name="exchange-copy-nightly", description="Nightly sync", enabled=True, payload={ "mode": "selected", "table": None, "tables": ["parsers_manufacturer"], "truncate_before_copy": True, "notify_on_error": True, }, schedule={ "type": "crontab", "minute": "15", "hour": "3", "day_of_week": "*", "day_of_month": "*", "month_of_year": "*", }, )