153 Add option to send mock bpa bundles to both appin and clin

mock-bpa-test/_test_util.py

@@ -19,37 +19,48 @@
 # the prime contract 80NM0018D0004 between the Caltech and NASA under
 # subcontract 1700763.
 #
-from enum import Enum
+from enum import IntEnum, unique
+from dataclasses import dataclass
+from typing import Any
 
 
-class DataFormat(Enum):
+@unique
+class DataFormat(IntEnum):
     BUNDLEARRAY = 0
     HEX = 1
     ERR = 2
     NONE = 3
 
-# "structure" to hold a simple test case
 
+@unique
+class BundleDestLoc(IntEnum):
+    APPIN = 0
+    CLIN = 1
 
+
+@dataclass
+# Holds 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 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: 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 working
-        self.is_working = is_working
-
-        self.input_data_format = input_data_format
-        self.expected_output_format = expected_output_format
+    # list representation of bundle
+    input_data: Any
+
+    # either list representation of expected output bundle OR a string to search log output for match
+    expected_output: DataFormat
+
+    # decimal digit representing uint32 for policy configuration OR path to JSON-encoded ION-like policy rules
+    policy_config: str
+
+    # path to JWK-encoded key set
+    key_set: str
+
+    # data format of input
+    input_data_format: DataFormat
+
+    # data format of output
+    expected_output_format: DataFormat
+
+    # True if test working (can be removed once all tests are working)
+    is_working: bool = True
+
+    # destination location of the bundle
+    bundle_dest_loc: BundleDestLoc = BundleDestLoc.CLIN

mock-bpa-test/policy_provider_test.json

@@ -7,7 +7,7 @@
         "rule_id": "1",
         "role": "s",
         "tgt": 1,
-        "loc": "clin",
+        "loc": "appin",
         "sc_id": 1
       },
       "spec": {
@@ -33,7 +33,7 @@
         ]
       },
       "es_ref": "d_integrity",
-      "_temp_not_ion_spec_policy_action_on_fail": "delete_bundle"
+      "policy_action_on_fail": "delete_bundle"
     },
     "event_set": {
       "es_ref": "d_integrity",
@@ -52,7 +52,7 @@
         "rule_id": "2",
         "role": "s",
         "tgt": 1,
-        "loc": "clin",
+        "loc": "appin",
         "sc_id": 2
       },
       "spec": {
@@ -78,7 +78,7 @@
         ]
       },
       "es_ref": "d_confidentiality",
