Coverage for volexport/api_export.py: 96%

1from fastapi import APIRouter, HTTPException, Request 

2from pydantic import BaseModel, Field, SecretStr, field_serializer 

3from .config2 import config2 

4from .tgtd import Tgtd 

5from .lvm2 import LV 

6 

7router = APIRouter() 

8 

9 

10class ExportRequest(BaseModel): 

11 name: str = Field(description="Volume name to export", examples=["volume1"]) 

12 acl: list[str] | None = Field(description="Source IP Addresses to allow access") 

13 readonly: bool = Field(default=False, description="read-only if true", examples=[True, False]) 

14 user: str | None = Field(default=None, description="user name for access. auto-generate if null") 

15 passwd: SecretStr | None = Field(default=None, description="password for access. auto-generate if null") 

16 

17 

18class ExportResponse(BaseModel): 

19 protocol: str = Field(description="access protocol", examples=["iscsi"]) 

20 addresses: list[str] = Field(description="IP addresses of the target") 

21 targetname: str = Field(description="target name", examples=["iqn.2025-08.volexport:abcde"]) 

22 tid: int = Field(description="target ID") 

23 user: str = Field(description="user name for access", examples=["admin"]) 

24 passwd: SecretStr = Field(description="password for access", examples=["password123"]) 

25 lun: int = Field(description="LUN number", examples=[1, 2, 3]) 

26 acl: list[str] = Field(description="Access Control List (ACL) for the export") 

27 

28 @field_serializer("passwd", when_used="json") 

29 def dump_secret(self, v): 

30 return v.get_secret_value() 

31 

32 

33class ClientInfo(BaseModel): 

34 address: list[str] = Field(description="IP addresses of the client") 

35 initiator: str = Field(description="Initiator name", examples=["iqn.2025-08.volimport:client1"]) 

36 

37 

38class ExportReadResponse(BaseModel): 

39 protocol: str = Field(description="access protocol", examples=["iscsi"]) 

40 connected: list[ClientInfo] = Field(description="List of connected clients") 

41 targetname: str = Field(description="target name", examples=["iqn.2025-08.volexport:abcde"]) 

42 tid: int = Field(description="target ID") 

43 volumes: list[str] = Field(description="List of volumes exported", examples=["volume1"]) 

44 users: list[str] = Field(description="List of users with access", examples=["admin", "user1"]) 

45 acl: list[str] = Field(description="Access Control List (ACL) for the export") 

46 

47 

48class ExportStats(BaseModel): 

49 targets: int = Field(description="Number of export targets", examples=[5]) 

50 clients: int = Field(description="Number of connected clients", examples=[10]) 

51 volumes: int = Field(description="Number of volumes exported", examples=[15]) 

52 

53 

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

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

56 data["volumes"] = [LV(config2.VG).volume_path2vol(x) for x in data["volumes"]] 

57 return data 

58 

59 

60@router.get("/export", description="List all exports") 

61def list_export(volume: str | None = None) -> list[ExportReadResponse]: 

62 res = [ExportReadResponse.model_validate(_fixpath(x)) for x in Tgtd().export_list()] 

63 if volume: 

64 res = [x for x in res if volume in x.volumes] 

65 return res 

66 

67 

68@router.post("/export", description="Create a new export") 

69def create_export(req: Request, arg: ExportRequest) -> ExportResponse: 

70 filename = LV(config2.VG, arg.name).volume_vol2path() 

71 if not arg.acl: 

72 assert req.client is not None 

73 arg.acl = [req.client.host] 

74 return ExportResponse.model_validate( 

75 Tgtd().export_volume( 

76 filename=filename, 

77 acl=arg.acl, 

78 readonly=arg.readonly, 

79 user=arg.user, 

80 passwd=arg.passwd.get_secret_value() if arg.passwd else None, 

81 ) 

82 ) 

83 

84 

85@router.get("/export/{name}", description="Read export details by name or TID") 

86def read_export(name) -> ExportReadResponse: 

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

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

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

90 return ExportReadResponse.model_validate(res[0]) 

91 

92 

93@router.delete("/export/{name}", description="Delete an export by name or TID") 

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

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

96 

97 

98@router.get("/address", description="Get addresses of the target") 

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

100 return Tgtd().myaddress() 

101 

102 

103@router.get("/stats/export", description="Get statistics of exports") 

104def stats_export() -> ExportStats: 

105 info = Tgtd().export_list() 

106 return ExportStats( 

107 targets=len(info), 

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

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

110 )