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"

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())