metadata pdu update

us-irs · Nov 27, 2024 · 78ccc7e · 78ccc7e
1 parent de5c764
commit 78ccc7e
Show file tree

Hide file tree

Showing 3 changed files with 70 additions and 27 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 # [unreleased]
 
+## Changed
+
+- `MetadataPdu` options have to be specified as an optional list of abstract TLVs now.
+   A new getter method `options_as_tlv` can be used to retrieve a list of concrete TLV objects.
+
 # [v0.25.0] 2024-10-29
 
 ## Changed

diff --git a/spacepackets/cfdp/pdu/metadata.py b/spacepackets/cfdp/pdu/metadata.py
@@ -12,7 +12,7 @@
     AbstractFileDirectiveBase,
 )
 from spacepackets.cfdp.conf import PduConfig, LargeFileFlag
-from spacepackets.cfdp.tlv import CfdpTlv
+from spacepackets.cfdp.tlv import CfdpTlv, TlvList
 from spacepackets.cfdp.lv import CfdpLv
 from spacepackets.cfdp.defs import ChecksumType, CrcFlag, Direction
 from spacepackets.crc import CRC16_CCITT_FUNC
@@ -53,7 +53,7 @@ def __init__(
         self,
         pdu_conf: PduConfig,
         params: MetadataParams,
-        options: Optional[List[CfdpTlv]] = None,
+        options: Optional[TlvList] = None,
     ):
         pdu_conf = copy.copy(pdu_conf)
         self.params = params
@@ -109,14 +109,23 @@ def __empty(cls) -> MetadataPdu:
         )
 
     @property
-    def options(self) -> Optional[List[CfdpTlv]]:
+    def options(self) -> Optional[TlvList]:
         return self._options
 
     @options.setter
-    def options(self, options: Optional[List[CfdpTlv]]):
+    def options(self, options: Optional[TlvList]):
         self._options = options
         self._calculate_directive_field_len()
 
+    def options_as_tlv(self) -> Optional[List[CfdpTlv]]:
+        """Returns :py:meth:`options` converted to a list of concrete :py:class:`CfdpTlv` objects."""
+        if self.options is None:
+            return None
+        cfdp_tlv_list = []
+        for option in self.options:
+            cfdp_tlv_list.append(CfdpTlv(option.tlv_type, option.value))
+        return cfdp_tlv_list
+
     @property
     def directive_param_field_len(self):
         return self.pdu_file_directive.directive_param_field_len

diff --git a/tests/cfdp/pdus/test_metadata.py b/tests/cfdp/pdus/test_metadata.py
@@ -21,13 +21,24 @@
 class TestMetadata(TestCase):
     def setUp(self) -> None:
         self.pdu_conf = PduConfig.default()
+        self.pdu_header_len = self.pdu_conf.header_len()
         self.metadata_params = MetadataParams(
             closure_requested=False,
             file_size=2,
             source_file_name="test.txt",
             dest_file_name="test2.txt",
             checksum_type=ChecksumType.MODULAR,
         )
+        self.file_name = "hallo.txt"
+        self.option_0 = FileStoreRequestTlv(
+            action_code=FilestoreActionCode.CREATE_FILE_SNM,
+            first_file_name=self.file_name,
+        )
+        self.option_1 = FaultHandlerOverrideTlv(
+            condition_code=ConditionCode.POSITIVE_ACK_LIMIT_REACHED,
+            handler_code=FaultHandlerCode.ABANDON_TRANSACTION,
+        )
+        self.assertEqual(self.option_1.packet_len, 3)
 
     def test_metadata_simple(self):
         metadata_pdu = MetadataPdu(pdu_conf=self.pdu_conf, params=self.metadata_params)
@@ -58,25 +69,20 @@ def test_metadata_with_crc(self):
         self.assertEqual(len(metadata_raw), expected_len)
 
     def test_metadata_pdu(self):
-        file_name = "hallo.txt"
-        option_0 = FileStoreRequestTlv(
-            action_code=FilestoreActionCode.CREATE_FILE_SNM, first_file_name=file_name
-        )
-
-        self.assertEqual(option_0.packet_len, 13)
+        self.assertEqual(self.option_0.packet_len, 13)
         expected_bytes = bytearray()
         expected_bytes.extend(bytes([0x00, 0x0B, 0x00, 0x09]))
-        expected_bytes.extend(file_name.encode())
-        self.assertEqual(option_0.pack(), expected_bytes)
+        expected_bytes.extend(self.file_name.encode())
+        self.assertEqual(self.option_0.pack(), expected_bytes)
 
         # Create completey new packet
         pdu_with_option = MetadataPdu(
             pdu_conf=self.pdu_conf,
             params=self.metadata_params,
-            options=[option_0],
+            options=[self.option_0],
         )
         header_len = pdu_with_option.pdu_file_directive.header_len
