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)