sp-rand: better test data representation.
Change-Id: I6706f259f23c18493979b7193627d574ba4afa66
diff --git a/tools/sp-rand/unit_tests/test_cleanup.py b/tools/sp-rand/unit_tests/test_cleanup.py
index c54c53e..c3efb2d 100644
--- a/tools/sp-rand/unit_tests/test_cleanup.py
+++ b/tools/sp-rand/unit_tests/test_cleanup.py
@@ -5,7 +5,7 @@
import pathlib
import tempfile
-from typing import DefaultDict, Dict, List, NamedTuple, Tuple # noqa: H301
+from typing import DefaultDict, Dict, List, NamedTuple # noqa: H301
import pytest
@@ -15,37 +15,58 @@
from sp_rand import defs as r_defs
-T_ATTACHED = {
- False: ["something-else", "testvol-42"],
- True: ["testvol-9000", "another.snapshot", "testvol-616", "some-testvol-12"],
-}
+class VolSnapData(NamedTuple):
+ """Test data: what volumes and snapshots should be shown as attached."""
+
+ volumes: List[str]
+ snapshots: List[str]
+
+
+class ExportsData(NamedTuple):
+ """Test data: which volumes are exported to an initiator."""
+
+ portal_group: str
+ volume_name: str
+
+
+class TargetData(NamedTuple):
+ """Test data: an iSCSI target record for a StorPool volume."""
+
+ name: str
+ volume: str
+
+
+T_ATTACHED = VolSnapData(
+ volumes=["something-else", "testvol-42"],
+ snapshots=["testvol-9000", "another.snapshot", "testvol-616", "some-testvol-12"],
+)
T_EXPORTED = {
"beleriand": [
- ("neighbors", "testvol-451"),
- ("ours", "something-else"),
- ("ours", "testvol-616"),
+ ExportsData(portal_group="neighbors", volume_name="testvol-451"),
+ ExportsData(portal_group="ours", volume_name="something-else"),
+ ExportsData(portal_group="ours", volume_name="testvol-616"),
],
- "gondor": [("ours", "testvol-12")],
+ "gondor": [ExportsData(portal_group="ours", volume_name="testvol-12")],
}
T_TARGETS = {
- 1: {"name": "ouriqn:target-something-else", "volume": "something-else"},
- 2: {"name": "ouriqn:target-testvol-12", "volume": "testvol-12"},
- 3: {"name": "ouriqn:target-testvol-616", "volume": "testvol-616"},
- 4: {"name": "neiqn:whee-testvol-451", "volume": "testvol-451"},
+ 1: TargetData(name="ouriqn:target-something-else", volume="something-else"),
+ 2: TargetData(name="ouriqn:target-testvol-12", volume="testvol-12"),
+ 3: TargetData(name="ouriqn:target-testvol-616", volume="testvol-616"),
+ 4: TargetData(name="neiqn:whee-testvol-451", volume="testvol-451"),
}
-T_VOLS = {
- False: ["something-else", "testvol-42", "testvol-451"],
- True: [
+T_VOLS = VolSnapData(
+ volumes=["something-else", "testvol-42", "testvol-451"],
+ snapshots=[
"another.snapshot",
"testvol-616",
"testvol-9000",
"some-testvol-12",
"yet.another.thing",
],
-}
+)
def test_parse_file() -> None:
@@ -120,24 +141,28 @@
class MockApi:
"""Mock the StorPool API."""
- attached: Dict[bool, List[str]]
- exported: Dict[str, List[Tuple[str, str]]]
+ attached: VolSnapData
+ exported: Dict[str, List[ExportsData]]
invoked: DefaultDict[str, int]
- targets: Dict[int, Dict[str, str]]
- vols: Dict[bool, List[str]]
+ targets: Dict[int, TargetData]
+ vols: VolSnapData
def __init__(
self,
- attached: Dict[bool, List[str]],
- exported: Dict[str, List[Tuple[str, str]]],
- targets: Dict[int, Dict[str, str]],
- vols: Dict[bool, List[str]],
+ attached: VolSnapData,
+ exported: Dict[str, List[ExportsData]],
+ targets: Dict[int, TargetData],
+ vols: VolSnapData,
) -> None:
"""Store deep copies of the passed definitions."""
- self.attached = {key: list(value) for key, value in attached.items()}
+ self.attached = VolSnapData(
+ volumes=list(attached.volumes), snapshots=list(attached.snapshots)
+ )
self.exported = {key: list(value) for key, value in exported.items()}
- self.targets = {key: dict(value) for key, value in targets.items()}
- self.vols = {key: list(value) for key, value in vols.items()}
+ self.targets = dict(targets)
+ self.vols = VolSnapData(
+ volumes=list(vols.volumes), snapshots=list(vols.snapshots)
+ )
self.invoked = collections.defaultdict(lambda: 0)
# pylint: disable=invalid-name
@@ -149,18 +174,18 @@
sptypes.AttachmentDesc(
volume=name, snapshot=False, client=11, rights="rw", pos=0
)
- for name in self.attached[False]
+ for name in self.attached.volumes
] + [
sptypes.AttachmentDesc(
volume=name, snapshot=True, client=11, rights="ro", pos=0
)
- for name in self.attached[True]
+ for name in self.attached.snapshots
]
def iSCSIConfig(self) -> sptypes.iSCSIConfig:
"""Return the current view of the iSCSI configuration."""
self.invoked["iSCSIConfig"] += 1
- target_dict = {tgt["volume"]: tgt["name"] for tgt in self.targets.values()}
+ target_dict = {tgt.volume: tgt.name for tgt in self.targets.values()}
return sptypes.iSCSIConfig(
iscsi=sptypes.iSCSIConfigData(
@@ -173,7 +198,8 @@
nets=[],
exports=[
sptypes.iSCSIExport(
- portalGroup=exp[0], target=target_dict[exp[1]]
+ portalGroup=exp.portal_group,
+ target=target_dict[exp.volume_name],
)
for exp in explist
],
@@ -184,8 +210,8 @@
targets={
tid: sptypes.iSCSITarget(
currentControllerId=65535,
- name=tgt["name"],
- volume=tgt["volume"],
+ name=tgt.name,
+ volume=tgt.volume,
)
for tid, tgt in self.targets.items()
},
@@ -201,14 +227,16 @@
tgtdel = cmd.deleteTarget
if expdel is not None:
initiator = self.exported[expdel.initiator]
- data = (expdel.portalGroup, expdel.volumeName)
+ data = ExportsData(
+ portal_group=expdel.portalGroup, volume_name=expdel.volumeName
+ )
assert data in initiator
initiator.remove(data)
elif tgtdel is not None:
found = [
tid
for tid, tgt in self.targets.items()
- if tgt["volume"] == tgtdel.volumeName
+ if tgt.volume == tgtdel.volumeName
]
assert len(found) == 1
del self.targets[found[0]]
@@ -222,38 +250,35 @@
for item in req.reassign:
if isinstance(item, sptypes.VolumeReassignDesc):
assert item.volume.startswith("testvol-")
- self.attached[False].remove(item.volume)
+ self.attached.volumes.remove(item.volume)
else:
assert isinstance(item, sptypes.SnapshotReassignDesc)
assert item.snapshot.startswith("testvol-")
- self.attached[True].remove(item.snapshot)
+ self.attached.snapshots.remove(item.snapshot)
assert item.detach == [11]
def volumesList(self) -> List[MockVolumeSummary]:
"""Return the current view of the available volumes."""
self.invoked["volumesList"] += 1
- return [MockVolumeSummary(name=name) for name in self.vols[False]]
+ return [MockVolumeSummary(name=name) for name in self.vols.volumes]
def snapshotsList(self) -> List[MockVolumeSummary]:
"""Return the current view of the available snapshots."""
self.invoked["snapshotsList"] += 1
- return [MockVolumeSummary(name=name) for name in self.vols[True]]
-
- def _volumeDelete(self, name: str, snapshot: bool) -> None:
- """Remove a snapshot or a volume from our view."""
- assert name not in self.attached[snapshot]
- self.vols[snapshot].remove(name)
+ return [MockVolumeSummary(name=name) for name in self.vols.snapshots]
def volumeDelete(self, name: str) -> None:
"""Remove a volume from our view."""
self.invoked["volumeDelete"] += 1
- self._volumeDelete(name, False)
+ assert name not in self.attached.volumes
+ self.vols.volumes.remove(name)
def snapshotDelete(self, name: str) -> None:
"""Remove a snapshot from our view."""
self.invoked["snapshotDelete"] += 1
- self._volumeDelete(name, True)
+ assert name not in self.attached.snapshots
+ self.vols.snapshots.remove(name)
def test_remove_volumes() -> None:
@@ -265,23 +290,31 @@
r_cleanup.remove_volumes(cfg, mock_api, "testvol-")
- assert mock_api.attached == {
- key: [name for name in value if not name.startswith("testvol-")]
- for key, value in T_ATTACHED.items()
- }
+ assert mock_api.attached == VolSnapData(
+ volumes=[
+ name for name in T_ATTACHED.volumes if not name.startswith("testvol-")
+ ],
+ snapshots=[
+ name for name in T_ATTACHED.snapshots if not name.startswith("testvol-")
+ ],
+ )
assert mock_api.exported == {
- initiator: [item for item in exports if not item[1].startswith("testvol-")]
+ initiator: [
+ item for item in exports if not item.volume_name.startswith("testvol-")
+ ]
for initiator, exports in T_EXPORTED.items()
}
assert mock_api.targets == {
tid: tgt
for tid, tgt in T_TARGETS.items()
- if not tgt["volume"].startswith("testvol-")
+ if not tgt.volume.startswith("testvol-")
}
- assert mock_api.vols == {
- key: [name for name in value if not name.startswith("testvol-")]
- for key, value in T_VOLS.items()
- }
+ assert mock_api.vols == VolSnapData(
+ volumes=[name for name in T_VOLS.volumes if not name.startswith("testvol-")],
+ snapshots=[
+ name for name in T_VOLS.snapshots if not name.startswith("testvol-")
+ ],
+ )
assert mock_api.invoked == {
"attachmentsList": 1,