import os import unittest from datetime import datetime from types import SimpleNamespace from unittest.mock import patch os.environ.setdefault("DEBUG", "false") from fastapi import HTTPException from app.db.mock_models import Vehicle from app.services.domain import order_service class LockingQuery: def __init__(self, result): self.result = result self.with_for_update_called = False def filter(self, *args, **kwargs): return self def with_for_update(self): self.with_for_update_called = True return self def first(self): return self.result class LockingSession: def __init__(self, result): self.query_instance = LockingQuery(result) def query(self, model): return self.query_instance class FakeSession: def __init__(self, vehicle=None, user=None): self.vehicle = vehicle self.user = user self.added = [] self.committed = False self.closed = False self.rolled_back = False self.refreshed = [] def query(self, model): if model is order_service.Vehicle: return LockingQuery(self.vehicle) if model is order_service.User: return LockingQuery(self.user) raise AssertionError(f"unexpected model query: {model}") def execute(self, statement, params=None): sql_text = str(statement) if "GET_LOCK" in sql_text: return SimpleNamespace(scalar=lambda: 1) if "RELEASE_LOCK" in sql_text: return SimpleNamespace(scalar=lambda: 1) raise AssertionError(f"unexpected execute call: {sql_text}") def add(self, item): self.added.append(item) def commit(self): self.committed = True def rollback(self): self.rolled_back = True def refresh(self, item): self.refreshed.append(item) def close(self): self.closed = True class OrderServiceReservationTests(unittest.IsolatedAsyncioTestCase): def test_get_vehicle_for_update_uses_row_lock(self): vehicle = Vehicle(id=8, modelo="Toyota Corolla 2024", categoria="suv", preco=76087.0) session = LockingSession(vehicle) locked_vehicle = order_service._get_vehicle_for_update(session, 8) self.assertIs(locked_vehicle, vehicle) self.assertTrue(session.query_instance.with_for_update_called) async def test_realizar_pedido_raises_conflict_when_locked_vehicle_is_already_reserved(self): vehicle = Vehicle(id=8, modelo="Toyota Corolla 2024", categoria="suv", preco=76087.0) session = FakeSession(vehicle=vehicle) existing_order = SimpleNamespace(numero_pedido="PED-RESERVA-001", status="Ativo") async def fake_hydrate_mock_customer_from_cpf(cpf: str, user_id: int | None = None): return {"cpf": cpf, "user_id": user_id} async def fake_validar_cliente_venda(cpf: str, valor_veiculo: float): return {"aprovado": True} with patch.object(order_service, "SessionMockLocal", return_value=session), patch.object( order_service, "hydrate_mock_customer_from_cpf", new=fake_hydrate_mock_customer_from_cpf, ), patch.object( order_service, "validar_cliente_venda", new=fake_validar_cliente_venda, ), patch.object(order_service, "_get_vehicle_for_update", return_value=vehicle) as locked_vehicle, patch.object( order_service, "_get_active_order_for_vehicle", return_value=existing_order, ): with self.assertRaises(HTTPException) as ctx: await order_service.realizar_pedido(cpf="123.456.789-09", vehicle_id=8) self.assertEqual(ctx.exception.status_code, 409) self.assertEqual(ctx.exception.detail["code"], "vehicle_already_reserved") locked_vehicle.assert_called_once_with(session, 8) self.assertEqual(session.added, []) self.assertTrue(session.closed) async def test_realizar_pedido_rejects_cpf_linked_to_other_user(self): vehicle = Vehicle(id=8, modelo="Toyota Corolla 2024", categoria="suv", preco=76087.0) session = FakeSession(vehicle=vehicle) async def fake_hydrate_mock_customer_from_cpf(cpf: str, user_id: int | None = None): raise ValueError("cpf_already_linked") async def fake_validar_cliente_venda(cpf: str, valor_veiculo: float): raise AssertionError("nao deveria consultar credito quando o CPF ja pertence a outro usuario") with patch.object(order_service, "SessionMockLocal", return_value=session), patch.object( order_service, "hydrate_mock_customer_from_cpf", new=fake_hydrate_mock_customer_from_cpf, ), patch.object( order_service, "validar_cliente_venda", new=fake_validar_cliente_venda, ): with self.assertRaises(HTTPException) as ctx: await order_service.realizar_pedido(cpf="123.456.789-09", vehicle_id=8, user_id=99) self.assertEqual(ctx.exception.status_code, 409) self.assertEqual(ctx.exception.detail["code"], "cpf_already_linked") self.assertTrue(session.closed) async def test_realizar_pedido_uses_locked_vehicle_before_persisting_order(self): vehicle = Vehicle(id=8, modelo="Toyota Corolla 2024", categoria="suv", preco=76087.0) session = FakeSession(vehicle=vehicle) fake_uuid = SimpleNamespace(hex="abc123def456") fixed_now = datetime(2026, 3, 16, 17, 30, 0) async def fake_hydrate_mock_customer_from_cpf(cpf: str, user_id: int | None = None): return {"cpf": cpf, "user_id": user_id} async def fake_validar_cliente_venda(cpf: str, valor_veiculo: float): return {"aprovado": True} with patch.object(order_service, "SessionMockLocal", return_value=session), patch.object( order_service, "hydrate_mock_customer_from_cpf", new=fake_hydrate_mock_customer_from_cpf, ), patch.object( order_service, "validar_cliente_venda", new=fake_validar_cliente_venda, ), patch.object(order_service, "_get_vehicle_for_update", return_value=vehicle) as locked_vehicle, patch.object( order_service, "_get_active_order_for_vehicle", return_value=None, ), patch.object(order_service, "uuid4", return_value=fake_uuid), patch.object(order_service, "utc_now", return_value=fixed_now): result = await order_service.realizar_pedido(cpf="123.456.789-09", vehicle_id=8) locked_vehicle.assert_called_once_with(session, 8) self.assertTrue(session.committed) self.assertEqual(len(session.added), 1) created_order = session.added[0] self.assertEqual(created_order.vehicle_id, 8) self.assertEqual(created_order.cpf, "12345678909") self.assertEqual(result["numero_pedido"], "PED-20260316173000-ABC123") self.assertEqual(result["vehicle_id"], 8) self.assertEqual(result["status_veiculo"], "Reservado") self.assertTrue(session.closed) if __name__ == "__main__": unittest.main()