diff --git a/README.rst b/README.rst index 39694cf..a0e9615 100644 --- a/README.rst +++ b/README.rst @@ -232,6 +232,35 @@ If you want to persist indefinitly all declared variables on the shell, set this **BEWARE**: *leaving this disabled is not recomended on production code!* +ADMIN_SHELL_CALLBACK +^^^^^^^^^^^^^^^^^^^^^^^^^^^ +*type* : **string** + +*default* : **None** + +This setting allows you to specify a callback function that will be called after each code execution in the admin shell. The callback function should be provided as a string in the format "path.to.module.function_name". + +The callback function will receive a dictionary with the following keys: +- 'request': The current request object +- 'user': The user who executed the code +- 'code': The code that was executed +- 'response': The result of the code execution +- 'timestamp': The time when the code was executed + +This can be useful for logging or auditing purposes (e.g. saving executed code and its result to database). For example: + +.. code-block:: python + + def my_callback(callback_data): + print(f"User {callback_data['user']} executed code at {callback_data['timestamp']}") + + # In your settings.py + ADMIN_SHELL_CALLBACK = 'path.to.my_callback' + +If the callback function is defined but cannot be imported or raises an exception, a warning will be logged. + +**Note**: Be careful when implementing the callback function, as it will be called for every code execution in the admin shell. + Code examples ------------- diff --git a/django_admin_shell/settings.py b/django_admin_shell/settings.py index a41c7b7..5861e75 100644 --- a/django_admin_shell/settings.py +++ b/django_admin_shell/settings.py @@ -72,3 +72,8 @@ def from_settings_or_default(name, default): 'ADMIN_SHELL_CLEAR_SCOPE_ON_CLEAR_HISTORY', False ) + +ADMIN_SHELL_CALLBACK = from_settings_or_default( + 'ADMIN_SHELL_CALLBACK', + None +) diff --git a/django_admin_shell/tests/test_view.py b/django_admin_shell/tests/test_view.py index 717d9c1..150644f 100644 --- a/django_admin_shell/tests/test_view.py +++ b/django_admin_shell/tests/test_view.py @@ -208,3 +208,62 @@ def test_clear_scope(self): assert ADMIN_SHELL_SESSION_KEY in self.client_auth.session assert self.client_auth.session[ADMIN_SHELL_SESSION_KEY] == [] assert 'a' not in self.view.runner.importer.get_scope() + + @override_settings(DEBUG=True) + @mock.patch( + 'django_admin_shell.views.ADMIN_SHELL_CALLBACK', + 'django_admin_shell.tests.utils.callback') + @mock.patch('django_admin_shell.tests.utils.callback') + def test_callback_function(self, mock_callback) -> None: + """ + Show that the callback function is called with the correct arguments. + """ + self.user.is_staff = True + self.user.is_superuser = True + self.user.save() + + response = self.client_auth.post(self.url, {"code": 'print("Hello, World!")'}) + + assert mock_callback.called + + callback_data = mock_callback.call_args[0][0] + assert 'request' in callback_data + assert 'user' in callback_data + assert 'code' in callback_data + assert 'response' in callback_data + assert 'timestamp' in callback_data + assert callback_data['code'] == 'print("Hello, World!")' + assert callback_data['response']['status'] == 'success' + assert callback_data['response']['out'] == 'Hello, World!\n' + + assert response.status_code == 302 + session = self.client_auth.session[ADMIN_SHELL_SESSION_KEY] + assert len(session) == 1 + assert session[0]['code'] == 'print("Hello, World!")' + assert session[0]['status'] == 'success' + assert session[0]['out'] == 'Hello, World!\n' + + @override_settings(DEBUG=True) + @mock.patch( + 'django_admin_shell.views.ADMIN_SHELL_CALLBACK', + 'django_admin_shell.tests.utils.callback') + @mock.patch('django_admin_shell.tests.utils.callback') + def test_callback_function_error(self, mock_callback) -> None: + self.user.is_staff = True + self.user.is_superuser = True + self.user.save() + + def mock_callback_error(callback_data): + raise Exception("Test exception") + + mock_callback.side_effect = mock_callback_error + + with self.assertWarns(RuntimeWarning): + response = self.client_auth.post(self.url, {"code": 'print("Hello, World!")'}) + + assert response.status_code == 302 + session = self.client_auth.session[ADMIN_SHELL_SESSION_KEY] + assert len(session) == 1 + assert session[0]['code'] == 'print("Hello, World!")' + assert session[0]['status'] == 'success' + assert session[0]['out'] == 'Hello, World!\n' diff --git a/django_admin_shell/tests/utils.py b/django_admin_shell/tests/utils.py new file mode 100644 index 0000000..234707d --- /dev/null +++ b/django_admin_shell/tests/utils.py @@ -0,0 +1,3 @@ +# Just for testing +def callback(callback_data): + pass diff --git a/django_admin_shell/views.py b/django_admin_shell/views.py index 3893afc..6c7d0f8 100644 --- a/django_admin_shell/views.py +++ b/django_admin_shell/views.py @@ -1,12 +1,16 @@ # encoding: utf-8 from django.apps import apps from django.views.generic import FormView + +from django.utils.module_loading import import_string + from .forms import ShellForm from django.http import ( HttpResponseForbidden, HttpResponseNotFound ) from django.conf import settings +from django.utils import timezone try: # Only for python 2 @@ -24,7 +28,8 @@ ADMIN_SHELL_IMPORT_DJANGO, ADMIN_SHELL_IMPORT_DJANGO_MODULES, ADMIN_SHELL_IMPORT_MODELS, - ADMIN_SHELL_CLEAR_SCOPE_ON_CLEAR_HISTORY + ADMIN_SHELL_CLEAR_SCOPE_ON_CLEAR_HISTORY, + ADMIN_SHELL_CALLBACK ) import django @@ -234,11 +239,14 @@ def get(self, request, *args, **kwargs): return super(ShellView, self).get(request, *args, **kwargs) def form_valid(self, form): + code = form.cleaned_data.get("code", "") + result = None if len(code.strip()) > 0: result = self.runner.run_code(code) self.add_to_outout(result) self.save_output() + self.call_callback(self.request, result, code) return super(ShellView, self).form_valid(form) def get_context_data(self, **kwargs): @@ -251,3 +259,36 @@ def get_context_data(self, **kwargs): ctx['django_version'] = get_dj_version() ctx['auto_import'] = str(self.runner.importer) return ctx + + def call_callback(self, request, response, code) -> None: + callback_string = ADMIN_SHELL_CALLBACK + if not callback_string: + return + try: + callback = import_string(callback_string) + except Exception as e: + warnings.warn( + f"Error in trying to import callback function: {str(e)}", + RuntimeWarning + ) + return + if not callable(callback): + warnings.warn( + f"ADMIN_SHELL_CALLBACK is set but is not callable: {callback_string}", + RuntimeWarning + ) + return + try: + callback_data = { + 'request': request, + 'user': request.user, + 'code': code, + 'response': response, + 'timestamp': timezone.now() + } + callback(callback_data) + except Exception as e: + warnings.warn( + f"Error in ADMIN_SHELL_CALLBACK: {str(e)}", + RuntimeWarning + )