bugy#200: implemented script callback for script executions

samples/scripts/callback_test.py

@@ -0,0 +1,17 @@
+import os
+import sys
+
+output_file = sys.argv[1]
+
+with open(output_file, 'w') as f:
+    for arg in sys.argv[2:]:
+        f.write(arg + '\n')
+
+    f.write('\n\n')
+
+    fields = ['event_type', 'execution_id', 'pid', 'script_name', 'user', 'exit_code']
+
+    for field in fields:
+        value = os.environ.get(field)
+
+        f.write(field + ': ' + str(value) + '\n')

src/communications/destination_script.py

@@ -0,0 +1,47 @@
+import communications.destination_base as destination_base
+from model.model_helper import read_obligatory
+from utils import process_utils
+from utils.string_utils import values_to_string
+
+
+def _create_communicator(params_dict):
+    return ScriptCommunicator(params_dict)
+
+
+class ScriptDestination(destination_base.Destination):
+    def __init__(self, params_dict):
+        self._communicator = _create_communicator(params_dict)
+
+    def send(self, title, body, files=None):
+        environment_variables = None
+
+        if isinstance(body, dict):
+            parameters = list(body.values())
+            environment_variables = values_to_string(dict(body))
+        elif isinstance(body, list):
+            parameters = body
+        else:
+            raise Exception('Only dict or list bodies are supported')
+
+        parameters = values_to_string(parameters)
+
+        if files:
+            raise Exception('Files are not supported for scripts')
+
+        self._communicator.send(parameters, environment_variables=environment_variables)
+
+    def __str__(self, *args, **kwargs):
+        return type(self).__name__ + ' for ' + str(self._communicator)
+
+
+class ScriptCommunicator:
+    def __init__(self, params_dict):
+        command_config = read_obligatory(params_dict, 'command', ' for Script callback')
+        self.command = process_utils.split_command(command_config)
+
+    def send(self, parameters, environment_variables=None):
+        full_command = self.command + parameters
+        process_utils.invoke(full_command, environment_variables=environment_variables)
+
+    def __str__(self, *args, **kwargs):
+        return 'Script: ' + str(self.command)

src/features/executions_callback_feature.py

@@ -2,6 +2,7 @@
 from collections import OrderedDict
 
 from communications.communicaton_service import CommunicationsService
+from communications.destination_script import ScriptDestination
 from execution.execution_service import ExecutionService
 from model.model_helper import read_bool_from_config, read_list
 
@@ -23,6 +24,8 @@ def _init_destinations(destinations_config):
         elif destination_type == 'http':
             import communications.destination_http as http
             destination = http.HttpDestination(destination_config)
+        elif destination_type == 'script':
+            destination = ScriptDestination(destination_config)
         else:
             raise Exception('Unknown destination type: ' + destination_type)
 
@@ -103,3 +106,7 @@ def prepare_notification_object(self, execution_id, event_type):
         notification_object.move_to_end('event_type', False)
 
         return notification_object
+
+    # tests only
+    def _wait(self):
+        self._communication_service._wait()

src/tests/communications/communication_test_utils.py

@@ -45,17 +45,22 @@ def creator(config):
 
             return creator
 
-        self.http_patch = patch('communications.destination_http._create_communicator')
-        http_patched_func = self.http_patch.start()
-        http_patched_func.side_effect = communicator_generator('http', MockHttpCommunicator)
+        self.patches = []
 
-        self.email_patch = patch('communications.destination_email._create_communicator')
-        email_patched_func = self.email_patch.start()
-        email_patched_func.side_effect = communicator_generator('email', MockEmailCommunicator)
+        communicator_types = {
+            'email': MockEmailCommunicator,
+            'http': MockHttpCommunicator,
+            'script': MockScriptCommunicator}
+
+        for communicator_type, clazz in communicator_types.items():
+            communicator_patch = patch('communications.destination_' + communicator_type + '._create_communicator')
+            http_patched_func = communicator_patch.start()
+            http_patched_func.side_effect = communicator_generator(communicator_type, clazz)
+            self.patches.append(communicator_patch)
 
     def stop(self):
-        self.http_patch.stop()
-        self.email_patch.stop()
+        for communicator_patch in self.patches:
+            communicator_patch.stop()
 
 
 class MockDestination(Destination):
