from datetime import timedelta from django.test import TestCase, override_settings from django.urls import reverse from django.utils import timezone from rest_framework.test import APIClient from apps.licensing.models import ClientDevice, DeviceBindingAudit, DeviceSession from apps.licensing.services import record_device_heartbeat from apps.users.models import ApiKey, User, UserWallet class DeviceRegistrationApiTests(TestCase): def setUp(self): self.user = User.objects.create_user( username="device-user", email="device@example.com", password="test-password", ) UserWallet.objects.create(user=self.user, points_balance=20) self.api_key, self.raw_api_key = ApiKey.create_for_user(self.user, name="desktop") self.client = APIClient() self.register_url = reverse("api-client-device-register") self.heartbeat_url = reverse("api-client-device-heartbeat") self.payload = { "product_code": ClientDevice.ProductCode.CMSHOPEE, "device_id": "v1:installed-device-abcdef", "device_id_version": "v1", "installation_public_key": "test-installation-public-key", "platform": ClientDevice.Platform.WINDOWS, "client_version": "0.1.0", } def register_device(self): self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.raw_api_key}") return self.client.post(self.register_url, self.payload, format="json") def test_register_requires_api_key(self): response = self.client.post(self.register_url, self.payload, format="json") self.assertEqual(response.status_code, 401) self.assertEqual(response.data["error"]["code"], "unauthorized") self.assertFalse(ClientDevice.objects.exists()) def test_register_creates_hashed_device_and_one_time_session_token(self): response = self.register_device() self.assertEqual(response.status_code, 201) self.assertEqual(response.data["device"]["product_code"], "cmshopee") self.assertIn("device_session_token", response.data) self.assertNotIn("installation_public_key", response.data) self.assertNotIn("device_id", response.data) device = ClientDevice.objects.get(user=self.user) session = DeviceSession.objects.get(device=device) raw_token = response.data["device_session_token"] self.assertNotEqual(device.device_fingerprint, self.payload["device_id"]) self.assertNotEqual(device.public_key_fingerprint, self.payload["installation_public_key"]) self.assertNotEqual(session.token_hash, raw_token) self.assertTrue(session.matches_token(raw_token)) self.assertEqual( DeviceBindingAudit.objects.filter( device=device, action=DeviceBindingAudit.Action.REGISTERED, ).count(), 1, ) def test_repeat_registration_reuses_device_and_rotates_session(self): first = self.register_device() second = self.register_device() self.assertEqual(first.status_code, 201) self.assertEqual(second.status_code, 200) self.assertEqual(ClientDevice.objects.filter(user=self.user).count(), 1) device = ClientDevice.objects.get(user=self.user) self.assertEqual( DeviceSession.objects.filter(device=device, revoked_at__isnull=True).count(), 1, ) self.assertEqual(DeviceSession.objects.filter(device=device).count(), 2) self.assertNotEqual( first.data["device_session_token"], second.data["device_session_token"], ) @override_settings(DEVICE_SESSION_TTL_SECONDS=172800) def test_heartbeat_uses_device_session_and_activity_is_throttled(self): registration = self.register_device() token = registration.data["device_session_token"] self.client.credentials(HTTP_X_DEVICE_SESSION=token) response = self.client.post(self.heartbeat_url, {}, format="json") self.assertEqual(response.status_code, 200) self.assertFalse(response.data["activity_updated"]) device = ClientDevice.objects.get(user=self.user) session = DeviceSession.objects.get(device=device, revoked_at__isnull=True) updated = record_device_heartbeat( session, now=device.last_seen_at + timedelta(days=1, seconds=1), ) self.assertTrue(updated) device.refresh_from_db() self.assertEqual( DeviceBindingAudit.objects.filter( device=device, action=DeviceBindingAudit.Action.HEARTBEAT, ).count(), 1, ) def test_invalid_or_revoked_device_session_is_rejected(self): self.client.credentials(HTTP_X_DEVICE_SESSION="dvs_cmhub_invalid") invalid = self.client.post(self.heartbeat_url, {}, format="json") self.assertEqual(invalid.status_code, 401) self.assertEqual(invalid.data["error"]["code"], "device_session_invalid") registration = self.register_device() device = ClientDevice.objects.get(user=self.user) device.status = ClientDevice.Status.REVOKED device.save(update_fields=("status",)) self.client.credentials(HTTP_X_DEVICE_SESSION=registration.data["device_session_token"]) revoked = self.client.post(self.heartbeat_url, {}, format="json") self.assertEqual(revoked.status_code, 403) self.assertEqual(revoked.data["error"]["code"], "device_revoked") def test_existing_balance_api_remains_compatible_without_device_header(self): self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.raw_api_key}") response = self.client.get(reverse("api-balance")) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["points_balance"], 20) @override_settings(DEVICE_ACTIVITY_UPDATE_SECONDS=60) def test_heartbeat_updates_when_device_is_stale(self): registration = self.register_device() device = ClientDevice.objects.get(user=self.user) stale_time = timezone.now() - timedelta(minutes=2) ClientDevice.objects.filter(pk=device.pk).update(last_seen_at=stale_time) self.client.credentials(HTTP_X_DEVICE_SESSION=registration.data["device_session_token"]) response = self.client.post(self.heartbeat_url, {}, format="json") self.assertEqual(response.status_code, 200) self.assertTrue(response.data["activity_updated"]) device.refresh_from_db() self.assertGreater(device.last_seen_at, stale_time) @override_settings(DEVICE_ACTIVITY_UPDATE_SECONDS=60) def test_repeat_registration_updates_stale_device_activity(self): self.register_device() device = ClientDevice.objects.get(user=self.user) stale_time = timezone.now() - timedelta(minutes=2) ClientDevice.objects.filter(pk=device.pk).update(last_seen_at=stale_time) response = self.register_device() self.assertEqual(response.status_code, 200) device.refresh_from_db() self.assertGreater(device.last_seen_at, stale_time)