Initial importv0.1.0

1 files changed, 222 insertions, 0 deletions

diff --git a/tests/server.py b/tests/server.py
new file mode 100755
index 0000000..c77d7f6
--- /dev/null
+++ b/tests/server.py
@@ -0,0 +1,222 @@
+#!/usr/bin/env python3
+"""Local stand-in for the cep.is API, used by the test driver.
+
+Behaviour is selected by the first path segment, so a test points the client
+at a different ``base_url`` to get a different server: ``/fail2`` fails twice
+then answers, ``/always503`` never answers, and so on. Every prefix counts the
+requests it saw; ``/_stats/<prefix>/health`` reports that count in the ``ceps``
+field so the C driver can assert on it without speaking HTTP itself.
+"""
+
+import json
+import sys
+import threading
+from datetime import datetime, timedelta, timezone
+from email.utils import format_datetime
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs, urlparse
+
+COLUMNS = [
+    "cep", "logradouro", "complemento", "unidade", "bairro", "localidade",
+    "uf", "estado", "regiao", "ibge", "gia", "ddd", "siafi",
+]
+
+DB = {
+    "01001000": {
+        "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",
+        "geo": {"lat": -23.567202, "lon": -46.649659, "raio_m": 1084,
+                "precisao": "logradouro"},
+    },
+    "20040020": {
+        "cep": "20040020", "logradouro": "Avenida Rio Branco",
+        "complemento": "", "unidade": "", "bairro": "Centro",
+        "localidade": "Rio de Janeiro", "uf": "RJ",
+        "estado": "Rio de Janeiro", "regiao": "Sudeste", "ibge": "3304557",
+        "gia": "", "ddd": "21", "siafi": "6001",
+    },
+    "69900000": {
+        "cep": "69900000", "logradouro": "", "complemento": "", "unidade": "",
+        "bairro": "", "localidade": "Rio Branco", "uf": "AC",
+        "estado": "Acre", "regiao": "Norte", "ibge": "1200401", "gia": "",
+        "ddd": "68", "siafi": "0605",
+        "geo": {"lat": -9.97499, "lon": -67.8243, "raio_m": 18420,
+                "precisao": "localidade"},
+    },
+    "12345678": {
+        "cep": "12345678", "logradouro": 'Rua A, B "quoted"',
+        "complemento": "até 100", "unidade": "", "bairro": "Vila, Nova",
+        "localidade": "Campinas", "uf": "SP", "estado": "São Paulo",
+        "regiao": "Sudeste", "ibge": "3509502", "gia": "2446", "ddd": "19",
+        "siafi": "6291",
+    },
+}
+
+for _n in range(600):
+    _c = "%08d" % (70000000 + _n)
+    DB[_c] = {
+        "cep": _c, "logradouro": "Rua %d" % _n, "complemento": "",
+        "unidade": "", "bairro": "Centro", "localidade": "Cidade %d" % _n,
+        "uf": "SP", "estado": "São Paulo", "regiao": "Sudeste",
+        "ibge": "3550308", "gia": "", "ddd": "11", "siafi": "7107",
+    }
+
+HITS = {}
+LOCK = threading.Lock()
+
+
+def bump(prefix):
+    with LOCK:
+        HITS[prefix] = HITS.get(prefix, 0) + 1
+        return HITS[prefix]
+
+
+def csv_escape(value):
+    if any(c in value for c in ',"\r\n'):
+        return '"' + value.replace('"', '""') + '"'
+    return value
+
+
+def csv_row(cep):
+    entry = DB.get(cep)
+    if entry is None:
+        return ",".join([cep] + [""] * (len(COLUMNS) - 1))
+    return ",".join(csv_escape(entry.get(c, "")) for c in COLUMNS)
+
+
+class Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def log_message(self, fmt, *args):
+        pass
+
+    def send(self, code, body, ctype="application/json; charset=utf-8",
+             headers=None):
+        raw = body.encode("utf-8") if isinstance(body, str) else body
+        self.send_response(code)
+        self.send_header("Content-Type", ctype)
+        self.send_header("Content-Length", str(len(raw)))
+        for key, value in (headers or {}).items():
+            self.send_header(key, value)
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def do_GET(self):
+        parsed = urlparse(self.path)
+        parts = [p for p in parsed.path.split("/") if p]
+
+        if parts and parts[0] == "_stats":
+            name = parts[1] if len(parts) > 1 else ""
+            with LOCK:
+                count = HITS.get(name, 0)
+            self.send(200, json.dumps({
+                "status": "ok", "dne": "test", "ceps": count, "geo": 0,
+                "base_gerada_em": "2026-07-24T03:48:28Z",
+            }))
+            return
+
+        prefix = ""
+        if parts and not parts[0].isdigit() and parts[0] not in ("csv", "health"):
+            prefix = parts.pop(0)
+        count = bump(prefix)
+
+        if prefix == "always503":
+            self.send(503, '{"erro":"interno","mensagem":"falha interna"}')
+            return
+        if prefix == "always429":
+            self.send(429, '{"erro":"limite","mensagem":"devagar"}',
+                      headers={"Retry-After": "1"})
+            return
+        if prefix == "date429":
+            when = datetime.now(timezone.utc) + timedelta(seconds=2)
+            self.send(429, '{"erro":"limite","mensagem":"devagar"}',
+                      headers={"Retry-After": format_datetime(when,
+                                                              usegmt=True)})
+            return
+        if prefix == "hugeretry":
+            self.send(429, '{"erro":"limite","mensagem":"devagar"}',
+                      headers={"Retry-After": "3600"})
+            return
+        if prefix == "fail2" and count <= 2:
+            self.send(503, '{"erro":"interno","mensagem":"falha interna"}')
+            return
+        if prefix == "flaky" and count == 1:
+            self.send(500, '{"erro":"interno","mensagem":"falha interna"}')
+            return
+        if prefix == "always400":
+            self.send(400, '{"erro":"cep_invalido",'
+                           '"mensagem":"CEP deve ter 8 dígitos"}')
+            return
+        if prefix == "badjson":
+            self.send(200, "<html>captive portal</html>", ctype="text/html")
+            return
+        if prefix == "badgeo":
+            entry = dict(DB["01001000"])
+            entry["geo"] = {"lat": "nao-e-numero", "precisao": "logradouro"}
+            entry["campo_novo"] = {"nested": [1, 2, 3]}
+            self.send(200, json.dumps(entry, ensure_ascii=False))
+            return
+        if prefix == "surrogate":
+            entry = dict(DB["01001000"])
+            entry["logradouro"] = "Rua \U0001F600 Feliz"
+            self.send(200, json.dumps(entry, ensure_ascii=True))
+            return
+        if prefix == "escaped":
+            self.send(200, json.dumps(DB["01001000"], ensure_ascii=True))
+            return
+
+        if parts and parts[0] == "health":
+            body = {"status": "ok", "dne": "26071", "ceps": 1596168,
+                    "geo": 894489, "base_gerada_em": "2026-07-24T03:48:28Z"}
+            if prefix == "degraded":
+                body["status"] = "degraded"
+                self.send(503, json.dumps(body))
+                return
+            self.send(200, json.dumps(body))
+            return
+
+        if parts and parts[0] == "csv":
+            query = parse_qs(parsed.query)
+            ceps = [c for c in query.get("ceps", [""])[0].split(",") if c]
+            if len(ceps) > 500:
+                self.send(400, '{"erro":"limite","mensagem":"máximo 500"}')
+                return
+            if prefix == "short":
+                ceps = ceps[:1]
+            rows = [",".join(COLUMNS)] + [csv_row(c) for c in ceps]
+            body = "\r\n".join(rows) + "\r\n"
+            if prefix == "bom":
+                body = "\ufeff" + body
+            self.send(200, body, ctype="text/csv; charset=utf-8")
+            return
+
+        if not parts:
+            self.send(404, '{"erro":"cep_nao_encontrado",'
+                           '"mensagem":"CEP não existe na base do DNE"}')
+            return
+
+        cep = parts[0]
+        if not (len(cep) == 8 and cep.isdigit()):
+            self.send(400, '{"erro":"cep_invalido",'
+                           '"mensagem":"CEP deve ter 8 dígitos"}')
+            return
+        entry = DB.get(cep)
+        if entry is None:
+            self.send(404, '{"erro":"cep_nao_encontrado",'
+                           '"mensagem":"CEP não existe na base do DNE"}')
+            return
+        self.send(200, json.dumps(entry, ensure_ascii=False))
+
+
+def main():
+    port = int(sys.argv[1]) if len(sys.argv) > 1 else 0
+    server = ThreadingHTTPServer(("127.0.0.1", port), Handler)
+    print(server.server_address[1], flush=True)
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()