Coverage for volexport/api_volume.py: 87%

1import datetime 

2from fastapi import APIRouter, HTTPException 

3from pydantic import BaseModel 

4from .config import config 

5from .lvm2 import LV, VG 

6 

7router = APIRouter() 

8 

9 

10class VolumeCreateRequest(BaseModel): 

11 name: str 

12 size: int 

13 

14 

15class VolumeCreateResponse(BaseModel): 

16 name: str 

17 size: int 

18 device: str 

19 

20 

21class VolumeReadResponse(BaseModel): 

22 name: str 

23 created: datetime.datetime 

24 size: int 

25 used: int 

26 

27 

28class VolumeUpdateRequest(BaseModel): 

29 size: int | None = None 

30 readonly: bool | None = None 

31 

32 

33class PoolStats(BaseModel): 

34 total: int 

35 used: int 

36 free: int 

37 volumes: int 

38 

39 

40@router.get("/volume") 

41def list_volume(): 

42 return [VolumeReadResponse.model_validate(x) for x in LV(config.VG).volume_list()] 

43 

44 

45@router.post("/volume") 

46def create_volume(arg: VolumeCreateRequest): 

47 return VolumeCreateResponse.model_validate(LV(config.VG, arg.name).create(size=arg.size)) 

48 

49 

50@router.get("/volume/{name}") 

51def read_volume(name): 

52 res = LV(config.VG, name).volume_read() 

53 if res is None: 

54 raise HTTPException(status_code=404, detail="volume not found") 

55 return VolumeReadResponse.model_validate(res) 

56 

57 

58@router.delete("/volume/{name}") 

59def delete_volume(name) -> dict: 

60 LV(config.VG, name).delete() 

61 return {} 

62 

63 

64@router.post("/volume/{name}") 

65def update_volume(name, arg: VolumeUpdateRequest) -> VolumeReadResponse: 

66 if arg.readonly is not None: 

67 LV(config.VG, name).read_only(arg.readonly) 

68 if arg.size is not None: 

69 LV(config.VG, name).resize(arg.size) 

70 return VolumeReadResponse.model_validate(LV(config.VG, name).volume_read()) 

71 

72 

73@router.get("/stats/volume") 

74def stats_volume() -> PoolStats: 

75 info = VG(config.VG).get() 

76 if info is None: 

77 raise HTTPException(status_code=404, detail="pool not found") 

78 vols = info.get("Cur LV", 0) 

79 pesize = int(info["PE Size"].removesuffix(" B")) 

80 total_pe = int(info["Total PE"]) 

81 alloc_pe = int(info["Alloc PE / Size"].split()[0]) 

82 free_pe = int(info["Free PE / Size"].split()[0]) 

83 return PoolStats(total=pesize * total_pe, used=pesize * alloc_pe, free=pesize * free_pe, volumes=vols)