Mobilizon-Reshare-condividi-sui-social/tests/conftest.py

import importlib.resources
import os
import time
from collections import UserList
from datetime import datetime, timedelta, timezone
from typing import Union
from uuid import UUID

import arrow
import pytest
import responses
from tortoise import Tortoise

import mobilizon_reshare
from mobilizon_reshare.config.command import CommandConfig
from mobilizon_reshare.config.config import get_settings
from mobilizon_reshare.dataclasses import MobilizonEvent, EventPublicationStatus
from mobilizon_reshare.models.event import Event
from mobilizon_reshare.models.notification import Notification, NotificationStatus
from mobilizon_reshare.models.publication import Publication, PublicationStatus
from mobilizon_reshare.models.publisher import Publisher
from mobilizon_reshare.publishers.abstract import (
    AbstractPlatform,
    AbstractEventFormatter,
)
from mobilizon_reshare.publishers.exceptions import PublisherError, InvalidResponse
from tests import today

with importlib.resources.path(
    mobilizon_reshare, ".secrets.toml"
) as bundled_secrets_path:
    os.environ["SECRETS_FOR_DYNACONF"] = str(bundled_secrets_path)


def generate_publication_status(published) -> PublicationStatus:
    return PublicationStatus.COMPLETED if published else PublicationStatus.FAILED


@pytest.fixture(autouse=True)
def set_timezone():
    os.environ["TZ"] = "Coordinated Universal Time"
    time.tzset()


def generate_event_status(published):
    return (
        EventPublicationStatus.COMPLETED
        if published
        else EventPublicationStatus.WAITING
    )


def generate_notification_status(published):
    return NotificationStatus.COMPLETED if published else NotificationStatus.FAILED


@pytest.fixture(scope="session", autouse=True)
def set_dynaconf_environment() -> None:
    os.environ["ENV_FOR_DYNACONF"] = "testing"
    os.environ["FORCE_ENV_FOR_DYNACONF"] = "testing"

    yield None

    os.environ["ENV_FOR_DYNACONF"] = ""
    os.environ["FORCE_ENV_FOR_DYNACONF"] = ""


@pytest.fixture
def event_generator():
    def _event_generator(
        begin_date=arrow.Arrow(year=2021, month=1, day=1, hour=11, minute=30),
        published=False,
        publication_time=None,
        mobilizon_id=UUID(int=12345),
        last_update_time=arrow.Arrow(year=2021, month=1, day=1, hour=11, minute=30),
    ):

        return MobilizonEvent(
            name="test event",
            description="description of the event",
            begin_datetime=begin_date,
            end_datetime=begin_date.shift(hours=2),
            mobilizon_link="http://some_link.com/123",
            mobilizon_id=mobilizon_id,
            thumbnail_link="http://some_link.com/123.jpg",
            location="location",
            status=generate_event_status(published),
            publication_time=publication_time
            or (begin_date.shift(days=-1) if published else None),
            last_update_time=last_update_time,
        )

    return _event_generator


@pytest.fixture()
def event() -> MobilizonEvent:
    begin_date = arrow.get(
        datetime(
            year=2021,
            month=1,
            day=1,
            hour=11,
            minute=30,
            tzinfo=timezone(timedelta(hours=1)),
        )
    )
    return MobilizonEvent(
        name="test event",
        description="description of the event",
        begin_datetime=begin_date,
        end_datetime=begin_date.shift(hours=1),
        mobilizon_link="http://some_link.com/123",
        mobilizon_id=UUID(int=12345),
        thumbnail_link="http://some_link.com/123.jpg",
        location="location",
        last_update_time=begin_date,
    )


@pytest.fixture
async def stored_event(event) -> Event:
    model = event.to_model()
    await model.save()
    await model.fetch_related("publications")
    return model


@pytest.fixture(scope="function", autouse=True)
async def initialize_db_tests(request) -> None:
    config = {
        "connections": {
            "default": os.environ.get("TORTOISE_TEST_DB", "sqlite://:memory:")
        },
        "apps": {
            "models": {
                "models": [
                    "mobilizon_reshare.models.event",
                    "mobilizon_reshare.models.notification",
                    "mobilizon_reshare.models.publication",
                    "mobilizon_reshare.models.publisher",
                    "aerich.models",
                ],
                "default_connection": "default",
            },
        },
        # always store UTC time in database
        "use_tz": True,
    }

    async def _init_db() -> None:
        await Tortoise.init(config)
        try:
            await Tortoise._drop_databases()
        except:  # noqa
            pass

        await Tortoise.init(config, _create_db=True)
        await Tortoise.generate_schemas(safe=False)

    await _init_db()
    yield
    await Tortoise._drop_databases()


