OpenSlides/server/tests/integration/utils/test_consumers.py

from importlib import import_module
from typing import Optional, Tuple
from unittest.mock import patch

import pytest
from asgiref.sync import sync_to_async
from django.conf import settings
from django.contrib.auth import BACKEND_SESSION_KEY, HASH_SESSION_KEY, SESSION_KEY

from openslides.asgi import application
from openslides.core.config import config
from openslides.utils.autoupdate import (
    AutoupdateElement,
    inform_deleted_data,
    inform_elements,
)
from openslides.utils.cache import element_cache
from openslides.utils.websocket import (
    WEBSOCKET_CHANGE_ID_TOO_HIGH,
    WEBSOCKET_WRONG_FORMAT,
)

from ...unit.utils.cache_provider import (
    Collection1,
    Collection2,
    PersonalizedCollection,
    get_cachable_provider,
)
from ..helpers import TConfig, TProjector, TUser
from ..websocket import WebsocketCommunicator


@pytest.fixture(autouse=True)
async def prepare_element_cache(settings):
    """
    Resets the element cache.

    Uses a cacheable_provider for tests with example data.
    """
    await element_cache.cache_provider.clear_cache()
    orig_cachable_provider = element_cache.cachable_provider
    element_cache.cachable_provider = get_cachable_provider(
        [
            Collection1(),
            Collection2(),
            PersonalizedCollection(),
            TConfig(),
            TUser(),
            TProjector(),
        ]
    )
    element_cache._cachables = None
    await element_cache.async_ensure_cache(default_change_id=10)
    yield
    # Reset the cachable_provider
    element_cache.cachable_provider = orig_cachable_provider
    element_cache._cachables = None
    await element_cache.cache_provider.clear_cache()


@pytest.fixture
async def get_communicator():
    communicator: Optional[WebsocketCommunicator] = None

    def get_communicator(query_string="", headers=None):
        nonlocal communicator  # use the outer communicator variable
        if query_string:
            query_string = f"?{query_string}"
        communicator = WebsocketCommunicator(
            application, f"/ws/{query_string}", headers=headers
        )
        return communicator

    yield get_communicator
    if communicator:
        await communicator.disconnect()


@pytest.fixture
async def communicator(get_communicator):
    yield get_communicator()


@pytest.fixture
async def set_config():
    """
    Set a config variable in the element_cache without hitting the database.
    """

    async def _set_config(key, value):
        with patch("openslides.utils.autoupdate.save_history"):
            collection_string = config.get_collection_string()
            config_id = config.key_to_id[key]  # type: ignore
            full_data = {"id": config_id, "key": key, "value": value}
            await sync_to_async(inform_elements)(
                [
                    AutoupdateElement(
                        id=config_id,
                        collection_string=collection_string,
                        full_data=full_data,
                        disable_history=True,
                    )
                ]
            )

    return _set_config


