feat(demo): add isolated demo environments

apps/demos/tests/test_demo_api.py

@@ -0,0 +1,77 @@
+from django.contrib.auth import get_user_model
+from django.test import override_settings
+from django.utils import timezone
+from rest_framework.test import APITestCase
+
+from apps.clients.models import Client
+from apps.demos.models import DemoEnvironment
+from apps.demos.services import cleanup_expired_demo_environments
+from apps.projects.models import Project, ProjectAccess
+from apps.tags.models import Tag
+from apps.time_entries.models import TimeEntry
+from apps.workspaces.models import WorkspaceMembership, WorkspaceUserRate
+
+User = get_user_model()
+DEMO_START_URL = "/api/demo/start/"
+
+
+@override_settings(DEMO_ENABLED=True, DEMO_ENVIRONMENT_TTL_HOURS=24, DEMO_CLEANUP_BATCH_SIZE=100)
+class DemoStartApiTests(APITestCase):
+    def test_demo_start_creates_isolated_seeded_environment(self):
+        response = self.client.post(DEMO_START_URL)
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIn("access", response.data)
+        self.assertIn("refresh", response.data)
+        self.assertEqual(DemoEnvironment.objects.count(), 1)
+
+        environment = DemoEnvironment.objects.select_related("owner_user", "workspace").get()
+        self.assertTrue(environment.owner_user.is_demo)
+        self.assertEqual(environment.owner_user.demo_expires_at, environment.expires_at)
+        self.assertGreaterEqual(WorkspaceMembership.objects.filter(workspace=environment.workspace).count(), 4)
+        self.assertGreaterEqual(Client.objects.filter(workspace=environment.workspace).count(), 3)
+        self.assertGreaterEqual(Project.objects.filter(workspace=environment.workspace).count(), 5)
+        self.assertGreaterEqual(Tag.objects.filter(workspace=environment.workspace).count(), 4)
+        self.assertGreaterEqual(TimeEntry.objects.filter(workspace=environment.workspace).count(), 8)
+        self.assertGreaterEqual(WorkspaceUserRate.objects.filter(workspace=environment.workspace).count(), 4)
+        self.assertGreaterEqual(ProjectAccess.objects.filter(project__workspace=environment.workspace).count(), 1)
+
+    def test_two_demo_starts_do_not_share_workspace_data(self):
+        first = self.client.post(DEMO_START_URL)
+        second = self.client.post(DEMO_START_URL)
+
+        self.assertEqual(first.status_code, 201)
+        self.assertEqual(second.status_code, 201)
+        environments = list(DemoEnvironment.objects.order_by("created_at"))
+        self.assertEqual(len(environments), 2)
+        self.assertNotEqual(environments[0].workspace_id, environments[1].workspace_id)
+        self.assertNotEqual(environments[0].owner_user_id, environments[1].owner_user_id)
+
+    def test_demo_user_cannot_search_external_users_or_send_otp(self):
+        self.client.post(DEMO_START_URL)
+        environment = DemoEnvironment.objects.select_related("owner_user").get()
+        real_user = User.objects.create_user(mobile="09111111111", password="Testpass123!")
+        self.client.force_authenticate(environment.owner_user)
+
+        search_response = self.client.get(f"/api/users/search/?mobile={real_user.mobile}")
+        self.assertEqual(search_response.status_code, 403)
+
+        otp_response = self.client.post(
+            "/api/users/otp/send/",
+            {"mobile": environment.owner_user.mobile, "mode": "login"},
+            format="json",
+        )
+        self.assertEqual(otp_response.status_code, 400)
+
+    def test_cleanup_deletes_expired_demo_and_keeps_real_users(self):
+        self.client.post(DEMO_START_URL)
+        environment = DemoEnvironment.objects.select_related("workspace").get()
+        real_user = User.objects.create_user(mobile="09122222222", password="Testpass123!")
+        DemoEnvironment.objects.filter(id=environment.id).update(expires_at=timezone.now() - timezone.timedelta(minutes=1))
+
+        cleaned = cleanup_expired_demo_environments()
+
+        self.assertEqual(cleaned, 1)
+        self.assertFalse(DemoEnvironment.all_objects.filter(id=environment.id).exists())
+        self.assertFalse(TimeEntry.all_objects.filter(workspace_id=environment.workspace_id).exists())
+        self.assertTrue(User.objects.filter(id=real_user.id).exists())