@pytest.fixture()
def event_model_generator():
    def _event_model_generator(
        idx=1,
        begin_date=datetime(
            year=2021,
            month=1,
            day=1,
            hour=11,
            minute=30,
            tzinfo=timezone(timedelta(hours=0)),
        ),
    ):
        return Event(
            name=f"event_{idx}",
            description=f"desc_{idx}",
            mobilizon_id=UUID(int=idx),
            mobilizon_link=f"moblink_{idx}",
            thumbnail_link=f"thumblink_{idx}",
            location=f", loc_{idx}, ",
            begin_datetime=begin_date,
            end_datetime=begin_date + timedelta(hours=2),
            last_update_time=begin_date,
        )

    return _event_model_generator


@pytest.fixture()
def publisher_model_generator():
    def _publisher_model_generator(idx=1, name=None):
        return Publisher(
            name=name or f"publisher_{idx}", account_ref=f"account_ref_{idx}"
        )

    return _publisher_model_generator


@pytest.fixture()
def publication_model_generator():
    def _publication_model_generator(
        status=PublicationStatus.COMPLETED,
        publication_time=datetime(year=2021, month=1, day=1, hour=11, minute=30),
        event_id=None,
        publisher_id=None,
    ):
        return Publication(
            status=status,
            timestamp=publication_time,
            event_id=event_id,
            publisher_id=publisher_id,
        )

    return _publication_model_generator


@pytest.fixture()
def notification_model_generator():
    def _notification_model_generator(
        idx=1, published=False, publication_id=None, target_id=None
    ):
        return Notification(
            status=generate_notification_status(published),
            message=f"message_{idx}",
            publication_id=publication_id,
            target_id=target_id,
        )

    return _notification_model_generator


async def _generate_publishers(specification):

    publishers = []
    for i, publisher_name in enumerate(specification["publisher"]):
        publisher = Publisher(
            id=UUID(int=i), name=publisher_name, account_ref=f"account_ref_{i}"
        )
        publishers.append(publisher)
        await publisher.save()

    return publishers


async def _generate_events(specification):
    events = []
    if "event" in specification.keys():
        for i in range(specification["event"]):
            begin_date = today + timedelta(days=i)
            event = Event(
                id=UUID(int=i),
                name=f"event_{i}",
                description=f"desc_{i}",
                mobilizon_id=UUID(int=i),
                mobilizon_link=f"https://example.org/moblink_{i}",
                thumbnail_link=f"https://example.org/thumblink_{i}",
                location=f", loc_{i}, ",
                begin_datetime=begin_date,
                end_datetime=begin_date + timedelta(hours=2),
                last_update_time=begin_date,
            )
            events.append(event)
            await event.save()
    return events


async def _generate_publications(events, publishers, specification):
    if "publications" in specification.keys():
        for i, publication in enumerate(specification["publications"]):
            status = publication.get("status", PublicationStatus.COMPLETED)
            timestamp = publication.get("timestamp", today + timedelta(hours=i))
            await Publication.create(
                id=UUID(int=i),
                status=status,
                timestamp=timestamp,
                event_id=events[publication["event_idx"]].id,
                publisher_id=publishers[publication["publisher_idx"]].id,
            )


@pytest.fixture(scope="module")
def generate_models():
    async def _generate_models(specification: dict[str, Union[int, list]]):
        publishers = await _generate_publishers(specification)
        events = await _generate_events(specification)
        await _generate_publications(events, publishers, specification)

    return _generate_models


event_0 = MobilizonEvent(
    name="event_0",
    description="desc_0",
    mobilizon_id=UUID(int=0),
    mobilizon_link="https://example.org/moblink_0",
    thumbnail_link="https://example.org/thumblink_0",
    location=", loc_0, ",
    status=EventPublicationStatus.WAITING,
    begin_datetime=arrow.get(today),
    end_datetime=arrow.get(today + timedelta(hours=2)),
    last_update_time=arrow.get(today),
)

event_1 = MobilizonEvent(
    name="event_1",
    description="desc_1",
    mobilizon_id=UUID(int=1),
    mobilizon_link="https://example.org/moblink_1",
    thumbnail_link="https://example.org/thumblink_1",
    location=", loc_1, ",
    status=EventPublicationStatus.WAITING,
    begin_datetime=arrow.get(today + timedelta(days=1)),
    end_datetime=arrow.get(today + timedelta(days=1) + timedelta(hours=2)),
    last_update_time=arrow.get(today + timedelta(days=1)),
)

event_2 = MobilizonEvent(
    name="event_2",
    description="desc_2",
    mobilizon_id=UUID(int=2),
    mobilizon_link="https://example.org/moblink_2",
    thumbnail_link="https://example.org/thumblink_2",
    location=", loc_2, ",
    status=EventPublicationStatus.WAITING,
    begin_datetime=arrow.get(today + timedelta(days=2)),
    end_datetime=arrow.get(today + timedelta(days=2) + timedelta(hours=2)),
    last_update_time=arrow.get(today + timedelta(days=2)),
)

