adding a unit test and updating existing tests
@@ -72,21 +72,48 @@ RESPOND_LIVEACTION = {
"action": "st2.inquiry.respond",
}
+OUTPUT_SCHEMA_RESULT = {
+ "stdout": "",
+ "stderr": "",
+ "result": {
+ "os_secret_param": "to_be_masked",
+ "os_non_secret_param": "not_to_be_masked",
+ },
+}
+
+OUTPUT_SCHEMA_LIVEACTION = {
+ "action": "core.ask",
+ "parameters": {},
+}
+
ACTIONEXECUTIONS = {
"execution_1": {
- "action": {"uid": "action:core:ask"},
+ "action": {"uid": "action:core:ask", "output_schema": {}},
"status": "succeeded",
"runner": {"name": "inquirer"},
"liveaction": INQUIRY_LIVEACTION,
"result": INQUIRY_RESULT,
},
"execution_2": {
- "action": {"uid": "action:st2:inquiry.respond"},
+ "action": {"uid": "action:st2:inquiry.respond", "output_schema": {}},
"status": "succeeded",
"runner": {"name": "python-script"},
"liveaction": RESPOND_LIVEACTION,
"result": {"exit_code": 0, "result": None, "stderr": "", "stdout": ""},
},
+ "execution_3": {
+ "action": {
+ "uid": "action:core:ask",
+ "output_schema": {
+ "os_secret_param": {"type": "string", "required": True, "secret": True},
+ "os_non_secret_param": {"type": "string", "required": True},
+ },
+ },
+ "status": "succeeded",
+ "runner": {"name": "inquirer"},
+ "liveaction": OUTPUT_SCHEMA_LIVEACTION,
+ "result": OUTPUT_SCHEMA_RESULT,
+ },
}
@@ -166,6 +193,23 @@ class ActionExecutionModelTest(DbTestCase):
for value in masked["parameters"]["response"].values():
self.assertEqual(value, MASKED_ATTRIBUTE_VALUE)
+ def test_ouput_schema_secret_param_masking(self):
+ """Test that the parameter marked secret as true in output schema is maked in the output result
+
+ here out of two parameters in output schema is marked secret as true and we are asserting only
+ that is masked in the output result.
+ """
+
+ masked = self.executions["execution_3"].mask_secrets(
+ self.executions["execution_3"].to_serializable_dict()
+ )
+ self.assertEqual(
+ masked["result"]["result"]["os_secret_param"], MASKED_ATTRIBUTE_VALUE
+ )
+ self.assertEqual(
+ masked["result"]["result"]["os_non_secret_param"], "not_to_be_masked"
+ )
+
@staticmethod
def _save_execution(execution):
return ActionExecution.add_or_update(execution)
