"""Regression coverage for the routed update-history list and Excel export.""" import csv from datetime import UTC, datetime from io import StringIO from unittest.mock import patch from urllib.parse import parse_qs, urlsplit from apps.parsers import views from apps.parsers.models import ParserLoadLog from django.test import override_settings from django.urls import reverse from rest_framework.test import APITestCase from tests.apps.parsers.factories import ParserLoadLogFactory from tests.apps.user.factories import UserFactory @override_settings(TIME_ZONE="UTC") class SystemLogsCheckupTest(APITestCase): def setUp(self): self.client.force_authenticate(UserFactory.create_superuser()) self.list_url = reverse("api_v1:system:parser-logs-list") self.export_url = reverse("api_v1:system:parser-logs-export") @staticmethod def create_log(updated_at, **kwargs): kwargs.setdefault("source", "inspections") log = ParserLoadLogFactory(**kwargs) ParserLoadLog.objects.filter(pk=log.pk).update(updated_at=updated_at) return log def csv_rows(self, params): response = self.client.get(self.export_url, params) self.assertEqual(response.status_code, 200) return list( csv.reader(StringIO(response.content.decode("utf-8-sig")), delimiter=";") ) def test_date_filter_is_inclusive_and_single_start_means_one_day(self): self.create_log(datetime(2026, 9, 6, 23, 59, 59, tzinfo=UTC)) first = self.create_log(datetime(2026, 9, 7, tzinfo=UTC)) last = self.create_log(datetime(2026, 9, 7, 23, 59, 59, 999999, tzinfo=UTC)) self.create_log(datetime(2026, 9, 8, tzinfo=UTC)) params = {"date_from": "2026-09-07", "ordering": "updated_at", "page_size": 1} response = self.client.get(self.list_url, params) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["count"], 2) self.assertEqual(response.data["results"][0]["id"], first.id) self.assertIsNone(response.data["previous"]) next_query = parse_qs(urlsplit(response.data["next"]).query) self.assertEqual(next_query["date_from"], ["2026-09-07"]) self.assertEqual(next_query["page"], ["2"]) page_two = self.client.get(self.list_url, {**params, "page": 2}) self.assertEqual(page_two.data["results"][0]["id"], last.id) self.assertIsNone(page_two.data["next"]) self.assertIsNotNone(page_two.data["previous"]) self.assertEqual( [int(row[0]) for row in self.csv_rows(params)[1:]], [first.id, last.id] ) def test_period_combines_with_source_status_and_search(self): moment = datetime(2026, 9, 7, 12, tzinfo=UTC) expected = self.create_log( moment, source="manufactures", status="failed", error_message="маркер" ) self.create_log( moment, source="manufactures", status="success", error_message="маркер" ) self.create_log( moment, source="inspections", status="failed", error_message="маркер" ) self.create_log( moment, source="manufactures", status="failed", error_message="другое" ) params = { "date_from": "2026-09-01", "date_to": "2026-09-07", "source": "manufacturers-and-products", "status": "failed", "search": "маркер", } response = self.client.get(self.list_url, params) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["count"], 1) self.assertEqual(response.data["results"][0]["id"], expected.id) self.assertEqual( [int(row[0]) for row in self.csv_rows(params)[1:]], [expected.id] ) def test_date_validation_matches_between_list_and_export(self): for params in ( {"date_from": "2026-02-30"}, {"date_to": "07.09.2026"}, {"date_from": "20260907"}, {"date_from": "2026-09-08", "date_to": "2026-09-07"}, {"ordering": "unknown"}, {"ordering": "--source"}, {"batch_id": "invalid"}, ): for url in (self.list_url, self.export_url): with self.subTest(params=params, url=url): self.assertEqual(self.client.get(url, params).status_code, 400) def test_ordering_is_global_stable_and_enriches_only_the_page(self): moment = datetime(2026, 9, 7, tzinfo=UTC) middle = self.create_log(moment, records_count=20) low = self.create_log(moment, records_count=10) tie = self.create_log(moment, records_count=20) high = self.create_log(moment, records_count=30) for ordering, expected in ( ("records_count", [low.id, middle.id, tie.id, high.id]), ("-records_count", [high.id, middle.id, tie.id, low.id]), ): ids = [] for page in (1, 2): with patch.object( views, "_serialize_parser_log_row", wraps=views._serialize_parser_log_row, ) as serialize: response = self.client.get( self.list_url, {"ordering": ordering, "page_size": 2, "page": page}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["count"], 4) self.assertEqual(serialize.call_count, 2) ids.extend(row["id"] for row in response.data["results"]) self.assertEqual(ids, expected) self.assertEqual( [int(row[0]) for row in self.csv_rows({"ordering": ordering})[1:]], expected, ) def test_sort_by_source_uses_public_slug_consistently_with_export(self): moment = datetime(2026, 9, 7, tzinfo=UTC) manufacture = self.create_log(moment, source="manufactures") industrial = self.create_log(moment, source="industrial") report = self.create_log(moment, source="fns_reports") for ordering, expected in ( ("source", [report.id, manufacture.id, industrial.id]), ("-source", [manufacture.id, industrial.id, report.id]), ): response = self.client.get(self.list_url, {"ordering": ordering}) self.assertEqual(response.status_code, 200) self.assertEqual([row["id"] for row in response.data["results"]], expected) self.assertEqual( [int(row[0]) for row in self.csv_rows({"ordering": ordering})[1:]], expected, ) def test_csv_preserves_cyrillic_delimiters_quotes_and_newlines(self): message = 'Ошибка; "детали"\r\nСледующая строка' self.create_log( datetime(2026, 9, 7, 12, 34, tzinfo=UTC), status="failed", error_message=message, ) response = self.client.get(self.export_url) self.assertEqual(response.status_code, 200) self.assertEqual(response["Content-Type"], "text/csv; charset=utf-8") self.assertEqual( response["Content-Disposition"], 'attachment; filename="update-history.csv"' ) self.assertTrue(response.content.startswith(b"\xef\xbb\xbf")) decoded = response.content.decode("utf-8-sig") self.assertNotIn("\n", decoded.replace("\r\n", "")) rows = list(csv.reader(StringIO(decoded), delimiter=";")) self.assertEqual( rows[0][:5], ["№", "Дата актуализации", "Источник", "Статус", "Количество записей"], ) self.assertEqual(rows[1][1], "07.09.2026 12:34") self.assertEqual(rows[1][3], "Ошибка") self.assertEqual(rows[1][rows[0].index("Ошибка")], message) def test_openapi_documents_period_for_list_and_export(self): response = self.client.get(reverse("schema-swagger-ui"), {"format": "openapi"}) self.assertEqual(response.status_code, 200) paths = response.data["paths"] for suffix in ("/system/logs/", "/system/logs/export/"): path = next(path for path in paths if path.endswith(suffix)) parameters = { parameter["name"]: parameter for parameter in paths[path]["get"]["parameters"] } for name in ("date_from", "date_to"): self.assertEqual(parameters[name]["type"], "string") self.assertEqual(parameters[name]["format"], "date")