import os import unittest from datetime import datetime, timedelta from app.core.time_utils import utc_now os.environ.setdefault("DEBUG", "false") from app.services.orchestration.conversation_policy import ConversationPolicy from app.services.orchestration.entity_normalizer import EntityNormalizer class FakeState: def __init__(self, entries=None, contexts=None): self.entries = entries or {} self.contexts = contexts or {} def get_entry(self, bucket: str, user_id: int | None, *, expire: bool = False): if user_id is None: return None return self.entries.get(bucket, {}).get(user_id) def get_user_context(self, user_id: int | None): if user_id is None: return None return self.contexts.get(user_id) def save_user_context(self, user_id: int | None, context: dict): if user_id is None: return self.contexts[user_id] = context class FakeService: def __init__(self, state): self.state = state self.normalizer = EntityNormalizer() def _get_user_context(self, user_id: int | None): return self.state.get_user_context(user_id) def _save_user_context(self, user_id: int | None, context: dict | None) -> None: if user_id is None or not isinstance(context, dict): return self.state.save_user_context(user_id, context) class ContextSummaryTests(unittest.TestCase): def test_build_context_summary_describes_open_review_flow(self): now = utc_now() state = FakeState( entries={ "pending_review_drafts": { 21: { "payload": { "placa": "ABC1234", "modelo": "Onix", "ano": 2024, }, "expires_at": now + timedelta(minutes=15), } }, "pending_review_confirmations": { 21: { "payload": { "placa": "ABC1234", "data_hora": "14/03/2026 16:30", }, "expires_at": now + timedelta(minutes=15), } }, }, contexts={ 21: { "active_domain": "review", "active_task": "review_schedule", "generic_memory": { "placa": "ABC1234", "orcamento_max": 70000, "perfil_veiculo": ["suv"], }, "shared_memory": {}, "order_queue": [{"domain": "sales", "message": "quero comprar um carro"}], "pending_order_selection": { "orders": [ {"domain": "review", "message": "agendar revisao"}, {"domain": "sales", "message": "comprar um carro"}, ] }, "pending_switch": {"target_domain": "sales"}, "last_stock_results": [], "selected_vehicle": None, "last_tool_result": {"tool_name": "listar_agendamentos_revisao", "result_type": "list"}, } }, ) summary = ConversationPolicy(service=FakeService(state)).build_context_summary(21) self.assertIn("Fluxo ativo: agendamento de revisao.", summary) self.assertIn("Memoria generica temporaria: placa=ABC1234, orcamento=R$ 70.000, perfil=suv.", summary) self.assertIn("Ultima tool executada: listar_agendamentos_revisao (list).", summary) self.assertIn("Aguardando escolha entre 2 pedido(s) detectado(s) na mesma mensagem.", summary) self.assertIn("Troca de contexto pendente para compra de veiculo.", summary) self.assertIn("Fila de pedidos pendentes: 1.", summary) self.assertIn("Rascunho aberto de agendamento de revisao.", summary) self.assertIn("Dados atuais: placa=ABC1234, modelo=Onix, ano=2024.", summary) self.assertIn("Faltando: data/hora, km, revisao previa na concessionaria.", summary) self.assertIn("Confirmacao pendente de horario sugerido para revisao.", summary) self.assertIn("Dados sugeridos: placa=ABC1234, data/hora=14/03/2026 16:30.", summary) def test_build_context_summary_describes_open_order_flow(self): now = utc_now() state = FakeState( entries={ "pending_order_drafts": { 10: { "payload": {"cpf": "12345678909"}, "expires_at": now + timedelta(minutes=15), } }, "pending_stock_selections": { 10: { "payload": [ { "id": 1, "modelo": "Honda Civic 2021", "categoria": "sedan", "preco": 48500.0, } ], "expires_at": now + timedelta(minutes=15), } }, }, contexts={ 10: { "active_domain": "sales", "active_task": "order_create", "generic_memory": { "cpf": "12345678909", "orcamento_max": 80000, }, "shared_memory": {}, "order_queue": [], "pending_order_selection": None, "pending_switch": None, "last_stock_results": [ {"id": 1, "modelo": "Honda Civic 2021", "categoria": "sedan", "preco": 48500.0}, {"id": 2, "modelo": "Toyota Yaris 2020", "categoria": "hatch", "preco": 49900.0}, ], "selected_vehicle": {"id": 1, "modelo": "Honda Civic 2021"}, "pending_single_vehicle_confirmation": {"id": 1, "modelo": "Honda Civic 2021"}, "last_tool_result": {"tool_name": "consultar_estoque", "result_type": "list"}, } }, ) summary = ConversationPolicy(service=FakeService(state)).build_context_summary(10) self.assertIn("Fluxo ativo: criacao de pedido.", summary) self.assertIn("Memoria generica temporaria: cpf=12345678909, orcamento=R$ 80.000.", summary) self.assertIn("Veiculo selecionado para compra: Honda Civic 2021.", summary) self.assertIn("Ultima consulta de estoque com 2 opcao(oes) disponivel(is).", summary) self.assertIn("Ultima tool executada: consultar_estoque (list).", summary) self.assertIn("Aguardando confirmacao explicita do veiculo Honda Civic 2021.", summary) self.assertIn("Rascunho aberto de criacao de pedido.", summary) self.assertIn("Dados atuais: cpf=12345678909.", summary) self.assertIn("Faltando: vehicle_id.", summary) self.assertIn("Aguardando escolha de veiculo em 1 opcao(oes) de estoque.", summary) def test_build_context_summary_uses_snapshot_when_bucket_is_missing(self): now = utc_now() state = FakeState( contexts={ 7: { "active_domain": "review", "active_task": "review_schedule", "generic_memory": {"placa": "ABC1C23"}, "shared_memory": {}, "flow_snapshots": { "review_schedule": { "payload": { "placa": "ABC1C23", "modelo": "Onix", "ano": 2024, }, "expires_at": now + timedelta(minutes=15), } }, "order_queue": [], "pending_order_selection": None, "pending_switch": None, "last_stock_results": [], "selected_vehicle": None, } } ) summary = ConversationPolicy(service=FakeService(state)).build_context_summary(7) self.assertIn("Fluxo ativo: agendamento de revisao.", summary) self.assertIn("Rascunho aberto de agendamento de revisao.", summary) self.assertIn("Dados atuais: placa=ABC1C23, modelo=Onix, ano=2024.", summary) self.assertIn("Faltando: data/hora, km, revisao previa na concessionaria.", summary) def test_build_context_summary_describes_open_rental_flow(self): now = utc_now() state = FakeState( entries={ "pending_rental_drafts": { 5: { "payload": { "placa": "RAA1A01", "rental_vehicle_id": 1, "data_inicio": "20/03/2026 10:00", }, "expires_at": now + timedelta(minutes=15), } }, "pending_rental_selections": { 5: { "payload": [ {"id": 1, "placa": "RAA1A01", "modelo": "Chevrolet Tracker", "categoria": "suv", "ano": 2024, "valor_diaria": 219.9, "status": "disponivel"}, ], "expires_at": now + timedelta(minutes=15), } }, }, contexts={ 5: { "active_domain": "rental", "active_task": "rental_create", "generic_memory": {}, "shared_memory": {}, "order_queue": [], "pending_order_selection": None, "pending_switch": None, "last_stock_results": [], "selected_vehicle": None, "last_rental_results": [ {"id": 1, "placa": "RAA1A01", "modelo": "Chevrolet Tracker", "categoria": "suv", "ano": 2024, "valor_diaria": 219.9, "status": "disponivel"}, ], "selected_rental_vehicle": {"id": 1, "placa": "RAA1A01", "modelo": "Chevrolet Tracker"}, "last_tool_result": {"tool_name": "consultar_frota_aluguel", "result_type": "list"}, } }, ) summary = ConversationPolicy(service=FakeService(state)).build_context_summary(5) self.assertIn("Fluxo ativo: abertura de locacao.", summary) self.assertIn("Veiculo selecionado para locacao: Chevrolet Tracker.", summary) self.assertIn("Ultima consulta de locacao com 1 opcao(oes) disponivel(is).", summary) self.assertIn("Rascunho aberto de locacao.", summary) self.assertIn("Faltando: data fim prevista.", summary) self.assertIn("Aguardando escolha de veiculo em 1 opcao(oes) de locacao.", summary) if __name__ == "__main__": unittest.main()