-      "_temp_not_ion_spec_policy_action_on_fail": "delete_bundle"
+      "policy_action_on_fail": "delete_bundle"
     },
     "event_set": {
       "es_ref": "d_confidentiality",

mock-bpa-test/test_bpa.py

@@ -35,7 +35,7 @@
 import cbor2
 
 from helpers import CmdRunner, compose_args
-from _test_util import _TestCase, DataFormat
+from _test_util import _TestCase, DataFormat, BundleDestLoc
 
 OWNPATH = os.path.dirname(os.path.abspath(__file__))
 LOGGER = logging.getLogger(__name__)
@@ -141,14 +141,16 @@ def _single_test(self, testcase: Optional[_TestCase]):
         tx_data = testcase.input_data if (
             testcase.input_data_format == DataFormat.HEX) else self._encode(testcase.input_data)
 
-        if (testcase.expected_output_format == DataFormat.BUNDLEARRAY):
+        test_sock = self._ol_sock if testcase.bundle_dest_loc == BundleDestLoc.APPIN else self._ul_sock
+
+        if testcase.expected_output_format == DataFormat.BUNDLEARRAY:
             expected_rx = testcase.expected_output if (
                 testcase.expected_output == "HEX") else self._encode(testcase.expected_output)
 
-            self._ul_sock.send(tx_data)
+            test_sock.send(tx_data)
             LOGGER.debug('waiting')
 
-            rx_data = self._wait_for(self._ul_sock)
+            rx_data = self._wait_for(test_sock)
 
             LOGGER.info('\nTransferred data:\n%s\n', binascii.hexlify(tx_data))
             LOGGER.info('\nReceived data:\n%s\n', binascii.hexlify(rx_data))
@@ -160,12 +162,12 @@ def _single_test(self, testcase: Optional[_TestCase]):
 
             self.assertEqual(binascii.hexlify(expected_rx), binascii.hexlify(rx_data))
 
-        elif (testcase.expected_output_format == DataFormat.NONE):
-            self._ul_sock.send(tx_data)
+        elif testcase.expected_output_format == DataFormat.NONE:
+            test_sock.send(tx_data)
             LOGGER.debug('waiting')
 
             with self.assertRaises(TimeoutError):
-                self._wait_for(self._ul_sock)
+                self._wait_for(test_sock)
 
             LOGGER.info('\nTransferred data:\n%s\n', binascii.hexlify(tx_data))
 
@@ -178,12 +180,12 @@ def _single_test(self, testcase: Optional[_TestCase]):
             LOGGER.debug("\nFOUND OCCURENCE: %s", found)
             self.assertNotEqual("", found)
 
-        elif (testcase.expected_output_format == DataFormat.ERR):
-            self._ul_sock.send(tx_data)
+        elif testcase.expected_output_format == DataFormat.ERR:
+            test_sock.send(tx_data)
             LOGGER.debug('waiting')
 
             with self.assertRaises(TimeoutError):
-                self._wait_for(self._ul_sock)
+                self._wait_for(test_sock)
 
             LOGGER.info('\nTransferred data:\n%s\n', binascii.hexlify(tx_data))
 

mock-bpa-test/test_ccsds.py

@@ -24,7 +24,7 @@
 import binascii
 import tempfile
 import json
-from _test_util import _TestCase, DataFormat
+from _test_util import _TestCase, DataFormat, BundleDestLoc
 from test_bpa import TestAgent
 
 
@@ -150,14 +150,14 @@ def load_ccsds():
                             'rule_id': str(i),
                             'role': sec_role,
                             'tgt': int(target),
-                            'loc': 'clin',
+                            'loc': 'appin',
                             'sc_id': sec_ctx,
                         },
                         'spec': {
                             'sc_id': sec_ctx,
                             'sc_parms': params
                         },
-                        '_temp_not_ion_spec_policy_action_on_fail': 'delete_bundle'
+                        'policy_action_on_fail': 'delete_bundle'
                     }
                 }
                 print(f'Appending new Policy Rule {pr}')
@@ -178,6 +178,7 @@ def load_ccsds():
                 expected_output=output if (
                     output_format == DataFormat.BUNDLEARRAY) else r".*Delete bundle due to failed security operation",
                 policy_config=finame,
+                bundle_dest_loc=BundleDestLoc.APPIN,
                 key_set="mock-bpa-test/key_set_1.json",
                 is_working=True,
                 input_data_format=input_format,

mock-bpa-test/test_json_policy.py

@@ -19,7 +19,7 @@
 # the prime contract 80NM0018D0004 between the Caltech and NASA under
 # subcontract 1700763.
 #
-from _test_util import _TestCase, DataFormat
+from _test_util import _TestCase, DataFormat, BundleDestLoc
 from test_bpa import TestAgent
 
 # Test Cases utilizing JSON policy definitions
@@ -46,6 +46,7 @@ def test_json_source_bib_bcb(self):
                     '3a09c1e63fe23a7f66a59c7303837241e070b02619fc59c5214a22f08cd70795e73e9a')]
             ],
             policy_config='mock-bpa-test/policy_provider_test.json',
+            bundle_dest_loc=BundleDestLoc.APPIN,
             key_set="mock-bpa-test/key_set_1.json",
             is_working=True,
             input_data_format=DataFormat.BUNDLEARRAY,