map_app/tests.py - master - DGermanA/Chel-DPT

map_app/tests.py
191 строка7 KB
Herman Dmitriev
init chel-dtp
28 май 2026, 11:21
28 май 2026, 11:21b53942d
import io
import json
import zipfile
from datetime import datetime, timedelta
from unittest.mock import mock_open, patch

import pandas as pd
from django.contrib.gis.geos import Point
from django.core.management import call_command
from django.test import Client, TestCase
from django.urls import reverse
from django.utils import timezone

from map_app.management.commands.load_dtp import Command as LoadDtpCommand
from map_app.models import Accident, Participant, Vehicle


class ModelTests(TestCase):
    """Тестирование моделей и трансформации данных"""

    def setUp(self):
        self.accident = Accident.objects.create(
            external_id=1,
            location=Point(61.4, 55.1),
            datetime=timezone.now(),
            category="Столкновение",
            severity="Тяжёлый",
        )

    def test_severity_display_property(self):
        """Проверка замены 'ё' на 'е' в свойстве severity_display"""
        self.assertEqual(self.accident.severity, "Тяжёлый")
        self.assertEqual(self.accident.severity_display, "Тяжелый")

        self.accident.severity = "Смертельный"
        self.accident.save()
        self.assertEqual(self.accident.severity_display, "Смертельный")

    def test_accident_str(self):
        """Проверка строкового представления (метод __str__)"""
        expected_str = f"ДТП 1 ({self.accident.datetime.date()})"
        self.assertEqual(str(self.accident), expected_str)


class ViewsAndCSVTests(TestCase):
    """Тестирование страниц и выгрузки ZIP"""

    def setUp(self):
        self.client = Client()
        self.accident = Accident.objects.create(
            external_id=100,
            location=Point(61.4, 55.1),
            datetime=timezone.now(),
            category="Опрокидывание",
            severity="Легкий",
            injured_count=2,
            meteo_temp=-5.5,
        )

    def test_pages_status_codes(self):
        """Проверка, что все основные страницы открываются (код 200)"""
        urls = [
            "/",
            reverse("data"),
            reverse("charts"),
            reverse("accident_detail", args=[self.accident.id]),
        ]
        for url in urls:
            response = self.client.get(url)
            self.assertEqual(response.status_code, 200, f"Ошибка при загрузке {url}")

    def test_export_csv(self):
        """Проверка скачивания ZIP-архива с базами данных"""

        response = self.client.get(reverse("export_accidents"))
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response["Content-Type"], "application/zip")

        zip_buffer = io.BytesIO(response.content)
        with zipfile.ZipFile(zip_buffer, "r") as zf:
            file_names = zf.namelist()
            self.assertEqual(len(file_names), 3)
            self.assertIn("1_accidents.csv", file_names)
            self.assertIn("2_vehicles.csv", file_names)
            self.assertIn("3_participants.csv", file_names)


class GeoJSONAPITests(TestCase):
    """Тестирование API карты (фильтрация)"""

    def setUp(self):
        self.client = Client()
        now = timezone.now()

        Accident.objects.create(
            external_id=1,
            location=Point(60.0, 55.0),
            datetime=now,
            category="Наезд на пешехода",
            severity="Смертельный",
            dead_count=1,
        )
        Accident.objects.create(
            external_id=2,
            location=Point(61.0, 56.0),
            datetime=now - timedelta(days=400),
            category="Столкновение",
            severity="Легкий",
            dead_count=0,
        )

    def test_api_category_filter(self):
        """Фильтрация по категории"""
        url = f"{reverse('api_accidents')}?category=Столкновение"
        response = self.client.get(url)
        data = response.json()

        self.assertEqual(len(data["features"]), 1)
        self.assertEqual(data["features"][0]["properties"]["category"], "Столкновение")

    def test_api_with_dead_filter(self):
        """Фильтрация 'Только с погибшими'"""
        url = f"{reverse('api_accidents')}?with_dead=true"
        response = self.client.get(url)
        data = response.json()

        self.assertEqual(len(data["features"]), 1)
        self.assertEqual(data["features"][0]["properties"]["severity"], "Смертельный")


class ManagementCommandsTests(TestCase):
    """Тестирование скриптов загрузки данных"""

    @patch("os.path.exists")
    @patch("builtins.open", new_callable=mock_open)
    def test_load_dtp_command(self, mock_file, mock_exists):
        """Тест импорта JSON"""
        mock_exists.return_value = True

        fake_json_data = {
            "features": [
                {
                    "properties": {
                        "id": 999,
                        "datetime": "2023-10-15 12:30:00",
                        "category": "Наезд",
                        "severity": "Легкий",
                        "vehicles": [
                            {"brand": "LADA", "participants": [{"role": "Водитель"}]}
                        ],
                    },
                    "geometry": {"coordinates": [60.5, 55.1]},
                }
            ]
        }
        mock_file.return_value.read.return_value = json.dumps(fake_json_data)

        cmd = LoadDtpCommand()
        cmd.handle()

        self.assertTrue(Accident.objects.filter(external_id=999).exists())
        self.assertEqual(Vehicle.objects.count(), 1)
        self.assertEqual(Participant.objects.count(), 1)

    @patch("pandas.read_excel")
    def test_load_weather_command(self, mock_read_excel):
        """Тест привязки погоды из Excel"""
        dt = timezone.make_aware(datetime(2023, 10, 15, 12, 10))
        acc = Accident.objects.create(
            external_id=555,
            location=Point(60.0, 55.0),
            datetime=dt,
            category="test",
            severity="test",
        )

        fake_df = pd.DataFrame(
            {
                "Время": ["15.10.2023 12:00", "15.10.2023 15:00"],
                "T": [15.5, 18.0],
                "Po": [740, 742],
            }
        )
        mock_read_excel.return_value = fake_df

        call_command("load_weather", "fake_path.xlsx")

        acc.refresh_from_db()

        self.assertEqual(acc.meteo_temp, 15.5)
        self.assertEqual(acc.meteo_pressure, 740)