feat: Django 6.0

Author: adinhodovic

django_admin_shellx_custom_admin/admin.py

@@ -1,21 +1,75 @@
+from functools import update_wrapper
+
 from django.contrib import admin
-from django.urls import reverse
+from django.contrib.admin import autodiscover
+from django.urls import NoReverseMatch, re_path, reverse, reverse_lazy
 
 
 class CustomAdminSite(admin.AdminSite):
+    def _resolve_terminal_admin_url(self, app):
+        try:
+            return reverse("admin:django_admin_shellx_terminalcommand_terminal")
+        except NoReverseMatch:
+            pass
+
+        try:
+            changelist_url = reverse(
+                "admin:django_admin_shellx_terminalcommand_changelist"
+            )
+            return f"{changelist_url.rstrip('/')}/terminal/"
+        except NoReverseMatch:
+            pass
+
+        for model in app.get("models", []):
+            if model.get("object_name") == "TerminalCommand" and model.get("admin_url"):
+                return f"{model['admin_url'].rstrip('/')}/terminal/"
+
+        return None
+
+    def get_urls(self):
+        # Ensure admin modules are registered before URL patterns are built.
+        # Some projects resolve admin URLs very early, which can otherwise
+        # produce an empty registry and missing model/app routes.
+        autodiscover()
+
+        urls = super().get_urls()
+
+        def wrap(view, cacheable=False):
+            def wrapper(*args, **kwargs):
+                return self.admin_view(view, cacheable)(*args, **kwargs)
+
+            wrapper.admin_site = self
+            wrapper.login_url = reverse_lazy("admin:login", current_app=self.name)
+            return update_wrapper(wrapper, view)
+
+        # Keep a permissive app_list route available to avoid reverse failures
+        # when admin URLs were resolved before all apps were registered.
+        if not any(getattr(pattern, "name", None) == "app_list" for pattern in urls):
+            fallback = re_path(
+                r"^(?P<app_label>.+)/$", wrap(self.app_index), name="app_list"
+            )
+            if self.final_catch_all_view and urls:
+                urls.insert(-1, fallback)
+            else:
+                urls.append(fallback)
+
+        return urls
+
     def get_app_list(self, request, app_label=None):
         app_list = super().get_app_list(request, app_label)
 
         for app in app_list:
             if app["app_label"] == "django_admin_shellx":
+                terminal_url = self._resolve_terminal_admin_url(app)
+                if not terminal_url:
+                    break
+
                 app["models"].insert(
                     0,
                     {
                         "name": "Terminal",
                         "object_name": "Terminal",
-                        "admin_url": reverse(
-                            "admin:django_admin_shellx_terminalcommand_terminal"
-                        ),
+                        "admin_url": terminal_url,
                         "view_only": True,
                     },
                 )

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "django-admin-shellx"
-version = "0.3.3"
+version = "0.4.0"
 description = "A Django Admin Shell"
 readme = "README.md"
 requires-python = ">=3.12"

tests/test_commands.py

@@ -166,3 +166,67 @@ async def test_command_increments_execution_count(settings, superuser_logged_in)
     assert terminal_command.execution_count == 2
 
     await communicator.disconnect()
+
+
+@pytest.mark.asyncio
+@pytest.mark.django_db(transaction=True)
+async def test_save_history_ignores_unmapped_prompt(superuser_logged_in):
+    communicator = DefaultTimeoutWebsocketCommunicator(
+        TerminalConsumer.as_asgi(), "/testws/"
+    )
+    cast(Any, communicator.scope)["user"] = superuser_logged_in
+    connected, _ = await communicator.connect()
+    assert connected
+
+    # Ensure we go past the initial bash messages returned (shell startup)
+    await communicator.receive_from()
+
+    await communicator.send_to(
+        text_data=json.dumps(
+            {
+                "action": "save_history",
+                "data": {"command": "this output does not contain a shell prompt"},
+            }
+        )
+    )
+    await communicator.receive_from()
+
+    log_entry_count, _ = await get_log_entry()
+    terminal_command_count, _ = await get_terminal_command()
+
+    assert log_entry_count == 0
+    assert terminal_command_count == 0
+
+    await communicator.disconnect()
+
+
+@pytest.mark.asyncio
+@pytest.mark.django_db(transaction=True)
+async def test_save_history_ignores_reverse_search(superuser_logged_in):
+    communicator = DefaultTimeoutWebsocketCommunicator(
+        TerminalConsumer.as_asgi(), "/testws/"
+    )
+    cast(Any, communicator.scope)["user"] = superuser_logged_in
+    connected, _ = await communicator.connect()
+    assert connected
+
+    # Ensure we go past the initial bash messages returned (shell startup)
+    await communicator.receive_from()
+
+    await communicator.send_to(
+        text_data=json.dumps(
+            {
+                "action": "save_history",
+                "data": {"command": "(reverse-i-search)`ls': ls"},
+            }
+        )
+    )
+    await communicator.receive_from()
+
+    log_entry_count, _ = await get_log_entry()
+    terminal_command_count, _ = await get_terminal_command()
+
+    assert log_entry_count == 0
+    assert terminal_command_count == 0
+
+    await communicator.disconnect()

tests/test_custom_admin.py

