shopship/apps/telemetry/tests.py

from datetime import datetime, timedelta
from datetime import timezone as dt_timezone

from django.test import TestCase
from django.urls import reverse
from django.utils import timezone
from rest_framework.test import APIClient
from rest_framework_simplejwt.tokens import RefreshToken

from apps.account.models import User
from apps.telemetry.admin import AppSessionAdmin
from apps.telemetry.api.services import TelemetryIngestService
from apps.telemetry.models import ApiCallLog, AppSession, ClickEvent, ClientErrorLog, PageView


def _make_user(phone="09120000000") -> User:
    return User.objects.create_user(phone=phone, is_active=True, state=User.State.ACTIVE)


def _iso(text: str) -> datetime:
    return datetime.fromisoformat(text).replace(tzinfo=dt_timezone.utc)


class TelemetryIngestServiceTests(TestCase):
    def setUp(self):
        self.user = _make_user()

    def test_ingest_creates_session_and_events(self):
        TelemetryIngestService.ingest(
            self.user,
            "5.6.7.8",
            {
                "session_id": "sess-1",
                "platform": "android",
                "app_version": "1.0.0",
                "events": [
                    {"type": "session_start", "at": _iso("2026-01-01T10:00:00")},
                    {
                        "type": "page_view",
                        "screen": "LauncherScreen",
                        "at": _iso("2026-01-01T10:00:05"),
                        "duration_seconds": 12.5,
                        "element": "",
                        "x": None,
                        "y": None,
                        "message": "",
                        "stack": "",
                    },
                    {
                        "type": "click",
                        "screen": "LauncherScreen",
                        "element": "invoice_tile",
                        "at": _iso("2026-01-01T10:00:10"),
                        "x": 0.4,
                        "y": 0.5,
                        "duration_seconds": None,
                        "message": "",
                        "stack": "",
                    },
                    {
                        "type": "error",
                        "screen": "InvoiceHomeScreen",
                        "message": "boom",
                        "stack": "trace...",
                        "at": _iso("2026-01-01T10:01:00"),
                        "duration_seconds": None,
                        "element": "",
                        "x": None,
                        "y": None,
                    },
                    {"type": "session_end", "at": _iso("2026-01-01T10:05:00")},
                ],
            },
        )
        session = AppSession.objects.get(id="sess-1")
        self.assertEqual(session.user, self.user)
        self.assertEqual(session.ip_address, "5.6.7.8")
        self.assertEqual(session.duration_seconds, 300)
        self.assertEqual(PageView.objects.filter(session=session).count(), 1)
        self.assertEqual(ClickEvent.objects.filter(session=session).count(), 1)
        self.assertEqual(ClientErrorLog.objects.filter(session=session).count(), 1)

    def test_ingest_is_idempotent_for_same_session_id(self):
        payload = {
            "session_id": "sess-2",
            "platform": "ios",
            "app_version": "1.0.0",
            "events": [{"type": "session_start", "at": _iso("2026-01-01T10:00:00")}],
        }
        TelemetryIngestService.ingest(self.user, "1.1.1.1", payload)
        TelemetryIngestService.ingest(self.user, "1.1.1.1", payload)
        self.assertEqual(AppSession.objects.filter(id="sess-2").count(), 1)


class TelemetryApiTests(TestCase):
    def setUp(self):
        self.user = _make_user()
        self.client = APIClient()
        token = str(RefreshToken.for_user(self.user).access_token)
        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")

    def test_requires_authentication(self):
        anon = APIClient()
        response = anon.post(reverse("telemetry:events"), {"session_id": "x", "events": []}, format="json")
        self.assertEqual(response.status_code, 401)

    def test_rejects_empty_events(self):
        response = self.client.post(
            reverse("telemetry:events"), {"session_id": "sess-3", "events": []}, format="json"
        )
        self.assertEqual(response.status_code, 400)

    def test_accepts_valid_batch(self):
        response = self.client.post(
            reverse("telemetry:events"),
            {
                "session_id": "sess-4",
                "platform": "android",
                "events": [
                    {"type": "session_start", "at": "2026-01-01T10:00:00Z"},
                    {"type": "page_view", "screen": "LedgerHomeScreen", "at": "2026-01-01T10:00:05Z"},
                ],
            },
            format="json",
        )
        self.assertEqual(response.status_code, 201)
        self.assertTrue(AppSession.objects.filter(id="sess-4", user=self.user).exists())


