Initial importv0.1.0

1 files changed, 434 insertions, 0 deletions

diff --git a/tests/test_cepis.py b/tests/test_cepis.py
new file mode 100644
index 0000000..dc27179
--- /dev/null
+++ b/tests/test_cepis.py
@@ -0,0 +1,434 @@
+"""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()