-        self.assertEqual(pdu_with_option.options, [option_0])
+        self.assertEqual(pdu_with_option.options, [self.option_0])
         expected_len = 10 + 9 + 8 + 5 + 13
         self.assertEqual(pdu_with_option.packet_len, expected_len)
         pdu_with_option_raw = pdu_with_option.pack()
@@ -86,19 +92,15 @@ def test_metadata_pdu(self):
         tlv_wrapper = TlvHolder(pdu_with_option_unpacked.options[0])  # type: ignore
         tlv_typed = tlv_wrapper.to_fs_request()
         self.assertIsNotNone(tlv_typed)
-        self.assertEqual(tlv_typed.pack(), option_0.pack())
+        self.assertEqual(tlv_typed.pack(), self.option_0.pack())
 
         pdu_with_option.source_file_name = None
         pdu_with_option.dest_file_name = None
         expected_len = header_len + 1 + 1 + 5 + 13
         self.assertEqual(pdu_with_option.directive_param_field_len, 1 + 1 + 5 + 13)
         self.assertEqual(pdu_with_option.packet_len, expected_len)
 
-        option_1 = FaultHandlerOverrideTlv(
-            condition_code=ConditionCode.POSITIVE_ACK_LIMIT_REACHED,
-            handler_code=FaultHandlerCode.ABANDON_TRANSACTION,
-        )
-        self.assertEqual(option_1.packet_len, 3)
+    def test_metadata_pdu_two_options(self):
         metadata_params = MetadataParams(
             closure_requested=False,
             file_size=2,
@@ -109,20 +111,38 @@ def test_metadata_pdu(self):
         pdu_with_two_options = MetadataPdu(
             pdu_conf=self.pdu_conf,
             params=metadata_params,
-            options=[option_0, option_1],
+            options=[self.option_0, self.option_1],
         )
+        options_abstract = pdu_with_two_options.options
+        options_concrete = pdu_with_two_options.options_as_tlv()
+        for idx, option in enumerate(options_abstract):
+            self.assertEqual(option.tlv_type, options_concrete[idx].tlv_type)
+            self.assertEqual(option.value, options_concrete[idx].value)
         pdu_with_two_options_raw = pdu_with_two_options.pack()
-        expected_len = header_len + 5 + 2 + option_0.packet_len + option_1.packet_len
+        header_len = pdu_with_two_options.header_len
+        expected_len = (
+            header_len + 5 + 2 + self.option_0.packet_len + self.option_1.packet_len
+        )
         self.assertEqual(pdu_with_two_options.packet_len, expected_len)
         self.assertEqual(len(pdu_with_two_options_raw), expected_len)
         pdu_with_two_options.source_file_name = "hello.txt"
         expected_len = (
-            header_len + 5 + 1 + 10 + option_0.packet_len + option_1.packet_len
+            header_len
+            + 5
+            + 1
+            + 10
+            + self.option_0.packet_len
+            + self.option_1.packet_len
         )
         self.assertEqual(pdu_with_two_options.packet_len, expected_len)
         pdu_with_two_options.dest_file_name = "hello2.txt"
         expected_len = (
-            header_len + 5 + 11 + 10 + option_0.packet_len + option_1.packet_len
+            header_len
+            + 5
+            + 11
+            + 10
+            + self.option_0.packet_len
+            + self.option_1.packet_len
         )
         self.assertEqual(pdu_with_two_options.packet_len, expected_len)
         pdu_with_no_options = pdu_with_two_options
@@ -132,14 +152,23 @@ def test_metadata_pdu(self):
             pdu_with_no_options.pack()
 
         self.pdu_conf.file_flag = LargeFileFlag.LARGE
+
+    def test_metadata_pdu_1(self):
+        metadata_params = MetadataParams(
+            closure_requested=False,
+            file_size=2,
+            source_file_name=None,
+            dest_file_name=None,
+            checksum_type=ChecksumType.MODULAR,
+        )
         pdu_file_size_large = MetadataPdu(
             pdu_conf=self.pdu_conf,
             params=metadata_params,
             options=None,
         )
-        self.assertEqual(pdu_file_size_large.pdu_file_directive.header_len, header_len)
-        self.assertEqual(pdu_file_size_large.packet_len, header_len + 2 + 9)
-        pdu_file_size_large.options = [option_0]
+        # 1 byte directive type, 7 bytes metadata content
+        self.assertEqual(pdu_file_size_large.packet_len, self.pdu_header_len + 1 + 7)
+        pdu_file_size_large.options = [self.option_0]
         pdu_file_size_large_raw = pdu_file_size_large.pack()
         pdu_file_size_large_raw = pdu_file_size_large_raw[:-2]
         with self.assertRaises(ValueError):