Admin Panel

from component_framework.core import registry
from component_framework.adapters.django_model import DjangoModelComponent
from django.contrib.auth import get_user_model

User = get_user_model()

PAGE_SIZE = 25


@registry.register("user_list")
class UserListComponent(DjangoModelComponent):
    """
    Searchable, paginated user management table for staff.

    State shape:
        query      (str):  current search string
        page       (int):  current 1-based page number
        users      (list): serialised user rows for the current page
        total      (int):  total matching user count
        page_count (int):  total pages
    """

    template_name = "admin_panel/user_list.html"
    model = User
    state_fields = ["username", "email", "is_active", "date_joined"]

    def mount(self):
        self.state.update({
            "query":      "",
            "page":       1,
            "users":      [],
            "total":      0,
            "page_count": 1,
        })
        self._load_page()

    def _load_page(self):
        """Run the DB query for the current page and search query."""
        qs = User.objects.order_by("-date_joined")

        query = self.state.get("query", "").strip()
        if query:
            qs = qs.filter(
                username__icontains=query
            ) | qs.filter(email__icontains=query)

        total = qs.count()
        page  = self.state["page"]
        start = (page - 1) * PAGE_SIZE
        users = qs[start : start + PAGE_SIZE].values(
            "id", "username", "email",
            "is_active", "is_staff", "date_joined",
        )

        self.state.update({
            "users":      [dict(u) for u in users],
            "total":      total,
            "page_count": max(1, -(-total // PAGE_SIZE)),  # ceil div
        })

    def on_search(self, query: str):
        """User typed into the search box — reset to page 1 and reload."""
        self.state["query"] = query
        self.state["page"]  = 1
        self._load_page()

    def on_toggle_status(self, user_id: int):
        """Toggle a user's is_active flag and re-render the row."""
        user = User.objects.get(pk=user_id)
        user.is_active = not user.is_active
        user.save(update_fields=["is_active"])

        # Reflect the change in state without a full page reload
        for row in self.state["users"]:
            if row["id"] == user_id:
                row["is_active"] = user.is_active
                break

    def on_change_page(self, page: int):
        """Navigate to a different page."""
        self.state["page"] = max(1, min(page, self.state["page_count"]))
        self._load_page()

from component_framework.adapters.django_views import ComponentView
from component_framework.core.permissions import IsStaff


class UserListView(ComponentView):
    """
    Staff-only endpoint for the user management table.

    - Non-staff users receive JSON 403 — HTMX handles the UI gracefully.
    - No redirect that would break the HTMX partial-swap flow.
    """

    component_name = "user_list"
    permission_classes = [IsStaff]

from django.urls import path
from .views import UserListView, MetricsView, UserEditView

urlpatterns = [
    path("components/users/",     UserListView.as_view(),   name="admin_users"),
    path("components/metrics/",   MetricsView.as_view(),  name="admin_metrics"),
    path("components/user-edit/", UserEditView.as_view(), name="admin_user_edit"),
]

<div id="user-list-{{ component_id }}"
     hx-target="this"
     hx-swap="outerHTML">

  {# Search bar — triggers on_search after 400 ms debounce #}
  <input
    type="search"
    name="query"
    value="{{ state.query }}"
    placeholder="Search users…"
    hx-post="{% url 'admin_users' %}"
    hx-vals='{"event": "search"}'
    hx-trigger="input changed delay:400ms" />

  <p class="result-count">{{ state.total }} users</p>

  <table>
    <thead><tr>
      <th>Username</th>
      <th>Email</th>
      <th>Joined</th>
      <th>Status</th>
      <th>Actions</th>
    </tr></thead>
    <tbody>
    {% for user in state.users %}
      <tr>
        <td>{{ user.username }}</td>
        <td>{{ user.email }}</td>
        <td>{{ user.date_joined|date:"Y-m-d" }}</td>
        <td>
          <span class="badge {% if user.is_active %}badge--active{% else %}badge--inactive{% endif %}">
            {% if user.is_active %}Active{% else %}Inactive{% endif %}
          </span>
        </td>
        <td>
          {# Toggle button — instant feedback via HTMX #}
          <button
            hx-post="{% url 'admin_users' %}"
            hx-vals='{"event": "toggle_status", "user_id": "{{ user.id }}"}'>
            {% if user.is_active %}Deactivate{% else %}Activate{% endif %}
          </button>
        </td>
      </tr>
    {% endfor %}
    </tbody>
  </table>

  {# Pagination #}
  <div class="pagination">
    {% if state.page > 1 %}
      <button
        hx-post="{% url 'admin_users' %}"
        hx-vals='{"event": "change_page", "page": "{{ state.page|add:"-1" }}"}'>
        ← Prev
      </button>
    {% endif %}
    <span>Page {{ state.page }} of {{ state.page_count }}</span>
    {% if state.page < state.page_count %}
      <button
        hx-post="{% url 'admin_users' %}"
        hx-vals='{"event": "change_page", "page": "{{ state.page|add:"1" }}"}'>
        Next →
      </button>
    {% endif %}
  </div>

</div>

from component_framework.core import registry
from component_framework.core.component import Component
from component_framework.adapters.django_views import CacheMixin
from .models import AuditEntry

MAX_ENTRIES = 50  # keep the last 50 entries in state


@registry.register("audit_log")
class AuditLogComponent(CacheMixin, Component):
    """
    Real-time audit log feed.

    Initial load is cached for 30 seconds — subsequent pushes arrive via
    WebSocket and bypass the cache entirely.

    State shape:
        entries (list): [{id, actor, action, target, timestamp}]
    """

    template_name = "admin_panel/audit_log.html"
    cache_ttl     = 30  # seconds — short TTL; WS keeps it live after that

    def get_cache_key(self, name, params, state) -> str:
        return "admin:audit_log:initial"

    def mount(self):
        entries = (
            AuditEntry.objects
            .select_related("actor")
            .order_by("-timestamp")[:MAX_ENTRIES]
            .values("id", "actor__username", "action", "target", "timestamp")
        )
        self.state["entries"] = [dict(e) for e in entries]

    def on_new_entry(self, entry: dict):
        """
        Called by the WebSocket consumer when a new audit entry is saved.

        Prepend the new entry and trim the list to MAX_ENTRIES so the
        in-browser state never grows unbounded.
        """
        self.state["entries"] = [entry] + self.state["entries"][:MAX_ENTRIES - 1]

from django.db.models.signals import post_save
from django.dispatch import receiver
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from .models import AuditEntry


@receiver(post_save, sender=AuditEntry)
def broadcast_audit_entry(sender, instance, created, **kwargs):
    """Broadcast every new audit entry to connected admin consoles."""
    if not created:
        return

    layer = get_channel_layer()
    async_to_sync(layer.group_send)(
        "admin-audit",
        {
            "type":  "audit.entry",
            "entry": {
                "id":             instance.pk,
                "actor__username": instance.actor.username,
                "action":         instance.action,
                "target":         instance.target,
                "timestamp":      instance.timestamp.isoformat(),
            },
        },
    )

from component_framework.core import ws_manager
import json


class AuditLogConsumer(ws_manager.ComponentConsumer):
    """
    WebSocket consumer that keeps the audit log widget live.

    The parent class handles:
      - authenticating the connection (scope["user"].is_staff)
      - dispatching component events from incoming WS messages
      - pushing rendered HTML back to the browser
    """

    component_name = "audit_log"

    async def websocket_connect(self, message):
        user = self.scope["user"]
        if not (user.is_authenticated and user.is_staff):
            await self.close()
            return
        await self.channel_layer.group_add("admin-audit", self.channel_name)
        await super().websocket_connect(message)

    async def audit_entry(self, event):
        """Receive a broadcast from the signal and dispatch as a component event."""
        await self.dispatch_event(
            event="new_entry",
            payload={"entry": event["entry"]},
        )

    async def websocket_disconnect(self, message):
        await self.channel_layer.group_discard("admin-audit", self.channel_name)
        await super().websocket_disconnect(message)

from django.utils import timezone
from django.db.models import Sum, Count
from .models import Subscription, UserActivity


@registry.register("metrics")
class MetricsComponent(CacheMixin, Component):
    """
    KPI dashboard: MAU, MRR, and churn rate.

    Expensive aggregations are cached for 5 minutes.  The on_refresh event
    calls self.invalidate_cache() so staff can force fresh numbers without
    waiting for the TTL to expire.

    State shape:
        mau        (int):   monthly active users
        mrr        (str):   monthly recurring revenue (Decimal as string)
        churn_rate (float): percentage of users who cancelled last 30 days
        refreshed_at (str): ISO timestamp of last DB read
    """

    template_name = "admin_panel/metrics.html"
    cache_ttl     = 300  # 5 minutes

    def get_cache_key(self, name, params, state) -> str:
        return "admin:metrics"

    def mount(self):
        self._compute_metrics()

    def _compute_metrics(self):
        """Run all three ORM aggregations and update state."""
        now   = timezone.now()
        month = now - timezone.timedelta(days=30)

        # Monthly Active Users
        mau = (
            UserActivity.objects
            .filter(last_seen__gte=month)
            .values("user_id")
            .distinct()
            .count()
        )

        # Monthly Recurring Revenue
        mrr = (
            Subscription.objects
            .filter(status="active")
            .aggregate(total=Sum("monthly_amount"))["total"]
            or 0
        )

        # Churn Rate (cancellations last 30 days / total subs at month start)
        total_start = Subscription.objects.filter(created__lt=month).count()
        churned     = Subscription.objects.filter(
            cancelled_at__gte=month, cancelled_at__lte=now
        ).count()
        churn_rate = round((churned / total_start * 100) if total_start else 0.0, 2)

        self.state.update({
            "mau":          mau,
            "mrr":          str(mrr),
            "churn_rate":   churn_rate,
            "refreshed_at": now.isoformat(),
        })

    def on_refresh(self):
        """Staff clicked the Refresh button — bust the cache and recompute."""
        self.invalidate_cache()
        self._compute_metrics()

from pydantic import BaseModel, EmailStr, field_validator
from component_framework.core.form import FormComponent
from component_framework.adapters.django_views import OptimisticMixin


class UserEditSchema(BaseModel):
    """Pydantic schema for the user-edit form."""
    username:      str
    email:         EmailStr
    first_name:    str
    last_name:     str
    is_active:     bool
    # Role flags — only serialised when the requesting user is a superuser.
    # The component strips them for non-superusers before validation.
    is_staff:      bool = False
    is_superuser:  bool = False

    @field_validator("username")
    @classmethod
    def username_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Username cannot be blank.")
        return v.strip()


@registry.register("user_edit")
class UserEditComponent(OptimisticMixin, FormComponent):
    """
    Inline edit form for a single Django User.

    permission_classes = [IsSuperuser] ensures that:
      - Staff users can submit the form but role fields are stripped.
      - Non-staff users receive JSON 403 immediately.

    Optimistic UI shows a "Saving…" state immediately on submit.
    """

    template_name    = "admin_panel/user_edit.html"
    schema           = UserEditSchema
    # Note: permission is enforced at the VIEW level (IsSuperuser)
    # so only superusers ever reach on_submit with role fields intact.

    def mount(self):
        user_id = self.params["user_id"]
        user    = User.objects.get(pk=user_id)
        self.state.update({
            "user_id":     user.pk,
            "username":    user.username,
            "email":       user.email,
            "first_name":  user.first_name,
            "last_name":   user.last_name,
            "is_active":   user.is_active,
            "is_staff":    user.is_staff,
            "is_superuser": user.is_superuser,
            "saved":       False,
            "error":       None,
        })

    def on_submit(self):
        """
        Validate with Pydantic, then save.

        self.validated_data is populated by FormComponent after schema
        validation.  If validation fails, self.errors is populated and
        on_submit is not called.
        """
        data = self.validated_data
        user = User.objects.get(pk=self.state["user_id"])

        user.username    = data.username
        user.email       = data.email
        user.first_name  = data.first_name
        user.last_name   = data.last_name
        user.is_active   = data.is_active
        user.is_staff    = data.is_staff
        user.is_superuser = data.is_superuser
        user.save()

        self.state["saved"] = True

    # ── Optimistic UI ────────────────────────────────────────────────────
    def get_optimistic_patch(self, event: str, payload: dict) -> dict | None:
        """Show a 'Saving…' indicator immediately on submit."""
        if event == "submit":
            return {"saving": True, "error": None}
        return None

from component_framework.core.permissions import IsSuperuser


class UserEditView(ComponentView):
    """
    Superuser-only endpoint for the inline user-edit form.

    IsStaff users can VIEW the admin panel but cannot POST to this endpoint.
    Any attempt returns JSON 403 — HTMX shows the error inline without
    a page reload.
    """

    component_name    = "user_edit"
    permission_classes = [IsSuperuser]

    def get_component_params(self, request, **kwargs):
        params = super().get_component_params(request, **kwargs)
        params["user_id"] = int(request.POST.get("user_id", 0))
        return params

from django.test import TestCase
from django.contrib.auth import get_user_model
from component_framework.testing import ComponentTestCase, MockRenderer
from admin_panel.components import UserListComponent

User = get_user_model()


class TestUserListComponent(ComponentTestCase, TestCase):

    def setUp(self):
        UserListComponent.renderer = MockRenderer()

        # Seed the database with test users
        User.objects.create_user(
            username="alice", email="alice@example.com", is_active=True
        )
        User.objects.create_user(
            username="bob",   email="bob@example.com",   is_active=True
        )
        User.objects.create_user(
            username="carol", email="carol@example.com", is_active=False
        )

    def test_mount_loads_all_users(self):
        component = self.make_component(UserListComponent)
        component.mount()

        self.assertEqual(component.state["total"], 3)
        self.assertEqual(len(component.state["users"]), 3)

    def test_search_filters_results(self):
        component = self.make_component(UserListComponent)
        component.mount()
        component.on_search("alice")

        self.assertEqual(component.state["total"], 1)
        self.assertEqual(component.state["users"][0]["username"], "alice")

    def test_search_resets_to_page_one(self):
        component = self.make_component(UserListComponent)
        component.mount()
        component.state["page"] = 3
        component.on_search("bob")

        self.assertEqual(component.state["page"], 1)

    def test_toggle_status_updates_state(self):
        component = self.make_component(UserListComponent)
        component.mount()

        alice = User.objects.get(username="alice")
        self.assertTrue(alice.is_active)

        component.on_toggle_status(alice.pk)

        alice.refresh_from_db()
        self.assertFalse(alice.is_active)

    def test_toggle_inactive_user_activates(self):
        component = self.make_component(UserListComponent)
        component.mount()

        carol = User.objects.get(username="carol")
        self.assertFalse(carol.is_active)

        component.on_toggle_status(carol.pk)

        carol.refresh_from_db()
        self.assertTrue(carol.is_active)

from unittest.mock import patch, MagicMock
from django.test import TestCase
from component_framework.testing import ComponentTestCase, MockRenderer
from admin_panel.components import MetricsComponent


class TestMetricsComponent(ComponentTestCase, TestCase):

    def setUp(self):
        MetricsComponent.renderer = MockRenderer()

    def test_mount_populates_kpis(self):
        with patch.object(
            MetricsComponent, "_compute_metrics",
            lambda self: self.state.update({
                "mau": 1234, "mrr": "9800.00", "churn_rate": 2.1,
            }),
        ):
            component = self.make_component(MetricsComponent)
            component.mount()

        self.assertEqual(component.state["mau"], 1234)
        self.assertEqual(component.state["mrr"], "9800.00")
        self.assertEqual(component.state["churn_rate"], 2.1)

    def test_cache_hit_skips_db(self):
        """
        When the cache returns a state dict, mount() should not be called.
        Verify by checking that _compute_metrics is never invoked.
        """
        cached_state = {
            "mau": 999, "mrr": "5000.00",
            "churn_rate": 1.5, "refreshed_at": "2025-01-01T00:00:00",
        }
        with patch(
            "django.core.cache.cache.get",
            return_value=cached_state,
        ) as mock_cache, \
        patch.object(
            MetricsComponent, "_compute_metrics"
        ) as mock_compute:
            component = self.make_component(MetricsComponent)
            component.mount()

        mock_compute.assert_not_called()
        self.assertEqual(component.state["mau"], 999)

    def test_refresh_invalidates_cache(self):
        """on_refresh() should bust the cache and recompute."""
        with patch.object(
            MetricsComponent, "invalidate_cache"
        ) as mock_inv, \
        patch.object(
            MetricsComponent, "_compute_metrics"
        ) as mock_compute:
            component = self.make_component(MetricsComponent)
            component.on_refresh()

        mock_inv.assert_called_once()
        mock_compute.assert_called_once()

pytest admin_panel/tests/ -v