mostovik-backend/tests/apps/parsers/test_gisp_client.py

from __future__ import annotations

import json

from apps.parsers.clients.gisp import GispProductsClient, GispProductsClientError
from django.test import SimpleTestCase

from tests.utils import Response
from tests.utils import TestHTTPServer as HTTPTestServer


def _gisp_item(number: int) -> dict:
    return {
        "_product_reg_number_2023": str(number),
        "_product_reg_number_2022": f"old-{number}",
        "_res_number": f"resolution-{number}",
        "_org_inn": "7701000001",
        "_org_ogrn": "1027700000001",
        "_org_name": "АО Завод",
        "_product_name": f"Станок {number}",
        "_product_spec": f"Модель {number}",
        "_product_okpd2": "28.41.1",
        "_product_tnved": "8456",
        "_basedondoc_name": "Заключение",
        "_basedondoc_num": "42",
        "_basedondoc_date": "2026-01-15",
    }


class GispProductsClientTest(SimpleTestCase):
    def test_rejects_page_size_above_safe_bound(self):
        with self.assertRaisesMessage(
            ValueError,
            "page_size must be between 1 and 100",
        ):
            GispProductsClient(page_size=101)

    def test_rejects_max_pages_above_safe_bound(self):
        with self.assertRaisesMessage(
            ValueError,
            "max_pages must be between 1 and 100",
        ):
            GispProductsClient(max_pages=101)

    def test_rejects_max_records_above_safe_bound(self):
        with self.assertRaisesMessage(
            ValueError,
            "max_records must be between 1 and 10000",
        ):
            GispProductsClient(max_records=10_001)

    def test_fetch_products_follows_skip_pagination_and_maps_items(self):
        requests: list[dict] = []

        def handle_page(_request, body: bytes) -> Response:
            payload = json.loads(body)
            requests.append(payload)
            options = payload["opt"]
            skip = options["skip"]
            take = options["take"]
            items = [_gisp_item(number) for number in range(skip + 1, 6)][:take]
            return Response(
                body=json.dumps(
                    {"ok": True, "total_count": 5, "items": items}
                ).encode(),
                headers={"Content-Type": "application/json"},
            )

        with HTTPTestServer() as server:
            server.add_route("POST", "/pp719v2/pub/prod/b/", handle_page)
            with GispProductsClient(
                base_url=server.base_url,
                page_size=2,
                max_pages=10,
                max_records=5,
                http_adapter=server.adapter,
            ) as client:
                result = client.fetch_products()

        self.assertEqual(
            [payload["opt"] for payload in requests],
            [
                {"skip": 0, "take": 2, "requireTotalCount": True},
                {"skip": 2, "take": 2, "requireTotalCount": True},
                {"skip": 4, "take": 1, "requireTotalCount": True},
            ],
        )
        self.assertEqual(result.total_count, 5)
        self.assertEqual(result.pages_fetched, 3)
        self.assertEqual(result.items_fetched, 5)
        self.assertFalse(result.truncated)
        self.assertEqual(len(result.products), 5)
        self.assertEqual(result.products[0].registry_number, "1")
        self.assertEqual(result.products[0].inn, "7701000001")
        self.assertEqual(result.products[0].product_name, "Станок 1")
        self.assertEqual(
            result.products[0].regulatory_document,
            "Заключение | 42 | 2026-01-15",
        )

    def test_fetch_products_stops_at_max_pages_and_reports_truncation(self):
        calls = 0

        def handle_page(_request, body: bytes) -> Response:
            nonlocal calls
            calls += 1
            options = json.loads(body)["opt"]
            items = [
                _gisp_item(number)
                for number in range(options["skip"] + 1, options["skip"] + 3)
            ]
            return Response(
                body=json.dumps(
                    {"ok": True, "total_count": 100, "items": items}
                ).encode(),
                headers={"Content-Type": "application/json"},
            )

        with HTTPTestServer() as server:
            server.add_route("POST", "/pp719v2/pub/prod/b/", handle_page)
            with GispProductsClient(
                base_url=server.base_url,
                page_size=2,
                max_pages=2,
                max_records=100,
                http_adapter=server.adapter,
            ) as client:
                result = client.fetch_products()

        self.assertEqual(calls, 2)
        self.assertEqual(result.items_fetched, 4)
        self.assertTrue(result.truncated)

    def test_fetch_products_rejects_invalid_response_shape(self):
        with HTTPTestServer() as server:
            server.add_route(
                "POST",
                "/pp719v2/pub/prod/b/",
                lambda _request, _body: Response(
                    body=b'{"ok":true,"total_count":1}',
                    headers={"Content-Type": "application/json"},
                ),
            )
            client = GispProductsClient(
                base_url=server.base_url,
                http_adapter=server.adapter,
            )

            with self.assertRaisesMessage(
                GispProductsClientError,
                "returned no items list",
            ):
                client.fetch_products()

    def test_fetch_products_rejects_empty_page_before_known_total(self):
        with HTTPTestServer() as server:
            server.add_route(
                "POST",
                "/pp719v2/pub/prod/b/",
                lambda _request, _body: Response(
                    body=b'{"ok":true,"total_count":10,"items":[]}',
                    headers={"Content-Type": "application/json"},
                ),
            )
            client = GispProductsClient(
                base_url=server.base_url,
                http_adapter=server.adapter,
            )

            with self.assertRaisesMessage(
                GispProductsClientError,
                "empty page before total_count",
            ):
                client.fetch_products()