"""Testes do demonstrador. Só aqui, porque os dados são fictícios, juntamos as bases para conferir o resultado federado. Uso: python testes.py (depois de gerar_dados_ficticios.py) """ import json from pathlib import Path import numpy as np import coordenador as C import no as N RAIZ = Path(__file__).resolve().parent def teste_federado_igual_ao_agrupado(): fit = C.ajustar_federado(["A", "B"], "completo") termos = C.PROTOCOLO["modelos"]["completo"] Xs, ys = zip(*(N.matriz(N.ler_pacientes(RAIZ / "hospitais" / h), termos) for h in ("A", "B"))) X, y = np.vstack(Xs), np.concatenate(ys) b = np.zeros(X.shape[1]) for _ in range(50): p = 1 / (1 + np.exp(-X @ b)) b += np.linalg.solve(X.T @ (X * (p * (1 - p))[:, None]), X.T @ (y - p)) assert np.allclose(fit["beta"], b, atol=1e-8), (fit["beta"], b) print("ok: o modelo federado é idêntico ao modelo com as bases juntas (diferença máxima " f"{np.abs(fit['beta'] - b).max():.1e})") def teste_bloqueios(): assert "recusado" in C.pedir("A", "exportar_linhas") assert "recusado" in C.pedir("A", "descritivo", {"variaveis": ["cep_ficticio"]}) alt = json.loads(json.dumps(C.PROTOCOLO)) alt["controles_de_saida"]["celula_minima"] = 1 assert "recusado" in C.pedir("A", "descritivo", {"variaveis": ["sexo"]}, alt) print("ok: exportação de linhas, variável fora do protocolo e protocolo alterado são recusados") def teste_supressao(): s = N.suprimir({"sim": 3, "nao": 40, "(não registrado)": 12}, 5, True) assert s["sim"] == "<5" and s["(não registrado)"] == "suprimido" and s["nao"] == 40, s s = N.suprimir({"sim": 0, "nao": 40}, 5, True) assert s == {"sim": 0, "nao": 40}, s print("ok: supressão de células pequenas e supressão complementar") def teste_saida_sem_linhas_individuais(): assert "recusado" in N.verificar_saida({"x": list(range(60))}, 50) for h in ("A", "B", "C"): for l in (RAIZ / "hospitais" / h / "auditoria_local.jsonl").read_text().splitlines(): assert set(json.loads(l)) - {"modelo"} == {"quando", "operacao", "protocolo", "codigo", "decisao", "motivo"} res = json.loads((RAIZ / "saida" / "resultado.json").read_text()) res.pop("bloqueios") # as mensagens de recusa citam o nome da variável pedida, de propósito texto = json.dumps(res, ensure_ascii=False) import re assert not re.search(r"[ABC]-1\d{5}", texto) and "cep_" not in texto.lower(), "identificadores apareceram na saída" print("ok: nenhuma saída contém listas em nível individual nem identificadores") def teste_minimo_de_casos(): alt_pedido = {"protocolo": C.PROTOCOLO, "operacao": "logistica_passo", "parametros": {"modelo": "completo", "beta": [0] * 11}} pasta = RAIZ / "hospitais" / "A" orig = (pasta / "pacientes_ficticios.csv").read_text() try: linhas = orig.splitlines() (pasta / "pacientes_ficticios.csv").write_text("\n".join(linhas[:42]) + "\n") # cabeçalhos + 40 pacientes r = N.executar(alt_pedido, pasta) assert "menos de 50" in r.get("recusado", ""), r finally: (pasta / "pacientes_ficticios.csv").write_text(orig) print("ok: hospital com poucos casos completos recusa o modelo") def teste_hospital_de_teste_nao_treina(): # revisão, ponto 1 r = C.pedir("C", "logistica_passo", {"modelo": "completo", "beta": [0.0] * 11}) assert "recusado" in r and "teste externo" in r["recusado"], r r = C.pedir("C", "avaliar", {"modelo": "completo", "beta": [0.0] * 11}) assert "recusado" in r and "congelado" in r["recusado"], r mc = json.loads((RAIZ / "saida" / "modelos_congelados" / "completo.json").read_text()) r = C.pedir("C", "avaliar", {"modelo_congelado": mc}) assert "recusado" in r and "uma única vez" in r["recusado"], r mc2 = dict(mc, beta=[b + 0.01 for b in mc["beta"]]) r = C.pedir("C", "avaliar", {"modelo_congelado": mc2}) assert "recusado" in r and "não registrado" in r["recusado"], r print("ok: o hospital de teste recusa treino, modelo não congelado, modelo alterado e segunda avaliação") def teste_supressao_cruzada(): # revisão, ponto 2: '<5' faltantes + n observado + total não pode revelar o número q = C.pedir("B", "qualidade") d = C.pedir("B", "descritivo", {"variaveis": ["rnl", "albumina_g_dl", "perda_peso_pct"]}) for v in ("rnl", "albumina_g_dl", "perda_peso_pct"): falt, n_obs, total = q["faltantes"][v], d["variaveis"][v]["n"], q["n_total"] if isinstance(falt, str): # suprimido: o observado não pode ser exato assert isinstance(n_obs, str), (v, falt, n_obs) else: assert falt + n_obs == total, (v, falt, n_obs, total) assert q["faltantes"]["rnl"] == "<5" and d["variaveis"]["rnl"]["n"] == "156–159", (q["faltantes"]["rnl"], d["variaveis"]["rnl"]) print("ok: faltantes suprimidos não podem ser recuperados pelo número de observados") def teste_categorias_vazias_viram_ausencia(): # revisão, ponto 3 pasta = RAIZ / "hospitais" / "A" pac = N.ler_pacientes(pasta) termos = C.PROTOCOLO["modelos"]["completo"] X0, _ = N.matriz(pac, termos) alterados = [dict(r) for r in pac] alterados[0].update(sexo="", ct="", ecog="") alterados[1].update(ct="T9") # categoria inválida alterados[2].update(albumina_g_dl="abc") # número inválido alterados[3].update(rnl="-1") # fora da faixa plausível completos_antes = sum(1 for r in pac[:4] if not np.isnan(N.matriz([r], termos)[0]).any() and len(N.matriz([r], termos)[0])) X1, _ = N.matriz(alterados, termos) assert len(X0) - len(X1) == completos_antes, (len(X0), len(X1), completos_antes) for r in alterados[:4]: assert len(N.matriz([r], termos)[0]) == 0 assert not N.valido(alterados[0], "sexo") and not N.valido(alterados[1], "ct") print("ok: categorias vazias ou inválidas e números inválidos viram ausência, não categoria de referência") def teste_avaliacao_fixada_e_coeficientes_validos(): for mudanca in ({"mesmos_casos_que": "simples"}, {"semente": 8}, {"reamostragens": 50}): r = C.pedir("A", "avaliar", {"modelo": "simples", "beta": [0.0, 0.1, 0.2], **mudanca}) assert "parâmetros de avaliação" in r.get("recusado", ""), r for beta in ([0.0], [0.0, float("inf"), 0.0]): r = C.pedir("A", "logistica_passo", {"modelo": "simples", "beta": beta}) assert "coeficientes inválidos" in r.get("recusado", ""), r for x in ("nan", "inf", "-inf", "abc", ""): assert np.isnan(N.num(x)), x X, y = N.matriz([], C.PROTOCOLO["modelos"]["completo"]) assert X.shape == (0, 11) and len(y) == 0 print("ok: avaliação fixada, coeficientes finitos e entrada vazia validados") def teste_versao_local_nao_pode_ser_trocada(): import subprocess import tempfile mc = json.loads((RAIZ / "saida/modelos_congelados/completo.json").read_text()) with tempfile.TemporaryDirectory() as temp: pasta = Path(temp) (pasta / "aprovacoes.json").write_text(json.dumps({"protocolos_aprovados": [mc["protocolo"]]})) arq = pasta / "modelo.json" arq.write_text(json.dumps(mc)) cmd = [C.sys.executable, str(RAIZ / "aprovacao_local.py"), str(pasta), str(arq)] assert subprocess.run(cmd, capture_output=True).returncode == 0 mc["beta"][0] += .01 arq.write_text(json.dumps(mc)) r = subprocess.run(cmd, capture_output=True, text=True) assert r.returncode != 0 and "outra versão" in r.stderr, r print("ok: outra versão do modelo não pode ser registrada após o congelamento") if __name__ == "__main__": teste_supressao() teste_bloqueios() teste_minimo_de_casos() teste_federado_igual_ao_agrupado() teste_saida_sem_linhas_individuais() teste_hospital_de_teste_nao_treina() teste_supressao_cruzada() teste_categorias_vazias_viram_ausencia() teste_avaliacao_fixada_e_coeficientes_validos() teste_versao_local_nao_pode_ser_trocada()