@@ -0,0 +1,70 @@
+import pytest
+from django.contrib.auth import get_user_model
+from django.test import override_settings
+from django.urls import NoReverseMatch, clear_url_caches, path, reverse
+
+from django_admin_shellx_custom_admin.admin import CustomAdminSite
+
+pytestmark = pytest.mark.django_db
+
+urlpatterns = []
+
+
+def test_custom_admin_includes_fallback_app_list_route():
+    site = CustomAdminSite(name="admin")
+    urls = site.get_urls()
+    app_list_patterns = [
+        pattern for pattern in urls if getattr(pattern, "name", None) == "app_list"
+    ]
+
+    assert app_list_patterns
+    assert "app_label" in str(app_list_patterns[-1].pattern)
+
+
+def test_custom_admin_resolves_terminal_url_from_model_entry(monkeypatch):
+    site = CustomAdminSite(name="admin")
+
+    app = {
+        "app_label": "django_admin_shellx",
+        "models": [
+            {
+                "object_name": "TerminalCommand",
+                "admin_url": "/admin/django_admin_shellx/terminalcommand/",
+            }
+        ],
+    }
+
+    def always_fail_reverse(*_args, **_kwargs):
+        raise NoReverseMatch("missing")
+
+    monkeypatch.setattr(
+        "django_admin_shellx_custom_admin.admin.reverse", always_fail_reverse
+    )
+
+    assert (  # pylint: disable=protected-access
+        site._resolve_terminal_admin_url(app)
+        == "/admin/django_admin_shellx/terminalcommand/terminal/"
+    )
+
+
+@override_settings(ROOT_URLCONF="tests.test_custom_admin")
+def test_custom_admin_app_list_reverse_after_late_registration():
+    global urlpatterns  # pylint: disable=global-statement
+
+    site = CustomAdminSite(name="custom_admin")
+
+    try:
+        # Resolve admin URLs before any models are registered, to emulate projects
+        # where admin URLs are loaded early.
+        urlpatterns = [path("admin/", site.urls)]
+        clear_url_caches()
+
+        site.register(get_user_model())
+
+        assert (
+            reverse("custom_admin:app_list", kwargs={"app_label": "auth"})
+            == "/admin/auth/"
+        )
+    finally:
+        urlpatterns = []
+        clear_url_caches()

tests/test_views.py

@@ -120,3 +120,49 @@ def test_toggle_favorite_anonymous(client):
         f"toggle_favorite/{tc.id}/"
     )
     assert res.url == expected_url
+
+
+def test_toggle_favorite_post_does_not_change_value(admin_client):
+    tc = TerminalCommandFactory(favorite=False)
+    res = admin_client.post(
+        reverse(
+            "admin:django_admin_shellx_terminalcommand_toggle_favorite",
+            kwargs={"pk": tc.id},
+        )
+    )
+
+    assert res.status_code == 200
+    assert res.json() == {
+        "status": "error",
+        "message": "Only GET requests are allowed",
+    }
+    tc.refresh_from_db()
+    assert not tc.favorite
+
+
+def test_staff_user_allowed_when_superuser_not_required(settings, user_client):
+    settings.DJANGO_ADMIN_SHELLX_SUPERUSER_ONLY = False
+    user_client.user.is_staff = True
+    user_client.user.save()
+
+    tc = TerminalCommandFactory(favorite=False)
+    res = user_client.get(
+        reverse(
+            "admin:django_admin_shellx_terminalcommand_toggle_favorite",
+            kwargs={"pk": tc.id},
+        )
+    )
+
+    assert res.status_code == 200
+    tc.refresh_from_db()
+    assert tc.favorite
+
+
+def test_admin_index_contains_terminal_link(admin_client):
+    res = admin_client.get(reverse("admin:index"))
+
+    assert res.status_code == 200
+    terminal_url = reverse("admin:django_admin_shellx_terminalcommand_terminal")
+    content = res.content.decode()
+    assert "Terminal" in content
+    assert terminal_url in content

tests/test_websockets.py

@@ -1,15 +1,26 @@
 import json
 from asyncio import sleep
+from typing import Any, cast
 
 import pytest
+from channels.db import database_sync_to_async
+from channels.testing import WebsocketCommunicator
+from django.contrib.admin.models import LogEntry
 
 from django_admin_shellx.consumers import TerminalConsumer
+from django_admin_shellx.models import TerminalCommand
 
+from .asgi import application
 from .conftest import BASIC_BASH_COMMANDS, DefaultTimeoutWebsocketCommunicator
 
 pytestmark = pytest.mark.django_db
 
 
+@database_sync_to_async
+def get_history_counts():
+    return LogEntry.objects.count(), TerminalCommand.objects.count()
+
+
 @pytest.mark.asyncio
 async def test_websocket_rejects_unauthenticated():
     communicator = DefaultTimeoutWebsocketCommunicator(
@@ -21,6 +32,41 @@ async def test_websocket_rejects_unauthenticated():
     await communicator.disconnect()
 
 
+@pytest.mark.asyncio
+async def test_websocket_rejects_non_superuser_when_required(settings, user_logged_in):
+    settings.DJANGO_ADMIN_SHELLX_SUPERUSER_ONLY = True
+
+    communicator = DefaultTimeoutWebsocketCommunicator(
+        TerminalConsumer.as_asgi(), "/testws/"
+    )
+    cast(Any, communicator.scope)["user"] = user_logged_in
+    connected, subprotocol = await communicator.connect()
+    assert not connected
+    assert subprotocol == 4403
+    await communicator.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_websocket_rejects_disallowed_origin():
+    communicator = WebsocketCommunicator(
+        application,
+        "/ws/terminal/123/",
+        headers=[
+            (b"host", b"localhost:8000"),
+            (b"origin", b"http://evil.example.com"),
+        ],
+    )
+
+    connected, _ = await communicator.connect()
+    assert not connected
+
+    log_entry_count, terminal_command_count = await get_history_counts()
+    assert log_entry_count == 0
+    assert terminal_command_count == 0
+
+    await communicator.disconnect()
+
+
 @pytest.mark.asyncio
 async def test_websocket_accepts_authenticated_user(settings, user_logged_in):
     settings.DJANGO_ADMIN_SHELLX_SUPERUSER_ONLY = False