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

command tests (#97) 

* filtering publications with inactive publishers

* filtering publications with inactive publishers

* WIP: Generate publications at runtime.

TODO:
- change `MobilizonEvent.compute_status`'s contract and break everything
- while we're at it we should remove `PublicationStatus.WAITING`
- test `storage.query.create_publications_for_publishers`

* cli: inspect_events: Unnest if-then-else.

* publishers: abstract: Remove `EventPublication.make`.

* fixed tests

* split query.py file

* added tests for get_unpublished_events

* added tests

* more tests

* added start test

* refactored start test

* added test start with db event

* added test recap

* added failed publication test

* added format test

Co-authored-by: Giacomo Leidi <goodoldpaul@autistici.org>
This commit is contained in:
Simone Robutti 2021-11-11 16:20:50 +01:00 committed by GitHub
parent 4dc1e4080a
commit 5335ed8cc3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 361 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
mobilizon_reshare/main/recap.py
View File

@ -1,3 +1,4 @@
import logging
from typing import Optional, List
from arrow import now
@ -16,6 +17,8 @@ from mobilizon_reshare.publishers.platforms.platform_mapping import (
)
from mobilizon_reshare.storage.query.read_query import events_with_status
logger = logging.getLogger(__name__)
async def select_events_to_recap() -> List[MobilizonEvent]:
return list(
@ -28,7 +31,9 @@ async def select_events_to_recap() -> List[MobilizonEvent]:
async def recap() -> Optional[BaseCoordinatorReport]:
# I want to recap only the events that have been succesfully published and that haven't happened yet
events_to_recap = await select_events_to_recap()
if events_to_recap:
logger.debug(f"Found {len(events_to_recap)} events to recap.")
recap_publications = [
RecapPublication(
get_publisher_class(publisher)(),
@ -43,3 +48,5 @@ async def recap() -> Optional[BaseCoordinatorReport]:
if report.status == EventPublicationStatus.FAILED:
PublicationFailureNotifiersCoordinator(report).notify_failure()
return reports
else:
logger.debug("Found no events")

2
mobilizon_reshare/main/start.py
View File

@ -53,6 +53,6 @@ async def start():
await save_publication_report(reports, models)
for report in reports.reports:
if not report.succesful:
PublicationFailureNotifiersCoordinator(report).notify_failure()
PublicationFailureNotifiersCoordinator(report,).notify_failure()
else:
logger.debug("No event to publish found")

17
poetry.lock generated
View File

@ -281,6 +281,17 @@ pytest = ">=5.4.0"
[package.extras]
testing = ["coverage", "hypothesis (>=5.7.1)"]
[[package]]
name = "pytest-lazy-fixture"
version = "0.6.3"
description = "It helps to use fixtures in pytest.mark.parametrize"
category = "dev"
optional = false
python-versions = "*"
[package.dependencies]
pytest = ">=3.2.5"
[[package]]
name = "python-dateutil"
version = "2.8.2"
@ -435,7 +446,7 @@ socks = ["PySocks (>=1.5.6,!=1.5.7,<2.0)"]
[metadata]
lock-version = "1.1"
python-versions = "^3.9"
content-hash = "4fe276575784de9ed9d3ee66eef2c75355dd66e7717bcc277c38755dd489b4ac"
content-hash = "763106b0d68a1b95c690e2ad828a4e847ad2532a3b13354c227f35b70f1c8ad7"
[metadata.files]
aiosqlite = [
@ -575,6 +586,10 @@ pytest-asyncio = [
{file = "pytest-asyncio-0.15.1.tar.gz", hash = "sha256:2564ceb9612bbd560d19ca4b41347b54e7835c2f792c504f698e05395ed63f6f"},
{file = "pytest_asyncio-0.15.1-py3-none-any.whl", hash = "sha256:3042bcdf1c5d978f6b74d96a151c4cfb9dcece65006198389ccd7e6c60eb1eea"},
]
pytest-lazy-fixture = [
{file = "pytest-lazy-fixture-0.6.3.tar.gz", hash = "sha256:0e7d0c7f74ba33e6e80905e9bfd81f9d15ef9a790de97993e34213deb5ad10ac"},
{file = "pytest_lazy_fixture-0.6.3-py3-none-any.whl", hash = "sha256:e0b379f38299ff27a653f03eaa69b08a6fd4484e46fd1c9907d984b9f9daeda6"},
]
python-dateutil = [
{file = "python-dateutil-2.8.2.tar.gz", hash = "sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86"},
{file = "python_dateutil-2.8.2-py2.py3-none-any.whl", hash = "sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9"},

1
pyproject.toml
View File

@ -27,6 +27,7 @@ responses = "^0.13"
pytest-asyncio = "^0.15"
asynctest = "^0.13"
pytest = "^6.2"
pytest-lazy-fixture = "^0.6.3"
[build-system]
requires = ["poetry-core>=1.0.0"]

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

99
tests/commands/conftest.py Normal file
View File

@ -0,0 +1,99 @@
import uuid
import pytest
from click.testing import CliRunner
import mobilizon_reshare.publishers
from mobilizon_reshare.models import event
from mobilizon_reshare.models.publisher import Publisher
import mobilizon_reshare.main.recap
from mobilizon_reshare.publishers import coordinator
def simple_event_element():
return {
"beginsOn": "2021-05-23T12:15:00Z",
"description": "Some description",
"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": str(uuid.uuid4()),
}
@pytest.fixture
def mobilizon_answer(elements):
return {"data": {"group": {"organizedEvents": {"elements": elements}}}}
@pytest.fixture
async def mock_publisher_config(monkeypatch, publisher_class, mock_formatter_class):
p = Publisher(name="test")
await p.save()
p2 = Publisher(name="test2")
await p2.save()
def _mock_active_pub():
return ["test", "test2"]
def _mock_pub_class(name):
return publisher_class
def _mock_format_class(name):
return mock_formatter_class
monkeypatch.setattr(event, "get_active_publishers", _mock_active_pub)
monkeypatch.setattr(
mobilizon_reshare.publishers.platforms.platform_mapping,
"get_publisher_class",
_mock_pub_class,
)
monkeypatch.setattr(
mobilizon_reshare.publishers.platforms.platform_mapping,
"get_formatter_class",
_mock_format_class,
)
monkeypatch.setattr(
mobilizon_reshare.main.recap, "get_active_publishers", _mock_active_pub
)
monkeypatch.setattr(
mobilizon_reshare.main.recap, "get_publisher_class", _mock_pub_class,
)
monkeypatch.setattr(
mobilizon_reshare.main.recap, "get_formatter_class", _mock_format_class,
)
return p
@pytest.fixture
async def mock_notifier_config(monkeypatch, publisher_class, mock_formatter_class):
def _mock_active_notifier():
return ["test", "test2"]
def _mock_notifier_class(name):
return publisher_class
def _mock_format_class(name):
return mock_formatter_class
monkeypatch.setattr(
coordinator, "get_notifier_class", _mock_notifier_class,
)
monkeypatch.setattr(
mobilizon_reshare.publishers.platforms.platform_mapping,
"get_formatter_class",
_mock_format_class,
)
monkeypatch.setattr(coordinator, "get_active_notifiers", _mock_active_notifier)
@pytest.fixture
def runner():
return CliRunner()

35
tests/commands/test_format.py Normal file
View File

@ -0,0 +1,35 @@
import uuid
import pytest
from mobilizon_reshare.cli.commands.format.format import format_event
from mobilizon_reshare.publishers.platforms.platform_mapping import (
get_formatter_class,
name_to_formatter_class,
)
@pytest.mark.parametrize("publisher_name", name_to_formatter_class.keys())
@pytest.mark.asyncio
async def test_format_event(runner, event, capsys, publisher_name):
event_model = event.to_model()
await event_model.save()
await format_event(
event_id=str(event_model.mobilizon_id), publisher_name=publisher_name
)
assert (
capsys.readouterr().out.strip()
== get_formatter_class(publisher_name)().get_message_from_event(event).strip()
)
@pytest.mark.asyncio
async def test_format_event_missing(runner, capsys):
event_id = uuid.uuid4()
await format_event(event_id=event_id, publisher_name="telegram")
assert (
capsys.readouterr().out.strip()
== f"Event with mobilizon_id {event_id} not found."
)

45
tests/commands/test_recap.py Normal file
View File

@ -0,0 +1,45 @@
from logging import DEBUG
from uuid import UUID
import arrow
import pytest
from mobilizon_reshare.main.recap import recap
from mobilizon_reshare.models.publication import PublicationStatus
from mobilizon_reshare.storage.query.model_creation import (
create_event_publication_models,
)
@pytest.mark.parametrize(
"publisher_class", [pytest.lazy_fixture("mock_publisher_invalid_class")]
)
@pytest.mark.asyncio
async def test_start_event_from_db(
caplog, mock_publisher_config, message_collector, event_generator,
):
for i in range(2):
event = event_generator(
mobilizon_id=UUID(int=i), begin_date=arrow.now().shift(days=2)
)
event_model = event.to_model()
await event_model.save()
publications = await create_event_publication_models(event_model)
for p in publications:
p.status = PublicationStatus.COMPLETED
await p.save()
with caplog.at_level(DEBUG):
# calling the recap command
report = await recap()
assert report.successful
assert "Found 2 events to recap" in caplog.text
recap_message = """Upcoming
test event
test event"""
assert message_collector == [recap_message] * 2 # two publishers * 1 recap

190
tests/commands/test_start.py
View File

@ -2,75 +2,32 @@ from logging import DEBUG
import pytest
import mobilizon_reshare.publishers.platforms.platform_mapping
from tests.commands.conftest import simple_event_element
from mobilizon_reshare.event.event import MobilizonEvent, EventPublicationStatus
from mobilizon_reshare.main.start import start
from mobilizon_reshare.models import event
from mobilizon_reshare.models.event import Event
from mobilizon_reshare.models.publication import PublicationStatus
from mobilizon_reshare.models.publisher import Publisher
simple_event_element = {
"beginsOn": "2021-05-23T12:15:00Z",
"description": "Some description",
"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]}}}
}
@pytest.mark.asyncio
@pytest.mark.parametrize(
"mobilizon_answer", [{"data": {"group": {"organizedEvents": {"elements": []}}}}],
"elements", [[]],
)
async def test_start_no_event(mock_mobilizon_success_answer, mobilizon_answer, caplog):
async def test_start_no_event(
mock_mobilizon_success_answer, mobilizon_answer, caplog, elements
):
with caplog.at_level(DEBUG):
assert await start() is None
assert "No event to publish found" in caplog.text
@pytest.fixture
async def mock_publisher_config(
monkeypatch, mock_publisher_class, mock_formatter_class
):
p = Publisher(name="test")
await p.save()
def _mock_active_pub():
return ["test"]
def _mock_pub_class(name):
return mock_publisher_class
def _mock_format_class(name):
return mock_formatter_class
monkeypatch.setattr(event, "get_active_publishers", _mock_active_pub)
monkeypatch.setattr(
mobilizon_reshare.publishers.platforms.platform_mapping,
"get_publisher_class",
_mock_pub_class,
)
monkeypatch.setattr(
mobilizon_reshare.publishers.platforms.platform_mapping,
"get_formatter_class",
_mock_format_class,
)
return p
@pytest.mark.parametrize(
"publisher_class", [pytest.lazy_fixture("mock_publisher_class")]
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"mobilizon_answer", [simple_event_response],
"elements",
[[simple_event_element()], [simple_event_element(), simple_event_element()]],
)
@pytest.mark.parametrize("publication_window", [(0, 24)])
async def test_start_new_event(
@ -81,12 +38,17 @@ async def test_start_new_event(
mock_publication_window,
message_collector,
):
with caplog.at_level(DEBUG):
# calling the start command
assert await start() is None
# since the mobilizon_answer contains at least one result, one event to publish must be found and published
# by the publisher coordinator
assert "Event to publish found" in caplog.text
assert message_collector == ["test event|Some description"]
assert message_collector == [
"test event|Some description",
"test event|Some description",
]
all_events = (
await Event.all()
@ -94,13 +56,125 @@ async def test_start_new_event(
.prefetch_related("publications__publisher")
)
assert len(all_events) == 1, all_events
# the start command should save all the events in the database
assert len(all_events) == len(
mobilizon_answer["data"]["group"]["organizedEvents"]["elements"]
), all_events
# it should create a publication for each publisher
publications = all_events[0].publications
assert len(publications) == 1, publications
assert len(publications) == 2, publications
assert publications[0].status == PublicationStatus.COMPLETED
# all the other events should have no publication
for e in all_events[1:]:
assert len(e.publications) == 0, e.publications
# all the publications for the first event should be saved as COMPLETED
for p in publications[1:]:
assert p.status == PublicationStatus.COMPLETED
# the derived status for the event should be COMPLETED
assert (
MobilizonEvent.from_model(all_events[0]).status
== EventPublicationStatus.COMPLETED
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"publisher_class", [pytest.lazy_fixture("mock_publisher_class")]
)
@pytest.mark.parametrize(
"elements", [[]],
)
@pytest.mark.parametrize("publication_window", [(0, 24)])
async def test_start_event_from_db(
mock_mobilizon_success_answer,
mobilizon_answer,
caplog,
mock_publisher_config,
mock_publication_window,
message_collector,
event_generator,
):
event = event_generator()
event_model = event.to_model()
await event_model.save()
with caplog.at_level(DEBUG):
# calling the start command
assert await start() is None
# since the db contains at least one event, this has to be picked and published
assert "Event to publish found" in caplog.text
assert message_collector == [
"test event|description of the event",
"test event|description of the event",
]
await event_model.fetch_related("publications", "publications__publisher")
# it should create a publication for each publisher
publications = event_model.publications
assert len(publications) == 2, publications
# all the publications for the first event should be saved as COMPLETED
for p in publications[1:]:
assert p.status == PublicationStatus.COMPLETED
# the derived status for the event should be COMPLETED
assert (
MobilizonEvent.from_model(event_model).status
== EventPublicationStatus.COMPLETED
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"publisher_class", [pytest.lazy_fixture("mock_publisher_invalid_class")]
)
@pytest.mark.parametrize(
"elements", [[]],
)
@pytest.mark.parametrize("publication_window", [(0, 24)])
async def test_start_publisher_failure(
mock_mobilizon_success_answer,
mobilizon_answer,
caplog,
mock_publisher_config,
mock_publication_window,
message_collector,
event_generator,
mock_notifier_config,
):
event = event_generator()
event_model = event.to_model()
await event_model.save()
with caplog.at_level(DEBUG):
# calling the start command
assert await start() is None
# since the db contains at least one event, this has to be picked and published
await event_model.fetch_related("publications", "publications__publisher")
# it should create a publication for each publisher
publications = event_model.publications
assert len(publications) == 2, publications
# all the publications for event should be saved as FAILED
for p in publications:
assert p.status == PublicationStatus.FAILED
assert p.reason == "credentials error"
assert "Event to publish found" in caplog.text
assert message_collector == [
f"Publication {p.id} failed with status: 1."
f"\nReason: credentials error\nPublisher: mock"
for p in publications
for _ in range(2)
] # 2 publications failed * 2 notifiers
# the derived status for the event should be FAILED
assert (
MobilizonEvent.from_model(event_model).status
== EventPublicationStatus.FAILED
)

25
tests/conftest.py
View File

@ -20,6 +20,7 @@ from mobilizon_reshare.publishers.abstract import (
AbstractPlatform,
AbstractEventFormatter,
)
from mobilizon_reshare.publishers.exceptions import PublisherError, InvalidResponse
def generate_publication_status(published):
@ -272,3 +273,27 @@ def mock_formatter_class():
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):
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()