Create func to check for too many devices

corehq/apps/ota/const.py

@@ -0,0 +1 @@
+DEVICES_PER_USER = 50

corehq/apps/ota/tests/test_utils.py

@@ -1,11 +1,15 @@
+import uuid
+from datetime import datetime
 from unittest.mock import patch
 
 from django.test import TestCase
 
-from casexml.apps.phone.models import OTARestoreCommCareUser, OTARestoreWebUser
+from freezegun import freeze_time
+
+from casexml.apps.phone.models import OTARestoreCommCareUser, OTARestoreWebUser, SyncLogSQL
 from corehq.apps.domain.models import Domain
 from corehq.apps.locations.tests.util import LocationHierarchyTestCase
-from corehq.apps.ota.utils import get_restore_user, is_permitted_to_restore
+from corehq.apps.ota.utils import get_restore_user, is_permitted_to_restore, too_many_devices
 from corehq.apps.users.dbaccessors import delete_all_users
 from corehq.apps.users.models import CommCareUser, WebUser
 from corehq.apps.users.util import format_username
@@ -425,3 +429,94 @@ def test_get_restore_user_as_user_for_commcare_user(self):
         )
         self.assertEqual(user.user_id, self.other_commcare_user._id)
         self.assertEqual(user.request_user_id, self.commcare_user.user_id)
+
+
+@freeze_time("2024-12-10 12:00:00")
+class TooManyDevicesTest(TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.domain = 'too-many-devices-test'
+
+    def test_true_if_user_device_count_exceeds_limit(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'device-{i}', date=datetime(2024, 12, 10, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 1):
+            self.assertTrue(too_many_devices('abc123'))
+
+    def test_false_if_user_device_count_equals_limit(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'device-{i}', date=datetime(2024, 12, 10, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 2):
+            self.assertFalse(too_many_devices('abc123'))
+
+    def test_false_if_user_device_count_is_under_limit(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'device-{i}', date=datetime(2024, 12, 10, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 3):
+            self.assertFalse(too_many_devices('abc123'))
+
+    def test_false_for_web_apps_logins(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'WebAppsLogin*{i}', date=datetime(2024, 12, 10, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 1):
+            self.assertFalse(too_many_devices('abc123'))
+
+    def test_false_if_dates_are_older_than_a_day(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'device-{i}', date=datetime(2024, 12, 9, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 1):
+            self.assertFalse(too_many_devices('abc123'))
+
+    def test_true_for_last_submitted_field(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'device-{i}', last_submitted=datetime(2024, 12, 9, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 1):
+            self.assertFalse(too_many_devices('abc123'))
+
+    def test_false_if_last_submitted_times_are_older_than_a_day(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'device-{i}', last_submitted=datetime(2024, 12, 9, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 1):
+            self.assertFalse(too_many_devices('abc123'))
+
+    def test_true_if_date_is_recent_but_last_submitted_is_old(self):
+        for i in range(2):
+            self._create_synclog(
+                self.domain,
+                'abc123',
+                f'device-{i}',
+                date=datetime(2024, 12, 10, 0, 0),
+                last_submitted=datetime(2024, 12, 9, 0, 0),
+            )
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 1):
+            self.assertTrue(too_many_devices('abc123'))
+
+    def test_false_for_empty_queryset(self):
+        self.assertFalse(too_many_devices('abc123'))
+
+    def test_false_for_different_user(self):
+        for i in range(2):
+            self._create_synclog(self.domain, 'abc123', f'device-{i}', date=datetime(2024, 12, 10, 0, 0))
+
+        with patch('corehq.apps.ota.utils.DEVICES_PER_USER', 1):
+            self.assertFalse(too_many_devices('def456'))
+
+    def _create_synclog(self, domain, user_id, device_id, **kwargs):
+        SyncLogSQL.objects.create(
+            domain=domain,
+            doc={},
+            synclog_id=uuid.uuid4(),
+            user_id=user_id,
+            device_id=device_id,
+            **kwargs
+        )

corehq/apps/ota/utils.py

@@ -1,10 +1,13 @@
 from functools import wraps
+from datetime import datetime, timedelta
 
+from django.db.models import Count, Q
 from django.utils.translation import gettext as _
 
 from couchdbkit import ResourceConflict
 
 from casexml.apps.case.xml import V2
+from casexml.apps.phone.models import SyncLogSQL
 from casexml.apps.phone.restore import RestoreConfig, RestoreParams
 from dimagi.utils.logging import notify_exception
 from dimagi.utils.web import json_response
@@ -19,6 +22,7 @@
 from corehq.apps.users.models import CommCareUser
 
 from .exceptions import RestorePermissionDenied
+from .const import DEVICES_PER_USER
 from .models import DemoUserRestore
 
 
@@ -219,3 +223,17 @@ def _inner(request, domain, *args, **kwargs):
 
         return response
     return _inner
+
+
+def too_many_devices(user_id):
+    end_time = datetime.now()
+    start_time = end_time - timedelta(days=1)
+    date_query = Q(date__gte=start_time, date__lt=end_time)
+    last_submitted_query = Q(last_submitted__gte=start_time, last_submitted__lt=end_time)
+    result = (
+        SyncLogSQL.objects.filter(date_query | last_submitted_query, user_id=user_id)
+        .exclude(device_id__startswith='WebAppsLogin')
+        .values('user_id')
+        .annotate(device_count=Count('device_id', distinct=True))
+    )
+    return bool(result and result[0]['device_count'] > DEVICES_PER_USER)