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

download events (#19) 

* added get events stub

* added event parsing

* removed datetime

* added config and tests

* fixed response format in test

* more tests

* added error handling for request

* improved dummy data

* added get_unpublished_events

* added test

* removed last_accessed

* mobilizon_url moved to fixture

* added config comments

* moved mobilizon group to config
This commit is contained in:
Simone Robutti 2021-05-30 21:47:36 +02:00 committed by GitHub
parent 044c66eb5b
commit fcd5116d1c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 561 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
mobilizon_bots/config/config.py
View File

@ -1,49 +1,82 @@
from typing import List
from dynaconf import Dynaconf, Validator
from mobilizon_bots.config import strategies, publishers, notifiers
from mobilizon_bots.config.notifiers import notifier_names
from mobilizon_bots.config.publishers import publisher_names
SETTINGS_FILE = [
"mobilizon_bots/settings.toml",
"mobilizon_bots/.secrets.toml",
"/etc/mobilizon_bots.toml",
"/etc/mobilizon_bots_secrets.toml",
]
ENVVAR_PREFIX = "MOBILIZON_BOTS"
base_validators = (
[Validator("selection.strategy", must_exist=True)]
+ [
Validator(
f"publisher.{publisher_name}.active", must_exist=True, is_type_of=bool
)
for publisher_name in publisher_names
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",
]
+ [
Validator(f"notifier.{notifier_name}.active", must_exist=True, is_type_of=bool)
for notifier_name in notifier_names
]
)
ENVVAR_PREFIX = "MOBILIZON_BOTS"
raw_settings = Dynaconf(
environments=True,
envvar_prefix=ENVVAR_PREFIX,
settings_files=SETTINGS_FILE,
validators=base_validators,
)
return Dynaconf(
environments=True,
envvar_prefix=ENVVAR_PREFIX,
settings_files=SETTINGS_FILE,
validators=validators or [],
)
strategy_validators = strategies.get_validators(raw_settings)
publisher_validators = publishers.get_validators(raw_settings)
notifier_validators = notifiers.get_validators(raw_settings)
settings = Dynaconf(
environments=True,
envvar_prefix=ENVVAR_PREFIX,
settings_files=SETTINGS_FILE,
validators=base_validators
+ strategy_validators
+ publisher_validators
+ notifier_validators,
)
# TODO use validation control in dynaconf 3.2.0 once released
settings.validators.validate()
def build_and_validate_settings(settings_files: List[str] = None):
"""
Creates a settings object to be used in the application. It collects and apply generic validators and validators
specific for each publisher, notifier and publication strategy.
"""
# we first do a preliminary load of the settings without validation. We will later use them to determine which
# publishers, notifiers and strategy have been selected
raw_settings = build_settings(settings_files=settings_files)
# These validators are always applied
base_validators = (
[
# strategy to decide events to publish
Validator("selection.strategy", must_exist=True, is_type_of=str),
# url of the main Mobilizon instance to download events from
Validator("source.mobilizon.url", must_exist=True, is_type_of=str),
Validator("source.mobilizon.group", must_exist=True, is_type_of=str),
]
+ [
Validator(
f"publisher.{publisher_name}.active", must_exist=True, is_type_of=bool
)
for publisher_name in publisher_names
]
+ [
Validator(
f"notifier.{notifier_name}.active", must_exist=True, is_type_of=bool
)
for notifier_name in notifier_names
]
)
# we retrieve validators that are conditional. Each module will analyze the settings and decide which validators
# need to be applied.
strategy_validators = strategies.get_validators(raw_settings)
publisher_validators = publishers.get_validators(raw_settings)
notifier_validators = notifiers.get_validators(raw_settings)
# we rebuild the settings, providing all the selected validators.
settings = build_settings(
settings_files,
base_validators
+ strategy_validators
+ publisher_validators
+ notifier_validators,
)
# TODO use validation control in dynaconf 3.2.0 once released
settings.validators.validate()
return settings
settings = build_and_validate_settings()

12
mobilizon_bots/event/event.py
View File

@ -1,8 +1,8 @@
from dataclasses import dataclass, asdict
from datetime import datetime
from enum import Enum
from typing import Optional
import arrow
from jinja2 import Template
@ -18,18 +18,18 @@ class MobilizonEvent:
"""Class representing an event retrieved from Mobilizon."""
name: str
description: str
begin_datetime: datetime
end_datetime: datetime
last_accessed: datetime
description: Optional[str]
begin_datetime: arrow.Arrow
end_datetime: arrow.Arrow
mobilizon_link: str
mobilizon_id: str
thumbnail_link: Optional[str] = None
location: Optional[str] = None
publication_time: Optional[datetime] = None
publication_time: Optional[arrow.Arrow] = None
publication_status: PublicationStatus = PublicationStatus.WAITING
def __post_init__(self):
assert self.begin_datetime < self.end_datetime
if self.publication_time:
assert self.publication_status in [

20
mobilizon_bots/event/event_selector.py
View File

@ -1,6 +1,6 @@
from abc import ABC, abstractmethod
from datetime import timedelta, datetime
from typing import List, Optional
import arrow
from mobilizon_bots.event.event import MobilizonEvent
@ -16,8 +16,10 @@ class EventSelectionStrategy(ABC):
class SelectNextEventStrategy(EventSelectionStrategy):
def __init__(self, minimum_break_between_events: timedelta = timedelta(days=0)):
self.minimum_break_between_events = minimum_break_between_events
def __init__(self, minimum_break_between_events_in_minutes: int):
self.minimum_break_between_events_in_minutes = (
minimum_break_between_events_in_minutes
)
def select(
self,
@ -27,15 +29,17 @@ class SelectNextEventStrategy(EventSelectionStrategy):
last_published_event = published_events[-1]
first_unpublished_event = unpublished_events[0]
assert last_published_event.publication_time < datetime.now(), (
now = arrow.now()
assert last_published_event.publication_time < now, (
f"Last published event has been published in the future\n"
f"{last_published_event.publication_time}\n"
f"{datetime.now()}"
f"{now}"
)
if (
last_published_event.publication_time + self.minimum_break_between_events
> datetime.now()
last_published_event.publication_time.shift(
minutes=self.minimum_break_between_events_in_minutes
)
> now
):
return None

3
mobilizon_bots/main.py
View File

@ -1,3 +1,4 @@
from mobilizon_bots.mobilizon.events import get_unpublished_events
from mobilizon_bots.storage.db import MobilizonBotsDB
@ -7,8 +8,8 @@ def main():
:return:
"""
db = MobilizonBotsDB().setup()
unpublished_events = Mobilizon().get_unpublished_events()
published_events = db.get_published_events()
unpublished_events = get_unpublished_events(published_events)
event = select_event_to_publish()
result = PublisherCoordinator(event).publish() if event else exit(0)
exit(0 if result.is_success() else 1)

0
mobilizon_bots/mobilizon/__init__.py Normal file
View File

103
mobilizon_bots/mobilizon/events.py Normal file
View File

@ -0,0 +1,103 @@
from http import HTTPStatus
from typing import List, Optional
import arrow
import requests
from mobilizon_bots.config.config import settings
from mobilizon_bots.event.event import MobilizonEvent, PublicationStatus
class MobilizonRequestFailed(Exception):
# TODO move to an error module
pass
def parse_location(data):
# TODO define a better logic (or a customizable strategy) to get the location
return (data.get("physicalAddress", {}) or {}).get("locality") or data.get(
"onlineAddress"
)
def parse_picture(data):
return (data.get("picture", {}) or {}).get("url")
def parse_event(data):
return MobilizonEvent(
name=data["title"],
description=data.get("description", None),
begin_datetime=arrow.get(data["beginsOn"]) if "beginsOn" in data else None,
end_datetime=arrow.get(data["endsOn"]) if "endsOn" in data else None,
mobilizon_link=data.get("url", None),
mobilizon_id=data["uuid"],
thumbnail_link=parse_picture(data),
location=parse_location(data),
publication_time=None,
publication_status=PublicationStatus.WAITING,
)
query_future_events = """{{
group(preferredUsername: "{group}") {{
organizedEvents(page:{page}, afterDatetime:"{afterDatetime}"){{
elements {{
title,
url,
beginsOn,
endsOn,
options {{
showStartTime,
showEndTime,
}},
uuid,
description,
onlineAddress,
physicalAddress {{
locality,
description,
region
}},
picture {{
url
}},
}},
}}
}}
}}"""
def get_unpublished_events(published_events: List[MobilizonEvent]):
# I take all the future events
future_events = get_mobilizon_future_events()
# I get the ids of all the published events coming from the DB
published_events_id = set(map(lambda x: x.mobilizon_id, published_events))
# I keep the future events only the ones that haven't been published
# Note: some events might exist in the DB and be unpublished. Here they should be ignored because the information
# in the DB might be old and the event might have been updated.
# We assume the published_events list doesn't contain such events.
return list(
filter(lambda x: x.mobilizon_id not in published_events_id, future_events)
)
def get_mobilizon_future_events(
page: int = 1, from_date: Optional[arrow.Arrow] = None
) -> List[MobilizonEvent]:
url = settings["source"]["mobilizon"]["url"]
query = query_future_events.format(
group=settings["source"]["mobilizon"]["group"],
page=page,
afterDatetime=from_date or arrow.now().isoformat(),
)
r = requests.post(url, json={"query": query})
if r.status_code != HTTPStatus.OK:
raise MobilizonRequestFailed(
f"Request for events failed with code:{r.status_code}"
)
return list(
map(parse_event, r.json()["data"]["group"]["organizedEvents"]["elements"])
)

4
mobilizon_bots/settings.toml
View File

@ -6,6 +6,10 @@ log_dir = "/var/log/mobots"
db_name = "events.db"
db_path = "@format {this.local_state_dir}/{this.db_name}"
[default.source.mobilizon]
url="https://some_mobilizon"
group="my_group"
[default.selection]
strategy = "next_event"

139
poetry.lock generated
View File

@ -9,6 +9,17 @@ python-versions = ">=3.6"
[package.dependencies]
typing_extensions = ">=3.7.2"
[[package]]
name = "arrow"
version = "1.1.0"
description = "Better dates & times for Python"
category = "main"
optional = false
python-versions = ">=3.6"
[package.dependencies]
python-dateutil = ">=2.7.0"
[[package]]
name = "atomicwrites"
version = "1.4.0"
@ -31,6 +42,22 @@ docs = ["furo", "sphinx", "zope.interface"]
tests = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "six", "zope.interface"]
tests_no_zope = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "six"]
[[package]]
name = "certifi"
version = "2020.12.5"
description = "Python package for providing Mozilla's CA Bundle."
category = "main"
optional = false
python-versions = "*"
[[package]]
name = "chardet"
version = "4.0.0"
description = "Universal encoding detector for Python 2 and 3"
category = "main"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
[[package]]
name = "colorama"
version = "0.4.4"
@ -56,6 +83,14 @@ toml = ["toml"]
vault = ["hvac"]
yaml = ["ruamel.yaml"]
[[package]]
name = "idna"
version = "2.10"
description = "Internationalized Domain Names in Applications (IDNA)"
category = "main"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
[[package]]
name = "iso8601"
version = "0.1.14"
@ -162,6 +197,17 @@ wcwidth = "*"
checkqa-mypy = ["mypy (==v0.761)"]
testing = ["argcomplete", "hypothesis (>=3.56)", "mock", "nose", "requests", "xmlschema"]
[[package]]
name = "python-dateutil"
version = "2.8.1"
description = "Extensions to the standard Python datetime module"
category = "main"
optional = false
python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,>=2.7"
[package.dependencies]
six = ">=1.5"
[[package]]
name = "pytz"
version = "2020.5"
@ -170,6 +216,48 @@ category = "main"
optional = false
python-versions = "*"
[[package]]
name = "requests"
version = "2.25.1"
description = "Python HTTP for Humans."
category = "main"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
[package.dependencies]
certifi = ">=2017.4.17"
chardet = ">=3.0.2,<5"
idna = ">=2.5,<3"
urllib3 = ">=1.21.1,<1.27"
[package.extras]
security = ["pyOpenSSL (>=0.14)", "cryptography (>=1.3.4)"]
socks = ["PySocks (>=1.5.6,!=1.5.7)", "win-inet-pton"]
[[package]]
name = "responses"
version = "0.13.3"
description = "A utility library for mocking out the `requests` Python library."
category = "dev"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
[package.dependencies]
requests = ">=2.0"
six = "*"
urllib3 = ">=1.25.10"
[package.extras]
tests = ["coverage (>=3.7.1,<6.0.0)", "pytest-cov", "pytest-localserver", "flake8", "pytest (>=4.6,<5.0)", "pytest (>=4.6)", "mypy"]
[[package]]
name = "six"
version = "1.16.0"
description = "Python 2 and 3 compatibility utilities"
category = "main"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*"
[[package]]
name = "tortoise-orm"
version = "0.17.2"
@ -199,6 +287,19 @@ category = "main"
optional = false
python-versions = "*"
[[package]]
name = "urllib3"
version = "1.26.4"
description = "HTTP library with thread-safe connection pooling, file post, and more."
category = "main"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*, <4"
[package.extras]
secure = ["pyOpenSSL (>=0.14)", "cryptography (>=1.3.4)", "idna (>=2.0.0)", "certifi", "ipaddress"]
socks = ["PySocks (>=1.5.6,!=1.5.7,<2.0)"]
brotli = ["brotlipy (>=0.6.0)"]
[[package]]
name = "wcwidth"
version = "0.2.5"
@ -210,13 +311,17 @@ python-versions = "*"
[metadata]
lock-version = "1.1"
python-versions = "^3.9"
content-hash = "2a9d2f27ed4c4197d4dd022c2e1f8968ef7fff2ab73145a4bd74712f35d88855"
content-hash = "be81df312539b9c6d1020d05e5001c8819f5ca47435a84e398b35cd39020e7bf"
[metadata.files]
aiosqlite = [
{file = "aiosqlite-0.16.1-py3-none-any.whl", hash = "sha256:1df802815bb1e08a26c06d5ea9df589bcb8eec56e5f3378103b0f9b223c6703c"},
{file = "aiosqlite-0.16.1.tar.gz", hash = "sha256:2e915463164efa65b60fd1901aceca829b6090082f03082618afca6fb9c8fdf7"},
]
arrow = [
{file = "arrow-1.1.0-py3-none-any.whl", hash = "sha256:8cbe6a629b1c54ae11b52d6d9e70890089241958f63bc59467e277e34b7a5378"},
{file = "arrow-1.1.0.tar.gz", hash = "sha256:b8fe13abf3517abab315e09350c903902d1447bd311afbc17547ba1cb3ff5bd8"},
]
atomicwrites = [
{file = "atomicwrites-1.4.0-py2.py3-none-any.whl", hash = "sha256:6d1784dea7c0c8d4a5172b6c620f40b6e4cbfdf96d783691f2e1302a7b88e197"},
{file = "atomicwrites-1.4.0.tar.gz", hash = "sha256:ae70396ad1a434f9c7046fd2dd196fc04b12f9e91ffb859164193be8b6168a7a"},
@ -225,6 +330,14 @@ attrs = [
{file = "attrs-20.3.0-py2.py3-none-any.whl", hash = "sha256:31b2eced602aa8423c2aea9c76a724617ed67cf9513173fd3a4f03e3a929c7e6"},
{file = "attrs-20.3.0.tar.gz", hash = "sha256:832aa3cde19744e49938b91fea06d69ecb9e649c93ba974535d08ad92164f700"},
]
certifi = [
{file = "certifi-2020.12.5-py2.py3-none-any.whl", hash = "sha256:719a74fb9e33b9bd44cc7f3a8d94bc35e4049deebe19ba7d8e108280cfd59830"},
{file = "certifi-2020.12.5.tar.gz", hash = "sha256:1a4995114262bffbc2413b159f2a1a480c969de6e6eb13ee966d470af86af59c"},
]
chardet = [
{file = "chardet-4.0.0-py2.py3-none-any.whl", hash = "sha256:f864054d66fd9118f2e67044ac8981a54775ec5b67aed0441892edb553d21da5"},
{file = "chardet-4.0.0.tar.gz", hash = "sha256:0d6f53a15db4120f2b08c94f11e7d93d2c911ee118b6b30a04ec3ee8310179fa"},
]
colorama = [
{file = "colorama-0.4.4-py2.py3-none-any.whl", hash = "sha256:9f47eda37229f68eee03b24b9748937c7dc3868f906e8ba69fbcbdd3bc5dc3e2"},
{file = "colorama-0.4.4.tar.gz", hash = "sha256:5941b2b48a20143d2267e95b1c2a7603ce057ee39fd88e7329b0c292aa16869b"},
@ -233,6 +346,10 @@ dynaconf = [
{file = "dynaconf-3.1.4-py2.py3-none-any.whl", hash = "sha256:e6f383b84150b70fc439c8b2757581a38a58d07962aa14517292dcce1a77e160"},
{file = "dynaconf-3.1.4.tar.gz", hash = "sha256:b2f472d83052f809c5925565b8a2ba76a103d5dc1dbb9748b693ed67212781b9"},
]
idna = [
{file = "idna-2.10-py2.py3-none-any.whl", hash = "sha256:b97d804b1e9b523befed77c48dacec60e6dcb0b5391d57af6a65a312a90648c0"},
{file = "idna-2.10.tar.gz", hash = "sha256:b307872f855b18632ce0c21c5e45be78c0ea7ae4c15c828c20788b26921eb3f6"},
]
iso8601 = [
{file = "iso8601-0.1.14-py2.py3-none-any.whl", hash = "sha256:e7e1122f064d626e17d47cd5106bed2c620cb38fe464999e0ddae2b6d2de6004"},
{file = "iso8601-0.1.14.tar.gz", hash = "sha256:8aafd56fa0290496c5edbb13c311f78fa3a241f0853540da09d9363eae3ebd79"},
@ -299,10 +416,26 @@ pytest = [
{file = "pytest-5.4.3-py3-none-any.whl", hash = "sha256:5c0db86b698e8f170ba4582a492248919255fcd4c79b1ee64ace34301fb589a1"},
{file = "pytest-5.4.3.tar.gz", hash = "sha256:7979331bfcba207414f5e1263b5a0f8f521d0f457318836a7355531ed1a4c7d8"},
]
python-dateutil = [
{file = "python-dateutil-2.8.1.tar.gz", hash = "sha256:73ebfe9dbf22e832286dafa60473e4cd239f8592f699aa5adaf10050e6e1823c"},
{file = "python_dateutil-2.8.1-py2.py3-none-any.whl", hash = "sha256:75bb3f31ea686f1197762692a9ee6a7550b59fc6ca3a1f4b5d7e32fb98e2da2a"},
]
pytz = [
{file = "pytz-2020.5-py2.py3-none-any.whl", hash = "sha256:16962c5fb8db4a8f63a26646d8886e9d769b6c511543557bc84e9569fb9a9cb4"},
{file = "pytz-2020.5.tar.gz", hash = "sha256:180befebb1927b16f6b57101720075a984c019ac16b1b7575673bea42c6c3da5"},
]
requests = [
{file = "requests-2.25.1-py2.py3-none-any.whl", hash = "sha256:c210084e36a42ae6b9219e00e48287def368a26d03a048ddad7bfee44f75871e"},
{file = "requests-2.25.1.tar.gz", hash = "sha256:27973dd4a904a4f13b263a19c866c13b92a39ed1c964655f025f3f8d3d75b804"},
]
responses = [
{file = "responses-0.13.3-py2.py3-none-any.whl", hash = "sha256:b54067596f331786f5ed094ff21e8d79e6a1c68ef625180a7d34808d6f36c11b"},
{file = "responses-0.13.3.tar.gz", hash = "sha256:18a5b88eb24143adbf2b4100f328a2f5bfa72fbdacf12d97d41f07c26c45553d"},
]
six = [
{file = "six-1.16.0-py2.py3-none-any.whl", hash = "sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254"},
{file = "six-1.16.0.tar.gz", hash = "sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926"},
]
tortoise-orm = [
{file = "tortoise-orm-0.17.2.tar.gz", hash = "sha256:1a742b2f15a31d47a8dea7706b478cc9a7ce9af268b61d77d0fa22cfbaea271a"},
{file = "tortoise_orm-0.17.2-py3-none-any.whl", hash = "sha256:b0c02be3800398053058377ddca91fa051eb98eebb704d2db2a3ab1c6a58e347"},
@ -312,6 +445,10 @@ typing-extensions = [
{file = "typing_extensions-3.10.0.0-py3-none-any.whl", hash = "sha256:779383f6086d90c99ae41cf0ff39aac8a7937a9283ce0a414e5dd782f4c94a84"},
{file = "typing_extensions-3.10.0.0.tar.gz", hash = "sha256:50b6f157849174217d0656f99dc82fe932884fb250826c18350e159ec6cdf342"},
]
urllib3 = [
{file = "urllib3-1.26.4-py2.py3-none-any.whl", hash = "sha256:2f4da4594db7e1e110a944bb1b551fdf4e6c136ad42e4234131391e21eb5b0df"},
{file = "urllib3-1.26.4.tar.gz", hash = "sha256:e7b021f7241115872f92f43c6508082facffbd1c048e3c6e2bb9c2a157e28937"},
]
wcwidth = [
{file = "wcwidth-0.2.5-py2.py3-none-any.whl", hash = "sha256:beb4802a9cebb9144e99086eff703a642a13d6a0052920003a230f3294bbe784"},
{file = "wcwidth-0.2.5.tar.gz", hash = "sha256:c4d647b99872929fdb7bdcaa4fbe7f01413ed3d98077df798530e5b04f116c83"},

3
pyproject.toml
View File

@ -10,9 +10,12 @@ dynaconf = "^3.1.4"
tortoise-orm = "^0.17"
aiosqlite = "^0.16"
Jinja2 = "^2.11.3"
requests = "^2.25.1"
arrow = "^1.1.0"
[tool.poetry.dev-dependencies]
pytest = "^5.3"
responses = "^0.13.3"
[build-system]
requires = ["poetry-core>=1.0.0"]

28
tests/conftest.py
View File

@ -1,7 +1,9 @@
from datetime import datetime, timedelta
import arrow
import pkg_resources
import pytest
from dynaconf import settings
from mobilizon_bots.config.config import build_and_validate_settings
from mobilizon_bots.event.event import MobilizonEvent, PublicationStatus
@ -9,10 +11,20 @@ def generate_publication_status(published):
return PublicationStatus.COMPLETED if published else PublicationStatus.WAITING
@pytest.fixture(scope="session", autouse=True)
def set_test_settings():
config_file = pkg_resources.resource_filename(
"tests.resources", "test_settings.toml"
)
settings.configure(FORCE_ENV_FOR_DYNACONF="testing")
build_and_validate_settings([config_file])
@pytest.fixture
def event_generator():
def _event_generator(
begin_date=datetime(year=2021, month=1, day=1, hour=11, minute=30),
begin_date=arrow.Arrow(year=2021, month=1, day=1, hour=11, minute=30),
published=False,
publication_time=None,
):
@ -21,15 +33,14 @@ def event_generator():
name="test event",
description="description of the event",
begin_datetime=begin_date,
end_datetime=begin_date + timedelta(hours=2),
last_accessed=datetime.now(),
end_datetime=begin_date.shift(hours=2),
mobilizon_link="http://some_link.com/123",
mobilizon_id="12345",
thumbnail_link="http://some_link.com/123.jpg",
location="location",
publication_status=generate_publication_status(published),
publication_time=publication_time
or (begin_date - timedelta(days=1) if published else None),
or (begin_date.shift(days=-1) if published else None),
)
return _event_generator
@ -40,9 +51,8 @@ def event() -> MobilizonEvent:
return MobilizonEvent(
name="test event",
description="description of the event",
begin_datetime=datetime(year=2021, month=1, day=1, hour=11, minute=30),
end_datetime=datetime(year=2021, month=1, day=1, hour=12, minute=30),
last_accessed=datetime.now(),
begin_datetime=arrow.Arrow(year=2021, month=1, day=1, hour=11, minute=30),
end_datetime=arrow.Arrow(year=2021, month=1, day=1, hour=12, minute=30),
mobilizon_link="http://some_link.com/123",
mobilizon_id="12345",
thumbnail_link="http://some_link.com/123.jpg",

46
tests/event/test_strategies.py
View File

@ -1,104 +1,98 @@
from datetime import datetime, timedelta
import arrow
import pytest
from mobilizon_bots.event.event_selector import SelectNextEventStrategy
@pytest.mark.parametrize(
"desired_break_window,days_passed_from_publication", [[2, 1], [3, 2]]
"desired_break_window_days,days_passed_from_publication", [[2, 1], [3, 2]]
)
def test_window_simple_no_event(
event_generator, desired_break_window, days_passed_from_publication
event_generator, desired_break_window_days, days_passed_from_publication
):
"Testing that the break between events is respected"
unpublished_events = [
event_generator(
published=False,
begin_date=datetime(year=2021, month=1, day=5, hour=11, minute=30),
begin_date=arrow.Arrow(year=2021, month=1, day=5, hour=11, minute=30),
)
]
published_events = [
event_generator(
published=True,
publication_time=datetime.now()
- timedelta(days=days_passed_from_publication),
publication_time=arrow.now().shift(days=-days_passed_from_publication),
)
]
selected_event = SelectNextEventStrategy(
minimum_break_between_events=timedelta(days=desired_break_window)
minimum_break_between_events_in_minutes=desired_break_window_days * 24 * 60
).select(published_events, unpublished_events)
assert selected_event is None
@pytest.mark.parametrize(
"desired_break_window,days_passed_from_publication", [[1, 2], [2, 10], [4, 4]]
"desired_break_window_days,days_passed_from_publication", [[1, 2], [2, 10], [4, 4]]
)
def test_window_simple_event_found(
event_generator, desired_break_window, days_passed_from_publication,
event_generator, desired_break_window_days, days_passed_from_publication,
):
"Testing that the break between events is respected and an event is found"
unpublished_events = [
event_generator(
published=False,
begin_date=datetime(year=2021, month=1, day=5, hour=11, minute=30),
begin_date=arrow.Arrow(year=2021, month=1, day=5, hour=11, minute=30),
)
]
published_events = [
event_generator(
published=True,
publication_time=datetime.now()
- timedelta(days=days_passed_from_publication),
publication_time=arrow.now().shift(days=-days_passed_from_publication),
)
]
selected_event = SelectNextEventStrategy(
minimum_break_between_events=timedelta(days=desired_break_window)
minimum_break_between_events_in_minutes=desired_break_window_days * 24 * 60
).select(published_events, unpublished_events)
assert selected_event is unpublished_events[0]
@pytest.mark.parametrize(
"desired_break_window,days_passed_from_publication", [[1, 2], [2, 10], [4, 4]]
"desired_break_window_days,days_passed_from_publication", [[1, 2], [2, 10], [4, 4]]
)
def test_window_multi_event_found(
event_generator, desired_break_window, days_passed_from_publication,
event_generator, desired_break_window_days, days_passed_from_publication,
):
"Testing that the break between events is respected when there are multiple events"
unpublished_events = [
event_generator(
published=False,
begin_date=datetime(year=2022, month=1, day=5, hour=11, minute=30),
begin_date=arrow.Arrow(year=2022, month=1, day=5, hour=11, minute=30),
),
event_generator(
published=False,
begin_date=datetime(year=2022, month=3, day=5, hour=11, minute=30),
begin_date=arrow.Arrow(year=2022, month=3, day=5, hour=11, minute=30),
),
event_generator(
published=False,
begin_date=datetime(year=2021, month=1, day=5, hour=11, minute=30),
begin_date=arrow.Arrow(year=2021, month=1, day=5, hour=11, minute=30),
),
]
published_events = [
event_generator(
published=True,
publication_time=datetime.now()
- timedelta(days=days_passed_from_publication),
publication_time=arrow.now().shift(days=-days_passed_from_publication),
),
event_generator(
published=True,
publication_time=datetime.now()
- timedelta(days=days_passed_from_publication + 2),
publication_time=arrow.now().shift(days=-days_passed_from_publication - 2),
),
event_generator(
published=True,
publication_time=datetime.now()
- timedelta(days=days_passed_from_publication + 4),
publication_time=arrow.now().shift(days=-days_passed_from_publication - 4),
),
]
selected_event = SelectNextEventStrategy(
minimum_break_between_events=timedelta(days=desired_break_window)
minimum_break_between_events_in_minutes=desired_break_window_days * 24 * 60
).select(published_events, unpublished_events)
assert selected_event is unpublished_events[0]

0
tests/mobilizon/__init__.py Normal file
View File

31
tests/mobilizon/conftest.py Normal file
View File

@ -0,0 +1,31 @@
import pytest
import responses
from mobilizon_bots.config.config import settings
@pytest.fixture
def mobilizon_url():
return 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_mobilizon_failure_answer(mobilizon_url):
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST, mobilizon_url, status=500,
)
yield

111
tests/mobilizon/test_events.py Normal file
View File

@ -0,0 +1,111 @@
import pytest
import arrow
from mobilizon_bots.event.event import PublicationStatus, MobilizonEvent
from mobilizon_bots.mobilizon.events import (
get_mobilizon_future_events,
MobilizonRequestFailed,
get_unpublished_events,
)
simple_event_element = {
"beginsOn": "2021-05-23T12:15:00Z",
"description": None,
"endsOn": "2021-05-23T15:15:00Z",
"onlineAddress": None,
"options": {"showEndTime": True, "showStartTime": True},
"physicalAddress": None,
"picture": None,
"title": "test event",
"url": "https://some_mobilizon/events/1e2e5943-4a5c-497a-b65d-90457b715d7b",
"uuid": "1e2e5943-4a5c-497a-b65d-90457b715d7b",
}
simple_event_response = {
"data": {"group": {"organizedEvents": {"elements": [simple_event_element]}}}
}
full_event_element = {
"beginsOn": "2021-05-25T15:15:00Z",
"description": "<p>a description</p>",
"endsOn": "2021-05-25T16:15:00Z",
"onlineAddress": "http://some_location",
"options": {"showEndTime": True, "showStartTime": True},
"physicalAddress": None,
"picture": None,
"title": "full event",
"url": "https://some_mobilizon/events/56e7ca43-1b6b-4c50-8362-0439393197e6",
"uuid": "56e7ca43-1b6b-4c50-8362-0439393197e6",
}
full_event_response = {
"data": {"group": {"organizedEvents": {"elements": [full_event_element]}}}
}
two_events_response = {
"data": {
"group": {
"organizedEvents": {"elements": [simple_event_element, full_event_element]}
}
}
}
simple_event = MobilizonEvent(
name="test event",
description=None,
begin_datetime=arrow.get("2021-05-23T12:15:00Z"),
end_datetime=arrow.get("2021-05-23T15:15:00Z"),
mobilizon_link="https://some_mobilizon/events/1e2e5943-4a5c-497a-b65d-90457b715d7b",
mobilizon_id="1e2e5943-4a5c-497a-b65d-90457b715d7b",
thumbnail_link=None,
location=None,
publication_time=None,
publication_status=PublicationStatus.WAITING,
)
full_event = MobilizonEvent(
name="full event",
description="<p>a description</p>",
begin_datetime=arrow.get("2021-05-25T15:15:00+00:00]"),
end_datetime=arrow.get("2021-05-25T16:15:00+00:00"),
mobilizon_link="https://some_mobilizon/events/56e7ca43-1b6b-4c50-8362-0439393197e6",
mobilizon_id="56e7ca43-1b6b-4c50-8362-0439393197e6",
thumbnail_link=None,
location="http://some_location",
publication_time=None,
publication_status=PublicationStatus.WAITING,
)
@pytest.mark.parametrize(
"mobilizon_answer, expected_result",
[
[{"data": {"group": {"organizedEvents": {"elements": []}}}}, []],
[simple_event_response, [simple_event]],
[full_event_response, [full_event]],
[two_events_response, [simple_event, full_event]],
],
)
def test_event_response(mock_mobilizon_success_answer, expected_result):
"""
Testing the request and parsing logic
"""
assert get_mobilizon_future_events() == expected_result
def test_failure(mock_mobilizon_failure_answer):
with pytest.raises(MobilizonRequestFailed):
get_mobilizon_future_events()
@pytest.mark.parametrize(
"mobilizon_answer, published_events,expected_result",
[
[{"data": {"group": {"organizedEvents": {"elements": []}}}}, [], []],
[simple_event_response, [], [simple_event]],
[two_events_response, [], [simple_event, full_event]],
[two_events_response, [simple_event], [full_event]],
],
)
def test_get_unpublished_events(
mock_mobilizon_success_answer, published_events, expected_result
):
assert get_unpublished_events(published_events) == expected_result

0
tests/resources/__init__.py Normal file
View File

40
tests/resources/test_settings.toml Normal file
View File

@ -0,0 +1,40 @@
[default]
debug = true
default = true
local_state_dir = "/var/mobots"
log_dir = "/var/log/mobots"
db_name = "events.db"
db_path = "@format {this.local_state_dir}/{this.db_name}"
[default.source.mobilizon]
url=""
group="my_group"
[default.selection]
strategy = "next_event"
[default.selection.strategy_options]
break_between_events_in_minutes =5
[default.publisher.telegram]
active=true
chat_id="xxx"
[default.publisher.facebook]
active=false
[default.publisher.zulip]
active=false
[default.publisher.twitter]
active=false
[default.publisher.mastodon]
active=false
[default.notifier.telegram]
active=true
chat_id="xxx"
[default.notifier.zulip]
active=false
[default.notifier.twitter]
active=false
[default.notifier.mastodon]
active=false

5
tests/test_mobilizon_bots.py
View File

@ -1,5 +0,0 @@
from mobilizon_bots import __version__
def test_version():
assert __version__ == "0.1.0"