event_3 = MobilizonEvent(
    name="event_3",
    description="desc_3",
    mobilizon_id=UUID(int=3),
    mobilizon_link="https://example.org/moblink_3",
    thumbnail_link="https://example.org/thumblink_3",
    location=", loc_3, ",
    status=EventPublicationStatus.WAITING,
    begin_datetime=arrow.get(today + timedelta(days=3)),
    end_datetime=arrow.get(today + timedelta(days=3) + timedelta(hours=2)),
    last_update_time=arrow.get(today + timedelta(days=3)),
)

event_3_updated = MobilizonEvent(
    name="event_3",
    description="desc_3",
    mobilizon_id=UUID(int=3),
    mobilizon_link="https://example.org/moblink_3",
    thumbnail_link="https://example.org/thumblink_3",
    location=", loc_6, ",
    status=EventPublicationStatus.WAITING,
    begin_datetime=arrow.get(today + timedelta(days=3)),
    end_datetime=arrow.get(today + timedelta(days=3) + timedelta(hours=2)),
    last_update_time=arrow.get(today + timedelta(days=4)),
)

event_6 = MobilizonEvent(
    name="event_6",
    description="desc_6",
    mobilizon_id=UUID(int=6),
    mobilizon_link="https://example.org/moblink_6",
    thumbnail_link="https://example.org/thumblink_6",
    location=", loc_6, ",
    status=EventPublicationStatus.WAITING,
    begin_datetime=arrow.get(today + timedelta(days=6)),
    end_datetime=arrow.get(today + timedelta(days=6) + timedelta(hours=2)),
    last_update_time=arrow.get(today + timedelta(days=6)),
)


@pytest.fixture()
def message_collector():
    class MessageCollector(UserList):
        def collect_message(self, message):
            self.append(message)

    return MessageCollector()


@pytest.fixture
def mock_publisher_class(message_collector):
    class MockPublisher(AbstractPlatform):
        name = "mock"

        def _send(self, message, event):
            message_collector.append(message)

        def _validate_response(self, response):
            pass

        def validate_credentials(self) -> None:
            pass

    return MockPublisher


@pytest.fixture
def mock_zulip_publisher_class(message_collector):
    class MockPublisher(AbstractPlatform):
        name = "zulip"

        def _send(self, message, event):
            message_collector.append(message)

        def _validate_response(self, response):
            pass

        def validate_credentials(self) -> None:
            pass

    return MockPublisher


@pytest.fixture
def mock_publisher_valid(message_collector, mock_publisher_class):

    return mock_publisher_class()


@pytest.fixture
def mock_zulip_publisher(message_collector, mock_zulip_publisher_class):

    return mock_zulip_publisher_class()


@pytest.fixture
def mobilizon_url():
    return get_settings()["source"]["mobilizon"]["url"]


@responses.activate
@pytest.fixture
def mock_mobilizon_success_answer(mobilizon_answer, mobilizon_url):
    with responses.RequestsMock() as rsps:

        rsps.add(
            responses.POST, mobilizon_url, json=mobilizon_answer, status=200,
        )
        yield


@responses.activate
@pytest.fixture
def mock_multiple_success_answer(multiple_answers, mobilizon_url):
    with responses.RequestsMock() as rsps:

        for answer in multiple_answers:
            rsps.add(
                responses.POST, mobilizon_url, json=answer, status=200,
            )

        yield


@pytest.fixture
def mock_formatter_class():
    class MockFormatter(AbstractEventFormatter):
        def validate_event(self, event) -> None:
            pass

        def get_message_from_event(self, event) -> str:
            return f"{event.name}|{event.description}"

        def validate_message(self, event) -> None:
            pass

        def get_recap_fragment(self, event):
            return event.name

        def get_recap_header(self):
            return "Upcoming"

    return MockFormatter


@pytest.fixture
def mock_formatter_valid(mock_formatter_class):

    return mock_formatter_class()


@pytest.fixture
def mock_publisher_invalid_class(message_collector):
    class MockPublisher(AbstractPlatform):

        name = "mock"

        def _send(self, message, event):
            message_collector.append(message)

        def _validate_response(self, response):
            return InvalidResponse("response error")

        def validate_credentials(self) -> None:
            raise PublisherError("credentials error")

    return MockPublisher


@pytest.fixture
def mock_publisher_invalid(mock_publisher_invalid_class):

    return mock_publisher_invalid_class()


@pytest.fixture
async def event_with_failed_publication(
    stored_event, mock_publisher_config, failed_publication
):
    return stored_event


@pytest.fixture
async def failed_publication(stored_event, mock_publisher) -> Publication:

    p = Publication(
        event=stored_event,
        status=PublicationStatus.FAILED,
        timestamp=arrow.now().datetime,
        publisher=mock_publisher,
    )
    await p.save()
    return p


@pytest.fixture
def command_config():
    return CommandConfig(dry_run=False)


@pytest.fixture()
async def mock_publisher(publisher_model_generator):
    publisher = await publisher_model_generator(name="mock")
    await publisher.save()
    return publisher