Aleksandr Meshchriakov
0824733d1a
Some checks failed
CI/CD Pipeline / Run Tests (pull_request) Successful in 1m35s
CI/CD Pipeline / Telegram Notify Success (push) Has been cancelled
CI/CD Pipeline / Run Tests (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (push) Has been cancelled
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been cancelled
CI/CD Pipeline / Code Quality Checks (pull_request) Has been cancelled
206 lines
8.6 KiB
Python
206 lines
8.6 KiB
Python
"""Команда генерации тестовых данных по реестрам через test factories."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from importlib import import_module
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from apps.core.management.commands.base import BaseAppCommand
|
||
from apps.parsers.models import ParserLoadLog
|
||
from django.conf import settings
|
||
from django.core.management.base import CommandError
|
||
from django.db.models import Max
|
||
|
||
|
||
class Command(BaseAppCommand):
|
||
"""Генерирует тестовые данные для всех реестров на основе фабрик тестов."""
|
||
|
||
help = (
|
||
"Генерация тестовых данных для registers/parsers на основе фабрик из tests"
|
||
)
|
||
use_transaction = True
|
||
|
||
def add_arguments(self, parser) -> None:
|
||
"""Добавление аргументов команды."""
|
||
super().add_arguments(parser)
|
||
parser.add_argument(
|
||
"--count",
|
||
type=int,
|
||
default=20,
|
||
help="Количество записей для каждого парсерного реестра",
|
||
)
|
||
parser.add_argument(
|
||
"--registers-count",
|
||
type=int,
|
||
default=5,
|
||
help="Количество записей в справочнике реестров (apps.registers)",
|
||
)
|
||
|
||
def execute_command(self, *args: Any, **options: Any) -> str:
|
||
"""Основная логика генерации тестовых данных."""
|
||
count = options["count"]
|
||
registers_count = options["registers_count"]
|
||
|
||
self._validate_positive("count", count)
|
||
self._validate_positive("registers_count", registers_count)
|
||
|
||
factories = self._load_factories()
|
||
|
||
register_factory = factories["register_factory"]
|
||
organization_factory = factories["organization_factory"]
|
||
register_upload_factory = factories["register_upload_factory"]
|
||
registry_membership_period_factory = factories[
|
||
"registry_membership_period_factory"
|
||
]
|
||
parser_load_log_factory = factories["parser_load_log_factory"]
|
||
proxy_factory = factories["proxy_factory"]
|
||
industrial_certificate_factory = factories["industrial_certificate_factory"]
|
||
manufacturer_factory = factories["manufacturer_factory"]
|
||
industrial_product_factory = factories["industrial_product_factory"]
|
||
inspection_factory = factories["inspection_factory"]
|
||
procurement_factory = factories["procurement_factory"]
|
||
|
||
run_token = uuid4().hex[:10]
|
||
run_seed = int(run_token[:6], 16) % 1_000_000
|
||
|
||
self.log_info("Создание тестовых реестров organizations/registers...")
|
||
registers = [
|
||
register_factory.create(name=f"Тестовый реестр {run_token}-{index + 1}")
|
||
for index in range(registers_count)
|
||
]
|
||
for register_index, register in enumerate(
|
||
self.progress_iter(
|
||
registers,
|
||
desc="Заполнение реестров организаций",
|
||
total=registers_count,
|
||
)
|
||
):
|
||
for item_index in range(count):
|
||
org_serial = register_index * count + item_index + 1
|
||
organization = organization_factory.create(
|
||
mn_ogrn=80_000_000_000_000 + run_seed * 100_000 + org_serial,
|
||
mn_inn=1_000_000_000
|
||
+ (run_seed * 10_000 + org_serial) % 8_000_000_000,
|
||
)
|
||
upload = register_upload_factory.create(registry=register)
|
||
registry_membership_period_factory.create(
|
||
registry=register,
|
||
organization=organization,
|
||
started_by_upload=upload,
|
||
)
|
||
|
||
self.log_info("Создание тестовых данных parser-реестров...")
|
||
for index in range(1, count + 1):
|
||
industrial_certificate_factory.create(
|
||
certificate_number=f"TST-CERT-{run_token}-{index:05d}"
|
||
)
|
||
|
||
for index in range(1, count + 1):
|
||
manufacturer_factory.create(
|
||
inn=f"{run_seed:06d}{index:09d}",
|
||
)
|
||
|
||
industrial_product_factory.create_batch(count)
|
||
|
||
for index in range(1, count + 1):
|
||
inspection_factory.create(
|
||
registration_number=f"TST-INSP-{run_token}-{index:05d}"
|
||
)
|
||
|
||
for index in range(1, count + 1):
|
||
procurement_factory.create(
|
||
purchase_number=f"{run_seed:06d}{index:013d}"
|
||
)
|
||
|
||
created_by_registry = {
|
||
"industrial_certificates": count,
|
||
"manufacturers": count,
|
||
"industrial_products": count,
|
||
"inspections": count,
|
||
"procurements": count,
|
||
}
|
||
|
||
self.log_info("Создание прокси и логов загрузок...")
|
||
for index in range(1, count + 1):
|
||
proxy_factory.create(
|
||
address=f"http://proxy-{run_token}-{index}.local:8080"
|
||
)
|
||
|
||
load_log_sources = [source for source, _ in ParserLoadLog.Source.choices]
|
||
for source in load_log_sources:
|
||
max_batch = (
|
||
ParserLoadLog.objects.filter(source=source).aggregate(
|
||
max_batch=Max("batch_id")
|
||
)["max_batch"]
|
||
or 0
|
||
)
|
||
for index in range(1, count + 1):
|
||
parser_load_log_factory.create(
|
||
source=source,
|
||
batch_id=max_batch + index,
|
||
)
|
||
|
||
summary_lines = [
|
||
"Тестовые данные успешно созданы.",
|
||
f"registers.Register: {registers_count}",
|
||
f"registers.RegisterUpload: {registers_count * count}",
|
||
f"registers.RegistryMembershipPeriod: {registers_count * count}",
|
||
f"parsers.Proxy: {count}",
|
||
f"parsers.ParserLoadLog: {len(load_log_sources) * count}",
|
||
]
|
||
summary_lines.extend(
|
||
f"parsers.{registry_name}: {created_count}"
|
||
for registry_name, created_count in created_by_registry.items()
|
||
)
|
||
return "\n".join(summary_lines)
|
||
|
||
@staticmethod
|
||
def _validate_positive(option_name: str, value: int) -> None:
|
||
"""Проверка, что числовой аргумент больше нуля."""
|
||
if value < 1:
|
||
raise CommandError(f"--{option_name.replace('_', '-')} должен быть >= 1")
|
||
|
||
@staticmethod
|
||
def _load_factories() -> dict[str, Any]:
|
||
"""Импорт фабрик из тестов (tests/apps/*/factories.py)."""
|
||
Command._ensure_project_root_on_path()
|
||
|
||
try:
|
||
registers_factories = import_module("tests.apps.registers.factories")
|
||
parsers_factories = import_module("tests.apps.parsers.factories")
|
||
except ModuleNotFoundError as exc:
|
||
raise CommandError(
|
||
"Не удалось импортировать тестовые фабрики из tests/apps/*/factories.py. "
|
||
"Проверьте, что каталог tests доступен в PYTHONPATH."
|
||
) from exc
|
||
|
||
return {
|
||
"register_factory": registers_factories.RegisterFactory,
|
||
"organization_factory": registers_factories.OrganizationFactory,
|
||
"register_upload_factory": registers_factories.RegisterUploadFactory,
|
||
"registry_membership_period_factory": (
|
||
registers_factories.RegistryMembershipPeriodFactory
|
||
),
|
||
"parser_load_log_factory": parsers_factories.ParserLoadLogFactory,
|
||
"proxy_factory": parsers_factories.ProxyFactory,
|
||
"industrial_certificate_factory": (
|
||
parsers_factories.IndustrialCertificateRecordFactory
|
||
),
|
||
"manufacturer_factory": parsers_factories.ManufacturerRecordFactory,
|
||
"industrial_product_factory": parsers_factories.IndustrialProductRecordFactory,
|
||
"inspection_factory": parsers_factories.InspectionRecordFactory,
|
||
"procurement_factory": parsers_factories.ProcurementRecordFactory,
|
||
}
|
||
|
||
@staticmethod
|
||
def _ensure_project_root_on_path() -> None:
|
||
"""Добавить корень проекта в PYTHONPATH для импорта tests.*."""
|
||
base_dir = Path(settings.BASE_DIR)
|
||
project_root = Path(getattr(settings, "PROJECT_ROOT", base_dir.parent))
|
||
project_root_str = str(project_root)
|
||
if project_root_str not in sys.path:
|
||
sys.path.append(project_root_str)
|