vitor
/
orquestrador
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

🧹 refactor(rental): remover suporte a multas de aluguel

Browse Source
main
Vitor Hugo Belorio Simão 2 weeks ago
parent a692a7023e
commit 9182ec29a3
18 changed files with 66 additions and 345 deletions
Show Stats Download Patch File Download Diff File

1
app/db/bootstrap.py
Unescape Escape View File

@ -16,7 +16,6 @@ from app.db.mock_models import (
IntegrationRoute,
Order,
RentalContract,
RentalFine,
RentalPayment,
RentalVehicle,
ReviewSchedule,

20
app/db/mock_models.py
Unescape Escape View File

@ -138,26 +138,6 @@ class RentalPayment(MockBase):
observacoes = Column(Text, nullable=True)
created_at = Column(DateTime, server_default=func.current_timestamp())
class RentalFine(MockBase):
__tablename__ = "rental_fines"
id = Column(Integer, primary_key=True, index=True)
protocolo = Column(String(50), unique=True, nullable=False, index=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
rental_contract_id = Column(Integer, ForeignKey("rental_contracts.id"), nullable=True, index=True)
contrato_numero = Column(String(50), nullable=True, index=True)
placa = Column(String(10), nullable=True, index=True)
auto_infracao = Column(String(60), nullable=True, index=True)
orgao_emissor = Column(String(120), nullable=True)
valor = Column(Float, nullable=False)
data_infracao = Column(DateTime, nullable=True, index=True)
vencimento = Column(DateTime, nullable=True, index=True)
status = Column(String(20), nullable=False, default="registrada", index=True)
observacoes = Column(Text, nullable=True)
created_at = Column(DateTime, server_default=func.current_timestamp())
class IntegrationRoute(MockBase):
__tablename__ = "integration_routes"
__table_args__ = (

52
app/db/tool_seed.py
Unescape Escape View File

@ -439,53 +439,6 @@ def get_tools_definitions():
"required": ["valor"],
},
},
{
"name": "registrar_multa_aluguel",
"description": (
"Use esta ferramenta quando o usuario enviar uma multa de transito de um "
"carro alugado ou pedir para registrar a multa no sistema. Ela registra "
"os dados da autuacao como placa, contrato, auto de infracao, orgao emissor, "
"valor, data da infracao e vencimento."
),
"parameters": {
"type": "object",
"properties": {
"placa": {
"type": "string",
"description": "Placa do veiculo relacionado a multa, quando disponivel.",
},
"contrato_numero": {
"type": "string",
"description": "Numero do contrato de aluguel, quando aparecer no documento.",
},
"auto_infracao": {
"type": "string",
"description": "Numero do auto de infracao ou identificador da multa.",
},
"orgao_emissor": {
"type": "string",
"description": "Orgao emissor da multa, por exemplo DETRAN ou prefeitura.",
},
"valor": {
"type": "number",
"description": "Valor da multa.",
},
"data_infracao": {
"type": "string",
"description": "Data da infracao. Aceita formatos como 17/03/2026 e 2026-03-17.",
},
"vencimento": {
"type": "string",
"description": "Data de vencimento da multa. Aceita formatos como 17/03/2026 e 2026-03-17.",
},
"observacoes": {
"type": "string",
"description": "Resumo livre do que foi identificado no documento.",
},
},
"required": ["valor"],
},
},
{
"name": "limpar_contexto_conversa",
"description": (
@ -560,6 +513,11 @@ def seed_tools():
try:
repo = ToolRepository(db)
existing = repo.get_all()
obsolete_tool_names = {"registrar_multa_aluguel"}
for tool in existing:
if tool.name in obsolete_tool_names:
repo.delete(tool.id)
existing = repo.get_all()
existing_names = {t.name for t in existing}
for tool_def in get_tools_definitions():
if tool_def["name"] in existing_names:

4
app/services/ai/llm_service.py
Unescape Escape View File

@ -115,7 +115,7 @@ class LLMService:
extracted_text = (payload.get("response") or "").strip() or (caption or "").strip()
return self._coerce_image_workflow_response(extracted_text)
# Define o prompt de extracao usado para comprovantes e multas em imagem.
# Define o prompt de extracao usado para comprovantes e documentos em imagem.
def _build_image_workflow_prompt(self, *, caption: str | None) -> str:
normalized_caption = (caption or "").strip() or "sem legenda"
return (
@ -129,8 +129,6 @@ class LLMService:
"Se a data de pagamento incluir hora e minuto visiveis na imagem, preserve a data e a hora no campo data_pagamento no formato DD/MM/AAAA HH:MM. "
"Nao reduza para somente a data quando a hora estiver visivel. "
"Se apenas a data estiver visivel, use somente a data. "
"Se for multa de transito relacionada a carro alugado, responda com uma frase objetiva em portugues no formato: "
"Registrar multa de aluguel: placa <...>; contrato <...>; auto_infracao <...>; orgao_emissor <...>; valor <...>; data_infracao <...>; vencimento <...>; observacoes <...>. "
"Se for outro documento automotivo util, resuma em uma frase com os dados importantes. "
f"Se nao conseguir identificar com seguranca, responda exatamente: {IMAGE_ANALYSIS_FAILURE_MESSAGE} "
"Use apenas dados observaveis e nao invente informacoes. "

77
app/services/domain/rental_service.py
Unescape Escape View File

@ -7,7 +7,7 @@ from uuid import uuid4
from app.core.time_utils import utc_now
from app.db.mock_database import SessionMockLocal
from app.db.mock_models import RentalContract, RentalFine, RentalPayment, RentalVehicle, User
from app.db.mock_models import RentalContract, RentalPayment, RentalVehicle, User
from app.services.domain.tool_errors import raise_tool_http_error
from app.services.integrations.events import (
RENTAL_OPENED_EVENT,
@ -557,78 +557,3 @@ async def registrar_pagamento_aluguel(
return result
finally:
db.close()
# Registra uma multa ligada ao aluguel usando os identificadores disponiveis.
async def registrar_multa_aluguel(
valor: float,
placa: str | None = None,
contrato_numero: str | None = None,
auto_infracao: str | None = None,
orgao_emissor: str | None = None,
data_infracao: str | None = None,
vencimento: str | None = None,
observacoes: str | None = None,
user_id: int | None = None,
) -> dict:
normalized_contract = _normalize_contract_number(contrato_numero)
plate = technical_normalizer.normalize_plate(placa)
notice_number = _normalize_text_field(auto_infracao)
db = SessionMockLocal()
try:
contract = _resolve_rental_contract(
db,
contrato_numero=normalized_contract,
placa=plate,
user_id=user_id,
active_only=False,
)
if contract is not None:
normalized_contract = contract.contrato_numero
plate = contract.placa
if not normalized_contract and not plate and not notice_number:
raise_tool_http_error(
status_code=400,
code="missing_fine_reference",
message=(
"Preciso da placa, do numero do contrato, do auto da infracao ou de uma locacao "
"vinculada ao usuario para registrar a multa."
),
retryable=True,
field="placa",
)
record = RentalFine(
protocolo=f"ALM-{utc_now().strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}",
user_id=user_id,
rental_contract_id=contract.id if contract is not None else None,
contrato_numero=normalized_contract,
placa=plate,
auto_infracao=notice_number,
orgao_emissor=_normalize_text_field(orgao_emissor),
valor=_normalize_money(valor),
data_infracao=_parse_optional_datetime(data_infracao, field_name="data_infracao"),
vencimento=_parse_optional_datetime(vencimento, field_name="vencimento"),
observacoes=_normalize_text_field(observacoes),
status="registrada",
)
db.add(record)
db.commit()
db.refresh(record)
return {
"protocolo": record.protocolo,
"rental_contract_id": record.rental_contract_id,
"contrato_numero": record.contrato_numero,
"placa": record.placa,
"auto_infracao": record.auto_infracao,
"orgao_emissor": record.orgao_emissor,
"valor": float(record.valor),
"data_infracao": record.data_infracao.isoformat() if record.data_infracao else None,
"vencimento": record.vencimento.isoformat() if record.vencimento else None,
"status": record.status,
}
finally:
db.close()

8
app/services/flows/rental_flow.py
Unescape Escape View File

@ -486,7 +486,7 @@ class RentalFlowMixin:
# Detecta quando o usuario quer iniciar uma nova locacao.
def _has_explicit_rental_request(self, message: str) -> bool:
normalized = self._normalize_text(message).strip()
if any(term in normalized for term in {"multa", "comprovante", "pagamento", "devolucao", "devolver"}):
if any(term in normalized for term in {"comprovante", "pagamento", "devolucao", "devolver"}):
return False
request_terms = {
"quero alugar",
@ -507,10 +507,10 @@ class RentalFlowMixin:
normalized = self._normalize_text(message).strip()
return any(term in normalized for term in {"devolver", "devolucao", "encerrar locacao", "fechar locacao"})
# Detecta quando a mensagem parece tratar de pagamento ou multa de aluguel.
def _has_rental_payment_or_fine_request(self, message: str) -> bool:
# Detecta quando a mensagem parece tratar de pagamento de aluguel.
def _has_rental_payment_message(self, message: str) -> bool:
normalized = self._normalize_text(message).strip()
return any(term in normalized for term in {"multa", "comprovante", "pagamento", "boleto", "pix"})
return any(term in normalized for term in {"comprovante", "pagamento", "boleto", "pix"})
# Interpreta selecoes numericas com base na ultima lista apresentada.
def _match_rental_vehicle_from_message_index(self, message: str, rental_results: list[dict]) -> dict | None:

40
app/services/flows/rental_flow_support.py
Unescape Escape View File

@ -6,7 +6,7 @@ from sqlalchemy import or_
from app.core.time_utils import utc_now
from app.db.mock_database import SessionMockLocal
from app.db.mock_models import RentalContract, RentalFine, RentalPayment
from app.db.mock_models import RentalContract, RentalPayment
from app.services.flows.flow_state_support import FlowStateSupport
from app.services.orchestration import technical_normalizer
from app.services.orchestration.orchestrator_config import PENDING_RENTAL_SELECTION_TTL_MINUTES
@ -69,30 +69,6 @@ class RentalFlowStateSupport(FlowStateSupport):
}
)
latest_fine = (
db.query(RentalFine)
.filter(
or_(
RentalFine.rental_contract_id == contract.id,
RentalFine.contrato_numero == contract.contrato_numero,
)
)
.order_by(RentalFine.created_at.desc())
.first()
)
if latest_fine is not None:
payload.update(
{
"auto_infracao": latest_fine.auto_infracao,
"data_infracao": latest_fine.data_infracao.isoformat()
if latest_fine.data_infracao
else None,
"vencimento": latest_fine.vencimento.isoformat() if latest_fine.vencimento else None,
}
)
if latest_fine.valor is not None:
payload["valor_multa"] = float(latest_fine.valor)
return self.sanitize_rental_contract_snapshot(payload)
except Exception:
return None
@ -243,20 +219,6 @@ class RentalFlowStateSupport(FlowStateSupport):
snapshot["favorecido"] = favorecido
snapshot.setdefault("status_pagamento", "registrado")
violation_date = str(payload.get("data_infracao") or "").strip()
if violation_date:
snapshot["data_infracao"] = violation_date
due_date = str(payload.get("vencimento") or "").strip()
if due_date:
snapshot["vencimento"] = due_date
infraction_notice = str(payload.get("auto_infracao") or "").strip()
if infraction_notice:
snapshot["auto_infracao"] = infraction_notice
if violation_date or infraction_notice:
fine_value = technical_normalizer.normalize_positive_number(payload.get("valor"))
if fine_value is not None:
snapshot["valor_multa"] = float(fine_value)
return snapshot
def get_last_rental_contract(self, user_id: int | None) -> dict | None:

2
app/services/orchestration/conversation_policy.py
Unescape Escape View File

@ -589,8 +589,6 @@ class ConversationPolicy:
return "rental:return"
if tool_name == "registrar_pagamento_aluguel" or "comprovante" in normalized or "pagamento" in normalized:
return "rental:payment"
if tool_name == "registrar_multa_aluguel" or "multa" in normalized:
return "rental:fine"
if (
intent == "rental_create"
or self.contains_any_term(normalized, {"aluguel", "alugar", "locacao", "locar"})

1
app/services/orchestration/orchestrator_config.py
Unescape Escape View File

@ -59,7 +59,6 @@ DETERMINISTIC_RESPONSE_TOOLS = {
"abrir_locacao_aluguel",
"registrar_devolucao_aluguel",
"registrar_pagamento_aluguel",
"registrar_multa_aluguel",
"limpar_contexto_conversa",
"continuar_proximo_pedido",
"descartar_pedidos_pendentes",

1
app/services/orchestration/orchestrator_context_manager.py
Unescape Escape View File

@ -136,7 +136,6 @@ class OrchestratorContextManager:
"abrir_locacao_aluguel",
"registrar_devolucao_aluguel",
"registrar_pagamento_aluguel",
"registrar_multa_aluguel",
} and isinstance(tool_result, dict):
self.service._store_last_rental_contract(user_id=user_id, payload=tool_result)
self.capture_tool_result_context(

76
app/services/orchestration/orquestrador_service.py
Unescape Escape View File

@ -902,7 +902,7 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
context=context,
)
)
or self._has_rental_payment_or_fine_request(message)
or self._has_rental_payment_message(message)
):
return None
@ -1042,7 +1042,6 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
if (
self._has_rental_return_management_request(message, user_id=user_id)
or self._has_rental_payment_request(message, user_id=user_id)
or self._has_rental_fine_request(message, user_id=user_id)
):
return False
if (
@ -1291,8 +1290,6 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
# Detecta pedidos para registrar pagamento de aluguel.
def _has_rental_payment_request(self, message: str, user_id: int | None = None) -> bool:
normalized_message = self._normalize_text(message).strip()
if "multa" in normalized_message:
return False
payment_terms = ("pagamento", "comprovante", "pix", "boleto")
if not any(term in normalized_message for term in payment_terms):
return False
@ -1303,19 +1300,6 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
or self._extract_rental_plate_from_text(message)
)
# Detecta pedidos para registrar multa vinculada ao aluguel.
def _has_rental_fine_request(self, message: str, user_id: int | None = None) -> bool:
normalized_message = self._normalize_text(message).strip()
if "multa" not in normalized_message:
return False
return bool(
"aluguel" in normalized_message
or "locacao" in normalized_message
or "auto_infracao" in normalized_message
or self._extract_rental_contract_number_from_text(message)
or self._extract_rental_plate_from_text(message)
)
# Decide se a mensagem pode virar uma acao de aluguel sem depender do planner.
def _is_deterministic_rental_management_candidate(self, message: str, user_id: int | None) -> bool:
has_policy = hasattr(self, "policy") and getattr(self, "policy") is not None
@ -1326,7 +1310,6 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
return bool(
self._has_rental_return_management_request(message, user_id=user_id)
or self._has_rental_payment_request(message, user_id=user_id)
or self._has_rental_fine_request(message, user_id=user_id)
)
# Monta os argumentos da devolucao a partir do texto enviado pelo usuario.
@ -1387,54 +1370,7 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
return self._merge_last_rental_reference(user_id=user_id, arguments=arguments)
# Monta os argumentos da multa de aluguel a partir da mensagem recebida.
def _build_rental_fine_arguments_from_message(self, message: str, user_id: int | None) -> dict:
arguments: dict = {}
contract_number = self._extract_rental_contract_number_from_text(message)
if contract_number:
arguments["contrato_numero"] = contract_number
plate = self._extract_rental_plate_from_text(message)
if plate:
arguments["placa"] = plate
notice_number = self._extract_rental_labeled_value(
message,
("auto_infracao", "auto de infracao", "auto da infracao"),
)
if notice_number:
arguments["auto_infracao"] = notice_number
issuing_agency = self._extract_rental_labeled_value(
message,
("orgao_emissor", "orgao emissor"),
)
if issuing_agency:
arguments["orgao_emissor"] = issuing_agency
amount_text = self._extract_rental_labeled_value(message, ("valor",))
amount = self._normalize_positive_number(amount_text)
if amount is not None:
arguments["valor"] = float(amount)
violation_date = self._extract_rental_labeled_value(message, ("data_infracao", "data da infracao"))
due_date = self._extract_rental_labeled_value(message, ("vencimento", "data_vencimento", "data de vencimento"))
datetimes = self._extract_rental_datetimes_from_text(message)
if not violation_date and datetimes:
violation_date = datetimes[0]
if not due_date and len(datetimes) >= 2:
due_date = datetimes[1]
if violation_date:
arguments["data_infracao"] = violation_date
if due_date:
arguments["vencimento"] = due_date
observations = self._extract_rental_labeled_value(message, ("observacoes", "observacao"))
if observations:
arguments["observacoes"] = observations
return self._merge_last_rental_reference(user_id=user_id, arguments=arguments)
# Executa devolucao, pagamento ou multa de aluguel quando os dados ja estiverem claros.
# Executa devolucao ou pagamento de aluguel quando os dados ja estiverem claros.
async def _try_handle_deterministic_rental_management(
self,
message: str,
@ -1449,12 +1385,6 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
tool_name = "registrar_devolucao_aluguel"
arguments = self._build_rental_return_arguments_from_message(message=message, user_id=user_id)
missing_response = None
elif self._has_rental_fine_request(message, user_id=user_id):
tool_name = "registrar_multa_aluguel"
arguments = self._build_rental_fine_arguments_from_message(message=message, user_id=user_id)
missing_response = None
if "valor" not in arguments:
missing_response = "Para registrar a multa de aluguel, preciso do valor informado no documento."
elif self._has_rental_payment_request(message, user_id=user_id):
tool_name = "registrar_pagamento_aluguel"
arguments = self._build_rental_payment_arguments_from_message(message=message, user_id=user_id)
@ -1497,7 +1427,7 @@ class OrquestradorService(ReviewFlowMixin, OrderFlowMixin, RentalFlowMixin):
return None
if (
self._has_rental_return_management_request(message, user_id=user_id)
or self._has_rental_payment_or_fine_request(message)
or self._has_rental_payment_message(message)
):
return None
if (

4
app/services/orchestration/prompt_builders.py
Unescape Escape View File

@ -16,7 +16,7 @@ def build_router_prompt(
"Voce e um assistente de concessionaria. "
"Sempre que a solicitacao depender de dados operacionais (estoque, validacao de cliente, "
"avaliacao de troca, agendamento de revisao, realizacao ou cancelamento de pedido, consulta de frota de aluguel, "
"abertura de locacao, devolucao de aluguel, registro de pagamento de aluguel ou registro de multa de aluguel), use a tool correta. "
"abertura de locacao, devolucao de aluguel ou registro de pagamento de aluguel), use a tool correta. "
"Se o usuario pedir para recomecar, esquecer contexto, cancelar fluxo atual, descartar fila pendente "
"ou continuar o proximo pedido, use a tool de orquestracao apropriada. "
"Mensagens de controle da conversa tem prioridade sobre qualquer fluxo em aberto. "
@ -37,7 +37,7 @@ def build_force_tool_prompt(
user_context = _build_user_context_line(user_id)
return (
"Reavalie a mensagem e priorize chamar tool se houver intencao operacional. "
"Considere tambem as operacoes de aluguel (consultar frota, abrir locacao, registrar devolucao, pagamento ou multa). "
"Considere tambem as operacoes de aluguel (consultar frota, abrir locacao, registrar devolucao ou pagamento). "
"Considere tambem tools de orquestracao para limpar contexto, cancelar fluxo, descartar fila ou continuar o proximo pedido. "
"Mesmo com fluxo incremental ativo, se a mensagem for de controle global da conversa, a tool de orquestracao deve vencer o rascunho atual. "
"Use texto apenas quando faltar dado obrigatorio.\n\n"

18
app/services/orchestration/response_formatter.py
Unescape Escape View File

@ -232,24 +232,6 @@ def fallback_format_tool_result(tool_name: str, tool_result: Any) -> str:
f"Data do pagamento: {data_pagamento}"
)
if tool_name == "registrar_multa_aluguel" and isinstance(tool_result, dict):
protocolo = tool_result.get("protocolo", "N/A")
contrato = tool_result.get("contrato_numero") or "N/A"
placa = tool_result.get("placa") or "N/A"
valor = format_currency_br(tool_result.get("valor"))
data_infracao = format_datetime_for_chat(tool_result.get("data_infracao", "N/A"))
vencimento = format_datetime_for_chat(tool_result.get("vencimento", "N/A"))
auto_infracao = tool_result.get("auto_infracao") or "N/A"
return (
"Multa de aluguel registrada com sucesso.\n"
f"Protocolo: {protocolo}\n"
f"Contrato: {contrato}\n"
f"Placa: {placa}\n"
f"Auto de infracao: {auto_infracao}\n"
f"Valor: {valor}\n"
f"Data da infracao: {data_infracao}\n"
f"Vencimento: {vencimento}"
)
if tool_name in {
"limpar_contexto_conversa",
"continuar_proximo_pedido",

2
app/services/tools/handlers.py
Unescape Escape View File

@ -7,7 +7,6 @@ from app.services.domain.rental_service import (
abrir_locacao_aluguel,
consultar_frota_aluguel,
registrar_devolucao_aluguel,
registrar_multa_aluguel,
registrar_pagamento_aluguel,
)
from app.services.domain.review_service import (
@ -39,7 +38,6 @@ __all__ = [
"listar_pedidos",
"abrir_locacao_aluguel",
"registrar_devolucao_aluguel",
"registrar_multa_aluguel",
"registrar_pagamento_aluguel",
"realizar_pedido",
"validar_cliente_venda",

2
app/services/tools/tool_registry.py
Unescape Escape View File

@ -18,7 +18,6 @@ from app.services.tools.handlers import (
consultar_frota_aluguel,
abrir_locacao_aluguel,
registrar_devolucao_aluguel,
registrar_multa_aluguel,
registrar_pagamento_aluguel,
realizar_pedido,
validar_cliente_venda,
@ -40,7 +39,6 @@ HANDLERS: Dict[str, Callable] = {
"abrir_locacao_aluguel": abrir_locacao_aluguel,
"registrar_devolucao_aluguel": registrar_devolucao_aluguel,
"registrar_pagamento_aluguel": registrar_pagamento_aluguel,
"registrar_multa_aluguel": registrar_multa_aluguel,
}

52
tests/test_rental_service.py
Unescape Escape View File

@ -12,7 +12,7 @@ from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.db.mock_database import MockBase
from app.db.mock_models import RentalContract, RentalFine, RentalPayment, RentalVehicle
from app.db.mock_models import RentalContract, RentalPayment, RentalVehicle
from app.services.domain import rental_service
@ -434,55 +434,5 @@ class RentalServiceTests(unittest.IsolatedAsyncioTestCase):
finally:
db.close()
async def test_registrar_multa_aluguel_persiste_registro(self):
SessionLocal = self._build_session_local()
with patch("app.services.domain.rental_service.SessionMockLocal", SessionLocal):
result = await rental_service.registrar_multa_aluguel(
placa="abc1d23",
auto_infracao="A123456",
valor=293.47,
data_infracao="17/03/2026",
vencimento="10/04/2026",
orgao_emissor="DETRAN-SP",
user_id=11,
)
db = SessionLocal()
try:
stored = db.query(RentalFine).one()
self.assertEqual(stored.placa, "ABC1D23")
self.assertEqual(stored.auto_infracao, "A123456")
self.assertEqual(result["status"], "registrada")
finally:
db.close()
async def test_registrar_multa_aluguel_vincula_contrato_ativo_pela_placa(self):
SessionLocal = self._build_session_local()
db = SessionLocal()
try:
vehicle = self._create_rental_vehicle(db, placa="ABC1D23", status="alugado")
contract = self._create_rental_contract(db, vehicle, user_id=11)
finally:
db.close()
with patch("app.services.domain.rental_service.SessionMockLocal", SessionLocal):
result = await rental_service.registrar_multa_aluguel(
placa="ABC1D23",
auto_infracao="A123456",
valor=293.47,
user_id=11,
)
db = SessionLocal()
try:
stored = db.query(RentalFine).one()
self.assertEqual(stored.rental_contract_id, contract.id)
self.assertEqual(stored.contrato_numero, contract.contrato_numero)
self.assertEqual(result["contrato_numero"], contract.contrato_numero)
finally:
db.close()
if __name__ == "__main__":
unittest.main()

45
tests/test_runtime_bootstrap.py
Unescape Escape View File

@ -1,10 +1,12 @@
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from app import main as main_module
from app.core.settings import Settings
from app.db import bootstrap as bootstrap_module
from app.db import init_db as init_db_module
from app.db import tool_seed as tool_seed_module
class SettingsParsingTests(unittest.TestCase):
@ -30,12 +32,14 @@ class SettingsParsingTests(unittest.TestCase):
class BootstrapRuntimeTests(unittest.TestCase):
@patch.object(bootstrap_module, "seed_tools")
@patch.object(bootstrap_module, "seed_mock_data")
@patch.object(bootstrap_module, "_ensure_mock_schema_evolution")
@patch.object(bootstrap_module.MockBase.metadata, "create_all")
@patch.object(bootstrap_module.Base.metadata, "create_all")
def test_bootstrap_databases_respects_seed_flags(
self,
tools_create_all,
mock_create_all,
ensure_mock_schema_evolution,
seed_mock_data,
seed_tools,
):
@ -53,12 +57,14 @@ class BootstrapRuntimeTests(unittest.TestCase):
@patch.object(bootstrap_module, "seed_tools")
@patch.object(bootstrap_module, "seed_mock_data")
@patch.object(bootstrap_module, "_ensure_mock_schema_evolution")
@patch.object(bootstrap_module.MockBase.metadata, "create_all")
@patch.object(bootstrap_module.Base.metadata, "create_all", side_effect=RuntimeError("tools db down"))
def test_bootstrap_databases_raises_when_any_backend_fails(
self,
tools_create_all,
mock_create_all,
ensure_mock_schema_evolution,
seed_mock_data,
seed_tools,
):
@ -77,6 +83,45 @@ class BootstrapRuntimeTests(unittest.TestCase):
bootstrap_databases.assert_called_once_with()
class ToolSeedTests(unittest.TestCase):
def test_seed_tools_remove_tool_legada_de_multa(self):
class FakeToolRepo:
def __init__(self, tools):
self.tools = list(tools)
self.deleted_ids = []
def get_all(self):
return list(self.tools)
def delete(self, tool_id):
self.deleted_ids.append(tool_id)
self.tools = [tool for tool in self.tools if tool.id != tool_id]
def update_by_name(self, **kwargs):
return kwargs
def create(self, **kwargs):
return kwargs
repo = FakeToolRepo(
[
SimpleNamespace(id=1, name="registrar_multa_aluguel"),
SimpleNamespace(id=2, name="consultar_estoque"),
]
)
fake_db = SimpleNamespace(close=lambda: None)
with patch.object(tool_seed_module, "SessionLocal", return_value=fake_db), patch.object(
tool_seed_module,
"ToolRepository",
return_value=repo,
):
tool_seed_module.seed_tools()
self.assertEqual(repo.deleted_ids, [1])
self.assertFalse(any(tool.name == "registrar_multa_aluguel" for tool in repo.get_all()))
class HttpStartupTests(unittest.IsolatedAsyncioTestCase):
async def test_startup_event_warms_llm_without_running_bootstrap(self):
with patch("app.main.LLMService") as llm_cls, patch(

6
tests/test_telegram_multimodal.py
Unescape Escape View File

@ -100,13 +100,13 @@ class TelegramMultimodalTests(unittest.IsolatedAsyncioTestCase):
) as orchestrator_cls, patch.object(
service,
"_build_orchestration_message_from_image",
AsyncMock(return_value="[imagem recebida no telegram]\nDados extraidos da imagem: Registrar multa de aluguel: placa ABC1D23; valor 293,47; auto_infracao A123456."),
AsyncMock(return_value="[imagem recebida no telegram]\nDados extraidos da imagem: Registrar pagamento de aluguel: placa ABC1D23; valor 293,47; data_pagamento 17/03/2026 14:30; identificador_comprovante NSU123."),
):
user_service_cls.return_value.get_or_create.return_value = SimpleNamespace(id=7)
orchestrator_cls.return_value.handle_message = AsyncMock(return_value="ok")
answer = await service._process_message(
text="segue a multa",
text="segue o comprovante",
sender={"id": 99, "first_name": "Vitor"},
chat_id=99,
image_attachments=[{"mime_type": "image/jpeg", "data": b"123"}],
@ -115,7 +115,7 @@ class TelegramMultimodalTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(answer, "ok")
orchestrator_cls.return_value.handle_message.assert_awaited_once()
kwargs = orchestrator_cls.return_value.handle_message.await_args.kwargs
self.assertIn("Registrar multa de aluguel", kwargs["message"])
self.assertIn("Registrar pagamento de aluguel", kwargs["message"])
self.assertEqual(kwargs["user_id"], 7)
async def test_process_message_returns_direct_failure_for_unreadable_image(self):

Write Preview
Loading…
Cancel
Save