Coverage for backend \ app \ Caja \ services \ cajaService.py: 80.58%

1from app.Caja.repositories.cajaRepository import CajaRepository 

2from app.configuracionGeneral.schemasGenerales import respuestaApi 

3from app.configuracionGeneral.seguridadJWT import identificarUsuarioString 

4from app.Usuarios.repositories.usuarioRepository import UsuarioRepository 

5from fastapi import HTTPException 

6 

7from app.Caja.schemas.cajaSchemas import * 

8 

9class CajaService: 

10 def __init__(self, dbSession): 

11 self.dbSession = dbSession 

12 self.repo = CajaRepository(dbSession) 

13 

14 def _attach_usuario(self, caja_obj): 

15 """Asegura que la relación `usuario` esté poblada en la entidad `CajaHistorial` para serializarla.""" 

16 if not getattr(caja_obj, "usuario", None): 

17 usuario_entity = UsuarioRepository(self.dbSession).obtenerUsuarioPorId(caja_obj.idUsuarioCaja) 

18 if usuario_entity: 

19 caja_obj.usuario = usuario_entity 

20 return caja_obj 

21 

22 def crearCajaHistorial(self, cajaCrear: CajaCrearSchema, usuario: dict): 

23 idUsuario = usuario.get("idUsuario") 

24 actor = identificarUsuarioString(usuario) 

25 # montoInicial es un número provisto por el cliente 

26 monto_calc = round(float(cajaCrear.montoInicial), 2) 

27 resultado = self.repo.crearCajaHistorial(monto_calc, idUsuario) 

28 if isinstance(resultado, dict) and resultado.get("error"): 

29 if resultado.get("error") == "caja_abierta_existente": 

30 raise HTTPException(status_code=400, detail=f"Ya existe una caja abierta para el usuario {actor}") 

31 if resultado.get("error") == "caja_ya_abierta_hoy": 

32 raise HTTPException(status_code=400, detail=f"Ya existe una caja para hoy para el usuario {actor}") 

33 # Añadir detalle con identificación del actor 

34 resultado.detalle = f"Apertura por {actor}; montoInicialDeclarado: {monto_calc}" 

35 self.dbSession.add(resultado) 

36 self.dbSession.commit() 

37 self.dbSession.refresh(resultado) 

38 # Adjuntar usuario para serializar en la respuesta 

39 resultado = self._attach_usuario(resultado) 

40 data = CajaHistorialRespuestaSchema.from_orm(resultado) 

41 mensaje = f"Caja abierta por {actor} (monto inicial: {data.montoInicialDeclarado})" 

42 return respuestaApi(success=True, message=mensaje, data=data) 

43 

44 def cerrarCaja(self, idCaja: int, cerrar: CajaCerrarSchema, usuario: dict): 

45 # permisos: solo Cajero (propietario) o Administrador pueden cerrar 

46 rol = usuario.get("rol") 

47 idUsuario = usuario.get("idUsuario") 

48 caja = self.repo.obtenerPorId(idCaja) 

49 if not caja: 

50 raise HTTPException(status_code=404, detail="Caja no encontrada") 

51 # Si la caja ya está cerrada, devolver respuestaApi con success=False, mensaje y la caja en data 

52 if caja.estadoCaja == "CERRADA": 

53 caja = self._attach_usuario(caja) 

54 data = CajaHistorialRespuestaSchema.from_orm(caja) 

55 return respuestaApi(success=False, message="La caja ya está cerrada", data=data) 

56 monto_final = round(float(cerrar.montoFinal), 2) 

57 actor = identificarUsuarioString(usuario) 

58 

59 # Cajero: solo puede cerrar SU caja abierta del día actual 

60 if rol != "Administrador": 

61 if caja.idUsuarioCaja != idUsuario: 

62 raise HTTPException(status_code=403, detail="No tiene permisos para cerrar esta caja") 

63 # Validar que la apertura sea hoy 

64 from datetime import datetime 

65 tz = datetime.now().astimezone() 

66 hoy = datetime.now(tz.tzinfo).date() 

67 if caja.fechaAperturaCaja.date() != hoy: 

68 raise HTTPException(status_code=400, detail=f"Solo puede cerrar cajas abiertas hoy. Usuario: {actor}") 

69 resultado = self.repo.cerrarCaja(idCaja, monto_final, closedBy=actor, admin=False) 

70 else: 

71 # Administrador: puede cerrar cualquier caja por id, enviando solo montoFinal 

72 resultado = self.repo.cerrarCaja(idCaja, monto_final, closedBy=actor, admin=True) 

73 

74 if resultado is None: 

75 raise HTTPException(status_code=404, detail="Caja no encontrada") 

76 if isinstance(resultado, dict) and resultado.get("error") == "cierre_fuera_de_dia": 

77 raise HTTPException(status_code=400, detail="La caja fue abierta en un día distinto") 

78 if isinstance(resultado, dict) and resultado.get("error") == "caja_no_abierta": 

79 raise HTTPException(status_code=400, detail="La caja no está abierta") 

80 if isinstance(resultado, dict) and resultado.get("error") == "caja_ya_cerrada": 

