158 lines
6.3 KiB
Python
158 lines
6.3 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from app.domain.contas.line_reference import extract_requested_line_reference
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
POLICY_DIR = ROOT / "contas_mcp" / "servers" / "contas_mcp_server"
|
|
|
|
|
|
def _load(name: str):
|
|
path = POLICY_DIR / name
|
|
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_extrai_final_falado_sem_transformar_em_msisdn():
|
|
ref = extract_requested_line_reference("quero cancelar o streaming do número da minha esposa, final quatro três dois um")
|
|
assert ref == {"kind": "suffix", "value": "4321", "raw": "quatro tres dois um"}
|
|
|
|
|
|
def test_alt1_bloqueia_outra_linha_e_mantem_linha_autenticada():
|
|
alt1 = _load("line_policy_alt1.py")
|
|
decision = alt1.apply_line_policy("cancelar_vas_avulso", {
|
|
"msisdn": "11999999999",
|
|
"requested_line_reference": {"kind": "suffix", "value": "4321"},
|
|
})
|
|
assert decision.allowed is False
|
|
assert decision.effective_msisdn == "11999999999"
|
|
assert decision.reason == "other_line_not_allowed"
|
|
|
|
|
|
def test_alt1_permite_referencia_a_mesma_linha():
|
|
alt1 = _load("line_policy_alt1.py")
|
|
decision = alt1.apply_line_policy("consultar_faturas", {
|
|
"msisdn": "11999994321",
|
|
"requested_line_reference": {"kind": "suffix", "value": "4321"},
|
|
})
|
|
assert decision.allowed is True
|
|
assert decision.effective_msisdn == "11999994321"
|
|
|
|
|
|
def test_alt2_resolve_dependente_unico_por_sufixo_retornado_pelo_servico_autorizador():
|
|
alt2 = _load("line_policy_alt2.py")
|
|
decision = alt2.apply_line_policy("cancelar_vas_avulso", {
|
|
"msisdn": "11999999999",
|
|
"requested_line_reference": {"kind": "suffix", "value": "4321"},
|
|
"authorized_lines_evidence": {
|
|
"success": True,
|
|
"authorized_lines": [
|
|
{"msisdn": "11999999999", "relationship": "titular", "status": "ACTIVE", "authorized": True},
|
|
{"msisdn": "11988884321", "relationship": "dependente", "status": "ACTIVE", "authorized": True},
|
|
],
|
|
"authorized_msisdns": ["11999999999", "11988884321"],
|
|
},
|
|
})
|
|
assert decision.allowed is True
|
|
assert decision.effective_msisdn == "11988884321"
|
|
|
|
|
|
def test_alt2_nao_trata_invoice_detail_como_fonte_de_autorizacao():
|
|
alt2 = _load("line_policy_alt2.py")
|
|
decision = alt2.apply_line_policy("cancelar_vas_avulso", {
|
|
"msisdn": "11999999999",
|
|
"requested_line_reference": {"kind": "suffix", "value": "4321"},
|
|
"invoice_detail": {
|
|
"parsed_content": {
|
|
"11999999999": {},
|
|
"11988884321": {"SVA Detalhe Total": []},
|
|
}
|
|
},
|
|
})
|
|
assert decision.allowed is False
|
|
assert decision.reason == "other_line_not_authorized_or_not_resolved"
|
|
|
|
|
|
def test_alt2_nao_permite_linha_arbitraria_fora_da_resposta_do_servico():
|
|
alt2 = _load("line_policy_alt2.py")
|
|
decision = alt2.apply_line_policy("cancelar_vas_avulso", {
|
|
"msisdn": "11999999999",
|
|
"requested_line_reference": {"kind": "full", "value": "11977774321"},
|
|
"authorized_lines_evidence": {
|
|
"success": True,
|
|
"authorized_msisdns": ["11999999999", "11988884321"],
|
|
},
|
|
})
|
|
assert decision.allowed is False
|
|
assert decision.reason == "other_line_not_authorized_or_not_resolved"
|
|
|
|
|
|
def test_alt2_falha_fechada_se_servico_autorizador_falhar():
|
|
alt2 = _load("line_policy_alt2.py")
|
|
decision = alt2.apply_line_policy("consultar_vas", {
|
|
"msisdn": "11999999999",
|
|
"requested_line_reference": {"kind": "suffix", "value": "4321"},
|
|
"authorized_lines_evidence": {"success": False, "status": "TIMEOUT"},
|
|
})
|
|
assert decision.allowed is False
|
|
assert decision.authorized_lines == ("11999999999",)
|
|
|
|
|
|
def test_alt2_falha_fechada_se_sufixo_for_ambiguo():
|
|
alt2 = _load("line_policy_alt2.py")
|
|
decision = alt2.apply_line_policy("consultar_vas", {
|
|
"msisdn": "11999999999",
|
|
"requested_line_reference": {"kind": "suffix", "value": "4321"},
|
|
"authorized_lines_evidence": {
|
|
"success": True,
|
|
"authorized_msisdns": ["11911114321", "11922224321"],
|
|
},
|
|
})
|
|
assert decision.allowed is False
|
|
assert decision.reason == "other_line_ambiguous"
|
|
|
|
|
|
def test_mock_consultar_linhas_autorizadas_expoe_dependente_4321():
|
|
from contas_mcp.servers.contas_mcp_server.authorized_lines_service import AuthorizedLinesMockService
|
|
|
|
service = AuthorizedLinesMockService(ROOT / "app" / "domain" / "contas" / "fixtures" / "authorized_lines.json")
|
|
result = service.consultar_linhas_autorizadas(
|
|
authenticated_msisdn="11999999999",
|
|
customer_key="11999999999",
|
|
contract_key="3000131180",
|
|
)
|
|
assert result["success"] is True
|
|
assert "11988884321" in result["authorized_msisdns"]
|
|
dependent = next(row for row in result["authorized_lines"] if row["msisdn"] == "11988884321")
|
|
assert dependent["relationship"] == "dependente"
|
|
|
|
|
|
def test_mcp_registra_consultar_linhas_autorizadas_e_alt2_consume_evidencia_explicita():
|
|
source = (POLICY_DIR / "main.py").read_text(encoding="utf-8")
|
|
assert '"consultar_linhas_autorizadas"' in source
|
|
assert 'args["authorized_lines_evidence"] = authorized_lines_service.consultar_linhas_autorizadas' in source
|
|
alt2_source = (POLICY_DIR / "line_policy_alt2.py").read_text(encoding="utf-8")
|
|
assert 'args.get("authorized_lines_evidence")' in alt2_source
|
|
assert 'args.get("invoice_detail")' not in alt2_source
|
|
|
|
|
|
def test_politica_ativa_entregue_eh_alt1():
|
|
active = (POLICY_DIR / "line_policy.py").read_text(encoding="utf-8")
|
|
alt1 = (POLICY_DIR / "line_policy_alt1.py").read_text(encoding="utf-8")
|
|
assert active == alt1
|
|
|
|
|
|
def test_line_policy_block_result_declares_generic_terminal_contract():
|
|
source = (POLICY_DIR / "main.py").read_text(encoding="utf-8")
|
|
block = source[source.index('"status": "LINE_POLICY_BLOCKED"'):source.index('"metadata": {', source.index('"status": "LINE_POLICY_BLOCKED"'))]
|
|
assert '"terminal": True' in block
|
|
assert '"terminal_action": "block"' in block
|
|
assert '"user_message": decision.user_message' in block
|