157 lines
5.5 KiB
Python
157 lines
5.5 KiB
Python
from __future__ import annotations
|
||
|
||
from urllib.parse import parse_qs
|
||
|
||
from apps.parsers.clients.fns import FNSApiClient, FNSApiClientError
|
||
from django.test import SimpleTestCase
|
||
|
||
from tests.utils import Response
|
||
from tests.utils import TestHTTPServer as HTTPTestServer
|
||
|
||
|
||
class FNSApiClientTest(SimpleTestCase):
|
||
def test_fetch_reports_by_inn_uses_exact_search_and_maps_financial_lines(self):
|
||
search_queries: list[dict[str, list[str]]] = []
|
||
|
||
def search(_request, _body: bytes) -> Response:
|
||
search_queries.append(parse_qs(_request.query))
|
||
return Response(
|
||
body=(
|
||
b'{"content":['
|
||
b'{"id":1,"inn":"7700000000","ogrn":"1027700000000"},'
|
||
b'{"id":42,"inn":"<strong>7701000001</strong>",'
|
||
b'"ogrn":"1027700000001"}]}'
|
||
),
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
|
||
bfo = [
|
||
{
|
||
"period": "2025",
|
||
"actualBfoDate": "2026-03-10",
|
||
"typeCorrections": [
|
||
{
|
||
"type": 12,
|
||
"correction": {
|
||
"id": 555,
|
||
"datePresent": "2026-03-10T12:00:00",
|
||
"balance": {
|
||
"id": 555,
|
||
"okud": "0710001",
|
||
"current1600": 120,
|
||
"previous1600": 100,
|
||
"beforePrevious1600": 90,
|
||
},
|
||
"financialResult": {
|
||
"id": 555,
|
||
"okud": "0710002",
|
||
"current2110": 300,
|
||
"previous2110": 250,
|
||
"current2400": 50,
|
||
"previous2400": 40,
|
||
},
|
||
},
|
||
}
|
||
],
|
||
},
|
||
{
|
||
"period": "2024",
|
||
"typeCorrections": [],
|
||
},
|
||
]
|
||
|
||
with HTTPTestServer() as server:
|
||
server.add_route("GET", "/advanced-search/organizations/search", search)
|
||
server.add_json(
|
||
"/nbo/organizations/42",
|
||
{
|
||
"id": 42,
|
||
"inn": "7701000001",
|
||
"ogrn": "1027700000001",
|
||
"fullName": "АО Завод",
|
||
},
|
||
)
|
||
server.add_json("/nbo/organizations/42/bfo/", bfo)
|
||
with FNSApiClient(
|
||
base_url=server.base_url,
|
||
max_periods=1,
|
||
http_adapter=server.adapter,
|
||
) as client:
|
||
self.assertIn(
|
||
"Mozilla/5.0", client.http_client.session.headers["User-Agent"]
|
||
)
|
||
reports = client.fetch_reports_by_inn("7701000001")
|
||
|
||
self.assertEqual(
|
||
search_queries,
|
||
[{"query": ["7701000001"], "page": ["0"], "size": ["10"]}],
|
||
)
|
||
self.assertEqual(len(reports), 1)
|
||
report = reports[0]
|
||
self.assertEqual(report.external_id, "girbo:555")
|
||
self.assertEqual(report.ogrn, "1027700000001")
|
||
self.assertEqual(report.year, 2025)
|
||
self.assertEqual(report.period_type, 12)
|
||
self.assertEqual(
|
||
[
|
||
(
|
||
line.form_code,
|
||
line.line_code,
|
||
line.period_start,
|
||
line.period_end,
|
||
)
|
||
for line in report.lines
|
||
],
|
||
[
|
||
("1", "1600", 100, 120),
|
||
("2", "2110", 250, 300),
|
||
("2", "2400", 40, 50),
|
||
],
|
||
)
|
||
kwargs = report.to_save_report_kwargs(batch_id=7)
|
||
self.assertEqual(kwargs["source"], "api")
|
||
self.assertEqual(kwargs["batch_id"], 7)
|
||
self.assertEqual(kwargs["file_hash"], report.file_hash)
|
||
self.assertEqual(len(kwargs["lines_data"]), 3)
|
||
|
||
def test_fetch_reports_by_organization_id_does_not_run_search(self):
|
||
with HTTPTestServer() as server:
|
||
server.add_json(
|
||
"/nbo/organizations/42",
|
||
{
|
||
"id": 42,
|
||
"inn": "7701000001",
|
||
"ogrn": "1027700000001",
|
||
"fullName": "АО Завод",
|
||
},
|
||
)
|
||
server.add_json("/nbo/organizations/42/bfo/", [])
|
||
client = FNSApiClient(
|
||
base_url=server.base_url,
|
||
http_adapter=server.adapter,
|
||
)
|
||
|
||
reports = client.fetch_reports_by_organization_id(42)
|
||
|
||
self.assertEqual(reports, [])
|
||
|
||
def test_fetch_reports_by_inn_fails_when_exact_match_is_absent(self):
|
||
with HTTPTestServer() as server:
|
||
server.add_json(
|
||
"/advanced-search/organizations/search",
|
||
{"content": [{"id": 1, "inn": "7700000000"}]},
|
||
)
|
||
client = FNSApiClient(
|
||
base_url=server.base_url,
|
||
http_adapter=server.adapter,
|
||
)
|
||
|
||
with self.assertRaisesMessage(FNSApiClientError, "not found"):
|
||
client.fetch_reports_by_inn("7701000001")
|
||
|
||
def test_rejects_invalid_inn_before_request(self):
|
||
client = FNSApiClient()
|
||
|
||
with self.assertRaisesMessage(ValueError, "10 or 12 digits"):
|
||
client.fetch_reports_by_inn("not-an-inn")
|