import unittest from datetime import datetime from types import SimpleNamespace from unittest.mock import patch from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.db.mock_database import MockBase from app.db.mock_models import ConversationTurn, User from app.services.orchestration.conversation_history_service import ConversationHistoryService class _FakeQuery: def __init__(self, result): self.result = result def filter(self, *args, **kwargs): return self def first(self): return self.result class _FakeSession: def __init__(self, user=None): self.user = user self.added = [] self.committed = False self.rolled_back = False self.closed = False def query(self, model): return _FakeQuery(self.user) def add(self, item): self.added.append(item) def commit(self): self.committed = True def rollback(self): self.rolled_back = True def close(self): self.closed = True class ConversationHistoryServiceTests(unittest.TestCase): def test_record_turn_persists_conversation_audit_record(self): session = _FakeSession( user=SimpleNamespace( id=7, channel="telegram", external_id="12345", username="cliente_teste", ) ) service = ConversationHistoryService() started_at = datetime(2026, 3, 16, 18, 0, 0) completed_at = datetime(2026, 3, 16, 18, 0, 5) with patch( "app.services.orchestration.conversation_history_service.SessionMockLocal", return_value=session, ): service.record_turn( request_id="req-123", conversation_id="user:7", user_id=7, user_message="quero comprar um carro", assistant_response="Encontrei 2 veiculo(s).", turn_status="completed", intent="order_create", domain="sales", action="collect_order_create", tool_name="consultar_estoque", tool_arguments={"preco_max": 80000}, error_detail=None, started_at=started_at, completed_at=completed_at, elapsed_ms=512.4, ) self.assertTrue(session.committed) self.assertTrue(session.closed) self.assertEqual(len(session.added), 1) record = session.added[0] self.assertEqual(record.request_id, "req-123") self.assertEqual(record.conversation_id, "user:7") self.assertEqual(record.user_id, 7) self.assertEqual(record.channel, "telegram") self.assertEqual(record.external_id, "***345") self.assertEqual(record.username, "cliente_teste") self.assertEqual(record.intent, "order_create") self.assertEqual(record.domain, "sales") self.assertEqual(record.action, "collect_order_create") self.assertEqual(record.tool_name, "consultar_estoque") self.assertEqual(record.tool_arguments, '{"preco_max":80000}') self.assertEqual(record.started_at, started_at) self.assertEqual(record.completed_at, completed_at) self.assertEqual(record.elapsed_ms, 512.4) def test_record_turn_masks_sensitive_fields_before_persisting(self): session = _FakeSession( user=SimpleNamespace( id=7, channel="telegram", external_id="987654321", username="cliente_teste", ) ) service = ConversationHistoryService() with patch( "app.services.orchestration.conversation_history_service.SessionMockLocal", return_value=session, ): service.record_turn( request_id="req-sensitive", conversation_id="user:7", user_id=7, user_message="Meu cpf 123.456.789-09 e a placa ABC1D23.", assistant_response="Recebi o identificador_comprovante=NSU123 para a placa ABC1D23.", turn_status="failed", intent="rental_payment", domain="rental", action="call_tool", tool_name="registrar_pagamento_aluguel", tool_arguments={ "cpf": "12345678909", "placa": "ABC1D23", "external_id": "987654321", "identificador_comprovante": "NSU123", "nested": { "placa": "ABC1D23", }, }, error_detail='{"external_id":"987654321","placa":"ABC1D23","identificador_comprovante":"NSU123"}', ) record = session.added[0] self.assertNotIn("123.456.789-09", record.user_message) self.assertNotIn("ABC1D23", record.user_message) self.assertIn("***.***.***-09", record.user_message) self.assertIn("ABC***3", record.user_message) self.assertNotIn("NSU123", record.assistant_response) self.assertIn("***123", record.assistant_response) self.assertEqual(record.external_id, "******321") self.assertNotIn("12345678909", record.tool_arguments) self.assertNotIn("ABC1D23", record.tool_arguments) self.assertNotIn("987654321", record.tool_arguments) self.assertNotIn("NSU123", record.tool_arguments) self.assertIn("***.***.***-09", record.tool_arguments) self.assertIn("ABC***3", record.tool_arguments) self.assertIn("******321", record.tool_arguments) self.assertIn("***123", record.tool_arguments) self.assertNotIn("987654321", record.error_detail) self.assertNotIn("ABC1D23", record.error_detail) self.assertNotIn("NSU123", record.error_detail) def test_list_turns_filters_and_orders_recent_first(self): engine = create_engine("sqlite:///:memory:") SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) MockBase.metadata.create_all(bind=engine) self.addCleanup(engine.dispose) db = SessionLocal() user = User(channel="telegram", external_id="999", name="Cliente Teste", username="cliente_teste") db.add(user) db.commit() db.refresh(user) user_id = user.id db.add_all( [ ConversationTurn( request_id="req-old", conversation_id=f"user:{user_id}", user_id=user_id, channel="telegram", external_id="999", username="cliente_teste", user_message="oi", assistant_response="ola", turn_status="completed", intent="general", domain="general", action="answer_user", tool_name=None, tool_arguments=None, started_at=datetime(2026, 3, 16, 10, 0, 0), completed_at=datetime(2026, 3, 16, 10, 0, 1), elapsed_ms=100.0, ), ConversationTurn( request_id="req-new", conversation_id=f"user:{user_id}", user_id=user_id, channel="telegram", external_id="999", username="cliente_teste", user_message="quero comprar", assistant_response="pedido criado", turn_status="completed", intent="order_create", domain="sales", action="call_tool", tool_name="realizar_pedido", tool_arguments='{"vehicle_id":1}', started_at=datetime(2026, 3, 16, 11, 0, 0), completed_at=datetime(2026, 3, 16, 11, 0, 2), elapsed_ms=230.0, ), ConversationTurn( request_id="req-failed", conversation_id="user:999", user_id=None, channel=None, external_id=None, username=None, user_message="erro", assistant_response=None, turn_status="failed", intent="general", domain="general", action="answer_user", tool_name=None, tool_arguments=None, error_detail="RuntimeError: boom", started_at=datetime(2026, 3, 16, 12, 0, 0), completed_at=datetime(2026, 3, 16, 12, 0, 1), elapsed_ms=300.0, ), ] ) db.commit() db.close() service = ConversationHistoryService() with patch( "app.services.orchestration.conversation_history_service.SessionMockLocal", SessionLocal, ): items = service.list_turns(user_id=user_id, turn_status="completed", limit=5) self.assertEqual(len(items), 2) self.assertEqual(items[0]["request_id"], "req-new") self.assertEqual(items[1]["request_id"], "req-old") self.assertEqual(items[0]["tool_arguments"], {"vehicle_id": 1}) self.assertEqual(items[0]["turn_status"], "completed") self.assertEqual(items[0]["user_id"], user_id) self.assertEqual(items[0]["external_id"], "***") def test_list_turns_masks_legacy_sensitive_fields_on_read(self): engine = create_engine("sqlite:///:memory:") SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) MockBase.metadata.create_all(bind=engine) self.addCleanup(engine.dispose) db = SessionLocal() db.add( ConversationTurn( request_id="req-sensitive", conversation_id="user:42", user_id=42, channel="telegram", external_id="987654321", username="cliente", user_message="Cpf 12345678909 e placa ABC1234.", assistant_response="identificador_comprovante=NSU123 recebido para ABC1234", turn_status="completed", tool_name="registrar_pagamento_aluguel", tool_arguments='{"cpf":"12345678909","placa":"ABC1234","external_id":"987654321","identificador_comprovante":"NSU123"}', error_detail='{"placa":"ABC1234","external_id":"987654321","identificador_comprovante":"NSU123"}', started_at=datetime(2026, 3, 16, 13, 0, 0), completed_at=datetime(2026, 3, 16, 13, 0, 1), ) ) db.commit() db.close() service = ConversationHistoryService() with patch( "app.services.orchestration.conversation_history_service.SessionMockLocal", SessionLocal, ): items = service.list_turns(request_id="req-sensitive", limit=5) self.assertEqual(len(items), 1) item = items[0] self.assertEqual(item["external_id"], "******321") self.assertNotIn("12345678909", item["user_message"]) self.assertNotIn("ABC1234", item["user_message"]) self.assertNotIn("NSU123", item["assistant_response"]) self.assertEqual(item["tool_arguments"]["cpf"], "***.***.***-09") self.assertEqual(item["tool_arguments"]["placa"], "ABC***4") self.assertEqual(item["tool_arguments"]["external_id"], "******321") self.assertEqual(item["tool_arguments"]["identificador_comprovante"], "***123") self.assertNotIn("ABC1234", item["error_detail"]) self.assertNotIn("987654321", item["error_detail"]) self.assertNotIn("NSU123", item["error_detail"]) def test_list_turns_can_filter_by_request_id(self): engine = create_engine("sqlite:///:memory:") SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) MockBase.metadata.create_all(bind=engine) self.addCleanup(engine.dispose) db = SessionLocal() db.add( ConversationTurn( request_id="req-only", conversation_id="user:42", user_id=None, user_message="teste", assistant_response="ok", turn_status="completed", started_at=datetime(2026, 3, 16, 13, 0, 0), completed_at=datetime(2026, 3, 16, 13, 0, 1), ) ) db.commit() db.close() service = ConversationHistoryService() with patch( "app.services.orchestration.conversation_history_service.SessionMockLocal", SessionLocal, ): items = service.list_turns(request_id="req-only", limit=5) self.assertEqual(len(items), 1) self.assertEqual(items[0]["request_id"], "req-only") self.assertEqual(items[0]["conversation_id"], "user:42")