Coverage for backend \ app \ Venta \ repositories \ ventaRepository.py: 29.82%

1from app.Venta.models.ventaModel import Venta 

2from app.Venta.models.detalleVentaModel import DetalleVenta 

3from app.Productos.models.productoModel import Producto 

4from app.Inventario.repositories.inventarioRepository import InventarioRepository 

5from app.Venta.repositories.promocionRepository import PromocionRepository 

6from app.Clientes.repositories.clienteRepository import ClienteRepository 

7from datetime import datetime, timezone, timedelta 

8from sqlalchemy.orm import joinedload 

9 

10class VentaRepository: 

11 def __init__(self, dbSession): 

12 self.dbSession = dbSession 

13 

14 def obtenerPorId(self, idVenta: int): 

15 # Cargar relaciones necesarias para respuestas completas 

16 return self.dbSession.query(Venta).options( 

17 joinedload(Venta.detalles), 

18 joinedload(Venta.usuario), 

19 joinedload(Venta.cliente) 

20 ).filter(Venta.idVenta == idVenta).first() 

21 

22 def listarVentasHoy(self, idUsuario: int = None, esAdmin: bool = False): 

23 tz = timezone(timedelta(hours=-5)) 

24 hoy = datetime.now(tz).date() 

25 inicio = datetime.combine(hoy, datetime.min.time()).astimezone(tz) 

26 fin = datetime.combine(hoy, datetime.max.time()).replace(hour=23, minute=59, second=59, microsecond=0).astimezone(tz) 

27 query = self.dbSession.query(Venta).options(joinedload(Venta.detalles)).filter(Venta.fechaVenta >= inicio, Venta.fechaVenta <= fin) 

28 if not esAdmin and idUsuario: 

29 query = query.filter(Venta.idUsuarioVenta == idUsuario) 

30 return query.all() 

31 

32 def listarTodas(self): 

33 return self.dbSession.query(Venta).options(joinedload(Venta.detalles)).all() 

34 

35 def crearVenta(self, venta: Venta, detalles: list): 

36 self.dbSession.add(venta) 

37 self.dbSession.commit() 

38 self.dbSession.refresh(venta) 

39 for d in detalles: 

40 d.idVenta = venta.idVenta 

41 self.dbSession.add(d) 

42 self.dbSession.commit() 

43 self.dbSession.refresh(venta) 

44 return venta 

45 

46 def sumarVentasEfectivoNoAnuladasPorCaja(self, idCaja: int): 

47 """Devuelve la suma de totalPagar de las ventas en efectivo y no anuladas para una caja dada.""" 

48 from sqlalchemy import func 

49 suma = self.dbSession.query(func.coalesce(func.sum(Venta.totalPagar), 0)).filter( 

50 Venta.idCaja == idCaja, 

51 Venta.metodoPago == "Efectivo", 

52 Venta.estadoVenta != "ANULADA" 

53 ).scalar() 

54 try: 

55 return float(suma or 0.0) 

56 except Exception: 

57 return 0.0 

58 

59 def anularVenta(self, idVenta: int): 

60 venta = self.obtenerPorId(idVenta) 

61 if not venta: 

62 return None 

63 if venta.estadoVenta == "ANULADA": 

64 return {"error": "venta_ya_anulada", "venta": venta} 

65 venta.estadoVenta = "ANULADA" 

66 # revertir inventario 

67 invRepo = InventarioRepository(self.dbSession) 

68 for d in venta.detalles: 

69 inventario = invRepo.obtenerPorProducto(d.idProducto) 

70 if inventario: 

71 inventario.cantidadDisponible = (inventario.cantidadDisponible or 0) + (d.cantidadVendida or 0) 

72 self.dbSession.add(inventario) 

73 self.dbSession.commit() 

74 self.dbSession.refresh(venta) 

75 return venta