Some checks failed
CI/CD Pipeline / Code Quality Checks (push) Failing after 5m5s
CI/CD Pipeline / Run Tests (push) Failing after 5m5s
CI/CD Pipeline / Build Docker Images (push) Has been skipped
CI/CD Pipeline / Push to Gitea Registry (push) Has been skipped
CI/CD Pipeline / Deploy to Server (push) Has been skipped
242 lines
6.5 KiB
Python
242 lines
6.5 KiB
Python
"""Fixture builders for integration tests (Faker-based)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import zipfile
|
|
from dataclasses import dataclass
|
|
from typing import Iterable
|
|
|
|
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 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_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",
|
|
]
|