Merge pull request #48 from us-irs/copy-pdu-conf

copy PDU conf on ctor calls

spacepackets/cfdp/pdu/ack.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 import enum
+import copy
 import struct
 
 from spacepackets.cfdp.pdu import PduHeader
@@ -40,6 +41,7 @@ def __init__(
         :param pdu_conf: PDU configuration parameters
         :raises ValueError: Directive code invalid. Only EOF and Finished PDUs can be acknowledged
         """
+        pdu_conf = copy.copy(pdu_conf)
         if directive_code_of_acked_pdu not in [
             DirectiveType.FINISHED_PDU,
             DirectiveType.EOF_PDU,

spacepackets/cfdp/pdu/eof.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 import struct
+import copy
 from typing import Optional
 
 from spacepackets.cfdp.pdu import PduHeader
@@ -35,6 +36,7 @@ def __init__(
         :param condition_code:
         :raise ValueError: Invalid input, file checksum not 4 bytes long
         """
+        pdu_conf = copy.copy(pdu_conf)
         if len(file_checksum) != 4:
             raise ValueError
         self.condition_code = condition_code

spacepackets/cfdp/pdu/file_data.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 import enum
+import copy
 from dataclasses import dataclass
 from typing import Union, Optional
 import struct
@@ -58,6 +59,7 @@ def __init__(self, pdu_conf: PduConfig, params: FileDataParams):
             and params.record_cont_state is None
         ):
             raise ValueError("Record continuation state must be specified")
+        pdu_conf = copy.copy(pdu_conf)
         pdu_conf.direction = Direction.TOWARDS_RECEIVER
         self._pdu_header = PduHeader(
             segment_metadata_flag=self._params.segment_metadata_flag,

spacepackets/cfdp/pdu/finished.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 import enum
+import copy
 import struct
 from dataclasses import dataclass, field
 from typing import List, Optional
@@ -73,6 +74,7 @@ def __init__(
         pdu_conf: PduConfig,
         params: FinishedParams,
     ):
+        pdu_conf = copy.copy(pdu_conf)
         pdu_conf.direction = Direction.TOWARDS_SENDER
         self.pdu_file_directive = FileDirectivePduBase(
             directive_code=DirectiveType.FINISHED_PDU,

spacepackets/cfdp/pdu/keep_alive.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import struct
+import copy
 
 from spacepackets.cfdp import CrcFlag
 from spacepackets.cfdp.defs import Direction
@@ -18,6 +19,7 @@ class KeepAlivePdu(AbstractFileDirectiveBase):
     """Encapsulates the Keep Alive file directive PDU, see CCSDS 727.0-B-5 p.85"""
 
     def __init__(self, pdu_conf: PduConfig, progress: int):
+        pdu_conf = copy.copy(pdu_conf)
         directive_param_field_len = 4
         if pdu_conf.file_flag == LargeFileFlag.LARGE:
             directive_param_field_len = 8

spacepackets/cfdp/pdu/metadata.py

@@ -2,6 +2,7 @@
 
 import dataclasses
 import struct
+import copy
 from typing import Optional
 
 from spacepackets.cfdp.pdu import PduHeader
@@ -54,6 +55,7 @@ def __init__(
         params: MetadataParams,
         options: Optional[TlvList] = None,
     ):
+        pdu_conf = copy.copy(pdu_conf)
         self.params = params
         if params.source_file_name is None:
             self._source_file_name_lv = CfdpLv(value=bytes())

spacepackets/cfdp/pdu/prompt.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 import enum
+import copy
 import struct
 
 from spacepackets.cfdp import CrcFlag
@@ -20,6 +21,7 @@ class PromptPdu(AbstractFileDirectiveBase):
     """Encapsulates the Prompt file directive PDU, see CCSDS 727.0-B-5 p.84"""
 
     def __init__(self, pdu_conf: PduConfig, response_required: ResponseRequired):
+        pdu_conf = copy.copy(pdu_conf)
         pdu_conf.direction = Direction.TOWARDS_RECEIVER
         self.pdu_file_directive = FileDirectivePduBase(
             directive_code=DirectiveType.PROMPT_PDU,

tests/cfdp/pdus/test_file_data.py

@@ -82,8 +82,8 @@ def test_large_filedata(self):
             record_cont_state=RecordContinuationState.START_AND_END,
             segment_metadata=bytes([0xAA, 0xBB]),
         )
-        fd_pdu_with_metadata = FileDataPdu(pdu_conf=self.pdu_conf, params=fd_params)
         self.pdu_conf.file_flag = LargeFileFlag.LARGE
+        fd_pdu_with_metadata = FileDataPdu(pdu_conf=self.pdu_conf, params=fd_params)
         fd_params = FileDataParams(
             file_data=self.file_data_bytes,
             offset=0,