path: root/tests/test_cepis.py

"""Test suite.

These bring up a real standard-library HTTP server and point the client at it.
No test touches the network, and no test monkeypatches `urllib`: the code path
exercised is the one that runs in production, request construction and error
body reading included.

It doubles as the test that `base_url` works — if it did not, none of this
would run.

The fixture field names stay Portuguese because they mirror the API's JSON.
"""

from __future__ import annotations

import json
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from cepis import (
    CEPNotFound,
    Client,
    InvalidCEP,
    InvalidResponse,
    NetworkError,
    RateLimited,
    ServerError,
    normalize_cep,
)

SE_ADDRESS = {
    "cep": "01001000",
    "logradouro": "Praça da Sé",
    "complemento": "lado ímpar",
    "unidade": "",
    "bairro": "Sé",
    "localidade": "São Paulo",
    "uf": "SP",
    "estado": "São Paulo",
    "regiao": "Sudeste",
    "ibge": "3550308",
    "gia": "1004",
    "ddd": "11",
    "siafi": "7107",
}

CSV_HEADER = (
    "cep,logradouro,complemento,unidade,bairro,localidade,uf,estado,regiao,ibge,gia,ddd,siafi"
)


class Router:
    """Per-test programmable routes, plus a request counter."""

    def __init__(self) -> None:
        self.routes: dict[str, object] = {}
        self.calls: list[str] = []
        self.lock = threading.Lock()

    def respond(self, path: str, response: object) -> None:
        """Register the response for a path.

        The response may be a `(status, body, headers)` tuple or a list of
        them — in which case each request consumes the next one, which is how
        retrying is tested. The last entry repeats once the list runs out.
        """
        self.routes[path] = response

    def next(self, path: str) -> tuple[int, str, dict[str, str]]:
        """Record the call and return the response registered for `path`."""
        with self.lock:
            self.calls.append(path)
            response = self.routes.get(path)
            if response is None:
                return 404, json.dumps({"erro": "cep_nao_encontrado"}), {}
            if isinstance(response, list):
                return response.pop(0) if len(response) > 1 else response[0]
            return response  # type: ignore[return-value]

    def count(self, prefix: str) -> int:
        """How many requests so far began with `prefix`."""
        with self.lock:
            return sum(1 for c in self.calls if c.startswith(prefix))


class FakeServer:
    """A throwaway HTTP server on a random loopback port.

    `poll_interval` is deliberately low. The default is 0.5s and `shutdown()`
    waits a full cycle of it, which charged half a second of teardown to every
    test and put the suite at 12s with nothing actually slow. A slow suite is a
    suite nobody runs.
    """

    def __init__(self) -> None:
        self.router = Router()
        router = self.router

        class Handler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def do_GET(self) -> None:
                """Serve the router's response. Name required by the base class."""
                status, body, headers = router.next(self.path)
                data = body.encode("utf-8")
                self.send_response(status)
                for key, value in headers.items():
                    self.send_header(key, value)
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

            def log_message(self, *args: object) -> None:
                """Silenced, to keep the test output clean."""

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.thread = threading.Thread(
            target=self.server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True
        )
        self.thread.start()

    @property
    def url(self) -> str:
        host, port = self.server.server_address[:2]
        return f"http://{host}:{port}"

    def stop(self) -> None:
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)


class BaseTest(unittest.TestCase):
    """Fixture with a fake server and a client whose backoff is near zero.

    The backoff is tiny because what is under test is the retry logic, not
    `time.sleep`'s ability to sleep.
    """

    def setUp(self) -> None:
        self.server = FakeServer()
        self.addCleanup(self.server.stop)
        self.client = Client(
            base_url=self.server.url,
            timeout=5.0,
            backoff_base=0.001,
            backoff_max=0.01,
        )

    @property
    def routes(self) -> Router:
        return self.server.router