class ApiCallLoggingMiddlewareTests(TestCase):
    """میدل‌ور باید خودکار (بدون هیچ تغییری توی اپ فلاتر) هر فراخوانی API را
    با کاربر، آی‌پی و زمانِ پاسخ لاگ کند."""

    def setUp(self):
        self.user = _make_user()
        self.client = APIClient()
        token = str(RefreshToken.for_user(self.user).access_token)
        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")

    def test_api_call_is_logged_with_user_and_status(self):
        ApiCallLog.objects.all().delete()
        self.client.get("/api/bizapps/customers/")
        log = ApiCallLog.objects.filter(path="/api/bizapps/customers/").first()
        self.assertIsNotNone(log)
        self.assertEqual(log.user, self.user)
        self.assertEqual(log.method, "GET")
        self.assertGreaterEqual(log.duration_ms, 0)

    def test_non_api_paths_are_not_logged(self):
        ApiCallLog.objects.all().delete()
        self.client.get("/admin/login/")
        self.assertEqual(ApiCallLog.objects.count(), 0)


class DashboardContextTests(TestCase):
    """داشبوردِ پنل ادمین — همان چیزی که کاربر خواسته: زمانِ هر کاربر توی
    اپ، صفحاتِ بازدیدشده، آی‌پی‌ها، تعداد دعوت، و عملکردِ اندپوینت‌ها."""

    def setUp(self):
        self.user = _make_user(phone="09121110000")
        self.invited = _make_user(phone="09121110001")
        self.invited.referred_by = self.user
        self.invited.referral_bonus_granted = True
        self.invited.save()

    def test_dashboard_aggregates_per_user_stats(self):
        # عمداً نسبت‌به الان (نه یک تاریخِ ثابت) — داشبورد فقط بازه‌ی ۳۰
        # روزِ اخیر را نشان می‌دهد، پس تاریخِ ثابتِ گذشته باعثِ فیلترشدنِ
        # همین دیتای تست می‌شد.
        base = timezone.now() - timedelta(hours=1)
        TelemetryIngestService.ingest(
            self.user,
            "9.9.9.9",
            {
                "session_id": "sess-dash",
                "platform": "android",
                "app_version": "1.0.0",
                "events": [
                    {"type": "session_start", "at": base},
                    {
                        "type": "page_view",
                        "screen": "InvoiceHomeScreen",
                        "at": base,
                        "duration_seconds": 20,
                        "element": "",
                        "x": None,
                        "y": None,
                        "message": "",
                        "stack": "",
                    },
                    {"type": "session_end", "at": base + timedelta(minutes=10)},
                ],
            },
        )
        ApiCallLog.objects.create(
            user=self.user, method="GET", path="/api/bizapps/invoices/", status_code=200, duration_ms=50
        )

        context = AppSessionAdmin(AppSession, None)._build_dashboard_context()
        rows = {r["user"].id: r for r in context["user_rows"]}
        self.assertIn(self.user.id, rows)
        row = rows[self.user.id]
        self.assertEqual(row["session_count"], 1)
        self.assertEqual(row["total_minutes"], 10.0)
        self.assertEqual(row["page_view_count"], 1)
        self.assertEqual(row["api_call_count"], 1)
        self.assertEqual(row["referral_count"], 1)
        self.assertEqual(row["ips"], ["9.9.9.9"])
        self.assertEqual(context["overview"]["sessions_window"], 1)