Mobilizon
/
Mobilizon-Reshare-condividi-sui-social
mirror of https://github.com/Tech-Workers-Coalition-Italia/mobilizon-reshare.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

first run (#36) 

* added settings as CLI option

* added group missing error

* comments

* testing empty events

* fixed transaction

* main is working

* added logging and fixed publication

* refactored some publication code

* removed redundant field

* storage: query: Work around https://github.com/tortoise/tortoise-orm/issues/419 .

* storage: query: Update testing environment check.

* review

* tests: models: Test `MobilizonEvent.compute_status`.

Co-authored-by: Giacomo Leidi <goodoldpaul@autistici.org>
This commit is contained in:
Simone Robutti 2021-07-12 22:17:49 +02:00 committed by GitHub
parent a33a1d7b3e
commit 9578f18078
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 337 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -175,3 +175,4 @@ crashlytics.properties
crashlytics-build.properties
fabric.properties
.idea
*/local_testing.toml

5
mobilizon_bots/cli.py
View File

@ -11,9 +11,10 @@ def mobilizon_bots():
@mobilizon_bots.command()
def start():
@click.option("--settings-file", type=click.Path(exists=True))
def start(settings_file):
asyncio.run(main())
asyncio.run(main([settings_file] if settings_file else None))
if __name__ == "__main__":

46
mobilizon_bots/config/config.py
View File

@ -1,3 +1,4 @@
import os
from typing import List
from dynaconf import Dynaconf, Validator
@ -11,12 +12,16 @@ def build_settings(
settings_files: List[str] = None, validators: List[Validator] = None
):
SETTINGS_FILE = settings_files or [
"mobilizon_bots/settings.toml",
"mobilizon_bots/.secrets.toml",
"/etc/mobilizon_bots.toml",
"/etc/mobilizon_bots_secrets.toml",
]
SETTINGS_FILE = (
settings_files
or os.environ.get("MOBILIZON_BOTS_SETTINGS_FILE")
or [
"mobilizon_bots/settings.toml",
"mobilizon_bots/.secrets.toml",
"/etc/mobilizon_bots.toml",
"/etc/mobilizon_bots_secrets.toml",
]
)
ENVVAR_PREFIX = "MOBILIZON_BOTS"
return Dynaconf(
@ -89,4 +94,31 @@ def build_and_validate_settings(settings_files: List[str] = None):
return settings
settings = build_and_validate_settings()
# this singleton and functions are necessary to put together
# the necessities of the testing suite, the CLI and still having a single entrypoint to the config.
# The CLI needs to provide the settings file at run time so we cannot work at import time.
# The normal Dynaconf options to specify the settings files are also not a valid option because of the two steps
# validation that prevents us to employ their mechanism to specify settings files. This could probably be reworked
# better in the future.
class CustomConfig:
_instance = None
def __new__(cls, settings_files: List[str] = None):
if cls._instance is None:
print("Creating the object")
cls._instance = super(CustomConfig, cls).__new__(cls)
cls.settings = build_settings(settings_files)
return cls._instance
def update(self, settings_files: List[str] = None):
self.settings = build_settings(settings_files)
def get_settings():
config = CustomConfig()
return config.settings
def update_settings_files(settings_files: List[str] = None):
CustomConfig().update(settings_files)
return CustomConfig().settings

3
mobilizon_bots/config/publishers.py
View File

@ -1,8 +1,9 @@
from dynaconf import Validator
telegram_validators = [
Validator("publisher.telegram.chat_id", must_exist=True),
Validator("publisher.telegram.msg_template_path", must_exist=True,),
Validator("publisher.telegram.msg_template_path", must_exist=True, default=None),
Validator("publisher.telegram.token", must_exist=True),
Validator("publisher.telegram.username", must_exist=True),
]

44
mobilizon_bots/event/event_selection_strategies.py
View File

@ -1,10 +1,14 @@
import logging
from abc import ABC, abstractmethod
from typing import List, Optional
import arrow
from mobilizon_bots.config.config import settings
from mobilizon_bots.config.config import get_settings
from mobilizon_bots.event.event import MobilizonEvent
logger = logging.getLogger(__name__)
class EventSelectionStrategy(ABC):
def select(
@ -18,6 +22,7 @@ class EventSelectionStrategy(ABC):
return self._select(published_events, unpublished_events)
def is_in_publishing_window(self) -> bool:
settings = get_settings()
window_beginning = settings["publishing"]["window"]["begin"]
window_end = settings["publishing"]["window"]["end"]
now_hour = arrow.now().datetime.hour
@ -40,25 +45,45 @@ class SelectNextEventStrategy(EventSelectionStrategy):
self,
published_events: List[MobilizonEvent],
unpublished_events: List[MobilizonEvent],
publisher_name: str = "telegram",
) -> Optional[MobilizonEvent]:
last_published_event = published_events[-1]
# if there are no unpublished events, there's nothing I can do
if not unpublished_events:
logger.debug("No event to publish.")
return None
first_unpublished_event = unpublished_events[0]
# if there's no published event (first execution) I return the next in queue
if not published_events:
logger.debug(
"First Execution with an available event. Picking next event in the queue."
)
return first_unpublished_event
last_published_event = published_events[-1]
now = arrow.now()
assert last_published_event.publication_time[publisher_name] < now, (
last_published_event_most_recent_publication_time = max(
last_published_event.publication_time.values()
)
assert last_published_event_most_recent_publication_time < now, (
f"Last published event has been published in the future\n"
f"{last_published_event.publication_time[publisher_name]}\n"
f"{last_published_event_most_recent_publication_time}\n"
f"{now}"
)
if (
last_published_event.publication_time[publisher_name].shift(
minutes=settings[
last_published_event_most_recent_publication_time.shift(
minutes=get_settings()[
"selection.strategy_options.break_between_events_in_minutes"
]
)
> now
):
logger.debug(
"Last event was published recently. No event is going to be published."
)
return None
return first_unpublished_event
@ -88,5 +113,8 @@ def select_event_to_publish(
published_events: List[MobilizonEvent], unpublished_events: List[MobilizonEvent],
):
strategy = STRATEGY_NAME_TO_STRATEGY_CLASS[settings["selection"]["strategy"]]()
strategy = STRATEGY_NAME_TO_STRATEGY_CLASS[
get_settings()["selection"]["strategy"]
]()
return strategy.select(published_events, unpublished_events)

45
mobilizon_bots/main.py
View File

@ -1,29 +1,27 @@
import logging.config
from pathlib import Path
from tortoise import run_async
from mobilizon_bots.config.config import update_settings_files
from mobilizon_bots.config.config import settings
from mobilizon_bots.config.publishers import get_active_publishers
from mobilizon_bots.event.event_selector import EventSelector, SelectNextEventStrategy
from mobilizon_bots.event.event_selection_strategies import select_event_to_publish
from mobilizon_bots.mobilizon.events import get_unpublished_events
from mobilizon_bots.publishers import get_active_publishers
from mobilizon_bots.publishers.coordinator import PublisherCoordinator
from mobilizon_bots.storage.db import MobilizonBotsDB
from mobilizon_bots.storage.query import get_published_events, create_unpublished_events
from mobilizon_bots.storage.query import (
get_unpublished_events as get_db_unpublished_events,
)
logger = logging.getLogger(__name__)
async def main():
async def main(settings_file):
"""
STUB
:return:
"""
logging.config.dictConfig(settings.logging)
active_publishers = get_active_publishers(settings)
settings = update_settings_files(settings_file)
logging.config.dictConfig(settings["logging"])
active_publishers = get_active_publishers()
db = MobilizonBotsDB(Path(settings.db_path))
await db.setup()
@ -33,20 +31,15 @@ async def main():
# Pull unpublished events from Mobilizon
unpublished_events = get_unpublished_events(published_events)
# Store in the DB only the ones we din't know about
# Store in the DB only the ones we didn't know about
await create_unpublished_events(unpublished_events, active_publishers)
unpublished_events = list(await get_db_unpublished_events())
event = select_event_to_publish(published_events, unpublished_events)
if event:
logger.debug(f"Event to publish found: {event.name}")
result = PublisherCoordinator(event).run()
event_selector = EventSelector(
unpublished_events=unpublished_events, published_events=published_events
)
# TODO: Here we should somehow handle publishers
strategy = SelectNextEventStrategy(minimum_break_between_events_in_minutes=360)
event = event_selector.select_event_to_publish(strategy)
result = PublisherCoordinator(event).publish() if event else exit(0)
exit(0 if result.is_success() else 1)
if __name__ == "__main__":
run_async(main())
logger.debug("Closing")
exit(0 if result.successful else 1)
else:
logger.debug("Closing")
exit(0)

19
mobilizon_bots/mobilizon/events.py
View File

@ -1,12 +1,16 @@
import json
import logging
from http import HTTPStatus
from typing import List, Optional
import arrow
import requests
from mobilizon_bots.config.config import settings
from mobilizon_bots.config.config import get_settings
from mobilizon_bots.event.event import MobilizonEvent, PublicationStatus
logger = logging.getLogger(__name__)
class MobilizonRequestFailed(Exception):
# TODO move to an error module
@ -86,9 +90,9 @@ def get_mobilizon_future_events(
page: int = 1, from_date: Optional[arrow.Arrow] = None
) -> List[MobilizonEvent]:
url = settings["source"]["mobilizon"]["url"]
url = get_settings()["source"]["mobilizon"]["url"]
query = query_future_events.format(
group=settings["source"]["mobilizon"]["group"],
group=get_settings()["source"]["mobilizon"]["group"],
page=page,
afterDatetime=from_date or arrow.now().isoformat(),
)
@ -98,6 +102,13 @@ def get_mobilizon_future_events(
f"Request for events failed with code:{r.status_code}"
)
response_json = r.json()
logger.debug(f"Response:\n{json.dumps(response_json, indent=4)}")
if "errors" in response_json:
raise MobilizonRequestFailed(
f"Request for events failed because of the following errors: "
f"{json.dumps(response_json['errors'],indent=4)}"
)
return list(
map(parse_event, r.json()["data"]["group"]["organizedEvents"]["elements"])
map(parse_event, response_json["data"]["group"]["organizedEvents"]["elements"])
)

6
mobilizon_bots/publishers/__init__.py
View File

@ -0,0 +1,6 @@
from mobilizon_bots.config.config import get_settings
import mobilizon_bots.config.publishers
def get_active_publishers():
return mobilizon_bots.config.publishers.get_active_publishers(get_settings())

13
mobilizon_bots/publishers/abstract.py
View File

@ -1,11 +1,11 @@
import inspect
import logging
from abc import ABC, abstractmethod
from dynaconf.utils.boxing import DynaBox
from jinja2 import Environment, FileSystemLoader, Template
from mobilizon_bots.config.config import settings
from mobilizon_bots.config.config import get_settings
from mobilizon_bots.event.event import MobilizonEvent
from .exceptions import PublisherError, InvalidAttribute
@ -49,7 +49,7 @@ class AbstractNotifier(ABC):
)
try:
t, n = cls._conf or tuple() # Avoid unpacking ``None``
return settings[t][n]
return get_settings()[t][n]
except (KeyError, ValueError):
raise InvalidAttribute(
f"Class {cls.__name__} has invalid ``_conf`` attribute"
@ -163,4 +163,9 @@ class AbstractPublisher(AbstractNotifier):
"""
Retrieves publisher's message template.
"""
return JINJA_ENV.get_template(self.conf.msg_template_path)
template_path = (
self.conf.msg_template_path
if hasattr(self.conf, "msg_template_path")
else self.default_template_path
)
return JINJA_ENV.get_template(template_path)

62
mobilizon_bots/publishers/coordinator.py
View File

@ -1,11 +1,11 @@
from dataclasses import dataclass, field
from typing import List
from mobilizon_bots.config.config import settings
from mobilizon_bots.config.publishers import get_active_publishers
from mobilizon_bots.event.event import MobilizonEvent, PublicationStatus
from .exceptions import PublisherError
from .abstract import AbstractPublisher
from .telegram import TelegramPublisher
from mobilizon_bots.publishers import get_active_publishers
from mobilizon_bots.publishers.abstract import AbstractPublisher
from mobilizon_bots.publishers.exceptions import PublisherError
from mobilizon_bots.publishers.telegram import TelegramPublisher
KEY2CLS = {"telegram": TelegramPublisher}
@ -15,62 +15,51 @@ class PublisherReport:
status: PublicationStatus
reason: str
publisher: AbstractPublisher
event: MobilizonEvent
@dataclass
class PublisherCoordinatorReport:
reports: list = field(default_factory=[])
reports: List[PublisherReport] = field(default_factory=[])
@property
def successful(self):
return all(r.status == PublicationStatus.COMPLETED for r in self.reports)
def __iter__(self):
return self.reports.__iter__()
class PublisherCoordinator:
def __init__(self, event: MobilizonEvent):
self.publishers = tuple(
KEY2CLS[pn](event) for pn in get_active_publishers(settings)
)
self.publishers = tuple(KEY2CLS[pn](event) for pn in get_active_publishers())
def run(self) -> PublisherCoordinatorReport:
invalid_credentials, invalid_event, invalid_msg = self._validate()
if invalid_credentials or invalid_event or invalid_msg:
return PublisherCoordinatorReport(
reports=invalid_credentials + invalid_event + invalid_msg
)
errors = invalid_credentials + invalid_event + invalid_msg
if errors:
return PublisherCoordinatorReport(reports=errors)
failed_publishers = self._post()
if failed_publishers:
return PublisherCoordinatorReport(reports=failed_publishers)
return self._post()
return PublisherCoordinatorReport(
reports=[
PublisherReport(
status=PublicationStatus.COMPLETED,
reason="",
publisher=p,
event=p.event,
)
for p in self.publishers
],
)
def _make_successful_report(self):
return [
PublisherReport(status=PublicationStatus.COMPLETED, reason="", publisher=p,)
for p in self.publishers
]
def _post(self):
failed_publishers = []
failed_publishers_reports = []
for p in self.publishers:
try:
p.post()
except PublisherError as e:
failed_publishers.append(
failed_publishers_reports.append(
PublisherReport(
status=PublicationStatus.FAILED,
reason=repr(e),
publisher=p,
event=p.event,
status=PublicationStatus.FAILED, reason=repr(e), publisher=p,
)
)
return failed_publishers
reports = failed_publishers_reports or self._make_successful_report()
return PublisherCoordinatorReport(reports)
def _validate(self):
invalid_credentials, invalid_event, invalid_msg = [], [], []
@ -81,7 +70,6 @@ class PublisherCoordinator:
status=PublicationStatus.FAILED,
reason="Invalid credentials",
publisher=p,
event=p.event,
)
)
if not p.is_event_valid():
@ -90,7 +78,6 @@ class PublisherCoordinator:
status=PublicationStatus.FAILED,
reason="Invalid event",
publisher=p,
event=p.event,
)
)
if not p.is_message_valid():
@ -99,7 +86,6 @@ class PublisherCoordinator:
status=PublicationStatus.FAILED,
reason="Invalid message",
publisher=p,
event=p.event,
)
)
return invalid_credentials, invalid_event, invalid_msg