@@ -88,3 +93,15 @@ def send(self, body, content_type=None):
         message = (None, body, None)
         self.messages.append(message)
         self.captured_arguments.append({'body': body, 'content_type': content_type})
+
+
+class MockScriptCommunicator:
+    def __init__(self, name) -> None:
+        self.messages = []
+        self.captured_arguments = []
+        self.name = name
+
+    def send(self, parameters, environment_variables=None):
+        message = (None, parameters, None)
+        self.messages.append(message)
+        self.captured_arguments.append({'parameters': parameters, 'environment_variables': environment_variables})

src/tests/communications/destination_script_test.py

@@ -0,0 +1,89 @@
+import unittest
+from collections import OrderedDict
+
+from communications.communication_model import File
+from communications.destination_script import ScriptDestination
+from tests.communications.communication_test_utils import mock_communicators
+
+
+@mock_communicators
+class TestScriptDestination(unittest.TestCase):
+    def test_send_when_list_body(self):
+        body = ['param1', 'param2']
+        self.do_send(body)
+
+        self.assert_messages([body])
+
+    def test_send_when_list_body_non_strings(self):
+        body = [1, None, False]
+        self.do_send(body)
+
+        self.assert_messages([['1', 'None', 'False']])
+
+    def test_send_when_dict_body(self):
+        body = OrderedDict([('param1', 'Hello'), ('param2', 'World')])
+        self.do_send(body)
+
+        self.assert_messages([['Hello', 'World']])
+
+    def test_send_when_dict_body_non_strings(self):
+        body = OrderedDict([('param1', 123.45), ('param2', None), ('param3', False)])
+        self.do_send(body)
+
+        self.assert_messages([['123.45', 'None', 'False']])
+
+    def test_send_3_times(self):
+        self.do_send(OrderedDict([('param1', 'Hello'), ('param2', 'World')]))
+        self.do_send([])
+        self.do_send(['123', '456', '789'])
+
+        self.assert_messages([['Hello', 'World'], [], ['123', '456', '789']])
+
+    def test_send_when_string_body(self):
+        self.assertRaisesRegex(Exception, 'Only dict or list bodies are supported', self.do_send, 'test message')
+
+    def test_send_when_files(self):
+        self.assertRaisesRegex(Exception, 'Files are not supported for scripts',
+                               self.do_send, ['param1'], [File('test_file')])
+
+    def test_env_when_list_body(self):
+        self.do_send(['param1', 'param2'])
+
+        self.assert_env_variables([None])
+
+    def test_env_when_dict_body(self):
+        body = OrderedDict([('param1', 'Hello'), ('param2', 'World')])
+        self.do_send(body)
+
+        self.assert_env_variables([{'param1': 'Hello', 'param2': 'World'}])
+
+    def test_env_when_dict_body_with_non_strings(self):
+        body = OrderedDict([('param1', 1), ('param2', None), ('param3', True)])
+        self.do_send(body)
+
+        self.assert_env_variables([{'param1': '1', 'param2': 'None', 'param3': 'True'}])
+
+    def test_env_when_send_3_times(self):
+        self.do_send(OrderedDict([('param1', 'Hello'), ('param2', 'World')]))
+        self.do_send([])
+        self.do_send({'abc': '123'})
+
+        self.assert_env_variables([{'param1': 'Hello', 'param2': 'World'}, None, {'abc': '123'}])
+
+    def assert_env_variables(self, env_variables):
+        communicator = self.get_communicators()[0]
+
+        actual_variables = [arg.get('environment_variables') for arg in communicator.captured_arguments]
+        self.assertEqual(env_variables, actual_variables)
+
+    def assert_messages(self, expected_bodies):
+        communicator = self.get_communicators()[0]
+
+        expected_messages = [(None, body, None) for body in expected_bodies]
+        self.assertEqual(expected_messages, communicator.messages)
+
+    def do_send(self, body, files=None):
+        self.destination.send('ignored', body, files)
+
+    def setUp(self):
+        self.destination = ScriptDestination({})

src/tests/features/executions_callback_feature_test.py

@@ -6,6 +6,7 @@
 from features.executions_callback_feature import ExecutionsCallbackFeature
 from tests.communications.communication_test_utils import mock_communicators
 from tests.test_utils import create_config_model
