Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

improve finished PDU API #44

Merged
merged 2 commits into from
Oct 8, 2023
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
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/).

# [unreleased]

## Added

- Added a `finished_params` property for the `FinishedPdu` class.

# [v0.19.0] 2023-10-05

## Fixed
Expand Down
8 changes: 6 additions & 2 deletions spacepackets/cfdp/pdu/finished.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations
import enum
import struct
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import List, Optional

from spacepackets.cfdp.pdu import PduHeader
Expand Down Expand Up @@ -34,7 +34,7 @@ class FinishedParams:
delivery_code: DeliveryCode
delivery_status: FileDeliveryStatus
condition_code: ConditionCode
file_store_responses: Optional[List[FileStoreResponseTlv]] = None
file_store_responses: List[FileStoreResponseTlv] = field(default_factory=lambda: [])
fault_location: Optional[EntityIdTlv] = None

@classmethod
Expand Down Expand Up @@ -98,6 +98,10 @@ def packet_len(self) -> int:
def file_store_responses(self) -> List[FileStoreResponseTlv]:
return self._params.file_store_responses

@property
def finished_params(self) -> FinishedParams:
return self._params

@property
def might_have_fault_location(self):
if self._params.condition_code in [
Expand Down
2 changes: 1 addition & 1 deletion spacepackets/cfdp/pdu/helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ def _cast_to_concrete_file_directive(
self, pdu_type: Type[Any], dir_type: DirectiveType
) -> Any:
if (
isinstance(self.base, AbstractFileDirectiveBase)
isinstance(self.pdu, AbstractFileDirectiveBase)
and self.pdu.pdu_type == PduType.FILE_DIRECTIVE # type: ignore
):
pdu_base = cast(AbstractFileDirectiveBase, self.pdu)
Expand Down
2 changes: 1 addition & 1 deletion spacepackets/cfdp/pdu/nak.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ def segment_requests(self, segment_requests: Optional[List[Tuple[int, int]]]):
if segment_requests is None:
self._segment_requests = []
else:
self._segment_requests: List[Tuple[int, int]] = segment_requests # type: ignore
self._segment_requests: List[Tuple[int, int]] = segment_requests # type: ignore
self._calculate_directive_field_len()

def pack(self) -> bytearray:
Expand Down
20 changes: 10 additions & 10 deletions tests/cfdp/pdus/test_pdu_holder.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ def test_file_data(self):

def test_holder_print(self):
nak_pdu = NakPdu(start_of_scope=0, end_of_scope=200, pdu_conf=self.pdu_conf)
self.pdu_wrapper.base = nak_pdu
self.pdu_wrapper.pdu = nak_pdu
print(self.pdu_wrapper)

def test_invalid_to_file_data(self):
Expand All @@ -57,7 +57,7 @@ def test_invalid_to_file_data(self):
checksum_type=ChecksumType.MODULAR,
)
metadata_pdu = MetadataPdu(pdu_conf=self.pdu_conf, params=params)
self.pdu_wrapper.base = metadata_pdu
self.pdu_wrapper.pdu = metadata_pdu
with self.assertRaises(TypeError) as cm:
self.pdu_wrapper.to_file_data_pdu()
exception = cm.exception
Expand All @@ -72,7 +72,7 @@ def test_metadata(self):
checksum_type=ChecksumType.MODULAR,
)
metadata_pdu = MetadataPdu(pdu_conf=self.pdu_conf, params=params)
self.pdu_wrapper.base = metadata_pdu
self.pdu_wrapper.pdu = metadata_pdu
metadata_casted_back = self.pdu_wrapper.to_metadata_pdu()
self.assertEqual(metadata_casted_back, metadata_pdu)

Expand All @@ -83,7 +83,7 @@ def test_invalid_cast(self):
segment_metadata_flag=False,
)
file_data_pdu = FileDataPdu(pdu_conf=self.pdu_conf, params=fd_params)
self.pdu_wrapper.base = file_data_pdu
self.pdu_wrapper.pdu = file_data_pdu
with self.assertRaises(TypeError) as cm:
self.pdu_wrapper.to_metadata_pdu()
exception = cm.exception
Expand All @@ -96,21 +96,21 @@ def test_ack_cast(self):
transaction_status=TransactionStatus.TERMINATED,
pdu_conf=self.pdu_conf,
)
self.pdu_wrapper.base = ack_pdu
self.pdu_wrapper.pdu = ack_pdu
ack_pdu_converted = self.pdu_wrapper.to_ack_pdu()
self.assertEqual(ack_pdu_converted, ack_pdu)

def test_nak_cast(self):
nak_pdu = NakPdu(start_of_scope=0, end_of_scope=200, pdu_conf=self.pdu_conf)
self.pdu_wrapper.base = nak_pdu
self.pdu_wrapper.pdu = nak_pdu
nak_pdu_converted = self.pdu_wrapper.to_nak_pdu()
self.assertEqual(nak_pdu_converted, nak_pdu)

def test_prompt_cast(self):
prompt_pdu = PromptPdu(
pdu_conf=self.pdu_conf, response_required=ResponseRequired.KEEP_ALIVE
)
self.pdu_wrapper.base = prompt_pdu
self.pdu_wrapper.pdu = prompt_pdu
prompt_pdu_converted = self.pdu_wrapper.to_prompt_pdu()
self.assertEqual(prompt_pdu_converted, prompt_pdu)

Expand All @@ -119,7 +119,7 @@ def test_eof_cast(self):
eof_pdu = EofPdu(
file_checksum=zero_checksum, file_size=0, pdu_conf=self.pdu_conf
)
self.pdu_wrapper.base = eof_pdu
self.pdu_wrapper.pdu = eof_pdu
eof_pdu_converted = self.pdu_wrapper.to_eof_pdu()
self.assertEqual(eof_pdu_converted, eof_pdu)

Expand All @@ -133,12 +133,12 @@ def test_finished_cast(self):
params=params,
pdu_conf=self.pdu_conf,
)
self.pdu_wrapper.base = finish_pdu
self.pdu_wrapper.pdu = finish_pdu
finish_pdu_converted = self.pdu_wrapper.to_finished_pdu()
self.assertEqual(finish_pdu_converted, finish_pdu)

def test_keep_alive_cast(self):
keep_alive_pdu = KeepAlivePdu(progress=0, pdu_conf=self.pdu_conf)
self.pdu_wrapper.base = keep_alive_pdu
self.pdu_wrapper.pdu = keep_alive_pdu
keep_alive_pdu_converted = self.pdu_wrapper.to_keep_alive_pdu()
self.assertEqual(keep_alive_pdu_converted, keep_alive_pdu)