class TestNormalisation(unittest.TestCase):
    def test_accepts_masked_and_unmasked(self) -> None:
        for value in ("01001000", "01001-000", " 01001-000 ", "01001.000"):
            self.assertEqual(normalize_cep(value), "01001000")

    def test_rejects_bad_format(self) -> None:
        for value in ("", "123", "0100100a", "010010000", "abcdefgh"):
            with self.assertRaises(InvalidCEP):
                normalize_cep(value)

    def test_rejects_wrong_type(self) -> None:
        with self.assertRaises(InvalidCEP):
            normalize_cep(1001000)  # type: ignore[arg-type]

    def test_is_a_valueerror(self) -> None:
        """Code already handling ValueError keeps working without this library."""
        with self.assertRaises(ValueError):
            normalize_cep("123")


class TestLookup(BaseTest):
    def test_successful_lookup(self) -> None:
        self.routes.respond("/01001000", (200, json.dumps(SE_ADDRESS), {}))
        address = self.client.lookup("01001-000")
        self.assertEqual(address.localidade, "São Paulo")
        self.assertEqual(address.uf, "SP")
        self.assertEqual(address.formatted_cep, "01001-000")
        self.assertIsNone(address.geo)

    def test_reads_the_coordinate_when_present(self) -> None:
        body = dict(SE_ADDRESS)
        body["geo"] = {"lat": -23.5, "lon": -46.6, "raio_m": 120, "precisao": "logradouro"}
        self.routes.respond("/01310100", (200, json.dumps(body), {}))
        address = self.client.lookup("01310100")
        assert address.geo is not None
        self.assertAlmostEqual(address.geo.lat, -23.5)
        self.assertTrue(address.geo.is_precise)

    def test_municipality_coordinate_is_not_precise(self) -> None:
        body = dict(SE_ADDRESS)
        body["geo"] = {"lat": -23.5, "lon": -46.6, "raio_m": 22000, "precisao": "localidade"}
        self.routes.respond("/01001000", (200, json.dumps(body), {}))
        address = self.client.lookup("01001000")
        assert address.geo is not None
        self.assertFalse(address.geo.is_precise)

    def test_malformed_geo_does_not_lose_the_address(self) -> None:
        body = dict(SE_ADDRESS)
        body["geo"] = {"lat": "not a number"}
        self.routes.respond("/01001000", (200, json.dumps(body), {}))
        address = self.client.lookup("01001000")
        self.assertIsNone(address.geo)
        self.assertEqual(address.localidade, "São Paulo")

    def test_new_api_field_does_not_break(self) -> None:
        body = dict(SE_ADDRESS)
        body["campo_do_futuro"] = "value"
        self.routes.respond("/01001000", (200, json.dumps(body), {}))
        address = self.client.lookup("01001000")
        self.assertEqual(address.raw["campo_do_futuro"], "value")

    def test_not_found(self) -> None:
        """KeyError quotes its str() by inheritance; the message must be clean."""
        self.routes.respond(
            "/99999999",
            (404, json.dumps({"erro": "cep_nao_encontrado", "mensagem": "gone"}), {}),
        )
        with self.assertRaises(CEPNotFound) as ctx:
            self.client.lookup("99999999")
        self.assertNotIn('"', str(ctx.exception))

    def test_not_found_is_not_retried(self) -> None:
        """404 is definitive; retrying it only burns someone's quota."""
        self.routes.respond("/99999999", (404, json.dumps({"erro": "x"}), {}))
        with self.assertRaises(CEPNotFound):
            self.client.lookup("99999999")
        self.assertEqual(self.routes.count("/99999999"), 1)

    def test_invalid_cep_never_leaves_the_machine(self) -> None:
        with self.assertRaises(InvalidCEP):
            self.client.lookup("123")
        self.assertEqual(len(self.routes.calls), 0)

    def test_api_400_becomes_invalid_cep(self) -> None:
        self.routes.respond(
            "/00000000",
            (400, json.dumps({"erro": "cep_invalido", "mensagem": "rejected by the API"}), {}),
        )
        with self.assertRaises(InvalidCEP) as ctx:
            self.client.lookup("00000000")
        self.assertIn("rejected by the API", str(ctx.exception))

    def test_response_that_is_not_json(self) -> None:
        self.routes.respond("/01001000", (200, "<html>captive portal</html>", {}))
        with self.assertRaises(InvalidResponse):
            self.client.lookup("01001000")