16
mobilizon_bots/publishers/telegram.py
View File

@ -1,3 +1,4 @@
import pkg_resources
import requests
from .abstract import AbstractPublisher
@ -15,6 +16,9 @@ class TelegramPublisher(AbstractPublisher):
"""
_conf = ("publisher", "telegram")
default_template_path = pkg_resources.resource_filename(
"mobilizon_bots.publishers.templates", "telegram.tmpl.j2"
)
def post(self) -> None:
conf = self.conf
@ -38,8 +42,7 @@ class TelegramPublisher(AbstractPublisher):
err.append("username")
if err:
self._log_error(
", ".join(err) + " is/are missing",
raise_error=InvalidCredentials,
", ".join(err) + " is/are missing", raise_error=InvalidCredentials,
)
res = requests.get(f"https://api.telegram.org/bot{token}/getMe")
@ -47,8 +50,7 @@ class TelegramPublisher(AbstractPublisher):
if not username == data.get("result", {}).get("username"):
self._log_error(
"Found a different bot than the expected one",
raise_error=InvalidBot,
"Found a different bot than the expected one", raise_error=InvalidBot,
)
def validate_event(self) -> None:
@ -61,8 +63,7 @@ class TelegramPublisher(AbstractPublisher):
res.raise_for_status()
except requests.exceptions.HTTPError as e:
self._log_error(
f"Server returned invalid data: {str(e)}",
raise_error=InvalidResponse,
f"Server returned invalid data: {str(e)}", raise_error=InvalidResponse,
)
try:
@ -75,8 +76,7 @@ class TelegramPublisher(AbstractPublisher):
if not data.get("ok"):
self._log_error(
f"Invalid request (response: {data})",
raise_error=InvalidResponse,
f"Invalid request (response: {data})", raise_error=InvalidResponse,
)
return data

0
mobilizon_bots/publishers/templates/__init__.py Normal file
View File

1
mobilizon_bots/storage/db.py
View File

@ -1,7 +1,6 @@
import asyncio
import atexit
import logging
from pathlib import Path
from tortoise import Tortoise

63
mobilizon_bots/storage/query.py
View File

@ -1,3 +1,5 @@
import sys
from typing import Iterable, Optional
from tortoise.transactions import atomic
@ -6,6 +8,13 @@ from mobilizon_bots.event.event import MobilizonEvent
from mobilizon_bots.models.event import Event
from mobilizon_bots.models.publication import Publication, PublicationStatus
from mobilizon_bots.models.publisher import Publisher
from mobilizon_bots.publishers.coordinator import PublisherCoordinatorReport
# This is due to Tortoise community fixtures to
# set up and tear down a DB instance for Pytest.
# See: https://github.com/tortoise/tortoise-orm/issues/419#issuecomment-696991745
# and: https://docs.pytest.org/en/stable/example/simple.html
CONNECTION_NAME = "models" if "pytest" in sys.modules else None
async def events_with_status(
@ -23,22 +32,32 @@ async def events_with_status(
async def get_published_events() -> Iterable[MobilizonEvent]:
return await events_with_status(
[
PublicationStatus.COMPLETED,
PublicationStatus.PARTIAL,
]
[PublicationStatus.COMPLETED, PublicationStatus.PARTIAL]
)
async def get_unpublished_events() -> Iterable[MobilizonEvent]:
return await events_with_status(
[
PublicationStatus.WAITING,
]
return await events_with_status([PublicationStatus.WAITING])
async def save_event(event):
event_model = event.to_model()
await event_model.save()
return event_model
async def save_publication(publisher_name, event_model, status: PublicationStatus):
publisher = await Publisher.filter(name=publisher_name).first()
await Publication.create(
status=status,
event_id=event_model.id,
publisher_id=publisher.id,
)
@atomic("models")
@atomic(CONNECTION_NAME)
async def create_unpublished_events(
unpublished_mobilizon_events: Iterable[MobilizonEvent],
active_publishers: Iterable[str],
@ -47,23 +66,25 @@ async def create_unpublished_events(
unpublished_event_models = set(
map(lambda event: event.mobilizon_id, await get_unpublished_events())
)
unpublished_events = filter(
lambda event: event.mobilizon_id not in unpublished_event_models,
unpublished_mobilizon_events,
unpublished_events = list(
filter(
lambda event: event.mobilizon_id not in unpublished_event_models,
unpublished_mobilizon_events,
)
)
for event in unpublished_events:
event_model = event.to_model()
await event_model.save()
for publisher_name in active_publishers:
publisher = await Publisher.filter(name=publisher_name).first()
await Publication.create(
status=PublicationStatus.WAITING,
event_id=event_model.id,
publisher_id=publisher.id,
event_model = await save_event(event)
for publisher in active_publishers:
await save_publication(
publisher, event_model, status=PublicationStatus.WAITING
)
async def create_publisher(name: str, account_ref: Optional[str] = None) -> None:
await Publisher.create(name=name, account_ref=account_ref)
async def save_publication_report(publication_report: PublisherCoordinatorReport):
for publisher_report in publication_report:
pass

17
tests/event/test_strategies.py
View File

@ -3,7 +3,7 @@ import pytest
from unittest.mock import patch
from mobilizon_bots.config.config import settings
from mobilizon_bots.config.config import get_settings
from mobilizon_bots.event.event_selection_strategies import (
SelectNextEventStrategy,
select_event_to_publish,
@ -12,7 +12,7 @@ from mobilizon_bots.event.event_selection_strategies import (
@pytest.fixture
def set_break_window_config(desired_break_window_days):
settings.update(
get_settings().update(
{
"selection.strategy_options.break_between_events_in_minutes": desired_break_window_days
* 24
@ -23,20 +23,27 @@ def set_break_window_config(desired_break_window_days):
@pytest.fixture
def set_strategy(strategy_name):
settings.update({"selection.strategy": strategy_name})
get_settings().update({"selection.strategy": strategy_name})
@pytest.fixture
def mock_publication_window(publication_window):
begin, end = publication_window
settings.update({"publishing.window.begin": begin, "publishing.window.end": end})
get_settings().update(
{"publishing.window.begin": begin, "publishing.window.end": end}
)
def test_window_no_event():
selected_event = SelectNextEventStrategy().select([], [])
assert selected_event is None
@pytest.mark.parametrize("current_hour", [10])
@pytest.mark.parametrize(
"desired_break_window_days,days_passed_from_publication", [[2, 1], [3, 2]]
)
def test_window_simple_no_event(
def test_window_simple_no_event_in_window(
event_generator,
desired_break_window_days,
days_passed_from_publication,

4
tests/mobilizon/conftest.py
View File

@ -1,12 +1,12 @@
import pytest
import responses
from mobilizon_bots.config.config import settings
from mobilizon_bots.config.config import get_settings
@pytest.fixture
def mobilizon_url():
return settings["source"]["mobilizon"]["url"]
return get_settings()["source"]["mobilizon"]["url"]
@responses.activate

29
tests/mobilizon/test_events.py
View File

@ -1,7 +1,7 @@
import pytest
import arrow
import pytest
from mobilizon_bots.event.event import PublicationStatus, MobilizonEvent
from mobilizon_bots.event.event import MobilizonEvent
from mobilizon_bots.mobilizon.events import (
get_mobilizon_future_events,
MobilizonRequestFailed,
@ -87,7 +87,30 @@ def test_event_response(mock_mobilizon_success_answer, expected_result):
assert get_mobilizon_future_events() == expected_result
def test_failure(mock_mobilizon_failure_answer):
def test_failure_404(mock_mobilizon_failure_answer):
with pytest.raises(MobilizonRequestFailed):
get_mobilizon_future_events()
@pytest.mark.parametrize(
"mobilizon_answer",
[
{
"data": {"group": None},
"errors": [
{
"code": "group_not_found",
"field": None,
"locations": [{"column": 13, "line": 2}],
"message": "Group not found",
"path": ["group"],
"status_code": 404,
}
],
},
],
)
def test_failure_wrong_group(mock_mobilizon_success_answer):
with pytest.raises(MobilizonRequestFailed):
get_mobilizon_future_events()

93
tests/models/test_event.py
View File

@ -149,3 +149,96 @@ async def test_mobilizon_event_from_model(
assert event.location == "loc_1"
assert event.publication_time[publisher_model.name] == publication.timestamp
assert event.publication_status == PublicationStatus.PARTIAL
@pytest.mark.asyncio
async def test_mobilizon_event_compute_status_failed(
event_model_generator, publication_model_generator, publisher_model_generator
):
event_model = event_model_generator()
await event_model.save()
publisher_model = publisher_model_generator()
await publisher_model.save()
publisher_model_2 = publisher_model_generator(idx=2)
await publisher_model_2.save()
publication = publication_model_generator(
status=PublicationStatus.FAILED,
event_id=event_model.id,
publisher_id=publisher_model.id,
)
await publication.save()
publication_2 = publication_model_generator(
status=PublicationStatus.COMPLETED,
event_id=event_model.id,
publisher_id=publisher_model_2.id,
)
await publication_2.save()
assert (
MobilizonEvent.compute_status([publication, publication_2])
== PublicationStatus.FAILED
)
@pytest.mark.asyncio
async def test_mobilizon_event_compute_status_partial(
event_model_generator, publication_model_generator, publisher_model_generator
):
event_model = event_model_generator()
await event_model.save()
publisher_model = publisher_model_generator()
await publisher_model.save()
publisher_model_2 = publisher_model_generator(idx=2)
await publisher_model_2.save()
publication = publication_model_generator(
status=PublicationStatus.WAITING,
event_id=event_model.id,
publisher_id=publisher_model.id,
)
await publication.save()
publication_2 = publication_model_generator(
status=PublicationStatus.COMPLETED,
event_id=event_model.id,
publisher_id=publisher_model_2.id,
)
await publication_2.save()
assert (
MobilizonEvent.compute_status([publication, publication_2])
== PublicationStatus.PARTIAL
)
@pytest.mark.asyncio
async def test_mobilizon_event_compute_status_waiting(
event_model_generator, publication_model_generator, publisher_model_generator
):
event_model = event_model_generator()
await event_model.save()
publisher_model = publisher_model_generator()
await publisher_model.save()
publisher_model_2 = publisher_model_generator(idx=2)
await publisher_model_2.save()
publication = publication_model_generator(
status=PublicationStatus.WAITING,
event_id=event_model.id,
publisher_id=publisher_model.id,
)
await publication.save()
publication_2 = publication_model_generator(
status=PublicationStatus.WAITING,
event_id=event_model.id,
publisher_id=publisher_model_2.id,
)
await publication_2.save()
assert (
MobilizonEvent.compute_status([publication, publication_2])
== PublicationStatus.WAITING
)