@pytest.mark.asyncio
async def test_normal_connection(get_communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    connected, __ = await get_communicator().connect()
    assert connected


@pytest.mark.asyncio
async def test_connection_with_change_id(get_communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    communicator = get_communicator("change_id=0")
    await communicator.connect()

    response = await communicator.receive_json_from()

    type = response.get("type")
    content = response.get("content")
    assert type == "autoupdate"
    assert "changed" in content
    assert "deleted" in content
    assert "from_change_id" in content
    assert "to_change_id" in content
    assert Collection1().get_collection_string() in content["changed"]
    assert Collection2().get_collection_string() in content["changed"]
    assert PersonalizedCollection().get_collection_string() in content["changed"]
    assert TConfig().get_collection_string() in content["changed"]
    assert TUser().get_collection_string() in content["changed"]


@pytest.mark.xfail  # This will fail until a proper solution in #4009
@pytest.mark.asyncio
async def test_connection_with_invalid_change_id(get_communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    communicator = get_communicator("change_id=invalid")
    connected, __ = await communicator.connect()

    assert connected is False


@pytest.mark.asyncio
async def test_connection_with_too_big_change_id(get_communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    communicator = get_communicator("change_id=100")

    connected, __ = await communicator.connect()

    assert connected is True
    await communicator.assert_receive_error(code=WEBSOCKET_CHANGE_ID_TOO_HIGH)


@pytest.mark.asyncio
async def test_changed_data_autoupdate(get_communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    communicator = get_communicator()
    await communicator.connect()

    # Change a config value
    await set_config("general_event_name", "Test Event")
    response = await communicator.receive_json_from()

    id = config.get_key_to_id()["general_event_name"]
    type = response.get("type")
    content = response.get("content")
    assert type == "autoupdate"
    assert content["changed"] == {
        "core/config": [{"id": id, "key": "general_event_name", "value": "Test Event"}]
    }


@pytest.mark.xfail  # This will fail until a proper solution in #4009
@pytest.mark.asyncio
async def test_anonymous_disabled(communicator):
    connected, __ = await communicator.connect()

    assert not connected


async def create_user_session_cookie(user_id: int) -> Tuple[bytes, bytes]:
    # login user with id 1
    engine = import_module(settings.SESSION_ENGINE)
    session = engine.SessionStore()  # type: ignore
    session[SESSION_KEY] = str(user_id)
    session[
        HASH_SESSION_KEY
    ] = "362d4f2de1463293cb3aaba7727c967c35de43ee"  # see helpers.TUser
    session[BACKEND_SESSION_KEY] = "django.contrib.auth.backends.ModelBackend"
    session.save()
    scn = settings.SESSION_COOKIE_NAME
    cookie_header = (b"cookie", f"{scn}={session.session_key}".encode())
    return cookie_header


@pytest.mark.asyncio
async def test_with_user(get_communicator):
    cookie_header = await create_user_session_cookie(1)
    communicator = get_communicator(headers=[cookie_header])

    connected, __ = await communicator.connect()

    assert connected


@pytest.mark.asyncio
async def test_skipping_autoupdate(set_config, get_communicator):
    cookie_header = await create_user_session_cookie(1)
    communicator = get_communicator(headers=[cookie_header])

    await communicator.connect()

    with patch("openslides.utils.autoupdate.save_history"):
        await sync_to_async(inform_elements)(
            [
                AutoupdateElement(
                    id=2,
                    collection_string=PersonalizedCollection().get_collection_string(),
                    full_data={"id": 2, "value": "new value 1", "user_id": 2},
                    disable_history=True,
                )
            ]
        )
        await sync_to_async(inform_elements)(
            [
                AutoupdateElement(
                    id=2,
                    collection_string=PersonalizedCollection().get_collection_string(),
                    full_data={"id": 2, "value": "new value 2", "user_id": 2},
                    disable_history=True,
                )
            ]
        )
    assert await communicator.receive_nothing()

    # Trigger autoupdate
    await set_config("general_event_name", "Test Event")
    response = await communicator.receive_json_from()
    content = response["content"]

    assert PersonalizedCollection().get_collection_string() not in content["deleted"]
    assert PersonalizedCollection().get_collection_string() not in content["changed"]
    assert config.get_collection_string() in content["changed"]
    assert (
        content["to_change_id"] - content["from_change_id"]
    ) == 2  # Skipped two autoupdates


@pytest.mark.asyncio
async def test_receive_deleted_data(get_communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    communicator = get_communicator()
    await communicator.connect()

    # Delete test element
    with patch("openslides.utils.autoupdate.save_history"):
        await sync_to_async(inform_deleted_data)(
            [(Collection1().get_collection_string(), 1)]
        )
    response = await communicator.receive_json_from()

    type = response.get("type")
    content = response.get("content")
    assert type == "autoupdate"
    assert content["deleted"] == {Collection1().get_collection_string(): [1]}


@pytest.mark.asyncio
async def test_send_notify(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {
            "type": "notify",
            "content": {"content": "foobar, what else.", "name": "message_name"},
            "id": "test",
        }
    )
    response = await communicator.receive_json_from()

    content = response["content"]
    assert isinstance(content, dict)
    assert content["content"] == "foobar, what else."
    assert content["name"] == "message_name"
    assert "senderChannelName" in content
    assert content["senderUserId"] == 0


@pytest.mark.asyncio
async def test_invalid_websocket_message_type(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to([])

    await communicator.assert_receive_error(code=WEBSOCKET_WRONG_FORMAT)


@pytest.mark.asyncio
async def test_invalid_websocket_message_no_id(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to({"type": "test", "content": "foobar"})

    await communicator.assert_receive_error(code=WEBSOCKET_WRONG_FORMAT)


@pytest.mark.asyncio
async def test_send_unknown_type(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {
            "type": "if_you_add_this_type_to_openslides_I_will_be_sad",
            "content": True,
            "id": "test_id",
        }
    )

    await communicator.assert_receive_error(
        code=WEBSOCKET_WRONG_FORMAT, in_response="test_id"
    )


@pytest.mark.asyncio
async def test_request_constants(communicator, settings, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {"type": "constants", "content": "", "id": "test_id"}
    )

    response = await communicator.receive_json_from()
    assert response["type"] == "constants"
    # See conftest.py for the content of 'content'
    assert response["content"] == {"constant1": "value1", "constant2": "value2"}


@pytest.mark.asyncio
async def test_send_get_elements(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {"type": "getElements", "content": {}, "id": "test_id"}
    )
    response = await communicator.receive_json_from()

    type = response.get("type")
    content = response.get("content")
    assert type == "autoupdate"
    assert "changed" in content
    assert "deleted" in content
    assert "from_change_id" in content
    assert "to_change_id" in content
    assert Collection1().get_collection_string() in content["changed"]
    assert Collection2().get_collection_string() in content["changed"]
    assert PersonalizedCollection().get_collection_string() in content["changed"]
    assert TConfig().get_collection_string() in content["changed"]
    assert TUser().get_collection_string() in content["changed"]


@pytest.mark.asyncio
async def test_send_get_elements_too_big_change_id(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {"type": "getElements", "content": {"change_id": 100}, "id": "test_id"}
    )
    await communicator.assert_receive_error(
        code=WEBSOCKET_CHANGE_ID_TOO_HIGH, in_response="test_id"
    )


@pytest.mark.asyncio
async def test_send_get_elements_too_small_change_id(communicator, set_config):
    # Note: this test depends on the default_change_id set in prepare_element_cache
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {"type": "getElements", "content": {"change_id": 1}, "id": "test_id"}
    )
    response = await communicator.receive_json_from()

    type = response.get("type")
    assert type == "autoupdate"
    assert response.get("in_response") == "test_id"
    assert response.get("content")["all_data"]


@pytest.mark.asyncio
async def test_send_connect_up_to_date(communicator, set_config):
    """
    Test, that a second request with change_id+1 from the first request does not
    send anything, becuase the client is up to date.
    """
    await set_config("general_system_enable_anonymous", True)
    element_cache.cache_provider.change_id_data = {}  # type: ignore
    await communicator.connect()
    await communicator.send_json_to(
        {"type": "getElements", "content": {"change_id": 0}, "id": "test_id"}
    )
    response1 = await communicator.receive_json_from()
    max_change_id = response1.get("content")["to_change_id"]

    await communicator.send_json_to(
        {
            "type": "getElements",
            "content": {"change_id": max_change_id},
            "id": "test_id",
        }
    )
    assert await communicator.receive_nothing()


@pytest.mark.xfail  # This test is broken
@pytest.mark.asyncio
async def test_send_connect_twice_with_clear_change_id_cache_same_change_id_then_first_request(
    communicator, set_config
):
    """
    Test, that a second request with the change_id from the first request, returns
    all data.

    A client should not do this but request for change_id+1
    """
    await set_config("general_system_enable_anonymous", True)
    await element_cache.cache_provider.clear_cache()
    await communicator.connect()
    await communicator.send_json_to(
        {"type": "getElements", "content": {"change_id": 0}, "id": "test_id"}
    )
    response1 = await communicator.receive_json_from()
    first_change_id = response1.get("content")["to_change_id"]

    await communicator.send_json_to(
        {
            "type": "getElements",
            "content": {"change_id": first_change_id},
            "id": "test_id",
        }
    )
    response2 = await communicator.receive_json_from()

    assert response2["type"] == "autoupdate"
    assert response2.get("content")["all_data"]


@pytest.mark.asyncio
async def test_request_changed_elements_no_douple_elements(communicator, set_config):
    """
    Test, that when an elements is changed twice, it is only returned
    onces when ask a range of change ids.

    Test when all_data is false
    """
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()
    # Change element twice
    await set_config("general_event_name", "Test Event")
    await set_config("general_event_name", "Other value")
    # Ask for all elements
    await communicator.send_json_to(
        {"type": "getElements", "content": {"change_id": 2}, "id": "test_id"}
    )

    response = await communicator.receive_json_from()
    type = response.get("type")
    content = response.get("content")
    assert type == "autoupdate"
    assert not response.get("content")["all_data"]
    config_ids = [e["id"] for e in content["changed"]["core/config"]]
    # test that config_ids are unique
    assert len(config_ids) == len(set(config_ids))


@pytest.mark.asyncio
async def test_send_invalid_get_elements(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {"type": "getElements", "content": {"change_id": "some value"}, "id": "test_id"}
    )
    response = await communicator.receive_json_from()

    type = response.get("type")
    assert type == "error"
    assert response.get("in_response") == "test_id"


@pytest.mark.asyncio
async def test_listen_to_projector(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()

    await communicator.send_json_to(
        {
            "type": "listenToProjectors",
            "content": {"projector_ids": [1]},
            "id": "test_id",
        }
    )
    response = await communicator.receive_json_from()

    type = response.get("type")
    content = response.get("content")
    assert type == "projector"
    assert content == {
        "change_id": 11,
        "data": {
            "1": [
                {
                    "data": {"name": "slide1", "event_name": "OpenSlides"},
                    "element": {"id": 1, "name": "test/slide1"},
                }
            ]
        },
    }


@pytest.mark.asyncio
async def test_update_projector(communicator, set_config):
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()
    await communicator.send_json_to(
        {
            "type": "listenToProjectors",
            "content": {"projector_ids": [1]},
            "id": "test_id",
        }
    )
    await communicator.receive_json_from()  # recieve initial projector data

    # Change a config value
    await set_config("general_event_name", "Test Event")

    # We need two messages: The autoupdate and the projector data in this order
    response = await communicator.receive_json_from()
    assert response.get("type") == "autoupdate"

    response = await communicator.receive_json_from()

    type = response.get("type")
    content = response.get("content")
    assert type == "projector"
    assert content == {
        "change_id": 12,
        "data": {
            "1": [
                {
                    "data": {"name": "slide1", "event_name": "Test Event"},
                    "element": {"id": 1, "name": "test/slide1"},
                }
            ]
        },
    }


@pytest.mark.asyncio
async def test_update_projector_to_current_value(communicator, set_config):
    """
    When a value does not change, the projector should not be updated.
    """
    await set_config("general_system_enable_anonymous", True)
    await communicator.connect()
    await communicator.send_json_to(
        {
            "type": "listenToProjectors",
            "content": {"projector_ids": [1]},
            "id": "test_id",
        }
    )
    await communicator.receive_json_from()

    # Change a config value to current_value
    await set_config("general_event_name", "OpenSlides")

    # We await an autoupdate, bot no projector data
    response = await communicator.receive_json_from()
    assert response.get("type") == "autoupdate"

    assert await communicator.receive_nothing()