280 lines
10 KiB
Python
280 lines
10 KiB
Python
from pathlib import Path
|
|
import yaml
|
|
import pytest
|
|
|
|
from contas_mcp.servers.contas_mcp_server import main
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def test_all_concrete_subject_domains_have_domain_prevalidation():
|
|
policies = yaml.safe_load((ROOT / 'config' / 'tool_policies.yaml').read_text(encoding='utf-8'))['tool_policies']
|
|
assert policies['contestar_cobranca']['pre_validation']['tool'] == 'validar_contestacao'
|
|
for tool in ('cancelar_vas_avulso', 'tratar_vas_estrategico'):
|
|
assert policies[tool]['pre_validation'] == {
|
|
'enabled': True,
|
|
'tool': 'validar_vas_subject',
|
|
'fail_open': False,
|
|
}
|
|
|
|
|
|
def test_vas_subject_schema_requires_concrete_entity_semantically():
|
|
tools = yaml.safe_load((ROOT / 'config' / 'tools.yaml').read_text(encoding='utf-8'))['tools']
|
|
for tool in ('cancelar_vas_avulso', 'tratar_vas_estrategico'):
|
|
description = tools[tool]['args_schema']['subject']['description'].lower()
|
|
assert 'concreto' in description
|
|
assert 'item real' in description
|
|
assert 'ausente/null' in description
|
|
|
|
|
|
def test_generic_vas_category_does_not_resolve_to_arbitrary_product():
|
|
catalog = [
|
|
{'name': 'TIM Fashion Mensal'},
|
|
{'name': 'Aya Audiobooks Premium'},
|
|
{'name': 'Neymar Jr'},
|
|
{'name': 'Tamboro Mensal'},
|
|
{'name': 'Youtube Premium'},
|
|
]
|
|
canonical, matches = main._resolve_catalog_entity('streaming', catalog)
|
|
assert canonical is None
|
|
assert matches == []
|
|
|
|
|
|
def test_partial_unique_concrete_vas_name_resolves():
|
|
catalog = [
|
|
{'name': 'TIM Fashion Mensal'},
|
|
{'name': 'Aya Audiobooks Premium'},
|
|
{'name': 'Neymar Jr'},
|
|
]
|
|
canonical, matches = main._resolve_catalog_entity('TIM Fashion', catalog)
|
|
assert canonical == 'TIM Fashion Mensal'
|
|
assert len(matches) == 1
|
|
|
|
|
|
def test_ambiguous_reference_is_not_silently_resolved():
|
|
catalog = [
|
|
{'name': 'Apple Music SVA Mes'},
|
|
{'name': 'Apple Music Dados Mes'},
|
|
]
|
|
canonical, matches = main._resolve_catalog_entity('Apple Music', catalog)
|
|
assert canonical is None
|
|
assert {item['name'] for item in matches} == {'Apple Music SVA Mes', 'Apple Music Dados Mes'}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vas_validator_returns_needs_parameter_for_generic_category(monkeypatch):
|
|
async def no_enrich(name, args):
|
|
return None
|
|
monkeypatch.setattr(main, '_enrich_invoice_context', no_enrich)
|
|
monkeypatch.setattr(main, '_vas_entity_catalog', lambda args: [
|
|
{'name': 'TIM Fashion Mensal'},
|
|
{'name': 'Aya Audiobooks Premium'},
|
|
{'name': 'Neymar Jr'},
|
|
{'name': 'Tamboro Mensal'},
|
|
])
|
|
result = await main._validate_vas_subject({
|
|
'msisdn': '11999999999',
|
|
'subject': 'streaming',
|
|
'target_tool': 'cancelar_vas_avulso',
|
|
})
|
|
assert result['eligible'] is False
|
|
assert result['status'] == 'NEEDS_PARAMETER'
|
|
assert result['parameter'] == 'subject'
|
|
assert result['reason'] == 'subject_not_resolved'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vas_validator_accepts_unique_concrete_entity(monkeypatch):
|
|
async def no_enrich(name, args):
|
|
return None
|
|
monkeypatch.setattr(main, '_enrich_invoice_context', no_enrich)
|
|
monkeypatch.setattr(main, '_vas_entity_catalog', lambda args: [
|
|
{'name': 'TIM Fashion Mensal'},
|
|
{'name': 'Tamboro Mensal'},
|
|
])
|
|
result = await main._validate_vas_subject({
|
|
'msisdn': '11999999999',
|
|
'subject': 'TIM Fashion',
|
|
'target_tool': 'cancelar_vas_avulso',
|
|
})
|
|
assert result['eligible'] is True
|
|
assert result['status'] == 'ELIGIBLE'
|
|
assert result['resolved_subject'] == 'TIM Fashion Mensal'
|
|
|
|
|
|
def test_vas_preflight_never_falls_through_when_subject_is_unresolved():
|
|
# Existing execution-time preflight is a second defensive layer. It must not
|
|
# let an unresolved generic entity proceed even if the pre-validation stage
|
|
# was bypassed by a caller outside the framework runtime.
|
|
args = {
|
|
'msisdn': '11999999999',
|
|
'subject': 'streaming',
|
|
'billing_analysis': {'currentInvoice': []},
|
|
}
|
|
result = main._preflight_subject('cancelar_vas_avulso', args)
|
|
assert result is not None
|
|
assert result['status'] == 'NEEDS_PARAMETER'
|
|
assert result['parameter'] == 'subject'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vas_validator_redirects_canonical_strategic_entity_before_execution(monkeypatch):
|
|
from types import SimpleNamespace
|
|
|
|
async def no_enrich(name, args):
|
|
args['billing_analysis'] = {'currentInvoice': []}
|
|
return None
|
|
|
|
monkeypatch.setattr(main, '_enrich_invoice_context', no_enrich)
|
|
monkeypatch.setattr(main, '_vas_entity_catalog', lambda args: [
|
|
{'name': 'Tamboro Mensal'},
|
|
{'name': 'Youtube Premium'},
|
|
])
|
|
monkeypatch.setattr(
|
|
main.invoice_resolver,
|
|
'resolve',
|
|
lambda subjects, billing: SimpleNamespace(
|
|
resolved=[SimpleNamespace(
|
|
canonical_name='Youtube Premium',
|
|
tool_category='vas_estrategico',
|
|
item_type='streaming',
|
|
)],
|
|
ambiguous=[],
|
|
out_of_scope=[],
|
|
),
|
|
)
|
|
|
|
result = await main._validate_vas_subject({
|
|
'msisdn': '11999999999',
|
|
'subject': 'youtube',
|
|
'target_tool': 'cancelar_vas_avulso',
|
|
})
|
|
|
|
assert result['eligible'] is True
|
|
assert result['resolved_subject'] == 'Youtube Premium'
|
|
decision = result['transaction_decision']
|
|
assert decision['resolved_arguments']['subject'] == 'Youtube Premium'
|
|
assert decision['target_tool'] == 'tratar_vas_estrategico'
|
|
assert decision['action_changed'] is True
|
|
assert decision['requires_reconfirmation'] is True
|
|
assert 'Youtube Premium' in decision['confirmation_message']
|
|
|
|
|
|
def test_vas_domain_policy_uses_invoice_detail_classe_for_strategic_item():
|
|
args = {
|
|
'invoice_detail': {
|
|
'parsed_content': {
|
|
'1199999999': {
|
|
'SVA Detalhe Total': [
|
|
{
|
|
'desc': 'Youtube Premium',
|
|
'value': 10,
|
|
'classe': 'estrategico',
|
|
'estrategico': True,
|
|
'verb': 'falar sobre',
|
|
},
|
|
{
|
|
'desc': 'Tamboro Mensal',
|
|
'value': 14.99,
|
|
'classe': 'avulso',
|
|
'verb': 'cancelar',
|
|
},
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
assert main._vas_domain_policy_from_invoice_detail('Youtube Premium', args) == ('vas_estrategico', 'estrategico')
|
|
assert main._vas_domain_policy_from_invoice_detail('Tamboro Mensal', args) == ('cancelar_vas_avulso', 'avulso')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vas_validator_real_invoice_detail_redirects_youtube_without_resolver_guess(monkeypatch):
|
|
async def enrich(name, args):
|
|
args['billing_analysis'] = {
|
|
'currentInvoice': [
|
|
{
|
|
'type': 'streaming',
|
|
'desc': 'Streamings',
|
|
'items': [{'desc': 'Youtube Premium', 'value': '10.0'}],
|
|
}
|
|
]
|
|
}
|
|
args['invoice_detail'] = {
|
|
'parsed_content': {
|
|
'1199999999': {
|
|
'SVA Detalhe Total': [
|
|
{
|
|
'desc': 'Youtube Premium',
|
|
'value': 10,
|
|
'classe': 'estrategico',
|
|
'estrategico': True,
|
|
'verb': 'falar sobre',
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
|
|
monkeypatch.setattr(main, '_enrich_invoice_context', enrich)
|
|
monkeypatch.setattr(main, '_vas_entity_catalog', lambda args: [{'name': 'Youtube Premium'}])
|
|
# If invoice-detail classification works, the generic resolver fallback is unnecessary.
|
|
monkeypatch.setattr(main.invoice_resolver, 'resolve', lambda *a, **k: (_ for _ in ()).throw(AssertionError('fallback should not run')))
|
|
|
|
result = await main._validate_vas_subject({
|
|
'msisdn': '11999999999',
|
|
'subject': 'youtube',
|
|
'target_tool': 'cancelar_vas_avulso',
|
|
})
|
|
|
|
decision = result['transaction_decision']
|
|
assert result['resolved_subject'] == 'Youtube Premium'
|
|
assert decision['domain_policy']['class'] == 'vas_estrategico'
|
|
assert decision['domain_policy']['item_type'] == 'estrategico'
|
|
assert decision['target_tool'] == 'tratar_vas_estrategico'
|
|
assert decision['action_changed'] is True
|
|
assert decision['requires_reconfirmation'] is True
|
|
|
|
|
|
def test_multi_item_execution_preflight_does_not_reresolve_joined_subject():
|
|
args = {
|
|
'msisdn': '11999999999',
|
|
'subject': 'Tamboro Mensal, Paramount+',
|
|
'items': [
|
|
{'name': 'Tamboro Mensal', 'msisdn': '11999999999'},
|
|
{'name': 'Paramount+', 'msisdn': '11999999999'},
|
|
],
|
|
}
|
|
assert main._preflight_subject('cancelar_vas_avulso', args) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_contestation_wrong_amount_is_recoverable_with_invoice_amount(monkeypatch):
|
|
async def no_enrich(name, args):
|
|
return None
|
|
|
|
monkeypatch.setattr(main, '_enrich_invoice_context', no_enrich)
|
|
monkeypatch.setattr(main, '_preflight_subject', lambda name, args: None)
|
|
monkeypatch.setattr(main, 'validate_contestation_items', lambda items, payload: (
|
|
items,
|
|
[{
|
|
'erro': 'valor_ajuste_maior_que_item',
|
|
'valor_item_fatura': '14.99',
|
|
'item_fatura_resolvido': 'Tamboro Mensal',
|
|
}],
|
|
"Valor de ajuste excede o valor cobrado",
|
|
))
|
|
result = await main._validate_contestation({
|
|
'msisdn': '11999999999',
|
|
'subject': 'Tamboro Mensal',
|
|
'valor': '29.98',
|
|
'billing_analysis': {},
|
|
'target_tool': 'contestar_cobranca',
|
|
})
|
|
assert result['eligible'] is False
|
|
assert result['status'] == 'NEEDS_PARAMETER'
|
|
assert result['parameter'] == 'valor'
|
|
assert result['resolved_value'] == '14.99'
|
|
assert 'R$ 14,99' in result['parameter_message']
|
|
assert 'R$ 29,98' in result['parameter_message']
|