Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).

- Relaxed the `fastcrc` requirement from `~= 0.3.0` to `>= 0.3` and the `deprecation`
requirement from `~= 2.1` to `>= 2.1`.
- Renamed `DirectoryParams.dir_file_name` to `requester_listing_file`, together with its
accessors and the `from_strs` and `from_paths` arguments. The field is the file at the filestore
of the requesting entity which receives the directory listing.

# [v0.32.0] 2026-03-03

Expand Down
36 changes: 21 additions & 15 deletions src/spacepackets/cfdp/tlv/msg_to_user.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,8 +234,8 @@ def get_dir_listing_request_params(self) -> DirectoryParams | None:
):
return None
dir_path_lv = CfdpLv.unpack(self.value[5:])
dir_file_name_lv = CfdpLv.unpack(self.value[5 + dir_path_lv.packet_len :])
return DirectoryParams(dir_path_lv, dir_file_name_lv)
listing_file_lv = CfdpLv.unpack(self.value[5 + dir_path_lv.packet_len :])
return DirectoryParams(dir_path_lv, listing_file_lv)

def get_dir_listing_response_params(self) -> tuple[bool, DirectoryParams] | None:
"""
Expand All @@ -256,8 +256,8 @@ def get_dir_listing_response_params(self) -> tuple[bool, DirectoryParams] | None
)
listing_success = ((self.value[5] >> 7) & 0b1) == 0
dir_path_lv = CfdpLv.unpack(self.value[6:])
dir_file_name_lv = CfdpLv.unpack(self.value[6 + dir_path_lv.packet_len :])
return listing_success, DirectoryParams(dir_path_lv, dir_file_name_lv)
listing_file_lv = CfdpLv.unpack(self.value[6 + dir_path_lv.packet_len :])
return listing_success, DirectoryParams(dir_path_lv, listing_file_lv)

def get_dir_listing_options(self) -> DirListingOptions | None:
if (
Expand Down Expand Up @@ -341,16 +341,22 @@ def __init__(self, transaction_id: TransactionId):

@dataclasses.dataclass
class DirectoryParams:
"""Parameters of the directory listing request and response, see CCSDS 727.0-B-5 6.3.3.2.

