Aleksandr Meshchriakov
a91ed1f1ae
Some checks failed
CI/CD Pipeline / Code Quality Checks (push) Failing after 3m10s
CI/CD Pipeline / Run Tests (push) Successful in 3m35s
CI/CD Pipeline / Telegram Notify Success (push) Has been skipped
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 2m26s
CI/CD Pipeline / Run Tests (pull_request) Successful in 2m46s
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been skipped
1024 lines
36 KiB
Python
1024 lines
36 KiB
Python
"""Tests for Checko API client using local HTTP server (no mocks)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from urllib.parse import parse_qs
|
||
|
||
import requests
|
||
from apps.parsers.clients.checko import (
|
||
BankRequest,
|
||
CaseRole,
|
||
CheckoAPIError,
|
||
CheckoClient,
|
||
CheckoConnectionError,
|
||
CheckoNotFoundError,
|
||
CheckoRateLimitError,
|
||
CheckoValidationError,
|
||
CompanyRequest,
|
||
ContractLaw,
|
||
ContractRole,
|
||
ContractsRequest,
|
||
EnforcementsRequest,
|
||
EntrepreneurRequest,
|
||
FinancesRequest,
|
||
InspectionsRequest,
|
||
LegalCasesRequest,
|
||
ObjectType,
|
||
PersonRequest,
|
||
SearchRequest,
|
||
SearchType,
|
||
SortOrder,
|
||
)
|
||
from apps.parsers.clients.checko.datasets import (
|
||
OKFS,
|
||
OKOPF,
|
||
OKPD,
|
||
OKPD2,
|
||
OKVED2,
|
||
AccountCodes,
|
||
CompanyStatuses,
|
||
EntrepreneurStatuses,
|
||
)
|
||
from django.test import SimpleTestCase
|
||
from requests.adapters import BaseAdapter
|
||
|
||
from tests.utils import Response, TestHTTPServer
|
||
from tests.utils.fixtures import fake
|
||
|
||
|
||
def _meta_ok() -> dict:
|
||
return {
|
||
"status": "ok",
|
||
"today_request_count": fake.random_int(min=1, max=10),
|
||
"balance": float(fake.pydecimal(left_digits=2, right_digits=2, positive=True)),
|
||
}
|
||
|
||
|
||
def _client_for(server: TestHTTPServer) -> CheckoClient:
|
||
return CheckoClient(
|
||
api_key="test_key",
|
||
base_url=f"{server.base_url}/v2",
|
||
http_adapter=server.adapter,
|
||
)
|
||
|
||
|
||
def _digits(length: int) -> str:
|
||
return "".join(str(fake.random_int(0, 9)) for _ in range(length))
|
||
|
||
|
||
class _RaisingAdapter(BaseAdapter):
|
||
def __init__(self, exc: Exception) -> None:
|
||
super().__init__()
|
||
self._exc = exc
|
||
|
||
def send(self, _request, **_kwargs):
|
||
raise self._exc
|
||
|
||
def close(self) -> None:
|
||
return
|
||
|
||
|
||
class CheckoClientInitTest(SimpleTestCase):
|
||
def test_client_initialization_default(self):
|
||
client = CheckoClient(api_key="test_key")
|
||
self.assertEqual(client.api_key, "test_key")
|
||
self.assertEqual(client.base_url, "https://api.checko.ru/v2")
|
||
self.assertEqual(client.timeout, 30)
|
||
self.assertIsNone(client.proxies)
|
||
|
||
def test_client_initialization_custom(self):
|
||
custom_url = f"https://{fake.domain_name()}"
|
||
proxy = f"http://{fake.ipv4()}:{fake.port_number()}"
|
||
client = CheckoClient(
|
||
api_key="test_key",
|
||
base_url=custom_url,
|
||
timeout=60,
|
||
proxies=[proxy],
|
||
)
|
||
|
||
self.assertEqual(client.base_url, custom_url)
|
||
self.assertEqual(client.timeout, 60)
|
||
self.assertEqual(client.proxies, [proxy])
|
||
|
||
def test_context_manager(self):
|
||
with CheckoClient(api_key="test_key") as client:
|
||
self.assertIsInstance(client, CheckoClient)
|
||
|
||
|
||
class CheckoClientValidationTest(SimpleTestCase):
|
||
def setUp(self):
|
||
self.client = CheckoClient(api_key="test_key")
|
||
|
||
def test_company_request_requires_identifier(self):
|
||
with self.assertRaises(CheckoValidationError) as context:
|
||
self.client.get_company(CompanyRequest())
|
||
self.assertIn("ogrn", str(context.exception).lower())
|
||
|
||
def test_entrepreneur_request_requires_identifier(self):
|
||
with self.assertRaises(CheckoValidationError) as context:
|
||
self.client.get_entrepreneur(EntrepreneurRequest())
|
||
self.assertIn("ogrn", str(context.exception).lower())
|
||
|
||
def test_search_request_min_query_length(self):
|
||
with self.assertRaises(CheckoValidationError) as context:
|
||
self.client.search(
|
||
SearchRequest(
|
||
by=SearchType.NAME, obj=ObjectType.ORGANIZATION, query="abc"
|
||
)
|
||
)
|
||
self.assertIn("4", str(context.exception))
|
||
|
||
def test_finances_request_requires_identifier(self):
|
||
with self.assertRaises(CheckoValidationError) as context:
|
||
self.client.get_finances(FinancesRequest())
|
||
self.assertIn("ogrn", str(context.exception).lower())
|
||
|
||
|
||
class CheckoRequestParamsTest(SimpleTestCase):
|
||
def test_basic_request_params(self):
|
||
company = CompanyRequest(
|
||
ogrn=_digits(13),
|
||
inn=_digits(10),
|
||
kpp=_digits(9),
|
||
okpo=_digits(8),
|
||
source=True,
|
||
)
|
||
params = company.to_params()
|
||
self.assertEqual(params["ogrn"], company.ogrn)
|
||
self.assertEqual(params["source"], "true")
|
||
|
||
entrepreneur = EntrepreneurRequest(
|
||
ogrn=_digits(15),
|
||
inn=_digits(12),
|
||
okpo=_digits(8),
|
||
source=True,
|
||
)
|
||
ent_params = entrepreneur.to_params()
|
||
self.assertEqual(ent_params["inn"], entrepreneur.inn)
|
||
self.assertEqual(ent_params["source"], "true")
|
||
|
||
person = PersonRequest(inn=_digits(12))
|
||
self.assertEqual(person.to_params()["inn"], person.inn)
|
||
|
||
finances = FinancesRequest(
|
||
ogrn=_digits(13),
|
||
inn=_digits(10),
|
||
kpp=_digits(9),
|
||
extended=True,
|
||
)
|
||
finances_params = finances.to_params()
|
||
self.assertEqual(finances_params["extended"], "true")
|
||
|
||
def test_extended_request_params(self):
|
||
search = SearchRequest(
|
||
by=SearchType.NAME,
|
||
obj=ObjectType.ORGANIZATION,
|
||
query=fake.pystr(min_chars=5, max_chars=10),
|
||
region=f"{fake.random_int(min=1, max=99):02d}",
|
||
okved=f"{fake.random_int(min=1, max=99)}.{fake.random_int(min=1, max=99)}",
|
||
opf=_digits(5),
|
||
active=True,
|
||
limit=fake.random_int(min=1, max=50),
|
||
page=fake.random_int(min=2, max=5),
|
||
)
|
||
search_params = search.to_params()
|
||
self.assertEqual(search_params["active"], "true")
|
||
|
||
contracts = ContractsRequest(
|
||
law=ContractLaw.FZ44,
|
||
ogrn=_digits(13),
|
||
inn=_digits(10),
|
||
kpp=_digits(9),
|
||
role=ContractRole.CUSTOMER,
|
||
limit=fake.random_int(min=1, max=50),
|
||
page=fake.random_int(min=2, max=5),
|
||
sort=SortOrder.DATE_DESC,
|
||
)
|
||
contract_params = contracts.to_params()
|
||
self.assertEqual(contract_params["role"], ContractRole.CUSTOMER.value)
|
||
|
||
inspections = InspectionsRequest(
|
||
ogrn=_digits(13),
|
||
inn=_digits(10),
|
||
kpp=_digits(9),
|
||
limit=fake.random_int(min=1, max=50),
|
||
page=fake.random_int(min=2, max=5),
|
||
sort=SortOrder.DATE_ASC,
|
||
)
|
||
inspection_params = inspections.to_params()
|
||
self.assertEqual(inspection_params["sort"], SortOrder.DATE_ASC.value)
|
||
|
||
enforcements = EnforcementsRequest(
|
||
ogrn=_digits(13),
|
||
inn=_digits(10),
|
||
kpp=_digits(9),
|
||
limit=fake.random_int(min=1, max=50),
|
||
page=fake.random_int(min=2, max=5),
|
||
sort=SortOrder.PRICE_DESC,
|
||
)
|
||
enforcement_params = enforcements.to_params()
|
||
self.assertEqual(enforcement_params["sort"], SortOrder.PRICE_DESC.value)
|
||
|
||
legal_cases = LegalCasesRequest(
|
||
ogrn=_digits(13),
|
||
inn=_digits(10),
|
||
kpp=_digits(9),
|
||
role=CaseRole.DEFENDANT,
|
||
actual=True,
|
||
active=True,
|
||
date_from=str(fake.date()),
|
||
date_to=str(fake.date()),
|
||
claim_amount_from=fake.random_int(min=1, max=1000),
|
||
claim_amount_to=fake.random_int(min=1001, max=5000),
|
||
limit=fake.random_int(min=1, max=50),
|
||
page=fake.random_int(min=2, max=5),
|
||
sort=SortOrder.DATE_ASC,
|
||
)
|
||
legal_params = legal_cases.to_params()
|
||
self.assertEqual(legal_params["role"], CaseRole.DEFENDANT.value)
|
||
|
||
bank = BankRequest(bic=_digits(9))
|
||
self.assertEqual(bank.to_params()["bic"], bank.bic)
|
||
|
||
|
||
class CheckoClientApiTest(SimpleTestCase):
|
||
def test_get_company_success(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
ogrn = "".join(str(fake.random_int(0, 9)) for _ in range(13))
|
||
short_name = fake.company()
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/company",
|
||
{
|
||
"data": {
|
||
"ogrn": ogrn,
|
||
"inn": inn,
|
||
"short_name": short_name,
|
||
"status": {"code": "100", "name": "Действующее"},
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_company(CompanyRequest(inn=inn))
|
||
|
||
self.assertEqual(response.meta.status, "ok")
|
||
self.assertEqual(response.data.inn, inn)
|
||
self.assertEqual(response.data.short_name, short_name)
|
||
self.assertEqual(response.data.status.code, "100")
|
||
|
||
def test_get_company_not_found(self):
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/company",
|
||
{
|
||
"data": None,
|
||
"meta": {"status": "error", "message": "Не найден"},
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
with self.assertRaises(CheckoNotFoundError):
|
||
client.get_company(
|
||
CompanyRequest(
|
||
inn="".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
)
|
||
)
|
||
|
||
def test_get_entrepreneur_success(self):
|
||
ogrnip = "".join(str(fake.random_int(0, 9)) for _ in range(15))
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(12))
|
||
full_name = fake.name()
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/entrepreneur",
|
||
{
|
||
"data": {
|
||
"ogrnip": ogrnip,
|
||
"inn": inn,
|
||
"full_name": full_name,
|
||
"status": {"code": "100", "name": "Действующий"},
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_entrepreneur(EntrepreneurRequest(inn=inn))
|
||
|
||
self.assertEqual(response.data.ogrnip, ogrnip)
|
||
self.assertEqual(response.data.full_name, full_name)
|
||
|
||
def test_search_organizations(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
ogrn = "".join(str(fake.random_int(0, 9)) for _ in range(13))
|
||
name = fake.company()
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/search",
|
||
{
|
||
"data": {
|
||
"records": [
|
||
{"inn": inn, "ogrn": ogrn, "short_name": name},
|
||
],
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.search(
|
||
SearchRequest(
|
||
by=SearchType.NAME,
|
||
obj=ObjectType.ORGANIZATION,
|
||
query=f"{name} {fake.word()}",
|
||
)
|
||
)
|
||
|
||
self.assertEqual(response.data.pagination.total_records, 1)
|
||
self.assertEqual(response.data.organizations[0].inn, inn)
|
||
|
||
def test_search_with_entrepreneurs(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(12))
|
||
ogrn = "".join(str(fake.random_int(0, 9)) for _ in range(15))
|
||
name = fake.name()
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/search",
|
||
{
|
||
"data": {
|
||
"entrepreneurs": [
|
||
{"inn": inn, "ogrn": ogrn, "full_name": name},
|
||
],
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.search(
|
||
SearchRequest(
|
||
by=SearchType.NAME,
|
||
obj=ObjectType.ENTREPRENEUR,
|
||
query=fake.pystr(min_chars=5, max_chars=10),
|
||
)
|
||
)
|
||
|
||
self.assertEqual(len(response.data.entrepreneurs), 1)
|
||
self.assertEqual(response.data.entrepreneurs[0].inn, inn)
|
||
|
||
def test_get_finances(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
ogrn = "".join(str(fake.random_int(0, 9)) for _ in range(13))
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/finances",
|
||
{
|
||
"data": {
|
||
"inn": inn,
|
||
"ogrn": ogrn,
|
||
"reports": [
|
||
{
|
||
"year": 2024,
|
||
"period": 12,
|
||
"lines": [
|
||
{"code": "1100", "value": 1000},
|
||
],
|
||
}
|
||
],
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_finances(FinancesRequest(inn=inn))
|
||
|
||
self.assertEqual(response.data.inn, inn)
|
||
self.assertTrue(response.data.reports)
|
||
|
||
def test_get_contracts(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
contract_number = "".join(str(fake.random_int(0, 9)) for _ in range(12))
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/contracts",
|
||
{
|
||
"data": {
|
||
"contracts": [
|
||
{
|
||
"registry_number": contract_number,
|
||
"price": "1000.00",
|
||
"publish_date": str(fake.date()),
|
||
}
|
||
],
|
||
"pagination": {
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_contracts(
|
||
ContractsRequest(inn=inn, law=ContractLaw.FZ44)
|
||
)
|
||
|
||
self.assertEqual(response.data.contracts[0].registry_number, contract_number)
|
||
|
||
def test_get_legal_cases(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
case_number = fake.bothify(text="A-####/####")
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/legal-cases",
|
||
{
|
||
"data": {
|
||
"cases": [
|
||
{
|
||
"case_number": case_number,
|
||
"court_name": fake.company(),
|
||
}
|
||
],
|
||
"pagination": {
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_legal_cases(LegalCasesRequest(inn=inn))
|
||
|
||
self.assertEqual(response.data.cases[0].case_number, case_number)
|
||
|
||
def test_api_error_handling(self):
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/company",
|
||
{"meta": {"status": "error", "message": "Ошибка API"}},
|
||
)
|
||
client = _client_for(server)
|
||
with self.assertRaises(CheckoAPIError):
|
||
client.get_company(
|
||
CompanyRequest(
|
||
inn="".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
)
|
||
)
|
||
|
||
def test_rate_limit_error_handling(self):
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/company",
|
||
{"meta": {"status": "error", "message": "Лимит запросов"}},
|
||
)
|
||
client = _client_for(server)
|
||
with self.assertRaises(CheckoRateLimitError):
|
||
client.get_company(
|
||
CompanyRequest(
|
||
inn="".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
)
|
||
)
|
||
|
||
def test_not_found_error_handling(self):
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/company",
|
||
{"meta": {"status": "error", "message": "не найден"}},
|
||
)
|
||
client = _client_for(server)
|
||
with self.assertRaises(CheckoNotFoundError):
|
||
client.get_company(
|
||
CompanyRequest(
|
||
inn="".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
)
|
||
)
|
||
|
||
|
||
class CheckoClientExtraEndpointsTest(SimpleTestCase):
|
||
def test_request_connection_error(self):
|
||
adapter = _RaisingAdapter(requests.exceptions.ConnectionError(fake.sentence()))
|
||
client = CheckoClient(
|
||
api_key="test_key",
|
||
base_url="https://api.checko.ru/v2",
|
||
http_adapter=adapter,
|
||
)
|
||
with self.assertRaises(CheckoConnectionError):
|
||
client.get_company(CompanyRequest(inn=_digits(10)))
|
||
|
||
def test_get_inspections(self):
|
||
inn = _digits(10)
|
||
inspection_id = fake.random_int(min=1, max=9999)
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/inspections",
|
||
{
|
||
"data": {
|
||
"inspections": [
|
||
{
|
||
"id": inspection_id,
|
||
"status": fake.word(),
|
||
}
|
||
],
|
||
"pagination": {
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_inspections(InspectionsRequest(inn=inn))
|
||
|
||
self.assertEqual(response.data.inspections[0].id, inspection_id)
|
||
|
||
def test_iter_inspections_pagination(self):
|
||
inn = _digits(10)
|
||
|
||
def inspections_handler(req, _body):
|
||
params = parse_qs(req.query)
|
||
page = int(params.get("page", ["1"])[0])
|
||
payload = {
|
||
"data": {
|
||
"inspections": [
|
||
{
|
||
"id": fake.random_int(min=1, max=9999),
|
||
"status": fake.word(),
|
||
}
|
||
],
|
||
"pagination": {
|
||
"total_records": 2,
|
||
"total_pages": 2,
|
||
"current_page": page,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
}
|
||
if page == 2:
|
||
payload["data"]["inspections"] = []
|
||
return Response(
|
||
status=200,
|
||
body=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_route("GET", "/v2/inspections", inspections_handler)
|
||
client = _client_for(server)
|
||
inspections = list(client.iter_inspections(InspectionsRequest(inn=inn)))
|
||
|
||
self.assertTrue(inspections)
|
||
|
||
def test_iter_inspections_single_page(self):
|
||
inn = _digits(10)
|
||
|
||
def inspections_handler(_req, _body):
|
||
payload = {
|
||
"data": {
|
||
"inspections": [
|
||
{
|
||
"id": fake.random_int(min=1, max=9999),
|
||
"status": fake.word(),
|
||
}
|
||
],
|
||
"pagination": {
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
}
|
||
return Response(
|
||
status=200,
|
||
body=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_route("GET", "/v2/inspections", inspections_handler)
|
||
client = _client_for(server)
|
||
inspections = list(client.iter_inspections(InspectionsRequest(inn=inn)))
|
||
|
||
self.assertEqual(len(inspections), 1)
|
||
|
||
def test_get_enforcements(self):
|
||
inn = _digits(10)
|
||
enforcement_number = fake.bothify(text="##-####")
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/enforcements",
|
||
{
|
||
"data": {
|
||
"enforcements": [
|
||
{"number": enforcement_number, "status": fake.word()}
|
||
],
|
||
"pagination": {
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
"total_debt": float(
|
||
fake.pydecimal(left_digits=4, right_digits=2, positive=True)
|
||
),
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_enforcements(EnforcementsRequest(inn=inn))
|
||
|
||
self.assertEqual(response.data.enforcements[0].number, enforcement_number)
|
||
|
||
def test_iter_enforcements_pagination(self):
|
||
inn = _digits(10)
|
||
|
||
def enforcements_handler(req, _body):
|
||
params = parse_qs(req.query)
|
||
page = int(params.get("page", ["1"])[0])
|
||
payload = {
|
||
"data": {
|
||
"enforcements": [{"number": fake.bothify(text="##-####")}],
|
||
"pagination": {
|
||
"total_records": 2,
|
||
"total_pages": 2,
|
||
"current_page": page,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
}
|
||
if page == 2:
|
||
payload["data"]["enforcements"] = []
|
||
return Response(
|
||
status=200,
|
||
body=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_route("GET", "/v2/enforcements", enforcements_handler)
|
||
client = _client_for(server)
|
||
enforcements = list(client.iter_enforcements(EnforcementsRequest(inn=inn)))
|
||
|
||
self.assertTrue(enforcements)
|
||
|
||
def test_iter_enforcements_single_page(self):
|
||
inn = _digits(10)
|
||
|
||
def enforcements_handler(_req, _body):
|
||
payload = {
|
||
"data": {
|
||
"enforcements": [{"number": fake.bothify(text="##-####")}],
|
||
"pagination": {
|
||
"total_records": 1,
|
||
"total_pages": 1,
|
||
"current_page": 1,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
}
|
||
return Response(
|
||
status=200,
|
||
body=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_route("GET", "/v2/enforcements", enforcements_handler)
|
||
client = _client_for(server)
|
||
enforcements = list(client.iter_enforcements(EnforcementsRequest(inn=inn)))
|
||
|
||
self.assertEqual(len(enforcements), 1)
|
||
|
||
def test_get_bank(self):
|
||
bic = _digits(9)
|
||
name = fake.company()
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/bank",
|
||
{
|
||
"data": {"bic": bic, "name": name, "city": fake.city()},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_bank(BankRequest(bic=bic))
|
||
|
||
self.assertEqual(response.data.bic, bic)
|
||
self.assertEqual(response.data.name, name)
|
||
|
||
def test_get_legal_cases_requires_identifier(self):
|
||
client = CheckoClient(api_key="test_key")
|
||
with self.assertRaises(CheckoValidationError):
|
||
client.get_legal_cases(LegalCasesRequest())
|
||
|
||
def test_map_ru_keys_none_returns_none(self):
|
||
from apps.parsers.clients.checko.client import _map_ru_keys
|
||
|
||
self.assertIsNone(_map_ru_keys(None))
|
||
|
||
|
||
class CheckoClientParsingHelpersTest(SimpleTestCase):
|
||
def setUp(self):
|
||
self.client = CheckoClient(api_key="test_key")
|
||
|
||
def test_parse_helpers_return_none(self):
|
||
self.assertIsNone(self.client._parse_okved(None))
|
||
self.assertIsNone(self.client._parse_pagination(None))
|
||
self.assertIsNone(self.client._parse_company_status(None))
|
||
self.assertIsNone(self.client._parse_entrepreneur_status(None))
|
||
self.assertIsNone(self.client._parse_entrepreneur_data(None))
|
||
self.assertIsNone(self.client._parse_person_data(None))
|
||
self.assertIsNone(self.client._parse_search_data(None))
|
||
self.assertIsNone(self.client._parse_contracts_data(None))
|
||
self.assertIsNone(self.client._parse_finances_data(None))
|
||
self.assertIsNone(self.client._parse_inspections_data(None))
|
||
self.assertIsNone(self.client._parse_enforcements_data(None))
|
||
self.assertIsNone(self.client._parse_legal_cases_data(None))
|
||
self.assertIsNone(self.client._parse_bank_data(None))
|
||
self.assertIsNone(self.client._parse_company_data(None))
|
||
|
||
def test_parse_entrepreneur_short_and_leader_list(self):
|
||
data = {
|
||
"ogrnip": _digits(15),
|
||
"inn": _digits(12),
|
||
"full_name": fake.name(),
|
||
}
|
||
parsed = self.client._parse_entrepreneur_short(data)
|
||
self.assertEqual(parsed.inn, data["inn"])
|
||
self.assertIsNone(self.client._parse_leader([]))
|
||
|
||
def test_parse_leader_empty_list_branch(self):
|
||
class _ToggleBoolList(list):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._first = True
|
||
|
||
def __bool__(self):
|
||
if self._first:
|
||
self._first = False
|
||
return True
|
||
return False
|
||
|
||
self.assertIsNone(self.client._parse_leader(_ToggleBoolList()))
|
||
|
||
def test_get_person(self):
|
||
inn = _digits(12)
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/person",
|
||
{
|
||
"data": {"inn": inn, "full_name": fake.name()},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
response = client.get_person(PersonRequest(inn=inn))
|
||
|
||
self.assertEqual(response.data.inn, inn)
|
||
|
||
def test_contracts_inspections_enforcements_require_identifier(self):
|
||
with self.assertRaises(CheckoValidationError):
|
||
self.client.get_contracts(ContractsRequest(law=ContractLaw.FZ44))
|
||
|
||
with self.assertRaises(CheckoValidationError):
|
||
self.client.get_inspections(InspectionsRequest())
|
||
|
||
with self.assertRaises(CheckoValidationError):
|
||
self.client.get_enforcements(EnforcementsRequest())
|
||
|
||
|
||
class CheckoRequestModelsTest(SimpleTestCase):
|
||
def test_company_request_to_params(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
req = CompanyRequest(inn=inn, source=True)
|
||
params = req.to_params()
|
||
self.assertEqual(params["inn"], inn)
|
||
self.assertEqual(params["source"], "true")
|
||
|
||
def test_search_request_to_params(self):
|
||
query = fake.word() + fake.word()
|
||
req = SearchRequest(
|
||
by=SearchType.NAME, obj=ObjectType.ORGANIZATION, query=query
|
||
)
|
||
params = req.to_params()
|
||
self.assertEqual(params["by"], "name")
|
||
self.assertEqual(params["obj"], "org")
|
||
self.assertEqual(params["query"], query)
|
||
|
||
def test_contracts_request_to_params(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
req = ContractsRequest(inn=inn, law=ContractLaw.FZ44)
|
||
params = req.to_params()
|
||
self.assertEqual(params["inn"], inn)
|
||
self.assertEqual(params["law"], "44")
|
||
|
||
def test_legal_cases_request_to_params(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
req = LegalCasesRequest(inn=inn, limit=50)
|
||
params = req.to_params()
|
||
self.assertEqual(params["inn"], inn)
|
||
self.assertEqual(params["limit"], "50")
|
||
|
||
|
||
class CheckoDatasetsTest(SimpleTestCase):
|
||
def test_okved2_get(self):
|
||
items = OKVED2.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(OKVED2.get(items[0].code))
|
||
|
||
def test_okved2_get_name(self):
|
||
items = OKVED2.all()
|
||
name = OKVED2.get_name(items[0].code) if items else None
|
||
self.assertIsInstance(name, str)
|
||
|
||
def test_okved2_search(self):
|
||
items = OKVED2.all()
|
||
term = items[0].name.split(" ")[0] if items else fake.word()
|
||
results = OKVED2.search(term)
|
||
self.assertIsInstance(results, list)
|
||
|
||
def test_okved2_exists(self):
|
||
items = OKVED2.all()
|
||
self.assertTrue(OKVED2.exists(items[0].code))
|
||
|
||
def test_okved2_get_children(self):
|
||
items = OKVED2.all()
|
||
prefix = items[0].code.split(".")[0] if items else ""
|
||
children = OKVED2.get_children(prefix)
|
||
self.assertIsInstance(children, list)
|
||
|
||
def test_okfs_get(self):
|
||
items = OKFS.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(OKFS.get(items[0].code))
|
||
|
||
def test_okfs_get_name(self):
|
||
items = OKFS.all()
|
||
name = OKFS.get_name(items[0].code) if items else None
|
||
self.assertIsInstance(name, str)
|
||
|
||
def test_okopf_get(self):
|
||
items = OKOPF.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(OKOPF.get(items[0].code))
|
||
|
||
def test_account_codes_get(self):
|
||
items = AccountCodes.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(AccountCodes.get(items[0].code))
|
||
|
||
def test_company_statuses_get(self):
|
||
items = CompanyStatuses.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(CompanyStatuses.get(items[0].code))
|
||
|
||
def test_entrepreneur_statuses_get(self):
|
||
items = EntrepreneurStatuses.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(EntrepreneurStatuses.get(items[0].code))
|
||
|
||
def test_okpd_get(self):
|
||
items = OKPD.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(OKPD.get(items[0].code))
|
||
|
||
def test_okpd2_get(self):
|
||
items = OKPD2.all()
|
||
self.assertTrue(items)
|
||
self.assertIsNotNone(OKPD2.get(items[0].code))
|
||
|
||
def test_okved2_parent_and_hierarchy(self):
|
||
items = OKVED2.all()
|
||
target = next((i for i in items if i.parent_code), items[0])
|
||
parent = OKVED2.get_parent(target.code)
|
||
hierarchy = OKVED2.get_hierarchy(target.code)
|
||
self.assertTrue(hierarchy)
|
||
if target.parent_code:
|
||
self.assertIsNotNone(parent)
|
||
|
||
def test_okpd2_parent_and_hierarchy(self):
|
||
items = OKPD2.all()
|
||
target = next((i for i in items if i.parent_code), items[0])
|
||
parent = OKPD2.get_parent(target.code)
|
||
hierarchy = OKPD2.get_hierarchy(target.code)
|
||
self.assertTrue(hierarchy)
|
||
if target.parent_code:
|
||
self.assertIsNotNone(parent)
|
||
|
||
def test_account_codes_helpers(self):
|
||
self.assertTrue(AccountCodes.get_balance_codes())
|
||
self.assertTrue(AccountCodes.get_profit_loss_codes())
|
||
self.assertTrue(AccountCodes.get_capital_codes())
|
||
self.assertTrue(AccountCodes.get_cash_flow_codes())
|
||
|
||
def test_statuses_fallback(self):
|
||
from apps.parsers.clients.checko.datasets import statuses
|
||
|
||
original_filename = statuses.CompanyStatuses._json_filename
|
||
original_data = statuses.CompanyStatuses._data
|
||
try:
|
||
statuses.CompanyStatuses._json_filename = "missing.json"
|
||
statuses.CompanyStatuses._data = None
|
||
items = statuses.CompanyStatuses.all()
|
||
self.assertTrue(items)
|
||
finally:
|
||
statuses.CompanyStatuses._json_filename = original_filename
|
||
statuses.CompanyStatuses._data = original_data
|
||
|
||
|
||
class CheckoClientIteratorsTest(SimpleTestCase):
|
||
def test_iter_contracts_pagination(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
contract_number = "".join(str(fake.random_int(0, 9)) for _ in range(12))
|
||
|
||
def contracts_handler(req, _body):
|
||
params = parse_qs(req.query)
|
||
page = int(params.get("page", ["1"])[0])
|
||
payload = {
|
||
"data": {
|
||
"contracts": [
|
||
{
|
||
"number": contract_number,
|
||
"sum": "1000.00",
|
||
"publish_date": str(fake.date()),
|
||
}
|
||
],
|
||
"pagination": {
|
||
"total_records": 2,
|
||
"total_pages": 2,
|
||
"current_page": page,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
}
|
||
if page == 2:
|
||
payload["data"]["contracts"] = []
|
||
return Response(
|
||
status=200,
|
||
body=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_route("GET", "/v2/contracts", contracts_handler)
|
||
client = _client_for(server)
|
||
contracts = list(
|
||
client.iter_contracts(ContractsRequest(inn=inn, law=ContractLaw.FZ44))
|
||
)
|
||
|
||
self.assertTrue(contracts)
|
||
|
||
|
||
class CheckoExceptionsTest(SimpleTestCase):
|
||
def test_api_error_str_includes_details(self):
|
||
exc = CheckoAPIError("boom", status_code=500, balance=10.5)
|
||
self.assertIn("boom", str(exc))
|
||
self.assertIn("status_code=500", str(exc))
|
||
self.assertIn("balance=10.5", str(exc))
|
||
|
||
def test_connection_error_str(self):
|
||
exc = CheckoConnectionError("conn failed", url="http://example.com")
|
||
self.assertIn("url=http://example.com", str(exc))
|
||
|
||
def test_iter_legal_cases_empty(self):
|
||
inn = "".join(str(fake.random_int(0, 9)) for _ in range(10))
|
||
|
||
with TestHTTPServer() as server:
|
||
server.add_json(
|
||
"/v2/legal-cases",
|
||
{
|
||
"data": {
|
||
"cases": [],
|
||
"pagination": {
|
||
"total_records": 0,
|
||
"total_pages": 0,
|
||
"current_page": 1,
|
||
},
|
||
},
|
||
"meta": _meta_ok(),
|
||
},
|
||
)
|
||
client = _client_for(server)
|
||
cases = list(client.iter_legal_cases(LegalCasesRequest(inn=inn)))
|
||
|
||
self.assertEqual(cases, [])
|