Coverage for backend \ app \ Inventario \ services \ inventarioService.py: 100.00%

1from app.Inventario.repositories.inventarioRepository import InventarioRepository 

2from app.configuracionGeneral.schemasGenerales import respuestaApi 

3from fastapi import HTTPException 

4from app.Inventario.schemas.inventarioSchemas import * 

5 

6class InventarioService: 

7 def __init__(self, dbSession): 

8 self.dbSession = dbSession 

9 self.inventarioRepository = InventarioRepository(dbSession) 

10 

11 def listarInventarios(self): 

12 invs = self.inventarioRepository.listarInventarios() 

13 if invs is None: 

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

15 data = [InventarioRespuestaSchema.from_orm(i) for i in invs] 

16 return respuestaApi(success=True, message="Inventarios encontrados", data=data) 

17 

18 def obtenerPorId(self, idInventario: int): 

19 inv = self.inventarioRepository.obtenerPorId(idInventario) 

20 if inv is None: 

21 raise HTTPException(status_code=404, detail="Inventario no encontrado") 

22 data = InventarioRespuestaSchema.from_orm(inv) 

23 return respuestaApi(success=True, message="Inventario encontrado", data=data) 

24 

25 def obtenerPorProducto(self, idProducto: int): 

26 inv = self.inventarioRepository.obtenerPorProducto(idProducto) 

27 if inv is None: 

28 raise HTTPException(status_code=404, detail="Inventario no encontrado") 

29 data = InventarioRespuestaSchema.from_orm(inv) 

30 return respuestaApi(success=True, message="Inventario encontrado", data=data) 

31 

32 def crearInventario(self, inventario: InventarioCrearSchema): 

33 # si ya existe inventario para el producto -> devolver existente (idempotente) 

34 existente = self.inventarioRepository.obtenerPorProducto(inventario.idProducto) 

35 if existente: 

36 data = InventarioRespuestaSchema.from_orm(existente) 

37 return respuestaApi(success=True, message="Inventario existente", data=data), 200 

38 creado = self.inventarioRepository.crearInventario(inventario) 

39 if isinstance(creado, dict) and "error" in creado: 

40 missing = creado["error"] 

41 mensajes = [] 

42 if "producto" in missing: 

43 mensajes.append("Producto no existe") 

44 if "producto_inactiva" in missing: 

45 mensajes.append("Producto deshabilitado") 

46 raise HTTPException(status_code=400, detail="; ".join(mensajes)) 

47 creado = InventarioRespuestaSchema.from_orm(creado) 

48 return respuestaApi(success=True, message="Inventario creado", data=creado), 201 

49 

50 def modificarInventario(self, idInventario: int, inventario: InventarioActualizarSchema, usuario: dict): 

51 # restricciones por rol: Bodeguero no puede modificar cantidadDisponible 

52 datos = inventario.model_dump(exclude_unset=True) 

53 rol = usuario.get("rol") 

54 if rol == "Bodeguero" and "cantidadDisponible" in datos: 

55 raise HTTPException(status_code=403, detail="No tienes permiso para modificar la cantidad disponible") 

56 actualizado = self.inventarioRepository.modificarInventario(idInventario, inventario) 

57 if actualizado is None: 

58 raise HTTPException(status_code=404, detail="Inventario no encontrado") 

59 actualizado = InventarioRespuestaSchema.from_orm(actualizado) 

60 return respuestaApi(success=True, message="Inventario actualizado", data=actualizado) 

61 

62 def deshabilitarInventario(self, idInventario: int): 

63 des = self.inventarioRepository.deshabilitarInventario(idInventario) 

64 if des is None: 

65 raise HTTPException(status_code=404, detail="Inventario no encontrado") 

66 if des is False: 

67 raise HTTPException(status_code=400, detail="No se puede deshabilitar inventario mientras el producto esté habilitado") 

68 des = InventarioRespuestaSchema.from_orm(des) 

69 return respuestaApi(success=True, message="Inventario deshabilitado", data=des)