Rework Python Mock BPA Tests (#113)

* init

* refactor init

* pr fixes

* Clear state before assert

My comment was unclear, this is what I meant

* A few very minor cleanups

* Fix setUp state

* Applying autopep8 to python sources

---------

Co-authored-by: Brian Sipos <brian.sipos@jhuapl.edu>

Author: jeronstone

.github/workflows/formatting.yaml

@@ -46,6 +46,6 @@ jobs:
         run: |
           sudo apt-get update && sudo apt-get install -y \
               clang-format python3-pip
-          pip3 install licenseheaders
+          pip3 install autopep8 licenseheaders
       - name: Check code formatting
         run: ./build.sh check-format

mock-bpa-test/_generate_simple_bundles.py

@@ -28,13 +28,15 @@
 from Crypto.Random import get_random_bytes
 
 # HMAC signs, encrypts, or decrypts payload string
-# This is a really messy function for now 
-def sign_and_encrypt(   payload_s:str, 
-                        sign:bool=True, sign_key_s:str='00112233445566778899AABBCCDDEEFF', 
-                        enc:bool=True, enc_key_s:str='00112233445566778899AABBCCDDEEFF', 
-                        denc:bool=False, denc_key_s:str='71776572747975696f70617364666768', denc_tag_s:str='5d37d992dbc6fc795ea597ed7e8a6078'):
-    
-    #print(f'{payload_s} | {sign} | {sign_key_s} | {enc} | {enc_key_s} | {denc} | {denc_key_s} | {denc_tag_s}\n')
+# This is a really messy function for now
+
+
+def sign_and_encrypt(payload_s: str,
+                     sign: bool = True, sign_key_s: str = '00112233445566778899AABBCCDDEEFF',
+                     enc: bool = True, enc_key_s: str = '00112233445566778899AABBCCDDEEFF',
+                     denc: bool = False, denc_key_s: str = '71776572747975696f70617364666768', denc_tag_s: str = '5d37d992dbc6fc795ea597ed7e8a6078'):
+
+    # print(f'{payload_s} | {sign} | {sign_key_s} | {enc} | {enc_key_s} | {denc} | {denc_key_s} | {denc_tag_s}\n')
 
     # TODO only convert what's needed in if statements below
     payload = bytes.fromhex(payload_s)
@@ -64,7 +66,8 @@ def sign_and_encrypt(   payload_s:str,
         cipher = AES.new(enc_key, AES.MODE_GCM, nonce=iv)
         cipher.update(aad)
         ciphertext, tag = cipher.encrypt_and_digest(payload)
-        results[1] = (ciphertext.hex(), tag.hex(), '69c411276fecddc4780df42c8a2af89296fabf34d7fae700', iv.hex())
+        results[1] = (ciphertext.hex(), tag.hex(),
+                      '69c411276fecddc4780df42c8a2af89296fabf34d7fae700', iv.hex())
 
     if denc:
         cipher = AES.new(denc_key, AES.MODE_GCM, nonce=iv)
@@ -81,10 +84,10 @@ def add_bib_to_bundle_over_x(bundle, x):
     if x == 0:
         sign = sign_and_encrypt(cbor2.dumps(bundle[0]).hex(), sign=True, enc=False, denc=False)[0]
         asb = [
-                [x], 1, 1, [2, [2, 1]],
-                [[1, 7], [3, 0]],
-                [[[1, bytes.fromhex(sign)]]]
-            ]
+            [x], 1, 1, [2, [2, 1]],
+            [[1, 7], [3, 0]],
+            [[[1, bytes.fromhex(sign)]]]
+        ]
         buf = io.BytesIO()
         for b in asb:
             cbor2.dump(b, buf)
@@ -101,7 +104,7 @@ def add_bib_to_bundle_over_x(bundle, x):
         # find the target block
         if blk[1] == x:
 
-            # get the HMAC signature 
+            # get the HMAC signature
             sign = sign_and_encrypt(blk[4], sign=True, enc=False, denc=False)[0]
 
             # create ASB and cbor dump as BIB btsd
@@ -119,6 +122,7 @@ def add_bib_to_bundle_over_x(bundle, x):
 
     return bundle
 
+
 def add_bcb_to_bundle_over_x(bundle, x):
     # find max blk num, we will make new BCB max+1. This guarantees the block num is free
     mx_blk_num = max(bundle, key=lambda blk: blk[1])[1]
@@ -133,9 +137,10 @@ def add_bcb_to_bundle_over_x(bundle, x):
             # create ASB and cbor dump as BCB btsd
             asb = [
                 [x], 2, 1, [2, [2, 1]],
-                     # IV                                       # wrapped key
-                [[1, bytes.fromhex(ciphertext[3])], [2, 1], [3, bytes.fromhex(ciphertext[2])], [4, 0]],
-                      # ciphertext
+                # IV                                       # wrapped key
+                [[1, bytes.fromhex(ciphertext[3])], [2, 1], [
+                    3, bytes.fromhex(ciphertext[2])], [4, 0]],
+                # ciphertext
                 [[[1, bytes.fromhex(ciphertext[1])]]]
             ]
             buf = io.BytesIO()
@@ -150,14 +155,15 @@ def add_bcb_to_bundle_over_x(bundle, x):
 
     return bundle
 
+
 b = [
     [7, 0, 0, [2, [1, 2]], [2, [2, 1]], [2, [2, 1]], [0, 40], 1000000],
     [1, 1, 0, 0, '526561647920746F2067656E657261746520612033322D62797465207061796C6F6164'],
 ]
 
 
-print (f"ORIGINAL BUNDLE: {b}")
+print(f"ORIGINAL BUNDLE: {b}")
 b = add_bib_to_bundle_over_x(b, 1)
 print(f'BUNDLE AFTER BIB: {b}')
-#b = add_bcb_to_bundle_over_x(b, 1)
+# b = add_bcb_to_bundle_over_x(b, 1)
 print(f'FINAL BUNDLE: {b}')

mock-bpa-test/_test_util.py

@@ -21,38 +21,35 @@
 #
 from enum import Enum
 
+
 class DataFormat(Enum):
-    BUNDLEARRAY=0
-    HEX=1
-    ERR=2
-    NONE=3
+    BUNDLEARRAY = 0
+    HEX = 1
+    ERR = 2
+    NONE = 3
 
 # "structure" to hold a simple test case
+
+
 class _TestCase:
     '''
     @param input_data: list representation of bundle
     @param expected_output: either list representation of expected output bundle OR a string to search log output for match 
-    @param policy_config: decimal digit representing uint32 for policy configuration
+    @param policy_config: decimal digit representing uint32 for policy configuration OR path to JSON-encoded ION-like policy rules
+    @param key_set: path to JWK-encoded key set
     @param is_working: True if test working
     @param input/output_data_format: data format of input/output
     '''
-    def __init__(self, input_data, expected_output, policy_config, is_working: bool, 
-                 input_data_format : DataFormat, expected_output_format : DataFormat):
+
+    def __init__(self, input_data, expected_output: DataFormat, policy_config: str, key_set: str, is_working: bool,
+                 input_data_format: DataFormat, expected_output_format: DataFormat):
         self.input_data = input_data
         self.expected_output = expected_output
         self.policy_config = policy_config
+        self.key_set = key_set
 
-        # can be removed once all tests are wworking
+        # can be removed once all tests are working
         self.is_working = is_working
 
         self.input_data_format = input_data_format
         self.expected_output_format = expected_output_format
-
-
-class _TestSet:
-    def __init__(self):
-        self.cases = {}
-
-NO_OUTPUT = 0
-FAILURE_CODE = -1
-DELETION = -2

mock-bpa-test/ccsds_tests.py

@@ -147,7 +147,7 @@ def _write_stdin(self, stream):
             stream.flush()
         LOGGER.debug('Stopping stdin thread')
 
-    def wait_for_line(self, timeout:float=5) -> str:
+    def wait_for_line(self, timeout: float = 5) -> str:
         ''' Wait for any received stdout line.
 
         :param timeout: The total time to wait for this line.
@@ -160,7 +160,7 @@ def wait_for_line(self, timeout:float=5) -> str:
             raise TimeoutError('no lines received before timeout')
         return text
 
-    def wait_for_text(self, pattern:str, timeout:float=5) -> str:
+    def wait_for_text(self, pattern: str, timeout: float = 5) -> str:
         ''' Iterate through the received stdout lines until a specific
         full matching line is seen.
 
@@ -187,7 +187,7 @@ def wait_for_text(self, pattern:str, timeout:float=5) -> str:
             if expr.match(text) is not None:
                 return text
 
-    def send_stdin(self, text:str):
+    def send_stdin(self, text: str):
         ''' Send an exact line of text to the process stdin.
 
         :param text: The line to send, which should include a newline