#!/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, "captive portal", {} 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 [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())