mirror of
https://github.com/Tech-Workers-Coalition-Italia/mobilizon-reshare.git
synced 2025-02-06 04:13:27 +01:00
More tests (#67)
* decoupled notifiers from event * stub * publishers working * fixed format CLI * fixed unit tests * renamed abstractnotifier * added another excluded character * restored bundled secrets file * test telegram escape * tested telegram event validation * added telegram response validation * added pragma * added zulip response validation test
This commit is contained in:
parent
bc212e7801
commit
b61a2c5c3c
@ -81,7 +81,7 @@ class AbstractPlatform(ABC, LoggerMixin, ConfLoaderMixin):
|
||||
|
||||
@abstractmethod
|
||||
def _send(self, message: str):
|
||||
raise NotImplementedError
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
def send(self, message: str):
|
||||
"""
|
||||
@ -96,7 +96,7 @@ class AbstractPlatform(ABC, LoggerMixin, ConfLoaderMixin):
|
||||
|
||||
@abstractmethod
|
||||
def _validate_response(self, response):
|
||||
raise NotImplementedError
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
def are_credentials_valid(self) -> bool:
|
||||
try:
|
||||
@ -112,7 +112,7 @@ class AbstractPlatform(ABC, LoggerMixin, ConfLoaderMixin):
|
||||
Should raise ``PublisherError`` (or one of its subclasses) if
|
||||
credentials are not valid.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
|
||||
class AbstractEventFormatter(LoggerMixin, ConfLoaderMixin):
|
||||
@ -123,13 +123,13 @@ class AbstractEventFormatter(LoggerMixin, ConfLoaderMixin):
|
||||
Should raise ``PublisherError`` (or one of its subclasses) if event
|
||||
is not valid.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
def _preprocess_event(self, event):
|
||||
"""
|
||||
Allows publishers to preprocess events before feeding them to the template
|
||||
"""
|
||||
pass
|
||||
pass # pragma: no cover
|
||||
|
||||
def get_message_from_event(self, event) -> str:
|
||||
"""
|
||||
@ -159,7 +159,7 @@ class AbstractEventFormatter(LoggerMixin, ConfLoaderMixin):
|
||||
Should raise ``PublisherError`` (or one of its subclasses) if message
|
||||
is not valid.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
def is_event_valid(self, event) -> bool:
|
||||
try:
|
||||
|
@ -26,20 +26,19 @@ class TelegramFormatter(AbstractEventFormatter):
|
||||
|
||||
@staticmethod
|
||||
def escape_message(message: str) -> str:
|
||||
message = (
|
||||
message.replace("-", "\\-")
|
||||
.replace(".", "\\.")
|
||||
.replace("(", "\\(")
|
||||
.replace("!", "\\!")
|
||||
.replace(")", "\\)")
|
||||
.replace("#", "")
|
||||
return (
|
||||
message.replace("-", r"\-")
|
||||
.replace(".", r"\.")
|
||||
.replace("(", r"\(")
|
||||
.replace("!", r"\!")
|
||||
.replace(")", r"\)")
|
||||
.replace("#", r"")
|
||||
)
|
||||
return message
|
||||
|
||||
def validate_event(self, event: MobilizonEvent) -> None:
|
||||
text = event.description
|
||||
if not (text and text.strip()):
|
||||
self._log_error("No text was found", raise_error=InvalidEvent)
|
||||
description = event.description
|
||||
if not (description and description.strip()):
|
||||
self._log_error("No description was found", raise_error=InvalidEvent)
|
||||
|
||||
def get_message_from_event(self, event: MobilizonEvent) -> str:
|
||||
return super(TelegramFormatter, self).get_message_from_event(event)
|
||||
|
@ -1,4 +1,11 @@
|
||||
from mobilizon_reshare.publishers.platforms.telegram import TelegramFormatter
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from mobilizon_reshare.publishers.exceptions import InvalidEvent, InvalidResponse
|
||||
from mobilizon_reshare.publishers.platforms.telegram import (
|
||||
TelegramFormatter,
|
||||
TelegramPublisher,
|
||||
)
|
||||
|
||||
|
||||
def test_message_length_success(event):
|
||||
@ -11,3 +18,64 @@ def test_message_length_failure(event):
|
||||
message = "a" * 10000
|
||||
event.description = message
|
||||
assert not TelegramFormatter().is_message_valid(event)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"message, result",
|
||||
[
|
||||
["", ""],
|
||||
["a#b", "ab"],
|
||||
["-", "\\-"],
|
||||
["(", "\\("],
|
||||
["!", "\\!"],
|
||||
[")", "\\)"],
|
||||
[")!", "\\)\\!"],
|
||||
],
|
||||
)
|
||||
def test_escape_message(message, result):
|
||||
assert TelegramFormatter().escape_message(message) == result
|
||||
|
||||
|
||||
def test_event_validation(event):
|
||||
event.description = None
|
||||
with pytest.raises(InvalidEvent):
|
||||
TelegramFormatter().validate_event(event)
|
||||
|
||||
|
||||
def test_validate_response():
|
||||
response = requests.Response()
|
||||
response.status_code = 200
|
||||
response._content = b"""{"ok":true}"""
|
||||
TelegramPublisher()._validate_response(response)
|
||||
|
||||
|
||||
def test_validate_response_invalid_json():
|
||||
response = requests.Response()
|
||||
response.status_code = 200
|
||||
response._content = b"""{"osxsa"""
|
||||
with pytest.raises(InvalidResponse) as e:
|
||||
TelegramPublisher()._validate_response(response)
|
||||
|
||||
e.match("json")
|
||||
|
||||
|
||||
def test_validate_response_invalid_request():
|
||||
response = requests.Response()
|
||||
response.status_code = 400
|
||||
response._content = b"""{"error":true}"""
|
||||
with pytest.raises(InvalidResponse) as e:
|
||||
|
||||
TelegramPublisher()._validate_response(response)
|
||||
|
||||
e.match("Server returned invalid data")
|
||||
|
||||
|
||||
def test_validate_response_invalid_response():
|
||||
response = requests.Response()
|
||||
response.status_code = 200
|
||||
response._content = b"""{"error":true}"""
|
||||
with pytest.raises(InvalidResponse) as e:
|
||||
|
||||
TelegramPublisher()._validate_response(response)
|
||||
|
||||
e.match("Invalid request")
|
||||
|
@ -1,6 +1,7 @@
|
||||
from functools import partial
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
import responses
|
||||
|
||||
from mobilizon_reshare.config.config import get_settings
|
||||
@ -8,7 +9,12 @@ from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers import get_active_publishers
|
||||
from mobilizon_reshare.publishers.abstract import EventPublication
|
||||
from mobilizon_reshare.publishers.coordinator import PublisherCoordinator
|
||||
from mobilizon_reshare.publishers.platforms.zulip import ZulipFormatter
|
||||
from mobilizon_reshare.publishers.exceptions import (
|
||||
InvalidEvent,
|
||||
InvalidResponse,
|
||||
ZulipError,
|
||||
)
|
||||
from mobilizon_reshare.publishers.platforms.zulip import ZulipFormatter, ZulipPublisher
|
||||
from mobilizon_reshare.storage.query import (
|
||||
get_publishers,
|
||||
update_publishers,
|
||||
@ -155,6 +161,12 @@ async def test_zulip_publisher_failure_client_error(
|
||||
assert list(report.reports.values())[0].reason == "400 Error - Invalid request"
|
||||
|
||||
|
||||
def test_event_validation(event):
|
||||
event.description = None
|
||||
with pytest.raises(InvalidEvent):
|
||||
ZulipFormatter().validate_event(event)
|
||||
|
||||
|
||||
def test_message_length_success(event):
|
||||
message = "a" * 500
|
||||
event.description = message
|
||||
@ -165,3 +177,31 @@ def test_message_length_failure(event):
|
||||
message = "a" * 10000
|
||||
event.description = message
|
||||
assert not ZulipFormatter().is_message_valid(event)
|
||||
|
||||
|
||||
def test_validate_response():
|
||||
response = requests.Response()
|
||||
response.status_code = 200
|
||||
response._content = b"""{"result":"ok"}"""
|
||||
ZulipPublisher()._validate_response(response)
|
||||
|
||||
|
||||
def test_validate_response_invalid_json():
|
||||
response = requests.Response()
|
||||
response.status_code = 200
|
||||
response._content = b"""{"osxsa"""
|
||||
with pytest.raises(InvalidResponse) as e:
|
||||
ZulipPublisher()._validate_response(response)
|
||||
|
||||
e.match("json")
|
||||
|
||||
|
||||
def test_validate_response_invalid_request():
|
||||
response = requests.Response()
|
||||
response.status_code = 400
|
||||
response._content = b"""{"result":"error", "msg":"wrong request"}"""
|
||||
with pytest.raises(ZulipError) as e:
|
||||
|
||||
ZulipPublisher()._validate_response(response)
|
||||
|
||||
e.match("wrong request")
|
||||
|
Loading…
x
Reference in New Issue
Block a user