path: root/tests/server.py

#!/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"

# The municipality a missing CEP would have belonged to, attached to the
# 404 for /25845000 below. The numbers are Areal/RJ's real ones: cep_ini is
# 25845001, one past the 25845000 the DNE publishes as the start of the
# range and that the route refuses, and the 118 CEPs are scattered over the
# ~4900 numbers up to cep_fim rather than filling them.
AREAL = {
    "localidade": "Areal",
    "uf": "RJ",
    "estado": "Rio de Janeiro",
    "regiao": "Sudeste",
    "ibge": "3300225",
    "cep_ini": "25845001",
    "cep_fim": "25849899",
    "ceps": 118,
}

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 == "/25845000":
        return 404, json_body(
            {
                "erro": "cep_nao_encontrado",
                "mensagem": "CEP não existe na base do DNE",
                "faixa": AREAL,
            }
        ), {}
    if path == "/25849900":
        return 404, json_body(
            {
                "erro": "cep_nao_encontrado",
                "mensagem": "CEP não existe na base do DNE",
            }
        ), {}
    if path == "/25845002":
        broken = dict(AREAL)
        broken["cep_fim"] = "nope"
        broken["ceps"] = "many"
        return 404, json_body(
            {"erro": "cep_nao_encontrado", "faixa": broken}
        ), {}
    if path == "/25845003":
        return 404, json_body(
            {"erro": "cep_nao_encontrado", "faixa": "Areal"}
        ), {}
    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())