test_uid.py

"""
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))