platform tests (#150)
* added twitter error handling * added facebook tests * added header format test * added multiple newlines check
This commit is contained in:
parent
9c77afa456
commit
420f823dd4
|
@ -13,6 +13,8 @@ from mobilizon_reshare.publishers.abstract import (
|
||||||
from mobilizon_reshare.publishers.exceptions import (
|
from mobilizon_reshare.publishers.exceptions import (
|
||||||
InvalidCredentials,
|
InvalidCredentials,
|
||||||
InvalidEvent,
|
InvalidEvent,
|
||||||
|
InvalidMessage,
|
||||||
|
PublisherError,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@ -37,7 +39,8 @@ class FacebookFormatter(AbstractEventFormatter):
|
||||||
self._log_error("No text was found", raise_error=InvalidEvent)
|
self._log_error("No text was found", raise_error=InvalidEvent)
|
||||||
|
|
||||||
def _validate_message(self, message) -> None:
|
def _validate_message(self, message) -> None:
|
||||||
pass
|
if len(message) >= 63200:
|
||||||
|
self._log_error("Message is too long", raise_error=InvalidMessage)
|
||||||
|
|
||||||
def _preprocess_event(self, event: MobilizonEvent):
|
def _preprocess_event(self, event: MobilizonEvent):
|
||||||
event.description = html_to_plaintext(event.description)
|
event.description = html_to_plaintext(event.description)
|
||||||
|
@ -52,18 +55,23 @@ class FacebookPlatform(AbstractPlatform):
|
||||||
|
|
||||||
name = "facebook"
|
name = "facebook"
|
||||||
|
|
||||||
def _get_api(self):
|
def _get_api(self) -> facebook.GraphAPI:
|
||||||
return facebook.GraphAPI(
|
return facebook.GraphAPI(
|
||||||
access_token=self.conf["page_access_token"], version="8.0"
|
access_token=self.conf["page_access_token"], version="8.0"
|
||||||
)
|
)
|
||||||
|
|
||||||
def _send(self, message: str, event: Optional[MobilizonEvent] = None):
|
def _send(self, message: str, event: Optional[MobilizonEvent] = None):
|
||||||
|
try:
|
||||||
self._get_api().put_object(
|
self._get_api().put_object(
|
||||||
parent_object="me",
|
parent_object="me",
|
||||||
connection_name="feed",
|
connection_name="feed",
|
||||||
message=message,
|
message=message,
|
||||||
link=event.mobilizon_link if event else None,
|
link=event.mobilizon_link if event else None,
|
||||||
)
|
)
|
||||||
|
except GraphAPIError:
|
||||||
|
self._log_error(
|
||||||
|
"Facebook send failed", raise_error=PublisherError,
|
||||||
|
)
|
||||||
|
|
||||||
def validate_credentials(self):
|
def validate_credentials(self):
|
||||||
|
|
||||||
|
|
|
@ -1,3 +1,4 @@
|
||||||
|
import re
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
import pkg_resources
|
import pkg_resources
|
||||||
|
@ -46,7 +47,11 @@ class TelegramFormatter(AbstractEventFormatter):
|
||||||
self._log_error("Message is too long", raise_error=InvalidMessage)
|
self._log_error("Message is too long", raise_error=InvalidMessage)
|
||||||
|
|
||||||
def _preprocess_message(self, message: str) -> str:
|
def _preprocess_message(self, message: str) -> str:
|
||||||
|
"""
|
||||||
|
This function converts HTML5 to Telegram's HTML dialect
|
||||||
|
:param message: a HTML5 string
|
||||||
|
:return: a HTML string compatible with Telegram
|
||||||
|
"""
|
||||||
html = BeautifulSoup(message, "html.parser")
|
html = BeautifulSoup(message, "html.parser")
|
||||||
# replacing paragraphs
|
# replacing paragraphs
|
||||||
for tag in html.findAll(["p", "br"]):
|
for tag in html.findAll(["p", "br"]):
|
||||||
|
@ -70,7 +75,8 @@ class TelegramFormatter(AbstractEventFormatter):
|
||||||
# cleaning html trailing whitespace
|
# cleaning html trailing whitespace
|
||||||
for tag in html.findAll("a"):
|
for tag in html.findAll("a"):
|
||||||
tag["href"] = tag["href"].replace(" ", "").strip().lstrip()
|
tag["href"] = tag["href"].replace(" ", "").strip().lstrip()
|
||||||
return str(html)
|
s = str(html)
|
||||||
|
return re.sub(r"\n{2,}", "\n\n", s).strip() # remove multiple newlines
|
||||||
|
|
||||||
|
|
||||||
class TelegramPlatform(AbstractPlatform):
|
class TelegramPlatform(AbstractPlatform):
|
||||||
|
|
|
@ -11,7 +11,6 @@ from mobilizon_reshare.publishers.abstract import (
|
||||||
)
|
)
|
||||||
from mobilizon_reshare.publishers.exceptions import (
|
from mobilizon_reshare.publishers.exceptions import (
|
||||||
InvalidCredentials,
|
InvalidCredentials,
|
||||||
InvalidEvent,
|
|
||||||
PublisherError,
|
PublisherError,
|
||||||
InvalidMessage,
|
InvalidMessage,
|
||||||
)
|
)
|
||||||
|
@ -33,9 +32,7 @@ class TwitterFormatter(AbstractEventFormatter):
|
||||||
)
|
)
|
||||||
|
|
||||||
def _validate_event(self, event: MobilizonEvent) -> None:
|
def _validate_event(self, event: MobilizonEvent) -> None:
|
||||||
text = event.description
|
pass # pragma: no cover
|
||||||
if not (text and text.strip()):
|
|
||||||
self._log_error("No text was found", raise_error=InvalidEvent)
|
|
||||||
|
|
||||||
def _validate_message(self, message) -> None:
|
def _validate_message(self, message) -> None:
|
||||||
# TODO this is not precise. It should count the characters according to Twitter's logic but
|
# TODO this is not precise. It should count the characters according to Twitter's logic but
|
||||||
|
@ -76,7 +73,7 @@ class TwitterPlatform(AbstractPlatform):
|
||||||
)
|
)
|
||||||
|
|
||||||
def _validate_response(self, res: Status) -> dict:
|
def _validate_response(self, res: Status) -> dict:
|
||||||
pass
|
pass # pragma: no cover
|
||||||
|
|
||||||
|
|
||||||
class TwitterPublisher(TwitterPlatform):
|
class TwitterPublisher(TwitterPlatform):
|
||||||
|
|
|
@ -25,7 +25,8 @@ end_date = begin_date.shift(hours=1)
|
||||||
def event() -> MobilizonEvent:
|
def event() -> MobilizonEvent:
|
||||||
return MobilizonEvent(
|
return MobilizonEvent(
|
||||||
name="test event",
|
name="test event",
|
||||||
description="<p>description of the event</p>",
|
description="<p><h1>description of the event</h1><h1>another header</h1></p>",
|
||||||
|
# "<ul><li>element</li></ul>",
|
||||||
begin_datetime=begin_date,
|
begin_datetime=begin_date,
|
||||||
end_datetime=end_date,
|
end_datetime=end_date,
|
||||||
mobilizon_link="http://some_link.com/123",
|
mobilizon_link="http://some_link.com/123",
|
||||||
|
@ -49,7 +50,7 @@ def event() -> MobilizonEvent:
|
||||||
📍 location
|
📍 location
|
||||||
|
|
||||||
|
|
||||||
description of the event
|
description of the event another header
|
||||||
|
|
||||||
Link: http://some_link.com/123
|
Link: http://some_link.com/123
|
||||||
""",
|
""",
|
||||||
|
@ -61,7 +62,9 @@ Link: http://some_link.com/123
|
||||||
🕒 01 January, {begin_date.format('HH:mm')} - 01 January, {end_date.format('HH:mm')}
|
🕒 01 January, {begin_date.format('HH:mm')} - 01 January, {end_date.format('HH:mm')}
|
||||||
📍 location
|
📍 location
|
||||||
|
|
||||||
description of the event
|
<b>description of the event</b>
|
||||||
|
|
||||||
|
<b>another header</b>
|
||||||
|
|
||||||
<a href="http://some_link.com/123">Link</a>""",
|
<a href="http://some_link.com/123">Link</a>""",
|
||||||
],
|
],
|
||||||
|
|
|
@ -0,0 +1,64 @@
|
||||||
|
from logging import DEBUG
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from facebook import GraphAPI, GraphAPIError
|
||||||
|
|
||||||
|
from mobilizon_reshare.publishers.exceptions import (
|
||||||
|
InvalidEvent,
|
||||||
|
InvalidMessage,
|
||||||
|
PublisherError,
|
||||||
|
)
|
||||||
|
from mobilizon_reshare.publishers.platforms.facebook import (
|
||||||
|
FacebookFormatter,
|
||||||
|
FacebookPublisher,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_message_length_success(event):
|
||||||
|
message = "a" * 500
|
||||||
|
event.description = message
|
||||||
|
assert FacebookFormatter().validate_event(event) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_message_length_failure(event):
|
||||||
|
message = "a" * 80000
|
||||||
|
event.description = message
|
||||||
|
|
||||||
|
with pytest.raises(InvalidMessage):
|
||||||
|
FacebookFormatter().validate_event(event)
|
||||||
|
|
||||||
|
|
||||||
|
def test_event_validation(event):
|
||||||
|
event.description = None
|
||||||
|
with pytest.raises(InvalidEvent):
|
||||||
|
FacebookFormatter().validate_event(event)
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_error(event):
|
||||||
|
|
||||||
|
with patch.object(
|
||||||
|
GraphAPI, "put_object", side_effect=GraphAPIError("some error")
|
||||||
|
) as mock:
|
||||||
|
with pytest.raises(PublisherError):
|
||||||
|
FacebookPublisher().send("abc", event)
|
||||||
|
mock.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
def test_validate_credentials_error(event):
|
||||||
|
|
||||||
|
with patch.object(
|
||||||
|
GraphAPI, "get_object", side_effect=GraphAPIError("some error")
|
||||||
|
) as mock:
|
||||||
|
with pytest.raises(PublisherError):
|
||||||
|
FacebookPublisher().validate_credentials()
|
||||||
|
mock.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
def test_validate_credentials(event, caplog):
|
||||||
|
|
||||||
|
with patch.object(GraphAPI, "get_object", return_value=None) as mock:
|
||||||
|
with caplog.at_level(DEBUG):
|
||||||
|
FacebookPublisher().validate_credentials()
|
||||||
|
mock.assert_called()
|
||||||
|
assert "Facebook credentials are valid" in caplog.text
|
|
@ -0,0 +1,44 @@
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from tweepy import API, TweepyException
|
||||||
|
|
||||||
|
from mobilizon_reshare.publishers.exceptions import (
|
||||||
|
InvalidMessage,
|
||||||
|
InvalidCredentials,
|
||||||
|
PublisherError,
|
||||||
|
)
|
||||||
|
from mobilizon_reshare.publishers.platforms.twitter import (
|
||||||
|
TwitterFormatter,
|
||||||
|
TwitterPublisher,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_message_length_success(event):
|
||||||
|
message = "a" * 300
|
||||||
|
event.description = message
|
||||||
|
assert TwitterFormatter().validate_event(event) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_message_length_failure(event):
|
||||||
|
message = "a" * 10000
|
||||||
|
event.name = message
|
||||||
|
|
||||||
|
with pytest.raises(InvalidMessage):
|
||||||
|
TwitterFormatter().validate_event(event)
|
||||||
|
|
||||||
|
|
||||||
|
def test_validate_credentials_error():
|
||||||
|
with patch.object(API, "verify_credentials", return_value=False) as mock:
|
||||||
|
with pytest.raises(InvalidCredentials):
|
||||||
|
TwitterPublisher().validate_credentials()
|
||||||
|
mock.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
def test_send_error(event):
|
||||||
|
with patch.object(
|
||||||
|
API, "update_status", side_effect=TweepyException("some error")
|
||||||
|
) as mock:
|
||||||
|
with pytest.raises(PublisherError):
|
||||||
|
TwitterPublisher().send("abc", event)
|
||||||
|
mock.assert_called()
|
Loading…
Reference in New Issue