81 caja_obj = resultado.get("caja") 

82 caja_obj = self._attach_usuario(caja_obj) 

83 data = CajaHistorialRespuestaSchema.from_orm(caja_obj) 

84 return respuestaApi(success=False, message="La caja ya está cerrada", data=data) 

85 resultado = self._attach_usuario(resultado) 

86 data = CajaHistorialRespuestaSchema.from_orm(resultado) 

87 mensaje = f"Caja cerrada por {actor} (monto cierre declarado: {data.montoCierreDeclarado})" 

88 return respuestaApi(success=True, message=mensaje, data=data) 

89 

90 

91 

92 def listarCajas(self, usuario: dict): 

93 rol = usuario.get("rol") 

94 idUsuario = usuario.get("idUsuario") 

95 esAdmin = rol == "Administrador" 

96 cajas = self.repo.listarCajas(idUsuario if not esAdmin else None, esAdmin) 

97 if not cajas: 

98 return respuestaApi(success=True, message="No se encontraron cajas", data=[]) 

99 # Adjuntar usuario en cada caja antes de serializar 

100 for c in cajas: 

101 self._attach_usuario(c) 

102 data = [CajaHistorialRespuestaSchema.from_orm(c) for c in cajas] 

103 return respuestaApi(success=True, message="Cajas encontradas", data=data) 

104 

105 def listarCajasHoy(self, usuario: dict): 

106 rol = usuario.get("rol") 

107 idUsuario = usuario.get("idUsuario") 

108 esAdmin = rol == "Administrador" 

109 cajas = self.repo.listarCajasHoy(idUsuario if not esAdmin else None, esAdmin) 

110 if not cajas: 

111 return respuestaApi(success=True, message="No se han abierto cajas hoy", data=[]) 

112 for c in cajas: 

113 self._attach_usuario(c) 

114 data = [CajaHistorialRespuestaSchema.from_orm(c) for c in cajas] 

115 return respuestaApi(success=True, message="Cajas del día encontradas", data=data) 

116 def listarTodasCajas(self, usuario: dict): 

117 rol = usuario.get("rol") 

118 if rol != "Administrador": 

119 raise HTTPException(status_code=403, detail="Solo Administrador puede listar todas las cajas") 

120 cajas = self.repo.listarCajas(None, True) 

121 if not cajas: 

122 return respuestaApi(success=True, message="No se encontraron cajas", data=[]) 

123 for c in cajas: 

124 self._attach_usuario(c) 

125 data = [CajaHistorialRespuestaSchema.from_orm(c) for c in cajas] 

126 return respuestaApi(success=True, message="Cajas encontradas", data=data) 

127 

128 def filtrarCaja(self, idUsuario: int, fecha: date, usuario: dict): 

129 # Solo Administrador puede usar este endpoint 

130 rol = usuario.get("rol") 

131 if rol != "Administrador": 

132 raise HTTPException(status_code=403, detail="Solo Administrador puede filtrar cajas") 

133 resultados = self.repo.filtrarCaja(idUsuario, fecha) 

134 if not resultados: 

135 return respuestaApi(success=True, message="No se encontraron cajas para los parámetros proporcionados", data=[]) 

136 for r in resultados: 

137 self._attach_usuario(r) 

138 data = [CajaHistorialRespuestaSchema.from_orm(r) for r in resultados] 

139 return respuestaApi(success=True, message="Cajas encontradas", data=data) 

140 

141 def reabrirCaja(self, idCaja: int, usuario: dict): 

142 # Solo Administrador puede reabrir cajas cerradas 

143 rol = usuario.get("rol") 

144 idUsuario = usuario.get("idUsuario") 

145 if rol != "Administrador": 

146 raise HTTPException(status_code=403, detail="Solo Administrador puede reabrir cajas") 

147 caja = self.repo.obtenerPorId(idCaja) 

148 if not caja: 

149 raise HTTPException(status_code=404, detail="Caja no encontrada") 

150 if caja.estadoCaja != "CERRADA": 

151 data = CajaHistorialRespuestaSchema.from_orm(caja) 

152 return respuestaApi(success=False, message="La caja no está cerrada", data=data) 

153 reabierta_por = identificarUsuarioString(usuario) 

154 resultado = self.repo.reabrirCaja(idCaja, reabiertaPor=reabierta_por) 

155 if resultado is None: 

156 raise HTTPException(status_code=404, detail="Caja no encontrada") 

157 if isinstance(resultado, dict) and resultado.get("error") == "caja_no_cerrada": 

158 caja_obj = resultado.get("caja") 

159 caja_obj = self._attach_usuario(caja_obj) 

160 data = CajaHistorialRespuestaSchema.from_orm(caja_obj) 

161 return respuestaApi(success=False, message=f"La caja no está cerrada. Usuario: {reabierta_por}", data=data) 

162 resultado = self._attach_usuario(resultado) 

163 data = CajaHistorialRespuestaSchema.from_orm(resultado) 

164 mensaje = f"Caja reabierta por {reabierta_por}" 

165 return respuestaApi(success=True, message=mensaje, data=data)