537 lines
18 KiB
Python
537 lines
18 KiB
Python
from unittest.mock import Mock, patch
|
|
from urllib.parse import urlencode
|
|
from uuid import uuid4
|
|
|
|
from django.core.cache import cache
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
from django.utils.translation import gettext as _
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
from newsreader.accounts.tests.factories import UserFactory
|
|
from newsreader.news.collection.exceptions import (
|
|
StreamException,
|
|
StreamTooManyException,
|
|
)
|
|
from newsreader.news.collection.twitter import TWITTER_AUTH_URL
|
|
|
|
|
|
class IntegrationsViewTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory(email="test@test.nl", password="test")
|
|
self.client.force_login(self.user)
|
|
|
|
self.url = reverse("accounts:settings:integrations")
|
|
|
|
|
|
class RedditIntegrationsTestCase(IntegrationsViewTestCase):
|
|
def test_reddit_authorization(self):
|
|
self.user.reddit_refresh_token = None
|
|
self.user.save()
|
|
|
|
response = self.client.get(self.url)
|
|
|
|
soup = BeautifulSoup(response.content, features="lxml")
|
|
button = soup.find("a", class_="link button button--reddit")
|
|
|
|
self.assertEquals(button.text.strip(), "Authorize account")
|
|
|
|
def test_reddit_refresh_token(self):
|
|
self.user.reddit_refresh_token = "jadajadajada"
|
|
self.user.reddit_access_token = None
|
|
self.user.save()
|
|
|
|
response = self.client.get(self.url)
|
|
|
|
soup = BeautifulSoup(response.content, features="lxml")
|
|
button = soup.find("a", class_="link button button--reddit")
|
|
|
|
self.assertEquals(button.text.strip(), "Refresh token")
|
|
|
|
def test_reddit_revoke(self):
|
|
self.user.reddit_refresh_token = "jadajadajada"
|
|
self.user.reddit_access_token = None
|
|
self.user.save()
|
|
|
|
response = self.client.get(self.url)
|
|
|
|
soup = BeautifulSoup(response.content, features="lxml")
|
|
buttons = soup.find_all("a", class_="link button button--reddit")
|
|
|
|
self.assertIn(
|
|
"Deauthorize account", [button.text.strip() for button in buttons]
|
|
)
|
|
|
|
|
|
class RedditTemplateViewTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory(email="test@test.nl", password="test")
|
|
self.client.force_login(self.user)
|
|
|
|
self.base_url = reverse("accounts:settings:reddit-template")
|
|
self.state = str(uuid4())
|
|
|
|
self.patch = patch("newsreader.news.collection.reddit.post")
|
|
self.mocked_post = self.patch.start()
|
|
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
|
|
def test_simple(self):
|
|
response = self.client.get(self.base_url)
|
|
|
|
self.assertEquals(response.status_code, 200)
|
|
self.assertContains(response, "Return to integrations page")
|
|
|
|
def test_successful_authorization(self):
|
|
self.mocked_post.return_value.json.return_value = {
|
|
"access_token": "1001010412",
|
|
"refresh_token": "134510143",
|
|
}
|
|
|
|
cache.set(f"{self.user.email}-reddit-auth", self.state)
|
|
|
|
params = {"state": self.state, "code": "Valid code"}
|
|
url = f"{self.base_url}?{urlencode(params)}"
|
|
|
|
response = self.client.get(url)
|
|
|
|
self.mocked_post.assert_called_once()
|
|
|
|
self.assertEquals(response.status_code, 200)
|
|
self.assertContains(response, "Your reddit account was successfully linked.")
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.reddit_access_token, "1001010412")
|
|
self.assertEquals(self.user.reddit_refresh_token, "134510143")
|
|
|
|
self.assertEquals(cache.get(f"{self.user.email}-reddit-auth"), None)
|
|
|
|
def test_error(self):
|
|
params = {"error": "Denied authorization"}
|
|
|
|
url = f"{self.base_url}?{urlencode(params)}"
|
|
|
|
response = self.client.get(url)
|
|
|
|
self.assertEquals(response.status_code, 200)
|
|
self.assertContains(response, "Denied authorization")
|
|
|
|
def test_invalid_state(self):
|
|
cache.set(f"{self.user.email}-reddit-auth", str(uuid4()))
|
|
|
|
params = {"code": "Valid code", "state": "Invalid state"}
|
|
|
|
url = f"{self.base_url}?{urlencode(params)}"
|
|
|
|
response = self.client.get(url)
|
|
|
|
self.assertEquals(response.status_code, 200)
|
|
self.assertContains(
|
|
response, "The saved state for Reddit authorization did not match"
|
|
)
|
|
|
|
def test_stream_error(self):
|
|
self.mocked_post.side_effect = StreamTooManyException
|
|
|
|
cache.set(f"{self.user.email}-reddit-auth", self.state)
|
|
|
|
params = {"state": self.state, "code": "Valid code"}
|
|
url = f"{self.base_url}?{urlencode(params)}"
|
|
|
|
response = self.client.get(url)
|
|
|
|
self.mocked_post.assert_called_once()
|
|
|
|
self.assertEquals(response.status_code, 200)
|
|
self.assertContains(response, "Too many requests")
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.reddit_access_token, None)
|
|
self.assertEquals(self.user.reddit_refresh_token, None)
|
|
|
|
self.assertEquals(cache.get(f"{self.user.email}-reddit-auth"), self.state)
|
|
|
|
def test_unexpected_json(self):
|
|
self.mocked_post.return_value.json.return_value = {"message": "Happy eastern"}
|
|
|
|
cache.set(f"{self.user.email}-reddit-auth", self.state)
|
|
|
|
params = {"state": self.state, "code": "Valid code"}
|
|
url = f"{self.base_url}?{urlencode(params)}"
|
|
|
|
response = self.client.get(url)
|
|
|
|
self.mocked_post.assert_called_once()
|
|
|
|
self.assertEquals(response.status_code, 200)
|
|
self.assertContains(response, "Access and refresh token not found in response")
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.reddit_access_token, None)
|
|
self.assertEquals(self.user.reddit_refresh_token, None)
|
|
|
|
self.assertEquals(cache.get(f"{self.user.email}-reddit-auth"), self.state)
|
|
|
|
|
|
class RedditTokenRedirectViewTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory(email="test@test.nl", password="test")
|
|
self.client.force_login(self.user)
|
|
|
|
self.patch = patch("newsreader.accounts.views.integrations.RedditTokenTask")
|
|
self.mocked_task = self.patch.start()
|
|
|
|
def tearDown(self):
|
|
cache.clear()
|
|
|
|
def test_simple(self):
|
|
response = self.client.get(reverse("accounts:settings:reddit-refresh"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.mocked_task.delay.assert_called_once_with(self.user.pk)
|
|
|
|
self.assertEquals(1, cache.get(f"{self.user.email}-reddit-refresh"))
|
|
|
|
def test_not_active(self):
|
|
cache.set(f"{self.user.email}-reddit-refresh", 1)
|
|
|
|
response = self.client.get(reverse("accounts:settings:reddit-refresh"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.mocked_task.delay.assert_not_called()
|
|
|
|
|
|
class RedditRevokeRedirectViewTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory(email="test@test.nl", password="test")
|
|
self.client.force_login(self.user)
|
|
|
|
self.patch = patch("newsreader.accounts.views.integrations.revoke_reddit_token")
|
|
self.mocked_revoke = self.patch.start()
|
|
|
|
def test_simple(self):
|
|
self.user.reddit_access_token = "jadajadajada"
|
|
self.user.reddit_refresh_token = "jadajadajada"
|
|
self.user.save()
|
|
|
|
self.mocked_revoke.return_value = True
|
|
|
|
response = self.client.get(reverse("accounts:settings:reddit-revoke"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.mocked_revoke.assert_called_once_with(self.user)
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.reddit_access_token, None)
|
|
self.assertEquals(self.user.reddit_refresh_token, None)
|
|
|
|
def test_no_refresh_token(self):
|
|
self.user.reddit_refresh_token = None
|
|
self.user.save()
|
|
|
|
response = self.client.get(reverse("accounts:settings:reddit-revoke"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.mocked_revoke.assert_not_called()
|
|
|
|
def test_unsuccessful_response(self):
|
|
self.user.reddit_access_token = "jadajadajada"
|
|
self.user.reddit_refresh_token = "jadajadajada"
|
|
self.user.save()
|
|
|
|
self.mocked_revoke.return_value = False
|
|
|
|
response = self.client.get(reverse("accounts:settings:reddit-revoke"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.reddit_access_token, "jadajadajada")
|
|
self.assertEquals(self.user.reddit_refresh_token, "jadajadajada")
|
|
|
|
def test_stream_exception(self):
|
|
self.user.reddit_access_token = "jadajadajada"
|
|
self.user.reddit_refresh_token = "jadajadajada"
|
|
self.user.save()
|
|
|
|
self.mocked_revoke.side_effect = StreamException
|
|
|
|
response = self.client.get(reverse("accounts:settings:reddit-revoke"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.reddit_access_token, "jadajadajada")
|
|
self.assertEquals(self.user.reddit_refresh_token, "jadajadajada")
|
|
|
|
|
|
class TwitterRevokeRedirectView(TestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory(email="test@test.nl", password="test")
|
|
self.client.force_login(self.user)
|
|
|
|
self.patch = patch("newsreader.accounts.views.integrations.post")
|
|
self.mocked_post = self.patch.start()
|
|
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
|
|
def test_simple(self):
|
|
self.user.twitter_oauth_token = "jadajadajada"
|
|
self.user.twitter_oauth_token_secret = "jadajadajada"
|
|
self.user.save()
|
|
|
|
response = self.client.get(reverse("accounts:settings:twitter-revoke"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertIsNone(self.user.twitter_oauth_token)
|
|
self.assertIsNone(self.user.twitter_oauth_token_secret)
|
|
|
|
def test_no_authorized_account(self):
|
|
self.user.twitter_oauth_token = None
|
|
self.user.twitter_oauth_token_secret = None
|
|
self.user.save()
|
|
|
|
response = self.client.get(reverse("accounts:settings:twitter-revoke"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.mocked_post.assert_not_called()
|
|
|
|
def test_stream_exception(self):
|
|
self.user.twitter_oauth_token = "jadajadajada"
|
|
self.user.twitter_oauth_token_secret = "jadajadajada"
|
|
self.user.save()
|
|
|
|
self.mocked_post.side_effect = StreamException
|
|
|
|
response = self.client.get(reverse("accounts:settings:twitter-revoke"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.twitter_oauth_token, "jadajadajada")
|
|
self.assertEquals(self.user.twitter_oauth_token_secret, "jadajadajada")
|
|
|
|
|
|
class TwitterAuthRedirectViewTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory(email="test@test.nl", password="test")
|
|
self.client.force_login(self.user)
|
|
|
|
self.patch = patch("newsreader.accounts.views.integrations.post")
|
|
self.mocked_post = self.patch.start()
|
|
|
|
def tearDown(self):
|
|
cache.clear()
|
|
|
|
def test_simple(self):
|
|
self.mocked_post.return_value = Mock(
|
|
text="oauth_token=foo&oauth_token_secret=bar"
|
|
)
|
|
|
|
response = self.client.get(reverse("accounts:settings:twitter-auth"))
|
|
|
|
self.assertRedirects(
|
|
response,
|
|
f"{TWITTER_AUTH_URL}/?oauth_token=foo",
|
|
fetch_redirect_response=False,
|
|
)
|
|
|
|
cached_token = cache.get(f"twitter-{self.user.email}-token")
|
|
cached_secret = cache.get(f"twitter-{self.user.email}-secret")
|
|
|
|
self.assertEquals(cached_token, "foo")
|
|
self.assertEquals(cached_secret, "bar")
|
|
|
|
def test_stream_exception(self):
|
|
self.mocked_post.side_effect = StreamException
|
|
|
|
response = self.client.get(reverse("accounts:settings:twitter-auth"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
cached_token = cache.get(f"twitter-{self.user.email}-token")
|
|
cached_secret = cache.get(f"twitter-{self.user.email}-secret")
|
|
|
|
self.assertIsNone(cached_token)
|
|
self.assertIsNone(cached_secret)
|
|
|
|
def test_unexpected_contents(self):
|
|
self.mocked_post.return_value = Mock(text="foo=bar&oauth_token_secret=bar")
|
|
|
|
response = self.client.get(reverse("accounts:settings:twitter-auth"))
|
|
|
|
self.assertRedirects(response, reverse("accounts:settings:integrations"))
|
|
|
|
cached_token = cache.get(f"twitter-{self.user.email}-token")
|
|
cached_secret = cache.get(f"twitter-{self.user.email}-secret")
|
|
|
|
self.assertIsNone(cached_token)
|
|
self.assertIsNone(cached_secret)
|
|
|
|
|
|
class TwitterTemplateViewTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory(email="test@test.nl", password="test")
|
|
self.client.force_login(self.user)
|
|
|
|
self.patch = patch("newsreader.accounts.views.integrations.post")
|
|
self.mocked_post = self.patch.start()
|
|
|
|
def tearDown(self):
|
|
cache.clear()
|
|
|
|
def test_simple(self):
|
|
cache.set_many(
|
|
{
|
|
f"twitter-{self.user.email}-token": "foo",
|
|
f"twitter-{self.user.email}-secret": "bar",
|
|
}
|
|
)
|
|
|
|
params = {"denied": "", "oauth_token": "foo", "oauth_verifier": "barfoo"}
|
|
|
|
self.mocked_post.return_value = Mock(
|
|
text="oauth_token=realtoken&oauth_token_secret=realsecret"
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"{reverse('accounts:settings:twitter-template')}?{urlencode(params)}"
|
|
)
|
|
|
|
self.assertContains(response, _("Twitter account is linked"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertEquals(self.user.twitter_oauth_token, "realtoken")
|
|
self.assertEquals(self.user.twitter_oauth_token_secret, "realsecret")
|
|
|
|
self.assertIsNone(cache.get(f"twitter-{self.user.email}-token"))
|
|
self.assertIsNone(cache.get(f"twitter-{self.user.email}-secret"))
|
|
|
|
def test_denied(self):
|
|
params = {"denied": "true", "oauth_token": "foo", "oauth_verifier": "barfoo"}
|
|
|
|
response = self.client.get(
|
|
f"{reverse('accounts:settings:twitter-template')}?{urlencode(params)}"
|
|
)
|
|
|
|
self.assertContains(response, _("Twitter authorization failed"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertIsNone(self.user.twitter_oauth_token)
|
|
self.assertIsNone(self.user.twitter_oauth_token_secret)
|
|
|
|
self.mocked_post.assert_not_called()
|
|
|
|
def test_mismatched_token(self):
|
|
cache.set_many(
|
|
{
|
|
f"twitter-{self.user.email}-token": "foo",
|
|
f"twitter-{self.user.email}-secret": "bar",
|
|
}
|
|
)
|
|
|
|
params = {"denied": "", "oauth_token": "boo", "oauth_verifier": "barfoo"}
|
|
|
|
response = self.client.get(
|
|
f"{reverse('accounts:settings:twitter-template')}?{urlencode(params)}"
|
|
)
|
|
|
|
self.assertContains(response, _("OAuth tokens failed to match"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertIsNone(self.user.twitter_oauth_token)
|
|
self.assertIsNone(self.user.twitter_oauth_token_secret)
|
|
|
|
self.mocked_post.assert_not_called()
|
|
|
|
def test_missing_secret(self):
|
|
cache.set_many({f"twitter-{self.user.email}-token": "foo"})
|
|
|
|
params = {"denied": "", "oauth_token": "foo", "oauth_verifier": "barfoo"}
|
|
|
|
response = self.client.get(
|
|
f"{reverse('accounts:settings:twitter-template')}?{urlencode(params)}"
|
|
)
|
|
|
|
self.assertContains(response, _("No matching tokens found for this user"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertIsNone(self.user.twitter_oauth_token_secret)
|
|
|
|
self.mocked_post.assert_not_called()
|
|
|
|
def test_stream_exception(self):
|
|
cache.set_many(
|
|
{
|
|
f"twitter-{self.user.email}-token": "foo",
|
|
f"twitter-{self.user.email}-secret": "bar",
|
|
}
|
|
)
|
|
|
|
params = {"denied": "", "oauth_token": "foo", "oauth_verifier": "barfoo"}
|
|
|
|
self.mocked_post.side_effect = StreamException
|
|
|
|
response = self.client.get(
|
|
f"{reverse('accounts:settings:twitter-template')}?{urlencode(params)}"
|
|
)
|
|
|
|
self.assertContains(response, _("Failed requesting access token"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertIsNone(self.user.twitter_oauth_token)
|
|
self.assertIsNone(self.user.twitter_oauth_token_secret)
|
|
|
|
self.assertIsNotNone(cache.get(f"twitter-{self.user.email}-token"))
|
|
self.assertIsNotNone(cache.get(f"twitter-{self.user.email}-secret"))
|
|
|
|
def test_unexpected_contents(self):
|
|
cache.set_many(
|
|
{
|
|
f"twitter-{self.user.email}-token": "foo",
|
|
f"twitter-{self.user.email}-secret": "bar",
|
|
}
|
|
)
|
|
|
|
params = {"denied": "", "oauth_token": "foo", "oauth_verifier": "barfoo"}
|
|
|
|
self.mocked_post.return_value = Mock(
|
|
text="foobar=boo&oauth_token_secret=realsecret"
|
|
)
|
|
|
|
response = self.client.get(
|
|
f"{reverse('accounts:settings:twitter-template')}?{urlencode(params)}"
|
|
)
|
|
|
|
self.assertContains(response, _("No credentials found in Twitter response"))
|
|
|
|
self.user.refresh_from_db()
|
|
|
|
self.assertIsNone(self.user.twitter_oauth_token)
|
|
self.assertIsNone(self.user.twitter_oauth_token_secret)
|
|
|
|
self.assertIsNotNone(cache.get(f"twitter-{self.user.email}-token"))
|
|
self.assertIsNotNone(cache.get(f"twitter-{self.user.email}-secret"))
|