class TestRetry(BaseTest):
    def test_retries_on_500_then_succeeds(self) -> None:
        self.routes.respond(
            "/01001000",
            [
                (500, json.dumps({"erro": "interno"}), {}),
                (200, json.dumps(SE_ADDRESS), {}),
            ],
        )
        address = self.client.lookup("01001000")
        self.assertEqual(address.localidade, "São Paulo")
        self.assertEqual(self.routes.count("/01001000"), 2)

    def test_exhausts_attempts_then_raises(self) -> None:
        """Three attempts is the documented default."""
        self.routes.respond("/01001000", (503, json.dumps({"erro": "interno"}), {}))
        with self.assertRaises(ServerError) as ctx:
            self.client.lookup("01001000")
        self.assertEqual(ctx.exception.status, 503)
        self.assertEqual(self.routes.count("/01001000"), 3)

    def test_max_attempts_one_disables_retry(self) -> None:
        client = Client(base_url=self.server.url, max_attempts=1, timeout=5.0)
        self.routes.respond("/01001000", (500, "{}", {}))
        with self.assertRaises(ServerError):
            client.lookup("01001000")
        self.assertEqual(self.routes.count("/01001000"), 1)

    def test_429_becomes_rate_limited_with_retry_after(self) -> None:
        self.routes.respond("/01001000", (429, "{}", {"Retry-After": "7"}))
        with self.assertRaises(RateLimited) as ctx:
            self.client.lookup("01001000")
        self.assertEqual(ctx.exception.retry_after, 7.0)

    def test_retry_after_honours_the_cap(self) -> None:
        """An absurd Retry-After from a bad proxy must not park the thread."""
        self.routes.respond("/01001000", (429, "{}", {"Retry-After": "3600"}))
        started = time.monotonic()
        with self.assertRaises(RateLimited):
            self.client.lookup("01001000")
        self.assertLess(time.monotonic() - started, 5.0)

    def test_network_error_when_nothing_listens(self) -> None:
        """Port 1 is closed, so the connection is refused outright."""
        client = Client(
            base_url="http://127.0.0.1:1",
            max_attempts=2,
            backoff_base=0.001,
            timeout=1.0,
        )
        with self.assertRaises(NetworkError):
            client.lookup("01001000")

    def test_total_timeout_bounds_the_whole_call(self) -> None:
        """The deadline must cut the run short of the 50 configured attempts."""
        client = Client(
            base_url=self.server.url,
            max_attempts=50,
            backoff_base=0.05,
            backoff_max=0.05,
            total_timeout=0.3,
        )
        self.routes.respond("/01001000", (500, "{}", {}))
        started = time.monotonic()
        with self.assertRaises(ServerError):
            client.lookup("01001000")
        self.assertLess(time.monotonic() - started, 3.0)
        self.assertLess(self.routes.count("/01001000"), 50)


