Skip to content

Commit

Permalink
added a few printout unittests
Browse files Browse the repository at this point in the history
  • Loading branch information
robamu committed Nov 10, 2023
1 parent c4d09f2 commit d1c1328
Show file tree
Hide file tree
Showing 2 changed files with 83 additions and 5 deletions.
74 changes: 74 additions & 0 deletions tests/cfdp/test_request.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
from pathlib import Path
from unittest import TestCase

from spacepackets.cfdp import CfdpLv, TransactionId
from spacepackets.cfdp.tlv import (
OriginatingTransactionId,
ProxyMessageType,
ProxyPutRequest,
ProxyPutRequestParams,
)
from spacepackets.util import ByteFieldU16

from tmtccmd.cfdp import PutRequest


class TestRequest(TestCase):
def setUp(self):
pass

def test_printout_0(self):
put_req = PutRequest(
destination_id=ByteFieldU16(5),
source_file=None,
dest_file=None,
trans_mode=None,
closure_requested=None,
)
print_str = str(put_req)
self.assertEqual(print_str, "Metadata Only Put Request with Destination ID 5")

def test_printout_1(self):
put_req = PutRequest(
destination_id=ByteFieldU16(5),
source_file=Path("/tmp/test.txt"),
dest_file=Path("/tmp/test2.txt"),
trans_mode=None,
closure_requested=None,
)
print_str = str(put_req)
self.assertTrue("Destination ID 5" in print_str)
self.assertTrue("/tmp/test.txt" in print_str)
self.assertTrue("/tmp/test2.txt" in print_str)
self.assertTrue("Transmission Mode from MIB" in print_str)
self.assertTrue("Closure Requested from MIB" in print_str)

def test_printout_2(self):
proxy_put = ProxyPutRequest(
ProxyPutRequestParams(
dest_entity_id=ByteFieldU16(2),
source_file_name=CfdpLv.from_str("/tmp/test.txt"),
dest_file_name=CfdpLv.from_str("/tmp/test2.txt"),
)
).to_generic_msg_to_user_tlv()
orig_id = TransactionId(
source_entity_id=ByteFieldU16(1), transaction_seq_num=ByteFieldU16(5)
)
orig_id_msg = OriginatingTransactionId(orig_id).to_generic_msg_to_user_tlv()
put_req = PutRequest(
destination_id=ByteFieldU16(5),
source_file=None,
dest_file=None,
trans_mode=None,
closure_requested=None,
msgs_to_user=[proxy_put, orig_id_msg],
)
print_str = str(put_req)
print(print_str)
self.assertTrue("Metadata Only Put Request with Destination ID 5" in print_str)
self.assertTrue(
f"Message to User 0: Proxy Operation {ProxyMessageType.PUT_REQUEST!r}"
)
self.assertTrue("/tmp/test.txt" in print_str)
self.assertTrue("/tmp/test2.txt" in print_str)
self.assertTrue("Message to User 1: Originating Transaction ID" in print_str)
14 changes: 9 additions & 5 deletions tmtccmd/cfdp/request.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,23 +69,26 @@ def __str__(self):
else:
closure_str = "No closure requested"
else:
closure_str = "Closure information from MIB"
closure_str = "Closure Requested from MIB"
if not self.metadata_only:
print_str = (
f"Destination ID: {self.destination_id.value}\n\t"
f"Destination ID {self.destination_id.value}\n\t"
f"{src_file_str}\n\t{dest_file_str}\n\t{trans_mode_str}\n\t{closure_str}"
)
else:
print_str = self.__str_for_metadata_only()
return print_str

def __str_for_metadata_only(self) -> str:
print_str = f"Metadata Only Put Request with Destination ID: {self.destination_id.value}\n"
print_str = (
f"Metadata Only Put Request with Destination ID {self.destination_id.value}"
)
if self.msgs_to_user is not None:
for idx, msg_to_user in enumerate(self.msgs_to_user):
msg_to_user = cast(MessageToUserTlv, msg_to_user)
if msg_to_user.is_reserved_cfdp_message():
reserved_msg = msg_to_user.to_reserved_msg_tlv()
assert reserved_msg is not None
print_str = PutRequest.__str_for_reserved_cfdp_msg(
idx, reserved_msg, print_str
)
Expand All @@ -97,19 +100,20 @@ def __str_for_reserved_cfdp_msg(
) -> str:
if reserved_msg.is_cfdp_proxy_operation():
proxy_msg_type = reserved_msg.get_cfdp_proxy_message_type()
print_str += f"Message to user {idx}: Proxy operation {proxy_msg_type!r}"
print_str += f"\nMessage to User {idx}: Proxy operation {proxy_msg_type!r}"
if proxy_msg_type == ProxyMessageType.PUT_REQUEST:
print_str = PutRequest.__str_for_put_req(reserved_msg, print_str)
elif reserved_msg.is_originating_transaction_id():
print_str += (
f"Message to user {idx}: Originating Transaction ID "
f"\nMessage to User {idx}: Originating Transaction ID "
f"{reserved_msg.get_originating_transaction_id()}"
)
return print_str

@staticmethod
def __str_for_put_req(reserved_msg: ReservedCfdpMessage, print_str: str) -> str:
put_request_params = reserved_msg.get_proxy_put_request_params()
assert put_request_params is not None
print_str += (
f"\n\tProxy Put Dest Entity ID: {put_request_params.dest_entity_id.value}"
)
Expand Down

0 comments on commit d1c1328

Please sign in to comment.