OpenSlides/server/tests/test_case.py

import random
import string

from asgiref.sync import async_to_sync
from django.contrib.auth import get_user_model
from django.test import TestCase as _TestCase
from rest_framework.test import APIClient

from openslides.core.config import config
from openslides.utils.autoupdate import inform_changed_data
from openslides.utils.cache import element_cache
from openslides.utils.utils import get_element_id
from tests.common_groups import GROUP_ADMIN_PK, GROUP_DELEGATE_PK
from tests.count_queries import AssertNumQueriesContext


class TestCase(_TestCase):
    maxDiff = None

    def setUp(self):
        self.admin = get_user_model().objects.get(username="admin")
        self.client = APIClient()
        self.client.login(username="admin", password="admin")
        self.advancedSetUp()

    def advancedSetUp(self):
        pass

    def create_guest_client(self):
        config["general_system_enable_anonymous"] = True
        return APIClient()

    """
    Adds testing for autoupdates after requests.
    """

    def get_last_autoupdate(self, user=None):
        """
        Get the last autoupdate as (changed_data, deleted_element_ids) for the given user.
        changed_elements is a dict with element_ids as keys and the actual element as value
        user_id=None if for full data, 0 for the anonymous and regular ids for users.
        """
        user_id = None if user is None else user.id
        current_change_id = async_to_sync(element_cache.get_current_change_id)()
        _, _changed_elements, deleted_element_ids = async_to_sync(
            element_cache.get_data_since
        )(user_id=user_id, change_id=current_change_id)

        changed_elements = {}
        for collection, elements in _changed_elements.items():
            for element in elements:
                changed_elements[get_element_id(collection, element["id"])] = element

        return (changed_elements, deleted_element_ids)

    def assertAutoupdate(self, model, user=None):
        self.assertTrue(
            model.get_element_id() in self.get_last_autoupdate(user=user)[0]
        )

    def assertDeletedAutoupdate(self, model, user=None):
        self.assertTrue(
            model.get_element_id() in self.get_last_autoupdate(user=user)[1]
        )

    def assertNoAutoupdate(self, model, user=None):
        self.assertFalse(
            model.get_element_id() in self.get_last_autoupdate(user=user)[0]
        )

    def assertNoDeletedAutoupdate(self, model, user=None):
        self.assertFalse(
            model.get_element_id() in self.get_last_autoupdate(user=user)[1]
        )

    def assertNumQueries(self, num, func=None, *args, verbose=False, **kwargs):
        context = AssertNumQueriesContext(self, num, verbose)
        if func is None:
            return context

        with context:
            func(*args, **kwargs)

    def assertHttpStatusVerbose(self, response, status):
        if response.status_code != status:
            print(response.data)
        self.assertEqual(response.status_code, status)

    """
    Create Helper functions
    """

    def create_user(self):
        password = "test_password_" + self._get_random_string()
        return (
            get_user_model().objects.create_user(
                username="test_user_" + self._get_random_string(), password=password
            ),
            password,
        )

    def make_admin_delegate(self):
        admin = get_user_model().objects.get(username="admin")
        admin.groups.add(GROUP_DELEGATE_PK)
        admin.groups.remove(GROUP_ADMIN_PK)
        inform_changed_data(admin)

    def _get_random_string(self, length=20):
        return "".join(
            random.choices(
                string.ascii_lowercase + string.ascii_uppercase + string.digits,
                k=length,
            )
        )