``requester_listing_file`` is the file path at the filestore local to the requesting entity.
The responder puts the directory listing into this file.
"""

dir_path: CfdpLv
dir_file_name: CfdpLv
requester_listing_file: CfdpLv

@classmethod
def from_strs(cls, dir_path: str, dir_file_name: str) -> DirectoryParams:
return cls(CfdpLv.from_str(dir_path), CfdpLv.from_str(dir_file_name))
def from_strs(cls, dir_path: str, requester_listing_file: str) -> DirectoryParams:
return cls(CfdpLv.from_str(dir_path), CfdpLv.from_str(requester_listing_file))

@classmethod
def from_paths(cls, dir_path: Path, dir_file_name: Path) -> DirectoryParams:
return cls(CfdpLv.from_path(dir_path), CfdpLv.from_path(dir_file_name))
def from_paths(cls, dir_path: Path, requester_listing_file: Path) -> DirectoryParams:
return cls(CfdpLv.from_path(dir_path), CfdpLv.from_path(requester_listing_file))

@property
def dir_path_as_str(self) -> str:
Expand All @@ -361,12 +367,12 @@ def dir_path_as_path(self) -> Path:
return Path(self.dir_path_as_str)

@property
def dir_file_name_as_str(self) -> str:
return self.dir_file_name.value.decode()
def requester_listing_file_as_str(self) -> str:
return self.requester_listing_file.value.decode()

@property
def dir_file_name_as_path(self) -> Path:
return Path(self.dir_file_name_as_str)
def requester_listing_file_as_path(self) -> Path:
return Path(self.requester_listing_file_as_str)


@dataclasses.dataclass
Expand All @@ -378,7 +384,7 @@ class DirListingOptions:
class DirectoryListingRequest(ReservedCfdpMessage):
def __init__(self, params: DirectoryParams):
"""Create a directory listing request."""
value = params.dir_path.pack() + params.dir_file_name.pack()
value = params.dir_path.pack() + params.requester_listing_file.pack()
super().__init__(DirectoryOperationMessageType.LISTING_REQUEST, value)


Expand All @@ -397,7 +403,7 @@ def __init__(self, listing_success: bool, dir_params: DirectoryParams):
value = (
bytes([0x00 if listing_success else 0x80])
+ dir_params.dir_path.pack()
+ dir_params.dir_file_name.pack()
+ dir_params.requester_listing_file.pack()
)
super().__init__(DirectoryOperationMessageType.LISTING_RESPONSE, value)

Expand Down
15 changes: 10 additions & 5 deletions tests/cfdp/tlvslvs/test_dir_params.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,21 @@ class TestDirParamsClass(TestCase):
def setUp(self):
self.dir_path = "/tmp"
self.dir_path_from_str = CfdpLv.from_str(self.dir_path)
self.dir_file_name = "~/dir-listing.txt"
self.dir_file_name_from_str = CfdpLv.from_str(self.dir_file_name)
self.dir_param = DirectoryParams(self.dir_path_from_str, self.dir_file_name_from_str)
self.dir_param_from_strs = DirectoryParams.from_strs(self.dir_path, self.dir_file_name)
self.listing_file = "~/dir-listing.txt"
self.listing_file_from_str = CfdpLv.from_str(self.listing_file)
self.dir_param = DirectoryParams(self.dir_path_from_str, self.listing_file_from_str)
self.dir_param_from_strs = DirectoryParams.from_strs(self.dir_path, self.listing_file)
# As POSIX to ensure this works on Windows as well.
self.dir_param_from_paths = DirectoryParams.from_paths(
Path(self.dir_path).as_posix(), Path(self.dir_file_name).as_posix()
Path(self.dir_path).as_posix(), Path(self.listing_file).as_posix()
)

def test_apis(self):
self.assertEqual(self.dir_param, self.dir_param_from_strs)
self.assertEqual(self.dir_param, self.dir_param_from_paths)
self.assertEqual(self.dir_param_from_strs, self.dir_param_from_paths)

def test_accessors(self):
self.assertEqual(self.dir_param.dir_path_as_str, self.dir_path)
self.assertEqual(self.dir_param.requester_listing_file_as_str, self.listing_file)
self.assertEqual(self.dir_param.requester_listing_file_as_path, Path(self.listing_file))
16 changes: 8 additions & 8 deletions tests/cfdp/tlvslvs/test_reserved_cfdp_msg.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,8 +65,8 @@ def setUp(self) -> None:
self.proxy_cancel_request = ProxyCancelRequest()

self.dir_path_lv = CfdpLv.from_str("/tmp")
self.dir_listing_name_lv = CfdpLv.from_str("/tmp/listing.txt")
self.dir_params = DirectoryParams(self.dir_path_lv, self.dir_listing_name_lv)
self.listing_file_lv = CfdpLv.from_str("/tmp/listing.txt")
self.dir_params = DirectoryParams(self.dir_path_lv, self.listing_file_lv)
self.dir_listing_req = DirectoryListingRequest(self.dir_params)
self.dir_listing_response = DirectoryListingResponse(True, self.dir_params)
self.dir_lst_opt_recursive = True
Expand Down Expand Up @@ -234,13 +234,13 @@ def test_dir_listing_req_pack(self):
dir_listing_req_raw = self.dir_listing_req.pack()
self._generic_raw_data_verification(
dir_listing_req_raw,
self.dir_path_lv.packet_len + self.dir_listing_name_lv.packet_len,
self.dir_path_lv.packet_len + self.listing_file_lv.packet_len,
DirectoryOperationMessageType.LISTING_REQUEST,
)
dir_path_lv = CfdpLv.unpack(dir_listing_req_raw[7:])
dir_listing_name_lv = CfdpLv.unpack(dir_listing_req_raw[7 + dir_path_lv.packet_len :])
listing_file_lv = CfdpLv.unpack(dir_listing_req_raw[7 + dir_path_lv.packet_len :])
self.assertEqual(dir_path_lv, self.dir_path_lv)
self.assertEqual(dir_listing_name_lv, self.dir_listing_name_lv)
self.assertEqual(listing_file_lv, self.listing_file_lv)

def test_dir_listing_req_unpack(self):
dir_listing_req_params = self.dir_listing_req.get_dir_listing_request_params()
Expand All @@ -265,15 +265,15 @@ def test_dir_listing_response_pack(self):
dir_listing_response_raw = self.dir_listing_response.pack()
self._generic_raw_data_verification(
dir_listing_response_raw,
1 + self.dir_path_lv.packet_len + self.dir_listing_name_lv.packet_len,
1 + self.dir_path_lv.packet_len + self.listing_file_lv.packet_len,
DirectoryOperationMessageType.LISTING_RESPONSE,
)
success_response = ((dir_listing_response_raw[7] >> 7) & 0b1) == 0
dir_path_lv = CfdpLv.unpack(dir_listing_response_raw[8:])
dir_listing_name_lv = CfdpLv.unpack(dir_listing_response_raw[8 + dir_path_lv.packet_len :])
listing_file_lv = CfdpLv.unpack(dir_listing_response_raw[8 + dir_path_lv.packet_len :])
self.assertTrue(success_response)
self.assertEqual(self.dir_path_lv, dir_path_lv)
self.assertEqual(self.dir_listing_name_lv, dir_listing_name_lv)
self.assertEqual(self.listing_file_lv, listing_file_lv)

def test_dir_listing_response_unpack(self):
dir_listing_response_params = self.dir_listing_req.get_dir_listing_request_params()
Expand Down
Loading