115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
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,
|
|
)
|
|
)
|