""" Tests for App.uid and Workspace.uid: - uid is generated by get_next_uid() and is unique under concurrency. - uid is read-only via the REST API (cannot be modified via PATCH/PUT). """ from concurrent.futures import ThreadPoolExecutor, as_completed import requests_mock from django.contrib.auth import get_user_model from django.core.cache import cache from django.db import connection, connections from api.models.app import App from api.models.workspace import Workspace, WorkspaceMember from api.tests import adapter, DryccTestCase, DryccTransactionTestCase User = get_user_model() class WorkspaceUidTest(DryccTestCase): """uid immutability via REST API for Workspace.""" fixtures = ['tests.json'] def setUp(self): self.user = User.objects.get(username='autotest') self.token = self.get_or_create_token(self.user) self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token) def tearDown(self): cache.clear() def test_workspace_uid_assigned_on_create(self): response = self.client.post( '/v2/workspaces', {'id': 'uidws01', 'email': 'uidws01@example.com'}, ) self.assertEqual(response.status_code, 201, response.data) self.assertIn('uid', response.data) self.assertIsNotNone(response.data['uid']) self.assertGreater(response.data['uid'], 0) def test_workspace_uid_immutable_via_patch(self): response = self.client.post( '/v2/workspaces', {'id': 'uidws02', 'email': 'uidws02@example.com'}, ) self.assertEqual(response.status_code, 201, response.data) original_uid = response.data['uid'] # Attempt to mutate uid via PATCH response = self.client.patch( '/v2/workspaces/uidws02', {'uid': 999999, 'email': 'new@example.com'}, ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual(response.data['uid'], original_uid) self.assertEqual(response.data['email'], 'new@example.com') # Re-fetch and confirm ws = Workspace.objects.get(id='uidws02') self.assertEqual(ws.uid, original_uid) def test_workspace_uid_immutable_via_put(self): response = self.client.post( '/v2/workspaces', {'id': 'uidws03', 'email': 'uidws03@example.com'}, ) self.assertEqual(response.status_code, 201, response.data) original_uid = response.data['uid'] response = self.client.put( '/v2/workspaces/uidws03', {'id': 'uidws03', 'uid': 888888, 'email': 'uidws03@example.com'}, ) # PUT may return 200 or 405 depending on viewset config; uid must not change. ws = Workspace.objects.get(id='uidws03') self.assertEqual(ws.uid, original_uid) class WorkspaceUidConcurrencyTest(DryccTransactionTestCase): """Concurrency: parallel Workspace.save() must produce unique uids.""" fixtures = ['tests.json'] def tearDown(self): cache.clear() # Close any connections opened by worker threads. connections.close_all() def test_workspace_uid_unique_under_concurrency(self): cache.clear() n = 16 def _create(i): try: ws = Workspace.objects.create( id=f'wsuid{i:03d}', email=f'wsuid{i:03d}@example.com', ) return ws.uid finally: connection.close() with ThreadPoolExecutor(max_workers=n) as ex: uids = [f.result() for f in as_completed([ex.submit(_create, i) for i in range(n)])] self.assertEqual(len(uids), n) self.assertEqual(len(set(uids)), n, f'duplicate uids detected: {uids}') self.assertTrue(all(u > 0 for u in uids)) @requests_mock.Mocker(real_http=True, adapter=adapter) class AppUidTest(DryccTestCase): """uid immutability for App. App.save() requires K8s. We exercise the REST API where possible, and fall back to ORM + serializer validation for environments without a K8s serviceaccount token. The serializer marking uid as read-only is the source of truth for API-level immutability. """ fixtures = ['tests.json'] def setUp(self): self.user = User.objects.get(username='autotest') self.token = self.get_or_create_token(self.user) self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token) self.workspace = Workspace.objects.create( id='uidappws', email='uidappws@example.com', ) WorkspaceMember.objects.create( workspace=self.workspace, user=self.user, role='admin', ) def tearDown(self): cache.clear() def test_app_uid_field_is_read_only_in_serializer(self, mock_requests): from api.serializers import AppSerializer self.assertIn('uid', AppSerializer.Meta.read_only_fields) def test_app_uid_ignored_on_serializer_update(self, mock_requests): """Even if a client sends uid in a PATCH body, the serializer drops it.""" from api.utils import get_next_uid from api.serializers import AppSerializer # Bypass App.save() (K8s dependency) by inserting via the queryset. App.objects.bulk_create([ App(id='uidapp01', workspace=self.workspace, uid=get_next_uid(App)), ]) app = App.objects.get(id='uidapp01') original_uid = app.uid serializer = AppSerializer( instance=app, data={'uid': 999999, 'workspace': self.workspace.id}, partial=True, ) self.assertTrue(serializer.is_valid(), serializer.errors) # uid is read-only -> dropped from validated_data, so no update path # can ever persist a client-supplied uid. self.assertNotIn('uid', serializer.validated_data) app.refresh_from_db() self.assertEqual(app.uid, original_uid) class AppUidConcurrencyTest(DryccTransactionTestCase): """Concurrency: parallel App.uid allocation via get_next_uid must be unique. App.save() requires K8s; we test the uid allocation pathway directly by creating App rows with the same code path used by save(): get_next_uid(App) plus bulk_create to bypass the K8s-dependent App.save() body. """ fixtures = ['tests.json'] def setUp(self): self.user = User.objects.get(username='autotest') self.workspace = Workspace.objects.create( id='uidappcon', email='uidappcon@example.com', ) WorkspaceMember.objects.create( workspace=self.workspace, user=self.user, role='admin', ) def tearDown(self): cache.clear() connections.close_all() def test_app_uid_unique_under_concurrency(self): from api.utils import get_next_uid cache.clear() n = 32 def _alloc(): try: return get_next_uid(App) finally: connection.close() with ThreadPoolExecutor(max_workers=n) as ex: uids = [f.result() for f in as_completed([ex.submit(_alloc) for _ in range(n)])] self.assertEqual(len(uids), n) self.assertEqual(len(set(uids)), n, f'duplicate uids detected: {uids}') self.assertTrue(all(u > 0 for u in uids))