+from utils.string_utils import values_to_string
 
 
 @mock_communicators
@@ -20,10 +21,15 @@ def test_init_email_communicator(self):
 
         self.assert_created_destinations(['email1'])
 
+    def test_init_script_communicator(self):
+        self.create_feature(['script'])
+
+        self.assert_created_destinations(['script1'])
+
     def test_init_mixed_communicators(self):
-        self.create_feature(['email', 'http', 'http', 'email'])
+        self.create_feature(['email', 'http', 'http', 'script', 'email'])
 
-        self.assert_created_destinations(['email1', 'http1', 'http2', 'email2'])
+        self.assert_created_destinations(['email1', 'http1', 'http2', 'script1', 'email2'])
 
     def test_unknown_communicator(self):
         self.assertRaisesRegex(Exception, 'Unknown destination type: socket',
@@ -112,6 +118,24 @@ def test_send_finished_callback_to_email_destination(self):
 
         self.assert_messages([123], 'execution_finished')
 
+    def test_send_started_callback_to_script_destination(self):
+        feature = self.create_feature(['script'])
+        feature.start()
+
+        self.add_execution(123, 'userX', 666, 13, 'my_script')
+        self.fire_started(123)
+
+        self.assert_messages([123], 'execution_started')
+
+    def test_send_finished_callback_to_script_destination(self):
+        feature = self.create_feature(['script'])
+        feature.start()
+
+        self.add_execution(123, 'userX', 666, 13, 'my_script')
+        self.fire_finished(123)
+
+        self.assert_messages([123], 'execution_finished')
+
     def test_started_callback_when_disabled(self):
         feature = self.create_feature(['http'], on_start=False)
         feature.start()
@@ -161,7 +185,8 @@ def create_feature(self, types, on_start=True, on_finish=True, notification_fiel
             config['notification_fields'] = notification_fields
 
         # noinspection PyTypeChecker
-        return ExecutionsCallbackFeature(self, config)
+        self.callback_feature = ExecutionsCallbackFeature(self, config)
+        return self.callback_feature
 
     def assert_created_destinations(self, expected_names):
         communicators = self.get_communicators()
@@ -208,11 +233,13 @@ def format_body(self, body, communicator):
             return destination_email._body_dict_to_message(body)
         elif communicator.name.startswith('http'):
             return json.dumps(body)
+        elif communicator.name.startswith('script'):
+            return values_to_string(list(body.values()))
 
         return body
 
     def build_title(self, event_type, communicator, execution_id):
-        if communicator.name.startswith('http'):
+        if communicator.name.startswith('http') or communicator.name.startswith('script'):
             return None
 
         if event_type == 'execution_started':
@@ -256,10 +283,16 @@ def fire_started(self, execution_id):
         for listener in self.start_listeners:
             listener(execution_id)
 
+        if self.callback_feature:
+            self.callback_feature._wait()
+
     def fire_finished(self, execution_id):
         for listener in self.finish_listeners:
             listener(execution_id)
 
+        if self.callback_feature:
+            self.callback_feature._wait()
+
 
 class _ExecutionInfo:
     def __init__(self, execution_id, owner, pid, exit_code, script_name) -> None:

src/utils/process_utils.py

@@ -10,14 +10,20 @@
 LOGGER = logging.getLogger('script_server.process_utils')
 
 
-def invoke(command, work_dir='.'):
+def invoke(command, work_dir='.', *, environment_variables=None):
     if isinstance(command, str):
         command = split_command(command, working_directory=work_dir)
 
+    if environment_variables is not None:
+        env = dict(os.environ, **environment_variables)
+    else:
+        env = None
+
     p = subprocess.Popen(command,
                          stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE,
-                         cwd=work_dir)
+                         cwd=work_dir,
+                         env=env)
 
     (output_bytes, error_bytes) = p.communicate()
 

src/utils/string_utils.py

@@ -51,3 +51,16 @@ def is_blank(value):
     if not value.strip():
         return True
     return False
+
+
+def values_to_string(value):
+    if not value:
+        return value
+
+    if isinstance(value, dict):
+        return {k: str(v) for k, v in value.items()}
+
+    if isinstance(value, list):
+        return [str(element) for element in value]
+
+    return value