Compare commits
15 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5c4d0b3aed | |||
| f20fc33349 | |||
| 3d2a8def57 | |||
| 80c1a17e4b | |||
| b5eda64a05 | |||
| 87cf273cd6 | |||
| c5466fd182 | |||
| fe17d3977c | |||
| f93e1cb341 | |||
| 2cf6c675e0 | |||
| 991cc564dd | |||
| 7ebc08356a | |||
| a58ca74362 | |||
| 8a73f911e2 | |||
| 87891a6331 |
+38
-8
@@ -19,9 +19,9 @@ Reviewing and merging pull requests is work, so whatever you can do to make this
|
||||
easier for the package maintainer not only speed up the process of getting your
|
||||
changes merged but also ensure they are. These few guidelines help significantly.
|
||||
If they are confusing or you need help understanding how to accomplish them,
|
||||
please ask for help in an issue.
|
||||
please ask for help in an issue.
|
||||
|
||||
- Please do make sure your chnageset represents a *discrete update*. If you would like to fix formatting, by all means, but don't mix that up with a bug fix. Those are separate PRs.
|
||||
- Please do make sure your changeset represents a *discrete update*. If you would like to fix formatting, by all means, but don't mix that up with a bug fix. Those are separate PRs.
|
||||
- Please do make sure that both your pull request description and your commits are meaningful and descriptive. Rebase first, if need be.
|
||||
- Please do make sure your changeset does not include more commits than necessary. Rebase first, if need be.
|
||||
- Please do make sure the changeset is not very big. If you have a large change propose it in an issue first.
|
||||
@@ -30,10 +30,27 @@ please ask for help in an issue.
|
||||
Testing
|
||||
=======
|
||||
|
||||
The best way to run the tests is with `tox <http://tox.readthedocs.org/en/latest/>`_::
|
||||
Running tests
|
||||
-------------
|
||||
|
||||
The simplest way to quickly and repeatedly run tests while developing a feature or fix
|
||||
is to use `pytest` in your current Python environment.
|
||||
|
||||
After installing the test dependencies::
|
||||
|
||||
pip install -r requirements.txt
|
||||
pip install -e .
|
||||
|
||||
Your can run the tests with `pytest`::
|
||||
|
||||
pytest --cov=src/pydiscourse
|
||||
|
||||
This will ensure you get coverage reporting.
|
||||
|
||||
The most comprehensive way to run the tests is with `tox <http://tox.readthedocs.org/en/latest/>`_::
|
||||
|
||||
pip install tox
|
||||
detox
|
||||
tox
|
||||
|
||||
Or it's slightly faster cousin `detox
|
||||
<https://pypi.python.org/pypi/detox>`_ which will parallelize test runs::
|
||||
@@ -41,16 +58,29 @@ Or it's slightly faster cousin `detox
|
||||
pip install detox
|
||||
detox
|
||||
|
||||
Alternatively, you can run the self test with the following commands::
|
||||
Writing tests
|
||||
-------------
|
||||
|
||||
pip install -r requirements.txt
|
||||
pip install -e .
|
||||
python setup.py test
|
||||
The primary modules of the library have coverage requirements, so you should
|
||||
write a test or tests when you add a new feature.
|
||||
|
||||
**At a bare minimum a test should show which Discourse API endpoint is called,
|
||||
using which HTTP method, and returning any necessary data for the new function/method.**
|
||||
|
||||
In most cases this can be accomplished quite simply by using the `discourse_request`
|
||||
fixture, which allows for mocking the HTTP request in the `requests` library. In some cases
|
||||
this may be insufficient, and you may want to directly use the `requests_mock` mocking
|
||||
fixture.
|
||||
|
||||
If in the course of writing your test you see a `requests_mock.NoMockAddress` exception
|
||||
raised then either the *method* or the *path* (including querystring) - or both! - in
|
||||
either your mock OR your new API client method is incorrect.
|
||||
|
||||
Live Testing
|
||||
============
|
||||
|
||||
You can test against a Discourse instance by following the [Official Discourse developement instructions][discoursedev].
|
||||
|
||||
For the impatient here is the quick and dirty version::
|
||||
|
||||
git clone git@github.com:discourse/discourse.git
|
||||
|
||||
+7
-2
@@ -1,2 +1,7 @@
|
||||
pytest==6.2.5
|
||||
pytest-cov==3.0.0
|
||||
pytest==7.4.0
|
||||
pytest-cov==4.1.0
|
||||
pytest-mock==3.11.1 # https://github.com/pytest-dev/pytest-mock/
|
||||
pytest-socket==0.6.0 # https://github.com/miketheman/pytest-socket
|
||||
requests-mock==1.11.0 # https://github.com/jamielennox/requests-mock
|
||||
pytest-subtests==0.11.0 # https://github.com/pytest-dev/pytest-subtests
|
||||
pytest-icdiff==0.6 # https://pypi.org/project/pytest-icdiff/
|
||||
|
||||
@@ -18,7 +18,6 @@ package_dir =
|
||||
=src
|
||||
install_requires =
|
||||
requests>=2.4.2
|
||||
typing; python_version<"3.6"
|
||||
classifiers =
|
||||
Development Status :: 5 - Production/Stable
|
||||
Environment :: Web Environment
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Python client for the Discourse API."""
|
||||
|
||||
__version__ = "1.5.0"
|
||||
__version__ = "1.5.1"
|
||||
|
||||
from pydiscourse.client import DiscourseClient
|
||||
|
||||
|
||||
@@ -27,6 +27,14 @@ POST = "POST"
|
||||
PUT = "PUT"
|
||||
|
||||
|
||||
def now() -> datetime:
|
||||
"""Returns the current UTC time.
|
||||
|
||||
This function enables simple mocking for freezing time.
|
||||
"""
|
||||
return datetime.utcnow()
|
||||
|
||||
|
||||
class DiscourseClient(object):
|
||||
"""Discourse API client"""
|
||||
|
||||
@@ -236,7 +244,7 @@ class DiscourseClient(object):
|
||||
????
|
||||
|
||||
"""
|
||||
suspend_until = (datetime.now() + timedelta(days=duration)).isoformat()
|
||||
suspend_until = (now() + timedelta(days=duration)).isoformat()
|
||||
return self._put(
|
||||
"/admin/users/{0}/suspend".format(userid),
|
||||
suspend_until=suspend_until,
|
||||
|
||||
@@ -43,15 +43,15 @@ def sso_validate(payload, signature, secret):
|
||||
raise DiscourseError("No SSO payload or signature.")
|
||||
|
||||
if not secret:
|
||||
raise DiscourseError("Invalid secret..")
|
||||
raise DiscourseError("Invalid secret.")
|
||||
|
||||
payload = unquote(payload)
|
||||
if not payload:
|
||||
raise DiscourseError("Invalid payload..")
|
||||
raise DiscourseError("Invalid payload.")
|
||||
|
||||
decoded = b64decode(payload.encode("utf-8")).decode("utf-8")
|
||||
if "nonce" not in decoded:
|
||||
raise DiscourseError("Invalid payload..")
|
||||
raise DiscourseError("Invalid payload.")
|
||||
|
||||
h = hmac.new(
|
||||
secret.encode("utf-8"), payload.encode("utf-8"), digestmod=hashlib.sha256
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
"""Test fixtures."""
|
||||
|
||||
import datetime
|
||||
|
||||
import pytest
|
||||
from pydiscourse import client
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def sso_secret():
|
||||
return "d836444a9e4084d5b224a60c208dce14"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def sso_nonce():
|
||||
return "cb68251eefb5211e58c00ff1395f0c0b"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def sso_payload():
|
||||
return "bm9uY2U9Y2I2ODI1MWVlZmI1MjExZTU4YzAwZmYxMzk1ZjBjMGI%3D%0A"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def sso_signature():
|
||||
return "2828aa29899722b35a2f191d34ef9b3ce695e0e6eeec47deb46d588d70c7cb56"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def name():
|
||||
return "sam"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def username():
|
||||
return "samsam"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def external_id():
|
||||
return "hello123"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def email():
|
||||
return "test@test.com"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def redirect_url():
|
||||
return "/session/sso_login?sso=bm9uY2U9Y2I2ODI1MWVlZmI1MjExZTU4YzAwZmYxMzk1ZjBjMGImbmFtZT1z%0AYW0mdXNlcm5hbWU9c2Ftc2FtJmVtYWlsPXRlc3QlNDB0ZXN0LmNvbSZleHRl%0Acm5hbF9pZD1oZWxsbzEyMw%3D%3D%0A&sig=1c884222282f3feacd76802a9dd94e8bc8deba5d619b292bed75d63eb3152c0b"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def discourse_host():
|
||||
return "http://testhost"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def discourse_api_username():
|
||||
return "testuser"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def discourse_api_key():
|
||||
return "testkey"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def discourse_client(discourse_host, discourse_api_username, discourse_api_key):
|
||||
return client.DiscourseClient(
|
||||
discourse_host, discourse_api_username, discourse_api_key
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def frozen_time(mocker):
|
||||
now = mocker.patch("pydiscourse.client.now")
|
||||
now.return_value = datetime.datetime(
|
||||
2023, 8, 13, 12, 30, 15, tzinfo=datetime.timezone.utc
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def discourse_request(discourse_host, discourse_client, requests_mock):
|
||||
"""Fixture for mocking Discourse API requests.
|
||||
|
||||
The only request arguments are the method and the path.
|
||||
|
||||
Example:
|
||||
|
||||
>>> def test_something(discourse_request):
|
||||
>>> request = discourse_request(
|
||||
>>> "put", # the method, case-insensitive
|
||||
>>> "/the-path.json?q=4", # the absolute path with query, NO host
|
||||
>>> headers={'content-type': 'text/plain'}, # override default headers
|
||||
>>> content=b"ERROR", # override bytestring response
|
||||
>>> )
|
||||
|
||||
If `content` is provided, that will be used as the response body.
|
||||
If `json` is provided, then the body will return the given JSON-
|
||||
compatable Python structure (e.g. dictionary).
|
||||
If neither is given then the return `json` will be an empty
|
||||
dictionary (`{}`).
|
||||
|
||||
Returns a function for inserting sensible default values.
|
||||
"""
|
||||
|
||||
def inner(method, path, headers=None, json=None, content=None):
|
||||
full_path = f"{discourse_host}{path}"
|
||||
if not headers:
|
||||
headers = {
|
||||
"Content-Type": "application/json; charset=utf-8",
|
||||
"Api-Key": discourse_client.api_key,
|
||||
"Api-Username": discourse_client.api_username,
|
||||
}
|
||||
|
||||
kwargs = {}
|
||||
if content:
|
||||
kwargs["content"] = content
|
||||
elif json:
|
||||
kwargs["json"] = json
|
||||
else:
|
||||
kwargs["json"] = {}
|
||||
|
||||
return requests_mock.request(method, full_path, headers=headers, **kwargs)
|
||||
|
||||
return inner
|
||||
+206
-193
@@ -1,221 +1,234 @@
|
||||
import sys
|
||||
import unittest
|
||||
"""Tests for the client methods."""
|
||||
|
||||
from unittest import mock
|
||||
|
||||
from pydiscourse import client
|
||||
import urllib.parse
|
||||
|
||||
|
||||
if sys.version_info < (3,):
|
||||
def test_empty_content_http_ok(discourse_host, discourse_client, requests_mock):
|
||||
"""Empty content should not raise error
|
||||
|
||||
def b(x):
|
||||
return x
|
||||
|
||||
|
||||
else:
|
||||
import codecs
|
||||
|
||||
def b(x):
|
||||
return codecs.latin_1_encode(x)[0]
|
||||
|
||||
|
||||
def prepare_response(request):
|
||||
# we need to mocked response to look a little more real
|
||||
request.return_value = mock.MagicMock(
|
||||
headers={"content-type": "application/json; charset=utf-8"}
|
||||
Critical to test against *bytestrings* rather than unicode
|
||||
"""
|
||||
requests_mock.get(
|
||||
f"{discourse_host}/users/admin/1/unsuspend",
|
||||
headers={"Content-Type": "text/plain; charset=utf-8"},
|
||||
content=b" ",
|
||||
)
|
||||
|
||||
resp = discourse_client._request("GET", "/users/admin/1/unsuspend", {})
|
||||
|
||||
class ClientBaseTestCase(unittest.TestCase):
|
||||
"""
|
||||
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.host = "http://testhost"
|
||||
self.api_username = "testuser"
|
||||
self.api_key = "testkey"
|
||||
|
||||
self.client = client.DiscourseClient(self.host, self.api_username, self.api_key)
|
||||
|
||||
def assertRequestCalled(self, request, verb, url, **params):
|
||||
self.assertTrue(request.called)
|
||||
|
||||
args, kwargs = request.call_args
|
||||
|
||||
self.assertEqual(args[0], verb)
|
||||
self.assertEqual(args[1], self.host + url)
|
||||
|
||||
headers = kwargs["headers"]
|
||||
self.assertEqual(headers.pop("Api-Username"), self.api_username)
|
||||
self.assertEqual(headers.pop("Api-Key"), self.api_key)
|
||||
|
||||
if verb == "GET":
|
||||
self.assertEqual(kwargs["params"], params)
|
||||
assert resp is None
|
||||
|
||||
|
||||
class TestClientRequests(ClientBaseTestCase):
|
||||
"""
|
||||
Tests for common request handling
|
||||
"""
|
||||
class TestUserManagement:
|
||||
def test_get_user(self, discourse_host, discourse_client, discourse_request):
|
||||
request = discourse_request(
|
||||
"get", "/users/someuser.json", json={"user": "someuser"}
|
||||
)
|
||||
discourse_client.user("someuser")
|
||||
|
||||
@mock.patch("pydiscourse.client.requests")
|
||||
def test_empty_content_http_ok(self, mocked_requests):
|
||||
"""Empty content should not raise error
|
||||
assert request.called_once
|
||||
|
||||
Critical to test against *bytestrings* rather than unicode
|
||||
"""
|
||||
mocked_response = mock.MagicMock()
|
||||
mocked_response.content = b(" ")
|
||||
mocked_response.status_code = 200
|
||||
mocked_response.headers = {"content-type": "text/plain; charset=utf-8"}
|
||||
|
||||
assert "content-type" in mocked_response.headers
|
||||
|
||||
mocked_requests.request = mock.MagicMock()
|
||||
mocked_requests.request.return_value = mocked_response
|
||||
|
||||
resp = self.client._request("GET", "/users/admin/1/unsuspend", {})
|
||||
self.assertIsNone(resp)
|
||||
|
||||
|
||||
@mock.patch("requests.request")
|
||||
class TestUser(ClientBaseTestCase):
|
||||
|
||||
def test_user(self, request):
|
||||
prepare_response(request)
|
||||
self.client.user("someuser")
|
||||
self.assertRequestCalled(request, "GET", "/users/someuser.json")
|
||||
|
||||
def test_create_user(self, request):
|
||||
prepare_response(request)
|
||||
self.client.create_user(
|
||||
def test_create_user(self, discourse_host, discourse_client, requests_mock):
|
||||
session_request = requests_mock.get(
|
||||
f"{discourse_host}/session/hp.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={"challenge": "challenge", "value": "value"},
|
||||
)
|
||||
user_request = requests_mock.post(
|
||||
f"{discourse_host}/users",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.create_user(
|
||||
"Test User", "testuser", "test@example.com", "notapassword"
|
||||
)
|
||||
self.assertEqual(request.call_count, 2)
|
||||
|
||||
# XXX incomplete
|
||||
assert session_request.called_once
|
||||
assert user_request.called_once
|
||||
|
||||
def test_update_email(self, request):
|
||||
prepare_response(request)
|
||||
email = "test@example.com"
|
||||
self.client.update_email("someuser", email)
|
||||
self.assertRequestCalled(
|
||||
request, "PUT", "/users/someuser/preferences/email", email=email
|
||||
def test_update_email(self, discourse_host, discourse_client, discourse_request):
|
||||
request = discourse_request("put", "/users/someuser/preferences/email")
|
||||
discourse_client.update_email("someuser", "newmeail@example.com")
|
||||
|
||||
assert request.called_once
|
||||
|
||||
def test_update_user(self, discourse_client, discourse_request):
|
||||
request = discourse_request("put", "/users/someuser")
|
||||
discourse_client.update_user("someuser", a="a", b="b")
|
||||
|
||||
assert request.called_once
|
||||
|
||||
def test_update_username(self, discourse_client, discourse_request):
|
||||
request = discourse_request("put", "/users/someuser/preferences/username")
|
||||
discourse_client.update_username("someuser", "newname")
|
||||
|
||||
assert request.called_once
|
||||
|
||||
def test_by_external_id(self, discourse_client, discourse_request):
|
||||
request = discourse_request(
|
||||
"get", "/users/by-external/123", json={"user": "123"}
|
||||
)
|
||||
discourse_client.by_external_id(123)
|
||||
|
||||
def test_update_user(self, request):
|
||||
prepare_response(request)
|
||||
self.client.update_user("someuser", a="a", b="b")
|
||||
self.assertRequestCalled(request, "PUT", "/users/someuser", a="a", b="b")
|
||||
assert request.called_once
|
||||
|
||||
def test_update_username(self, request):
|
||||
prepare_response(request)
|
||||
self.client.update_username("someuser", "newname")
|
||||
self.assertRequestCalled(
|
||||
request, "PUT", "/users/someuser/preferences/username", username="newname"
|
||||
def test_suspend_user(self, discourse_client, discourse_request, frozen_time):
|
||||
request = discourse_request("put", "/admin/users/123/suspend")
|
||||
discourse_client.suspend(123, 1, "Testing")
|
||||
|
||||
assert request.called_once
|
||||
assert request.last_request.method == "PUT"
|
||||
|
||||
request_payload = urllib.parse.parse_qs(request.last_request.text)
|
||||
|
||||
assert request_payload["reason"] == ["Testing"]
|
||||
assert request_payload["suspend_until"] == ["2023-08-14T12:30:15+00:00"]
|
||||
|
||||
def test_unsuspend_user(self, discourse_client, discourse_request):
|
||||
request = discourse_request("put", "/admin/users/123/unsuspend")
|
||||
discourse_client.unsuspend(123)
|
||||
|
||||
assert request.called_once
|
||||
|
||||
def test_user_bagdes(self, discourse_client, discourse_request):
|
||||
request = discourse_request("get", "/user-badges/myusername.json")
|
||||
discourse_client.user_badges("myusername")
|
||||
|
||||
assert request.called_once
|
||||
|
||||
|
||||
class TestTopics:
|
||||
def test_hot_topics(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/hot.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.hot_topics()
|
||||
assert request.called_once
|
||||
|
||||
def test_by_external_id(self, request):
|
||||
prepare_response(request)
|
||||
self.client.by_external_id(123)
|
||||
self.assertRequestCalled(request, "GET", "/users/by-external/123")
|
||||
|
||||
def test_suspend_user(self, request):
|
||||
prepare_response(request)
|
||||
self.client.suspend(123, 1, "Testing")
|
||||
self.assertRequestCalled(
|
||||
request, "PUT", "/admin/users/123/suspend", duration=1, reason="Testing"
|
||||
def test_latest_topics(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/latest.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.latest_topics()
|
||||
|
||||
def test_unsuspend_user(self, request):
|
||||
prepare_response(request)
|
||||
self.client.unsuspend(123)
|
||||
self.assertRequestCalled(request, "PUT", "/admin/users/123/unsuspend")
|
||||
assert request.called_once
|
||||
|
||||
def test_user_bagdes(self, request):
|
||||
prepare_response(request)
|
||||
self.client.user_badges("username")
|
||||
self.assertRequestCalled(
|
||||
request, "GET", "/user-badges/{}.json".format("username")
|
||||
def test_new_topics(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/new.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.new_topics()
|
||||
assert request.called_once
|
||||
|
||||
|
||||
@mock.patch("requests.request")
|
||||
class TestTopics(ClientBaseTestCase):
|
||||
|
||||
def test_hot_topics(self, request):
|
||||
prepare_response(request)
|
||||
self.client.hot_topics()
|
||||
self.assertRequestCalled(request, "GET", "/hot.json")
|
||||
|
||||
def test_latest_topics(self, request):
|
||||
prepare_response(request)
|
||||
self.client.latest_topics()
|
||||
self.assertRequestCalled(request, "GET", "/latest.json")
|
||||
|
||||
def test_new_topics(self, request):
|
||||
prepare_response(request)
|
||||
self.client.new_topics()
|
||||
self.assertRequestCalled(request, "GET", "/new.json")
|
||||
|
||||
def test_topic(self, request):
|
||||
prepare_response(request)
|
||||
self.client.topic("some-test-slug", 22)
|
||||
self.assertRequestCalled(request, "GET", "/t/some-test-slug/22.json")
|
||||
|
||||
def test_topics_by(self, request):
|
||||
prepare_response(request)
|
||||
r = self.client.topics_by("someuser")
|
||||
self.assertRequestCalled(request, "GET", "/topics/created-by/someuser.json")
|
||||
self.assertEqual(r, request().json()["topic_list"]["topics"])
|
||||
|
||||
def invite_user_to_topic(self, request):
|
||||
prepare_response(request)
|
||||
email = "test@example.com"
|
||||
self.client.invite_user_to_topic(email, 22)
|
||||
self.assertRequestCalled(
|
||||
request, "POST", "/t/22/invite.json", email=email, topic_id=22
|
||||
def test_topic(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/t/some-test-slug/22.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.topic("some-test-slug", 22)
|
||||
assert request.called_once
|
||||
|
||||
|
||||
@mock.patch("pydiscourse.client.requests.request")
|
||||
class MiscellaneousTests(ClientBaseTestCase):
|
||||
|
||||
def test_latest_posts(self, request):
|
||||
prepare_response(request)
|
||||
r = self.client.latest_posts(before=54321)
|
||||
self.assertRequestCalled(request, "GET", "/posts.json", before=54321)
|
||||
|
||||
def test_search(self, request):
|
||||
prepare_response(request)
|
||||
self.client.search("needle")
|
||||
self.assertRequestCalled(request, "GET", "/search.json", term="needle")
|
||||
|
||||
def test_categories(self, request):
|
||||
prepare_response(request)
|
||||
r = self.client.categories()
|
||||
self.assertRequestCalled(request, "GET", "/categories.json")
|
||||
self.assertEqual(r, request().json()["category_list"]["categories"])
|
||||
|
||||
def test_update_category(self, request):
|
||||
prepare_response(request)
|
||||
self.client.update_category(123, a="a", b="b")
|
||||
self.assertRequestCalled(request, "PUT", "/categories/123", a="a", b="b")
|
||||
|
||||
def test_users(self, request):
|
||||
prepare_response(request)
|
||||
self.client.users()
|
||||
self.assertRequestCalled(request, "GET", "/admin/users/list/active.json")
|
||||
|
||||
def test_badges(self, request):
|
||||
prepare_response(request)
|
||||
self.client.badges()
|
||||
self.assertRequestCalled(request, "GET", "/admin/badges.json")
|
||||
|
||||
def test_grant_badge_to(self, request):
|
||||
prepare_response(request)
|
||||
self.client.grant_badge_to("username", 1)
|
||||
self.assertRequestCalled(
|
||||
request, "POST", "/user_badges", username="username", badge_id=1
|
||||
def test_topics_by(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/topics/created-by/someuser.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={"topic_list": {"topics": []}},
|
||||
)
|
||||
discourse_client.topics_by("someuser")
|
||||
|
||||
assert request.called_once
|
||||
|
||||
def test_invite_user_to_topic(self, discourse_client, requests_mock):
|
||||
request = requests_mock.post(
|
||||
f"{discourse_client.host}/t/22/invite.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.invite_user_to_topic("test@example.com", 22)
|
||||
assert request.called_once
|
||||
|
||||
request_payload = urllib.parse.parse_qs(request.last_request.text)
|
||||
|
||||
assert request_payload["email"] == ["test@example.com"]
|
||||
assert request_payload["topic_id"] == ["22"]
|
||||
|
||||
|
||||
class TestEverything:
|
||||
def test_latest_posts(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/posts.json?before=54321",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.latest_posts(before=54321)
|
||||
assert request.called_once
|
||||
|
||||
def test_search(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/search.json?term=needle",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.search(term="needle")
|
||||
assert request.called_once
|
||||
|
||||
def test_categories(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/categories.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={"category_list": {"categories": []}},
|
||||
)
|
||||
discourse_client.categories()
|
||||
assert request.called_once
|
||||
|
||||
def test_update_category(self, discourse_client, requests_mock):
|
||||
# self.assertRequestCalled(request, "PUT", "/categories/123", a="a", b="b")
|
||||
request = requests_mock.put(
|
||||
f"{discourse_client.host}/categories/123",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.update_category(123, a="a", b="b")
|
||||
|
||||
request_payload = request.last_request.json()
|
||||
|
||||
assert request_payload["a"] == "a"
|
||||
assert request_payload["b"] == "b"
|
||||
|
||||
def test_users(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/admin/users/list/active.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.users()
|
||||
assert request.called_once
|
||||
|
||||
def test_badges(self, discourse_client, requests_mock):
|
||||
request = requests_mock.get(
|
||||
f"{discourse_client.host}/admin/badges.json",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.badges()
|
||||
assert request.called_once
|
||||
|
||||
def test_grant_badge_to(self, discourse_client, requests_mock):
|
||||
request = requests_mock.post(
|
||||
f"{discourse_client.host}/user_badges",
|
||||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||||
json={},
|
||||
)
|
||||
discourse_client.grant_badge_to("username", 1)
|
||||
|
||||
request_payload = urllib.parse.parse_qs(request.last_request.text)
|
||||
|
||||
assert request_payload["username"] == ["username"]
|
||||
assert request_payload["badge_id"] == ["1"]
|
||||
|
||||
+69
-56
@@ -1,76 +1,89 @@
|
||||
from base64 import b64decode
|
||||
|
||||
import unittest
|
||||
|
||||
from urllib.parse import unquote
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
|
||||
import pytest
|
||||
|
||||
from pydiscourse import sso
|
||||
from pydiscourse.exceptions import DiscourseError
|
||||
|
||||
|
||||
class SSOTestCase(unittest.TestCase):
|
||||
def test_sso_validate_missing_payload():
|
||||
with pytest.raises(DiscourseError) as excinfo:
|
||||
sso.sso_validate(None, "abc", "123")
|
||||
|
||||
def setUp(self):
|
||||
# values from https://meta.discourse.org/t/official-single-sign-on-for-discourse/13045
|
||||
self.secret = "d836444a9e4084d5b224a60c208dce14"
|
||||
self.nonce = "cb68251eefb5211e58c00ff1395f0c0b"
|
||||
self.payload = "bm9uY2U9Y2I2ODI1MWVlZmI1MjExZTU4YzAwZmYxMzk1ZjBjMGI%3D%0A"
|
||||
self.signature = "2828aa29899722b35a2f191d34ef9b3ce695e0e6eeec47deb46d588d70c7cb56"
|
||||
assert excinfo.value.args[0] == "No SSO payload or signature."
|
||||
|
||||
self.name = u"sam"
|
||||
self.username = u"samsam"
|
||||
self.external_id = u"hello123"
|
||||
self.email = u"test@test.com"
|
||||
self.redirect_url = u"/session/sso_login?sso=bm9uY2U9Y2I2ODI1MWVlZmI1MjExZTU4YzAwZmYxMzk1ZjBjMGImbmFtZT1z%0AYW0mdXNlcm5hbWU9c2Ftc2FtJmVtYWlsPXRlc3QlNDB0ZXN0LmNvbSZleHRl%0Acm5hbF9pZD1oZWxsbzEyMw%3D%3D%0A&sig=1c884222282f3feacd76802a9dd94e8bc8deba5d619b292bed75d63eb3152c0b"
|
||||
|
||||
def test_missing_args(self):
|
||||
with self.assertRaises(DiscourseError):
|
||||
sso.sso_validate(None, self.signature, self.secret)
|
||||
def test_sso_validate_empty_payload():
|
||||
with pytest.raises(DiscourseError) as excinfo:
|
||||
sso.sso_validate("", "abc", "123")
|
||||
|
||||
with self.assertRaises(DiscourseError):
|
||||
sso.sso_validate("", self.signature, self.secret)
|
||||
assert excinfo.value.args[0] == "Invalid payload."
|
||||
|
||||
with self.assertRaises(DiscourseError):
|
||||
sso.sso_validate(self.payload, None, self.secret)
|
||||
|
||||
def test_invalid_signature(self):
|
||||
with self.assertRaises(DiscourseError):
|
||||
sso.sso_validate(self.payload, "notavalidsignature", self.secret)
|
||||
def test_sso_validate_missing_signature():
|
||||
with pytest.raises(DiscourseError) as excinfo:
|
||||
sso.sso_validate("sig", None, "123")
|
||||
|
||||
def test_valid_nonce(self):
|
||||
nonce = sso.sso_validate(self.payload, self.signature, self.secret)
|
||||
self.assertEqual(nonce, self.nonce)
|
||||
assert excinfo.value.args[0] == "No SSO payload or signature."
|
||||
|
||||
def test_valid_redirect_url(self):
|
||||
url = sso.sso_redirect_url(
|
||||
self.nonce,
|
||||
self.secret,
|
||||
self.email,
|
||||
self.external_id,
|
||||
self.username,
|
||||
name="sam",
|
||||
)
|
||||
|
||||
self.assertIn("/session/sso_login", url[:20])
|
||||
@pytest.mark.parametrize("bad_secret", [None, ""])
|
||||
def test_sso_validate_missing_secret(bad_secret):
|
||||
with pytest.raises(DiscourseError) as excinfo:
|
||||
sso.sso_validate("payload", "signature", bad_secret)
|
||||
|
||||
# check its valid, using our own handy validator
|
||||
params = parse_qs(urlparse(url).query)
|
||||
payload = params["sso"][0]
|
||||
sso.sso_validate(payload, params["sig"][0], self.secret)
|
||||
assert excinfo.value.args[0] == "Invalid secret."
|
||||
|
||||
# check the params have all the data we expect
|
||||
payload = b64decode(payload.encode("utf-8")).decode("utf-8")
|
||||
payload = unquote(payload)
|
||||
payload = dict((p.split("=") for p in payload.split("&")))
|
||||
|
||||
self.assertEqual(
|
||||
payload,
|
||||
{
|
||||
"username": self.username,
|
||||
"nonce": self.nonce,
|
||||
"external_id": self.external_id,
|
||||
"name": self.name,
|
||||
"email": self.email,
|
||||
},
|
||||
)
|
||||
def test_sso_validate_invalid_signature(sso_payload, sso_signature, sso_secret):
|
||||
with pytest.raises(DiscourseError) as excinfo:
|
||||
sso.sso_validate("Ym9i", sso_signature, sso_secret)
|
||||
|
||||
assert excinfo.value.args[0] == "Invalid payload."
|
||||
|
||||
|
||||
def test_sso_validate_invalid_payload_nonce(sso_payload, sso_secret):
|
||||
with pytest.raises(DiscourseError) as excinfo:
|
||||
sso.sso_validate(sso_payload, "notavalidsignature", sso_secret)
|
||||
|
||||
assert excinfo.value.args[0] == "Payload does not match signature."
|
||||
|
||||
|
||||
def test_valid_nonce(sso_payload, sso_signature, sso_secret, sso_nonce):
|
||||
generated_nonce = sso.sso_validate(sso_payload, sso_signature, sso_secret)
|
||||
assert generated_nonce == sso_nonce
|
||||
|
||||
|
||||
def test_valid_redirect_url(
|
||||
sso_secret, sso_nonce, name, email, username, external_id, redirect_url
|
||||
):
|
||||
url = sso.sso_redirect_url(
|
||||
sso_nonce,
|
||||
sso_secret,
|
||||
email,
|
||||
external_id,
|
||||
username,
|
||||
name="sam",
|
||||
)
|
||||
|
||||
assert "/session/sso_login" in url[:20]
|
||||
|
||||
# check its valid, using our own handy validator
|
||||
params = parse_qs(urlparse(url).query)
|
||||
payload = params["sso"][0]
|
||||
sso.sso_validate(payload, params["sig"][0], sso_secret)
|
||||
|
||||
# check the params have all the data we expect
|
||||
payload = b64decode(payload.encode("utf-8")).decode("utf-8")
|
||||
payload = unquote(payload)
|
||||
payload = dict((p.split("=") for p in payload.split("&")))
|
||||
|
||||
assert payload == {
|
||||
"username": username,
|
||||
"nonce": sso_nonce,
|
||||
"external_id": external_id,
|
||||
"name": name,
|
||||
"email": email,
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
[tox]
|
||||
envlist = py37, py38, py39, py310, py311
|
||||
envlist = py38, py39, py310, py311
|
||||
|
||||
[gh-actions]
|
||||
python =
|
||||
@@ -11,7 +11,10 @@ python =
|
||||
[testenv]
|
||||
setenv =
|
||||
PYTHONPATH = {toxinidir}:{toxinidir}/pydiscourse
|
||||
commands = pytest {posargs} --cov=pydiscourse
|
||||
commands =
|
||||
pytest {posargs} --cov=pydiscourse
|
||||
coverage report -m --include='**/pydiscourse/client.py' --fail-under=45
|
||||
coverage report -m --include='**/pydiscourse/sso.py' --fail-under=100
|
||||
deps =
|
||||
-r{toxinidir}/requirements.txt
|
||||
|
||||
|
||||
Reference in New Issue
Block a user