Files
mostovik-backend/tests/apps/parsers/test_models.py
Aleksandr Meshchriakov c72343a375
All checks were successful
CI/CD Pipeline / Manual Action Help (push) Has been skipped
CI/CD Pipeline / Start Dev Containers in Dokploy (push) Has been skipped
CI/CD Pipeline / Drop and Recreate Dev Database (push) Has been skipped
CI/CD Pipeline / Quality Gate (push) Successful in 1m53s
CI/CD Pipeline / Build and Push Images (push) Successful in 2m42s
CI/CD Pipeline / Internal Notify (push) Successful in 1s
CI/CD Pipeline / Deploy Dev in Dokploy (push) Successful in 1s
fix parser schedule run issues
2026-04-28 13:58:55 +02:00

202 lines
6.7 KiB
Python

"""Tests for parsers models."""
from apps.parsers.models import (
GenericParserRecord,
IndustrialCertificateRecord,
IndustrialProductRecord,
ManufacturerRecord,
ParserLoadLog,
Proxy,
)
from django.test import TestCase
from .factories import (
IndustrialCertificateRecordFactory,
IndustrialProductRecordFactory,
ManufacturerRecordFactory,
ParserLoadLogFactory,
ProxyFactory,
fake,
generate_certificate_number,
generate_company_name,
generate_inn_legal,
generate_proxy_address,
)
class ProxyModelTest(TestCase):
"""Tests for Proxy model."""
def test_create_proxy(self):
"""Test creating proxy record."""
proxy = ProxyFactory()
self.assertIsInstance(proxy, Proxy)
self.assertIn("://", proxy.address) # http://, https://, socks5://
self.assertTrue(proxy.is_active)
self.assertEqual(proxy.fail_count, 0)
def test_proxy_str(self):
"""Test proxy string representation."""
address = generate_proxy_address()
proxy = ProxyFactory(address=address, is_active=True)
self.assertEqual(str(proxy), f"{address} (active)")
proxy.is_active = False
self.assertEqual(str(proxy), f"{address} (inactive)")
def test_proxy_ordering(self):
"""Test proxy ordering by fail_count."""
proxy1 = ProxyFactory(fail_count=5)
proxy2 = ProxyFactory(fail_count=0)
proxy3 = ProxyFactory(fail_count=10)
proxies = list(Proxy.objects.all())
self.assertEqual(proxies[0], proxy2) # fail_count=0
self.assertEqual(proxies[1], proxy1) # fail_count=5
self.assertEqual(proxies[2], proxy3) # fail_count=10
def test_proxy_unique_address(self):
"""Test proxy address uniqueness."""
address = generate_proxy_address()
ProxyFactory(address=address)
from django.db import IntegrityError
with self.assertRaises(IntegrityError):
ProxyFactory(address=address)
class ParserLoadLogModelTest(TestCase):
"""Tests for ParserLoadLog model."""
def test_create_load_log(self):
"""Test creating load log record."""
log = ParserLoadLogFactory()
self.assertIsInstance(log, ParserLoadLog)
self.assertIsNotNone(log.batch_id)
self.assertIn(log.source, [c[0] for c in ParserLoadLog.Source.choices])
def test_load_log_str(self):
"""Test load log string representation."""
batch_id = fake.random_int(min=1, max=999)
records_count = fake.random_int(min=1, max=5000)
log = ParserLoadLogFactory(
batch_id=batch_id,
source=ParserLoadLog.Source.INDUSTRIAL,
records_count=records_count,
)
self.assertIn(str(batch_id), str(log))
self.assertIn(str(records_count), str(log))
def test_load_log_timestamps(self):
"""Test load log has timestamps from mixin."""
log = ParserLoadLogFactory()
self.assertIsNotNone(log.created_at)
self.assertIsNotNone(log.updated_at)
class GenericParserRecordModelTest(TestCase):
"""Tests for generic parser records."""
def test_record_date_allows_source_specific_long_values(self):
field = GenericParserRecord._meta.get_field("record_date")
self.assertEqual(field.max_length, 255)
class IndustrialCertificateRecordModelTest(TestCase):
"""Tests for IndustrialCertificateRecord model."""
def test_create_certificate(self):
"""Test creating certificate record."""
cert = IndustrialCertificateRecordFactory()
self.assertIsInstance(cert, IndustrialCertificateRecord)
self.assertIsNotNone(cert.certificate_number)
self.assertIsNotNone(cert.inn)
self.assertIsNotNone(cert.ogrn)
def test_certificate_str(self):
"""Test certificate string representation."""
certificate_number = generate_certificate_number()
organisation_name = generate_company_name()
cert = IndustrialCertificateRecordFactory(
certificate_number=certificate_number,
organisation_name=organisation_name,
)
self.assertIn(certificate_number, str(cert))
self.assertIn(organisation_name[:50], str(cert))
def test_certificate_timestamps(self):
"""Test certificate has timestamps from mixin."""
cert = IndustrialCertificateRecordFactory()
self.assertIsNotNone(cert.created_at)
self.assertIsNotNone(cert.updated_at)
class ManufacturerRecordModelTest(TestCase):
"""Tests for ManufacturerRecord model."""
def test_create_manufacturer(self):
"""Test creating manufacturer record."""
manufacturer = ManufacturerRecordFactory()
self.assertIsInstance(manufacturer, ManufacturerRecord)
self.assertIsNotNone(manufacturer.full_legal_name)
self.assertIsNotNone(manufacturer.inn)
self.assertIsNotNone(manufacturer.ogrn)
def test_manufacturer_str(self):
"""Test manufacturer string representation."""
inn = generate_inn_legal()
company_name = generate_company_name()
manufacturer = ManufacturerRecordFactory(
inn=inn,
full_legal_name=company_name,
)
self.assertIn(inn, str(manufacturer))
self.assertIn(company_name[:50], str(manufacturer))
def test_manufacturer_timestamps(self):
"""Test manufacturer has timestamps from mixin."""
manufacturer = ManufacturerRecordFactory()
self.assertIsNotNone(manufacturer.created_at)
self.assertIsNotNone(manufacturer.updated_at)
class IndustrialProductRecordModelTest(TestCase):
"""Tests for IndustrialProductRecord model."""
def test_create_industrial_product(self):
"""Test creating industrial product record."""
product = IndustrialProductRecordFactory()
self.assertIsInstance(product, IndustrialProductRecord)
self.assertIsNotNone(product.registry_number)
self.assertIsNotNone(product.product_name)
self.assertIsNotNone(product.inn)
self.assertIsNotNone(product.ogrn)
def test_industrial_product_str(self):
"""Test industrial product string representation."""
registry_number = fake.bothify(text="MPP-########")
product_name = generate_company_name()
product = IndustrialProductRecordFactory(
registry_number=registry_number,
product_name=product_name,
)
self.assertIn(registry_number, str(product))
self.assertIn(product_name[:50], str(product))
def test_industrial_product_timestamps(self):
"""Test industrial product has timestamps from mixin."""
product = IndustrialProductRecordFactory()
self.assertIsNotNone(product.created_at)
self.assertIsNotNone(product.updated_at)