Initial importv0.1.0

1 files changed, 254 insertions, 0 deletions

diff --git a/tests/server.py b/tests/server.py
new file mode 100644
index 0000000..46185a0
--- /dev/null
+++ b/tests/server.py
@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+"""Fake cep.is for the COBOL test suite.
+
+Binds a loopback port, points CEPIS_TEST_URL at it, runs the command given
+on the argv and exits with that command's status. Standing an HTTP server up
+from COBOL is not practical; every route below is a fixture the driver in
+`cepis-tests.cob` asserts against, and no test touches the network.
+
+    python3 tests/server.py ./build/cepis-tests
+
+The counters behind `/count` are what let the driver prove that a 404 is not
+retried, that a 500 is, and that 501 CEPs become two batch requests.
+"""
+
+import email.utils
+import json
+import os
+import subprocess
+import sys
+import threading
+import time
+import urllib.parse
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+SE = {
+    "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_COLUMNS = [
+    "cep",
+    "logradouro",
+    "complemento",
+    "unidade",
+    "bairro",
+    "localidade",
+    "uf",
+    "estado",
+    "regiao",
+    "ibge",
+    "gia",
+    "ddd",
+    "siafi",
+]
+
+CSV_ROWS = {
+    "01001000": [
+        "Praça da Sé", "lado ímpar", "", "Sé", "São Paulo", "SP",
+        "São Paulo", "Sudeste", "3550308", "1004", "11", "7107",
+    ],
+    "20040020": [
+        "Avenida Rio Branco", "", "", "Centro", "Rio de Janeiro", "RJ",
+        "Rio de Janeiro", "Sudeste", "3304557", "", "21", "6001",
+    ],
+    "01001111": [
+        "Rua Um, Dois", 'aspas "aqui"', "", "Centro", "São Paulo", "SP",
+        "São Paulo", "Sudeste", "3550308", "", "11", "7107",
+    ],
+}
+
+SYNTHETIC = [
+    "Rua Sintetica", "", "", "Centro", "Cidade Sintetica", "SP",
+    "São Paulo", "Sudeste", "3550308", "", "11", "7107",
+]
+
+DROPPED_CEP = "00000099"
+
+LONG_STREET = "Avenida " + ("Muito Comprida " * 20)
+
+COUNTS = {}
+COUNT_LOCK = threading.Lock()
+
+
+def with_geo(precisao, raio):
+    body = dict(SE)
+    body["geo"] = {
+        "lat": -23.567202,
+        "lon": -46.649659,
+        "raio_m": raio,
+        "precisao": precisao,
+    }
+    return body
+
+
+def csv_escape(value):
+    if any(c in value for c in (",", '"', "\n", "\r")):
+        return '"' + value.replace('"', '""') + '"'
+    return value
+
+
+def csv_body(ceps):
+    lines = [",".join(CSV_COLUMNS)]
+    for cep in ceps:
+        if cep == DROPPED_CEP:
+            continue
+        if cep in CSV_ROWS:
+            values = [cep] + CSV_ROWS[cep]
+        elif cep.startswith("7") and len(cep) == 8:
+            values = [cep] + SYNTHETIC
+        else:
+            values = [cep] + [""] * 12
+        lines.append(",".join(csv_escape(v) for v in values))
+    return "\n".join(lines) + "\n"
+
+
+def json_body(payload):
+    return json.dumps(payload, ensure_ascii=False)
+
+
+def bump(path):
+    with COUNT_LOCK:
+        COUNTS[path] = COUNTS.get(path, 0) + 1
+        return COUNTS[path]
+
+
+def route(path, query, hits):
+    """Return (status, body, headers) for one request."""
+    if path == "/01001000":
+        return 200, json_body(SE), {}
+    if path == "/01310100":
+        return 200, json_body(with_geo("logradouro", 120)), {}
+    if path == "/01001001":
+        return 200, json_body(with_geo("localidade", 22000)), {}
+    if path == "/01001002":
+        body = dict(SE)
+        body["geo"] = {"lat": "not a number"}
+        return 200, json_body(body), {}
+    if path == "/01001003":
+        body = dict(SE)
+        body["campo_do_futuro"] = "value"
+        body["objeto_do_futuro"] = {"a": [1, 2, {"lat": 9}]}
+        return 200, json_body(body), {}
+    if path == "/01001004":
+        return 200, "<html>captive portal</html>", {}
+    if path == "/01001010":
+        body = dict(SE)
+        body["logradouro"] = LONG_STREET
+        return 200, json_body(body), {}
+    if path == "/01001005":
+        if hits == 1:
+            return 500, json_body({"erro": "interno"}), {}
+        return 200, json_body(SE), {}
+    if path in ("/01001006", "/01001011"):
+        return 503, json_body({"erro": "interno"}), {}
+    if path == "/01001012":
+        return 500, json_body({"erro": "interno"}), {}
+    if path == "/01001007":
+        return 429, json_body({"erro": "rate"}), {"Retry-After": "7"}
+    if path == "/01001008":
+        return 429, json_body({"erro": "rate"}), {"Retry-After": "3600"}
+    if path == "/01001009":
+        when = email.utils.formatdate(time.time() + 4, usegmt=True)
+        return 429, json_body({"erro": "rate"}), {"Retry-After": when}
+    if path == "/99999999":
+        return 404, json_body(
+            {"erro": "cep_nao_encontrado", "mensagem": "gone"}
+        ), {}
+    if path == "/00000000":
+        return 400, json_body(
+            {"erro": "cep_invalido", "mensagem": "rejected by the API"}
+        ), {}
+    if path == "/health":
+        return 200, json_body(
+            {
+                "status": "ok",
+                "dne": "26071",
+                "ceps": 1596168,
+                "geo": 894489,
+                "base_gerada_em": "2026-07-24T03:48:28Z",
+            }
+        ), {}
+    if path == "/degraded/health":
+        return 503, json_body(
+            {"status": "degradado", "dne": "", "ceps": 0, "geo": 0}
+        ), {}
+    if path == "/csv":
+        raw = urllib.parse.parse_qs(query).get("ceps", [""])[0]
+        ceps = [c for c in raw.split(",") if c]
+        if len(ceps) > 500:
+            return 400, json_body(
+                {"erro": "muitos_ceps", "mensagem": "no máximo 500"}
+            ), {}
+        return 200, csv_body(ceps), {}
+    return 404, json_body({"erro": "cep_nao_encontrado"}), {}
+
+
+class Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def do_GET(self):
+        parsed = urllib.parse.urlsplit(self.path)
+        path = parsed.path
+
+        if path == "/count":
+            wanted = urllib.parse.parse_qs(parsed.query).get("path", [""])[0]
+            with COUNT_LOCK:
+                total = sum(
+                    n for p, n in COUNTS.items() if p.startswith(wanted)
+                )
+            self.reply(200, str(total), {})
+            return
+
+        hits = bump(path)
+        status, body, headers = route(path, parsed.query, hits)
+        self.reply(status, body, headers)
+
+    def reply(self, status, body, headers):
+        data = body.encode("utf-8")
+        self.send_response(status)
+        for key, value in headers.items():
+            self.send_header(key, value)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def log_message(self, *args):
+        pass
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("usage: server.py <command> [args...]", file=sys.stderr)
+        return 2
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    host, port = server.server_address[:2]
+    thread = threading.Thread(
+        target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True
+    )
+    thread.start()
+
+    env = dict(os.environ)
+    env["CEPIS_TEST_URL"] = "http://{}:{}".format(host, port)
+    try:
+        return subprocess.call(sys.argv[1:], env=env)
+    finally:
+        server.shutdown()
+        server.server_close()
+
+
+if __name__ == "__main__":
+    sys.exit(main())