Aleksandr Meshchriakov
3d298ce352
Some checks failed
CI/CD Pipeline / Run Tests (pull_request) Successful in 1m53s
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 / Code Quality Checks (pull_request) Failing after 2m54s
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been skipped
315 lines
8.6 KiB
Python
315 lines
8.6 KiB
Python
"""Fixture builders for integration tests (Faker-based)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import zipfile
|
||
from collections.abc import Iterable
|
||
from dataclasses import dataclass
|
||
|
||
from faker import Faker
|
||
from openpyxl import Workbook
|
||
|
||
fake = Faker("ru_RU")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CertificateRow:
|
||
issue_date: str
|
||
certificate_number: str
|
||
expiry_date: str
|
||
certificate_file_url: str
|
||
organisation_name: str
|
||
inn: str
|
||
ogrn: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ManufacturerRow:
|
||
full_legal_name: str
|
||
inn: str
|
||
ogrn: str
|
||
address: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class IndustrialProductRow:
|
||
full_organisation_name: str
|
||
ogrn: str
|
||
inn: str
|
||
registry_number: str
|
||
product_name: str
|
||
product_model: str
|
||
okpd2_code: str
|
||
tnved_code: str
|
||
regulatory_document: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class InspectionRow:
|
||
registration_number: str
|
||
inn: str
|
||
ogrn: str
|
||
organisation_name: str
|
||
control_authority: str
|
||
inspection_type: str
|
||
inspection_form: str
|
||
start_date: str
|
||
end_date: str
|
||
status: str
|
||
legal_basis: str
|
||
result: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ProcurementRow:
|
||
purchase_number: str
|
||
purchase_name: str
|
||
customer_inn: str
|
||
customer_kpp: str
|
||
customer_ogrn: str
|
||
customer_name: str
|
||
max_price: str
|
||
publish_date: str
|
||
end_date: str
|
||
status: str
|
||
href: str
|
||
|
||
|
||
def _digits(length: int) -> str:
|
||
return "".join(str(fake.random_int(0, 9)) for _ in range(length))
|
||
|
||
|
||
def build_minpromtorg_certificates_excel(
|
||
count: int = 5,
|
||
) -> tuple[bytes, list[CertificateRow]]:
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(
|
||
[
|
||
"issue_date",
|
||
"certificate_number",
|
||
"expiry_date",
|
||
"certificate_file_url",
|
||
"organisation_name",
|
||
"inn",
|
||
"ogrn",
|
||
]
|
||
)
|
||
|
||
rows: list[CertificateRow] = []
|
||
for _ in range(count):
|
||
row = CertificateRow(
|
||
issue_date=str(fake.date()),
|
||
certificate_number=f"{fake.bothify(text='??-####-#####')}",
|
||
expiry_date=str(fake.date()),
|
||
certificate_file_url=fake.url(),
|
||
organisation_name=fake.company(),
|
||
inn=_digits(10),
|
||
ogrn=_digits(13),
|
||
)
|
||
rows.append(row)
|
||
ws.append(
|
||
[
|
||
row.issue_date,
|
||
row.certificate_number,
|
||
row.expiry_date,
|
||
row.certificate_file_url,
|
||
row.organisation_name,
|
||
row.inn,
|
||
row.ogrn,
|
||
]
|
||
)
|
||
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
wb.close()
|
||
return buf.getvalue(), rows
|
||
|
||
|
||
def build_minpromtorg_manufacturers_excel(
|
||
count: int = 5,
|
||
) -> tuple[bytes, list[ManufacturerRow]]:
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(["full_legal_name", "inn", "ogrn", "address"])
|
||
|
||
rows: list[ManufacturerRow] = []
|
||
for _ in range(count):
|
||
row = ManufacturerRow(
|
||
full_legal_name=fake.company(),
|
||
inn=_digits(10),
|
||
ogrn=_digits(13),
|
||
address=fake.address().replace("\n", ", "),
|
||
)
|
||
rows.append(row)
|
||
ws.append([row.full_legal_name, row.inn, row.ogrn, row.address])
|
||
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
wb.close()
|
||
return buf.getvalue(), rows
|
||
|
||
|
||
def build_minpromtorg_products_excel(
|
||
count: int = 5,
|
||
) -> tuple[bytes, list[IndustrialProductRow]]:
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(
|
||
[
|
||
"Реестр промышленной продукции, произведенной на территории РФ",
|
||
]
|
||
)
|
||
ws.append(
|
||
[
|
||
"Полное наименование организации",
|
||
"ОГРН",
|
||
"ИНН",
|
||
"Регистрационный номер записи",
|
||
"Наименование продукции",
|
||
"Модель или модификация",
|
||
"Код по ОКПД2",
|
||
"Код по ТН ВЭД",
|
||
"Наименование нормативного документа",
|
||
]
|
||
)
|
||
|
||
rows: list[IndustrialProductRow] = []
|
||
for _ in range(count):
|
||
row = IndustrialProductRow(
|
||
full_organisation_name=fake.company(),
|
||
ogrn=_digits(13),
|
||
inn=_digits(10),
|
||
registry_number=f"MPP-{_digits(8)}",
|
||
product_name=fake.sentence(nb_words=4),
|
||
product_model=fake.bothify(text="MODEL-###"),
|
||
okpd2_code=f"{fake.random_int(min=10, max=99)}.{fake.random_int(min=10, max=99)}",
|
||
tnved_code=_digits(10),
|
||
regulatory_document=fake.sentence(nb_words=5),
|
||
)
|
||
rows.append(row)
|
||
ws.append(
|
||
[
|
||
row.full_organisation_name,
|
||
row.ogrn,
|
||
row.inn,
|
||
row.registry_number,
|
||
row.product_name,
|
||
row.product_model,
|
||
row.okpd2_code,
|
||
row.tnved_code,
|
||
row.regulatory_document,
|
||
]
|
||
)
|
||
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
wb.close()
|
||
return buf.getvalue(), rows
|
||
|
||
|
||
def build_proverki_xml(count: int = 3) -> tuple[bytes, list[InspectionRow]]:
|
||
rows: list[InspectionRow] = []
|
||
parts = ["<?xml version='1.0' encoding='utf-8'?>", "<INSPECTIONS>"]
|
||
|
||
for _ in range(count):
|
||
row = InspectionRow(
|
||
registration_number=_digits(12),
|
||
inn=_digits(10),
|
||
ogrn=_digits(13),
|
||
organisation_name=fake.company(),
|
||
control_authority=fake.company(),
|
||
inspection_type=fake.word(),
|
||
inspection_form=fake.word(),
|
||
start_date=str(fake.date()),
|
||
end_date=str(fake.date()),
|
||
status=fake.word(),
|
||
legal_basis=fake.sentence(nb_words=4),
|
||
result=fake.sentence(nb_words=3),
|
||
)
|
||
rows.append(row)
|
||
parts.append(
|
||
"<INSPECTION "
|
||
f'ERPID="{row.registration_number}" '
|
||
f'INN="{row.inn}" '
|
||
f'OGRN="{row.ogrn}" '
|
||
f'ORG_NAME="{row.organisation_name}" '
|
||
f'FRGU_ORG_NAME="{row.control_authority}" '
|
||
f'ITYPE_NAME="{row.inspection_type}" '
|
||
f'ICARRYOUT_TYPE_NAME="{row.inspection_form}" '
|
||
f'START_DATE="{row.start_date}" '
|
||
f'END_DATE="{row.end_date}" '
|
||
f'STATUS="{row.status}" '
|
||
f'FZ_NAME="{row.legal_basis}" '
|
||
f'RESULT="{row.result}" />'
|
||
)
|
||
|
||
parts.append("</INSPECTIONS>")
|
||
xml = "".join(parts).encode("utf-8")
|
||
return xml, rows
|
||
|
||
|
||
def build_zakupki_xml(count: int = 3) -> tuple[bytes, list[ProcurementRow]]:
|
||
rows: list[ProcurementRow] = []
|
||
parts = ["<?xml version='1.0' encoding='utf-8'?>", "<root>"]
|
||
|
||
for _ in range(count):
|
||
row = ProcurementRow(
|
||
purchase_number=_digits(19),
|
||
purchase_name=fake.sentence(nb_words=6),
|
||
customer_inn=_digits(10),
|
||
customer_kpp=_digits(9),
|
||
customer_ogrn=_digits(13),
|
||
customer_name=fake.company(),
|
||
max_price=str(fake.pydecimal(left_digits=7, right_digits=2, positive=True)),
|
||
publish_date=str(fake.date()),
|
||
end_date=str(fake.date()),
|
||
status=fake.word(),
|
||
href=fake.url(),
|
||
)
|
||
rows.append(row)
|
||
parts.append(
|
||
"<notification>"
|
||
f"<purchaseNumber>{row.purchase_number}</purchaseNumber>"
|
||
f"<purchaseObjectInfo>{row.purchase_name}</purchaseObjectInfo>"
|
||
"<customer>"
|
||
f"<INN>{row.customer_inn}</INN>"
|
||
f"<KPP>{row.customer_kpp}</KPP>"
|
||
f"<OGRN>{row.customer_ogrn}</OGRN>"
|
||
f"<fullName>{row.customer_name}</fullName>"
|
||
"</customer>"
|
||
f"<maxPrice>{row.max_price}</maxPrice>"
|
||
f"<publishDate>{row.publish_date}</publishDate>"
|
||
f"<endDate>{row.end_date}</endDate>"
|
||
f"<state>{row.status}</state>"
|
||
f"<href>{row.href}</href>"
|
||
"</notification>"
|
||
)
|
||
|
||
parts.append("</root>")
|
||
xml = "".join(parts).encode("utf-8")
|
||
return xml, rows
|
||
|
||
|
||
def build_zip(files: Iterable[tuple[str, bytes]]) -> bytes:
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for name, content in files:
|
||
zf.writestr(name, content)
|
||
return buf.getvalue()
|
||
|
||
|
||
__all__ = [
|
||
"fake",
|
||
"CertificateRow",
|
||
"ManufacturerRow",
|
||
"InspectionRow",
|
||
"ProcurementRow",
|
||
"build_minpromtorg_certificates_excel",
|
||
"build_minpromtorg_manufacturers_excel",
|
||
"build_proverki_xml",
|
||
"build_zakupki_xml",
|
||
"build_zip",
|
||
]
|