class TestBatch(BaseTest):
    def _csv(self, rows: list[str]) -> str:
        return CSV_HEADER + "\n" + "\n".join(rows) + "\n"

    def test_preserves_order_and_marks_missing(self) -> None:
        """A missing CEP becomes None at its position; it does not vanish."""
        body = self._csv(
            [
                "01001000,Praça da Sé,lado ímpar,,Sé,São Paulo,SP,São Paulo,"
                "Sudeste,3550308,1004,11,7107",
                "99999999,,,,,,,,,,,,",
                "20040020,Avenida Rio Branco,,,Centro,Rio de Janeiro,RJ,"
                "Rio de Janeiro,Sudeste,3304557,,21,6001",
            ]
        )
        self.routes.respond("/csv?ceps=01001000%2C99999999%2C20040020", (200, body, {}))
        result = self.client.lookup_many(["01001000", "99999999", "20040020"])

        self.assertEqual(len(result), 3)
        assert result[0] is not None
        self.assertEqual(result[0].localidade, "São Paulo")
        self.assertIsNone(result[1])
        assert result[2] is not None
        self.assertEqual(result[2].uf, "RJ")

    def test_field_with_comma_is_parsed(self) -> None:
        body = self._csv(
            ['01001000,"Rua Um, Dois",,,Centro,São Paulo,SP,São Paulo,Sudeste,1,,11,1']
        )
        self.routes.respond("/csv?ceps=01001000", (200, body, {}))
        result = self.client.lookup_many(["01001000"])
        assert result[0] is not None
        self.assertEqual(result[0].logradouro, "Rua Um, Dois")

    def test_empty_list_makes_no_request(self) -> None:
        self.assertEqual(self.client.lookup_many([]), [])
        self.assertEqual(len(self.routes.calls), 0)

    def test_chunks_above_the_api_limit(self) -> None:
        """501 CEPs must become 2 requests, not 1 the API would reject."""
        ceps = [f"{i:08d}" for i in range(501)]
        rows_a = [f"{c},,,,,X,SP,São Paulo,Sudeste,1,,11,1" for c in ceps[:500]]
        rows_b = [f"{c},,,,,X,SP,São Paulo,Sudeste,1,,11,1" for c in ceps[500:]]

        calls: list[str] = []
        original = self.routes.next

        def spy(path: str):
            calls.append(path)
            return 200, self._csv(rows_a if len(calls) == 1 else rows_b), {}

        self.routes.next = spy  # type: ignore[method-assign]
        try:
            result = self.client.lookup_many(ceps)
        finally:
            self.routes.next = original  # type: ignore[method-assign]

        self.assertEqual(len(calls), 2)
        self.assertEqual(len(result), 501)

    def test_truncated_response_does_not_shift(self) -> None:
        """Padding beats shifting: a shift pairs the wrong address, silently."""
        body = self._csv(["01001000,,,,,São Paulo,SP,São Paulo,Sudeste,1,,11,1"])
        self.routes.respond("/csv?ceps=01001000%2C20040020", (200, body, {}))
        result = self.client.lookup_many(["01001000", "20040020"])
        self.assertEqual(len(result), 2)
        self.assertIsNone(result[1])


class TestHealth(BaseTest):
    def test_reads_health(self) -> None:
        body = json.dumps(
            {
                "status": "ok",
                "dne": "26071",
                "ceps": 1596168,
                "geo": 894489,
                "base_gerada_em": "2026-07-24T03:48:28Z",
            }
        )
        self.routes.respond("/health", (200, body, {}))
        health = self.client.health()
        self.assertTrue(health.ok)
        self.assertEqual(health.dne, "26071")
        self.assertEqual(health.ceps, 1596168)

    def test_degraded_does_not_raise(self) -> None:
        """Checking health is when you want a diagnosis, not an exception."""
        body = json.dumps({"status": "degradado", "dne": "", "ceps": 0, "geo": 0})
        self.routes.respond("/health", (503, body, {}))
        health = self.client.health()
        self.assertFalse(health.ok)


class TestConfiguration(unittest.TestCase):
    def test_trailing_slash_is_not_duplicated(self) -> None:
        self.assertEqual(Client(base_url="http://x/").base_url, "http://x")

    def test_invalid_parameters(self) -> None:
        with self.assertRaises(ValueError):
            Client(max_attempts=0)
        with self.assertRaises(ValueError):
            Client(timeout=0)

    def test_user_agent_identifies_the_library(self) -> None:
        self.assertIn("cepis-python/", Client().user_agent)


if __name__ == "__main__":
    unittest.main()