Coverage for volexport/api_export.py: 95%

1from fastapi import APIRouter, HTTPException 

2from pydantic import BaseModel 

3from .config import config 

4from .tgtd import Tgtd 

5from .lvm2 import LV 

6 

7router = APIRouter() 

8 

9 

10class ExportRequest(BaseModel): 

11 volname: str 

12 acl: list[str] 

13 readonly: bool = False 

14 

15 

16class ExportResponse(BaseModel): 

17 protocol: str 

18 addresses: list[str] 

19 targetname: str 

20 tid: int 

21 user: str 

22 passwd: str 

23 lun: int 

24 acl: list[str] 

25 

26 

27class ClientInfo(BaseModel): 

28 address: list[str] 

29 initiator: str 

30 

31 

32class ExportReadResponse(BaseModel): 

33 protocol: str 

34 connected: list[ClientInfo] 

35 targetname: str 

36 tid: int 

37 volumes: list[str] 

38 users: list[str] 

39 acl: list[str] 

40 

41 

42class ExportStats(BaseModel): 

43 targets: int 

44 clients: int 

45 volumes: int 

46 

47 

48def _fixpath(data: dict) -> dict: 

49 if "volumes" in data: 49 ↛ 51line 49 didn't jump to line 51 because the condition on line 49 was always true

50 data["volumes"] = [LV(config.VG).volume_path2vol(x) for x in data["volumes"]] 

51 return data 

52 

53 

54@router.get("/export") 

55def list_export() -> list[ExportReadResponse]: 

56 return [ExportReadResponse.model_validate(_fixpath(x)) for x in Tgtd().export_list()] 

57 

58 

59@router.post("/export") 

60def create_export(arg: ExportRequest) -> ExportResponse: 

61 filename = LV(config.VG, arg.volname).volume_vol2path() 

62 return ExportResponse.model_validate(Tgtd().export_volume(filename=filename, acl=arg.acl, readonly=arg.readonly)) 

63 

64 

65@router.get("/export/{name}") 

66def read_export(name) -> ExportReadResponse: 

67 res = [_fixpath(x) for x in Tgtd().export_list() if x["targetname"] == name or x["tid"] == name] 

68 if len(res) == 0: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 raise HTTPException(status_code=404, detail="export not found") 

70 return ExportReadResponse.model_validate(res[0]) 

71 

72 

73@router.delete("/export/{name}") 

74def delete_export(name, force: bool = False): 

75 return Tgtd().unexport_volume(targetname=name, force=force) 

76 

77 

78@router.get("/address") 

79def get_address() -> list[str]: 

80 return Tgtd().myaddress() 

81 

82 

83@router.get("/stats/export") 

84def stats_export() -> ExportStats: 

85 info = Tgtd().export_list() 

86 return ExportStats( 

87 targets=len(info), 

88 clients=sum([len(x["connected"]) for x in info]), 

89 volumes=sum([len(x["volumes"]) for x in info]), 

90 )