iSharkFly-Open/discoursep
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Format with black

Browse Source
This commit is contained in:
Ben Lopatin
2018-10-29 18:02:58 -04:00
parent b0b277c917
commit c0db7215c9
6 changed files with 401 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pydiscourse/__init__.py
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
__version__ = '0.7.0'
__version__ = "0.7.0"
from pydiscourse.client import DiscourseClient
pydiscourse/client.py
+252 -174
View File
@@ -7,11 +7,12 @@ import logging
import requests
from pydiscourse.exceptions import (
DiscourseError, DiscourseServerError, DiscourseClientError)
DiscourseError, DiscourseServerError, DiscourseClientError
)
from pydiscourse.sso import sso_payload
log = logging.getLogger('pydiscourse.client')
log = logging.getLogger("pydiscourse.client")
# HTTP verbs to be used as non string literals
DELETE = "DELETE"
@@ -55,7 +56,7 @@ class DiscourseClient(object):
dict of user information
"""
return self._get('/users/{0}.json'.format(username))['user']
return self._get("/users/{0}.json".format(username))["user"]
def user_all(self, user_id):
"""
@@ -66,7 +67,7 @@ class DiscourseClient(object):
Returns:
dict of user information
"""
return self._get('/admin/users/{0}.json'.format(user_id))
return self._get("/admin/users/{0}.json".format(user_id))
def create_user(self, name, username, email, password, **kwargs):
"""
@@ -87,12 +88,19 @@ class DiscourseClient(object):
????
"""
r = self._get('/users/hp.json')
challenge = r['challenge'][::-1] # reverse challenge, discourse security check
confirmations = r['value']
return self._post('/users', name=name, username=username, email=email,
password=password, password_confirmation=confirmations,
challenge=challenge, **kwargs)
r = self._get("/users/hp.json")
challenge = r["challenge"][::-1] # reverse challenge, discourse security check
confirmations = r["value"]
return self._post(
"/users",
name=name,
username=username,
email=email,
password=password,
password_confirmation=confirmations,
challenge=challenge,
**kwargs
)
def user_by_external_id(self, external_id):
"""
@@ -104,7 +112,8 @@ class DiscourseClient(object):
"""
response = self._get("/users/by-external/{0}".format(external_id))
return response['user']
return response["user"]
by_external_id = user_by_external_id
def log_out(self, userid):
@@ -116,7 +125,7 @@ class DiscourseClient(object):
Returns:
"""
return self._post('/admin/users/{0}/log_out'.format(userid))
return self._post("/admin/users/{0}/log_out".format(userid))
def trust_level(self, userid, level):
"""
@@ -128,7 +137,7 @@ class DiscourseClient(object):
Returns:
"""
return self._put('/admin/users/{0}/trust_level'.format(userid), level=level)
return self._put("/admin/users/{0}/trust_level".format(userid), level=level)
def suspend(self, userid, duration, reason):
"""
@@ -144,8 +153,9 @@ class DiscourseClient(object):
????
"""
return self._put('/admin/users/{0}/suspend'.format(userid),
duration=duration, reason=reason)
return self._put(
"/admin/users/{0}/suspend".format(userid), duration=duration, reason=reason
)
def unsuspend(self, userid):
"""
@@ -157,7 +167,7 @@ class DiscourseClient(object):
Returns:
None???
"""
return self._put('/admin/users/{0}/unsuspend'.format(userid))
return self._put("/admin/users/{0}/unsuspend".format(userid))
def list_users(self, type, **kwargs):
"""
@@ -171,7 +181,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/admin/users/list/{0}.json'.format(type), **kwargs)
return self._get("/admin/users/list/{0}.json".format(type), **kwargs)
def update_avatar_from_url(self, username, url, **kwargs):
"""
@@ -184,7 +194,9 @@ class DiscourseClient(object):
Returns:
"""
return self._post('/users/{0}/preferences/avatar'.format(username), file=url, **kwargs)
return self._post(
"/users/{0}/preferences/avatar".format(username), file=url, **kwargs
)
def update_avatar_image(self, username, img, **kwargs):
"""
@@ -199,8 +211,10 @@ class DiscourseClient(object):
Returns:
"""
files = {'file': img}
return self._post('/users/{0}/preferences/avatar'.format(username), files=files, **kwargs)
files = {"file": img}
return self._post(
"/users/{0}/preferences/avatar".format(username), files=files, **kwargs
)
def toggle_gravatar(self, username, state=True, **kwargs):
"""
@@ -213,11 +227,11 @@ class DiscourseClient(object):
Returns:
"""
url = '/users/{0}/preferences/avatar/toggle'.format(username)
url = "/users/{0}/preferences/avatar/toggle".format(username)
if bool(state):
kwargs['use_uploaded_avatar'] = 'true'
kwargs["use_uploaded_avatar"] = "true"
else:
kwargs['use_uploaded_avatar'] = 'false'
kwargs["use_uploaded_avatar"] = "false"
return self._put(url, **kwargs)
def pick_avatar(self, username, gravatar=True, generated=False, **kwargs):
@@ -232,7 +246,7 @@ class DiscourseClient(object):
Returns:
"""
url = '/users/{0}/preferences/avatar/pick'.format(username)
url = "/users/{0}/preferences/avatar/pick".format(username)
return self._put(url, **kwargs)
def update_avatar(self, username, url, **kwargs):
@@ -246,11 +260,14 @@ class DiscourseClient(object):
Returns:
"""
kwargs['type'] = 'avatar'
kwargs['synchronous'] = 'true'
upload_response = self._post('/uploads', url=url, **kwargs)
return self._put('/users/{0}/preferences/avatar/pick'.format(username),
upload_id=upload_response['id'], **kwargs)
kwargs["type"] = "avatar"
kwargs["synchronous"] = "true"
upload_response = self._post("/uploads", url=url, **kwargs)
return self._put(
"/users/{0}/preferences/avatar/pick".format(username),
upload_id=upload_response["id"],
**kwargs
)
def update_email(self, username, email, **kwargs):
"""
@@ -263,7 +280,9 @@ class DiscourseClient(object):
Returns:
"""
return self._put('/users/{0}/preferences/email'.format(username), email=email, **kwargs)
return self._put(
"/users/{0}/preferences/email".format(username), email=email, **kwargs
)
def update_user(self, username, **kwargs):
"""
@@ -275,7 +294,7 @@ class DiscourseClient(object):
Returns:
"""
return self._put('/users/{0}'.format(username), json=True, **kwargs)
return self._put("/users/{0}".format(username), json=True, **kwargs)
def update_username(self, username, new_username, **kwargs):
"""
@@ -288,8 +307,11 @@ class DiscourseClient(object):
Returns:
"""
return self._put('/users/{0}/preferences/username'.format(username),
username=new_username, **kwargs)
return self._put(
"/users/{0}/preferences/username".format(username),
username=new_username,
**kwargs
)
def set_preference(self, username=None, **kwargs):
"""
@@ -303,7 +325,7 @@ class DiscourseClient(object):
"""
if username is None:
username = self.api_username
return self._put(u'/users/{0}'.format(username), **kwargs)
return self._put(u"/users/{0}".format(username), **kwargs)
def sync_sso(self, **kwargs):
"""
@@ -317,9 +339,9 @@ class DiscourseClient(object):
Returns:
"""
sso_secret = kwargs.pop('sso_secret')
sso_secret = kwargs.pop("sso_secret")
payload = sso_payload(sso_secret, **kwargs)
return self._post('/admin/users/sync_sso?{0}'.format(payload), **kwargs)
return self._post("/admin/users/sync_sso?{0}".format(payload), **kwargs)
def generate_api_key(self, userid, **kwargs):
"""
@@ -331,7 +353,7 @@ class DiscourseClient(object):
Returns:
"""
return self._post('/admin/users/{0}/generate_api_key'.format(userid), **kwargs)
return self._post("/admin/users/{0}/generate_api_key".format(userid), **kwargs)
def delete_user(self, userid, **kwargs):
"""
@@ -347,7 +369,7 @@ class DiscourseClient(object):
Returns:
"""
return self._delete('/admin/users/{0}.json'.format(userid), **kwargs)
return self._delete("/admin/users/{0}.json".format(userid), **kwargs)
def users(self, filter=None, **kwargs):
"""
@@ -360,9 +382,9 @@ class DiscourseClient(object):
"""
if filter is None:
filter = 'active'
filter = "active"
return self._get('/admin/users/list/{0}.json'.format(filter), **kwargs)
return self._get("/admin/users/list/{0}.json".format(filter), **kwargs)
def private_messages(self, username=None, **kwargs):
"""
@@ -376,7 +398,7 @@ class DiscourseClient(object):
"""
if username is None:
username = self.api_username
return self._get('/topics/private-messages/{0}.json'.format(username), **kwargs)
return self._get("/topics/private-messages/{0}.json".format(username), **kwargs)
def private_messages_unread(self, username=None, **kwargs):
"""
@@ -390,7 +412,9 @@ class DiscourseClient(object):
"""
if username is None:
username = self.api_username
return self._get('/topics/private-messages-unread/{0}.json'.format(username), **kwargs)
return self._get(
"/topics/private-messages-unread/{0}.json".format(username), **kwargs
)
def hot_topics(self, **kwargs):
"""
@@ -401,7 +425,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/hot.json', **kwargs)
return self._get("/hot.json", **kwargs)
def latest_topics(self, **kwargs):
"""
@@ -412,7 +436,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/latest.json', **kwargs)
return self._get("/latest.json", **kwargs)
def new_topics(self, **kwargs):
"""
@@ -423,7 +447,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/new.json', **kwargs)
return self._get("/new.json", **kwargs)
def topic(self, slug, topic_id, **kwargs):
"""
@@ -436,7 +460,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/t/{0}/{1}.json'.format(slug, topic_id), **kwargs)
return self._get("/t/{0}/{1}.json".format(slug, topic_id), **kwargs)
def post(self, topic_id, post_id, **kwargs):
"""
@@ -449,7 +473,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/t/{0}/{1}.json'.format(topic_id, post_id), **kwargs)
return self._get("/t/{0}/{1}.json".format(topic_id, post_id), **kwargs)
def posts(self, topic_id, post_ids=None, **kwargs):
"""
@@ -464,8 +488,8 @@ class DiscourseClient(object):
"""
if post_ids:
kwargs['post_ids[]'] = post_ids
return self._get('/t/{0}/posts.json'.format(topic_id), **kwargs)
kwargs["post_ids[]"] = post_ids
return self._get("/t/{0}/posts.json".format(topic_id), **kwargs)
def topic_timings(self, topic_id, time, timings={}, **kwargs):
"""
@@ -482,12 +506,12 @@ class DiscourseClient(object):
Returns:
"""
kwargs['topic_id'] = topic_id
kwargs['topic_time'] = time
kwargs["topic_id"] = topic_id
kwargs["topic_time"] = time
for post_num, timing in timings.items():
kwargs['timings[{0}]'.format(post_num)] = timing
kwargs["timings[{0}]".format(post_num)] = timing
return self._post('/topics/timings', **kwargs)
return self._post("/topics/timings", **kwargs)
def topic_posts(self, topic_id, **kwargs):
"""
@@ -499,7 +523,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/t/{0}/posts.json'.format(topic_id), **kwargs)
return self._get("/t/{0}/posts.json".format(topic_id), **kwargs)
def update_topic(self, topic_url, title, **kwargs):
"""
@@ -513,11 +537,12 @@ class DiscourseClient(object):
Returns:
"""
kwargs['title'] = title
return self._put('{}'.format(topic_url), **kwargs)
kwargs["title"] = title
return self._put("{}".format(topic_url), **kwargs)
def create_post(self, content, category_id=None, topic_id=None,
title=None, tags=[], **kwargs):
def create_post(
self, content, category_id=None, topic_id=None, title=None, tags=[], **kwargs
):
"""
Args:
@@ -532,9 +557,15 @@ class DiscourseClient(object):
"""
if tags:
kwargs['tags[]'] = tags
return self._post('/posts', category=category_id, title=title,
raw=content, topic_id=topic_id, **kwargs)
kwargs["tags[]"] = tags
return self._post(
"/posts",
category=category_id,
title=title,
raw=content,
topic_id=topic_id,
**kwargs
)
def update_topic_status(self, topic_id, status, enabled, **kwargs):
"""
@@ -549,14 +580,14 @@ class DiscourseClient(object):
Returns:
"""
kwargs['status'] = status
kwargs["status"] = status
if bool(enabled):
kwargs['enabled'] = 'true'
kwargs["enabled"] = "true"
else:
kwargs['enabled'] = 'false'
return self._put('/t/{0}/status'.format(topic_id), **kwargs)
kwargs["enabled"] = "false"
return self._put("/t/{0}/status".format(topic_id), **kwargs)
def update_post(self, post_id, content, edit_reason='', **kwargs):
def update_post(self, post_id, content, edit_reason="", **kwargs):
"""
Args:
@@ -568,9 +599,9 @@ class DiscourseClient(object):
Returns:
"""
kwargs['post[raw]'] = content
kwargs['post[edit_reason]'] = edit_reason
return self._put('/posts/{0}'.format(post_id), **kwargs)
kwargs["post[raw]"] = content
kwargs["post[edit_reason]"] = edit_reason
return self._put("/posts/{0}".format(post_id), **kwargs)
def topics_by(self, username, **kwargs):
"""
@@ -582,8 +613,8 @@ class DiscourseClient(object):
Returns:
"""
url = '/topics/created-by/{0}.json'.format(username)
return self._get(url, **kwargs)['topic_list']['topics']
url = "/topics/created-by/{0}.json".format(username)
return self._get(url, **kwargs)["topic_list"]["topics"]
def invite_user_to_topic(self, user_email, topic_id):
"""
@@ -595,11 +626,8 @@ class DiscourseClient(object):
Returns:
"""
kwargs = {
'email': user_email,
'topic_id': topic_id,
}
return self._post('/t/{0}/invite.json'.format(topic_id), **kwargs)
kwargs = {"email": user_email, "topic_id": topic_id}
return self._post("/t/{0}/invite.json".format(topic_id), **kwargs)
def search(self, term, **kwargs):
"""
@@ -611,8 +639,8 @@ class DiscourseClient(object):
Returns:
"""
kwargs['term'] = term
return self._get('/search.json', **kwargs)
kwargs["term"] = term
return self._get("/search.json", **kwargs)
def badges(self, **kwargs):
"""
@@ -623,7 +651,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/admin/badges.json', **kwargs)
return self._get("/admin/badges.json", **kwargs)
def grant_badge_to(self, username, badge_id, **kwargs):
"""
@@ -636,7 +664,9 @@ class DiscourseClient(object):
Returns:
"""
return self._post('/user_badges', username=username, badge_id=badge_id, **kwargs)
return self._post(
"/user_badges", username=username, badge_id=badge_id, **kwargs
)
def user_badges(self, username, **kwargs):
"""
@@ -647,7 +677,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/user-badges/{}.json'.format(username))
return self._get("/user-badges/{}.json".format(username))
def user_emails(self, username, **kwargs):
"""
@@ -659,10 +689,11 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/u/{}/emails.json'.format(username))
return self._get("/u/{}/emails.json".format(username))
def create_category(self, name, color, text_color='FFFFFF',
permissions=None, parent=None, **kwargs):
def create_category(
self, name, color, text_color="FFFFFF", permissions=None, parent=None, **kwargs
):
"""
Args:
@@ -677,28 +708,29 @@ class DiscourseClient(object):
Returns:
"""
kwargs['name'] = name
kwargs['color'] = color
kwargs['text_color'] = text_color
kwargs["name"] = name
kwargs["color"] = color
kwargs["text_color"] = text_color
if permissions is None and 'permissions' not in kwargs:
permissions = {'everyone': '1'}
if permissions is None and "permissions" not in kwargs:
permissions = {"everyone": "1"}
for key, value in permissions.items():
kwargs['permissions[{0}]'.format(key)] = value
kwargs["permissions[{0}]".format(key)] = value
if parent:
parent_id = None
for category in self.categories():
if category['name'] == parent:
parent_id = category['id']
if category["name"] == parent:
parent_id = category["id"]
continue
if not parent_id:
raise DiscourseClientError(u'{0} not found'.format(parent))
kwargs['parent_category_id'] = parent_id
raise DiscourseClientError(u"{0} not found".format(parent))
return self._post('/categories', **kwargs)
kwargs["parent_category_id"] = parent_id
return self._post("/categories", **kwargs)
def categories(self, **kwargs):
"""
@@ -709,7 +741,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/categories.json', **kwargs)['category_list']['categories']
return self._get("/categories.json", **kwargs)["category_list"]["categories"]
def category(self, name, parent=None, **kwargs):
"""
@@ -723,9 +755,9 @@ class DiscourseClient(object):
"""
if parent:
name = u'{0}/{1}'.format(parent, name)
name = u"{0}/{1}".format(parent, name)
return self._get(u'/category/{0}.json'.format(name), **kwargs)
return self._get(u"/category/{0}.json".format(name), **kwargs)
def delete_category(self, category_id, **kwargs):
"""
@@ -738,7 +770,7 @@ class DiscourseClient(object):
Returns:
"""
return self._delete(u'/categories/{0}'.format(category_id), **kwargs)
return self._delete(u"/categories/{0}".format(category_id), **kwargs)
def site_settings(self, **kwargs):
"""
@@ -751,8 +783,10 @@ class DiscourseClient(object):
"""
for setting, value in kwargs.items():
setting = setting.replace(' ', '_')
self._request(PUT, '/admin/site_settings/{0}'.format(setting), {setting: value})
setting = setting.replace(" ", "_")
self._request(
PUT, "/admin/site_settings/{0}".format(setting), {setting: value}
)
def customize_site_texts(self, site_texts, **kwargs):
"""
@@ -766,8 +800,10 @@ class DiscourseClient(object):
"""
for site_text, value in site_texts.items():
kwargs = {'site_text': {'value': value}}
self._put('/admin/customize/site_texts/{0}'.format(site_text), json=True, **kwargs)
kwargs = {"site_text": {"value": value}}
self._put(
"/admin/customize/site_texts/{0}".format(site_text), json=True, **kwargs
)
def groups(self, **kwargs):
"""
@@ -822,7 +858,22 @@ class DiscourseClient(object):
"""
return self._get("/groups/{0}/members.json".format(group_name))
def create_group(self, name, title="", visible=True, alias_level=0, automatic_membership_retroactive=False, primary_group=False, automatic=False, automatic_membership_email_domains="", grant_trust_level=1, flair_url=None, flair_bg_color=None, flair_color=None, **kwargs):
def create_group(
self,
name,
title="",
visible=True,
alias_level=0,
automatic_membership_retroactive=False,
primary_group=False,
automatic=False,
automatic_membership_email_domains="",
grant_trust_level=1,
flair_url=None,
flair_bg_color=None,
flair_color=None,
**kwargs
):
"""
Args:
@@ -840,20 +891,22 @@ class DiscourseClient(object):
flair_color: Avatar Flair Color
"""
kwargs['name'] = name
kwargs['title'] = title
kwargs['visible'] = visible
kwargs['alias_level'] = alias_level
kwargs['automatic_membership_retroactive'] = automatic_membership_retroactive
kwargs['primary_group'] = primary_group
kwargs['automatic'] = automatic
kwargs['automatic_membership_email_domains'] = automatic_membership_email_domains
kwargs['grant_trust_level'] = grant_trust_level
kwargs['flair_url'] = flair_url
kwargs['flair_bg_color'] = flair_bg_color
kwargs['flair_color'] = flair_color
kwargs["name"] = name
kwargs["title"] = title
kwargs["visible"] = visible
kwargs["alias_level"] = alias_level
kwargs["automatic_membership_retroactive"] = automatic_membership_retroactive
kwargs["primary_group"] = primary_group
kwargs["automatic"] = automatic
kwargs[
"automatic_membership_email_domains"
] = automatic_membership_email_domains
kwargs["grant_trust_level"] = grant_trust_level
kwargs["flair_url"] = flair_url
kwargs["flair_bg_color"] = flair_bg_color
kwargs["flair_color"] = flair_color
# Discourse v.1.7.0
kwargs = {'group': kwargs}
kwargs = {"group": kwargs}
return self._post("/admin/groups", json=True, **kwargs)
@@ -882,7 +935,9 @@ class DiscourseClient(object):
JSON API response
"""
return self._put("/admin/groups/{0}/owners.json".format(groupid), usernames=username)
return self._put(
"/admin/groups/{0}/owners.json".format(groupid), usernames=username
)
def delete_group_owner(self, groupid, userid):
"""
@@ -898,22 +953,24 @@ class DiscourseClient(object):
JSON API response
"""
return self._delete("/admin/groups/{0}/owners.json".format(groupid), user_id=userid)
return self._delete(
"/admin/groups/{0}/owners.json".format(groupid), user_id=userid
)
def group_owners(self, group_name):
"""
Get all owners of a group by group name
"""
group = self._get("/groups/{0}/members.json".format(group_name))
return group['owners']
return group["owners"]
def group_members(self, group_name, offset=0, **kwargs):
"""
Get all members of a group by group name
"""
kwargs['offset'] = offset
kwargs["offset"] = offset
group = self._get("/groups/{0}/members.json".format(group_name), **kwargs)
return group['members']
return group["members"]
def add_group_member(self, groupid, username):
"""
@@ -930,7 +987,9 @@ class DiscourseClient(object):
DiscourseError if user is already member of group
"""
return self._put("/admin/groups/{0}/members.json".format(groupid), usernames=username)
return self._put(
"/admin/groups/{0}/members.json".format(groupid), usernames=username
)
def add_group_members(self, groupid, usernames):
"""
@@ -947,9 +1006,10 @@ class DiscourseClient(object):
DiscourseError if any of the users is already member of group
"""
usernames = ','.join(usernames)
return self._put("/admin/groups/{0}/members.json".format(groupid), usernames=usernames)
usernames = ",".join(usernames)
return self._put(
"/admin/groups/{0}/members.json".format(groupid), usernames=usernames
)
def add_user_to_group(self, groupid, userid):
"""
@@ -982,7 +1042,9 @@ class DiscourseClient(object):
JSON API response
"""
return self._delete("/admin/groups/{0}/members.json".format(groupid), user_id=userid)
return self._delete(
"/admin/groups/{0}/members.json".format(groupid), user_id=userid
)
def color_schemes(self, **kwargs):
"""
@@ -994,7 +1056,7 @@ class DiscourseClient(object):
Returns:
"""
return self._get('/admin/color_schemes.json', **kwargs)
return self._get("/admin/color_schemes.json", **kwargs)
def create_color_scheme(self, name, enabled, colors, **kwargs):
"""
@@ -1009,14 +1071,15 @@ class DiscourseClient(object):
Returns:
"""
kwargs['name'] = name
kwargs["name"] = name
if bool(enabled):
kwargs['enabled'] = 'true'
kwargs["enabled"] = "true"
else:
kwargs['enabled'] = 'false'
kwargs['colors'] = [{'name': name, 'hex': color}
for name, color in colors.items()]
kwargs = {'color_scheme': kwargs}
kwargs["enabled"] = "false"
kwargs["colors"] = [
{"name": name, "hex": color} for name, color in colors.items()
]
kwargs = {"color_scheme": kwargs}
return self._post("/admin/color_schemes.json", json=True, **kwargs)
def create_site_customization(self, name, enabled, stylesheet, **kwargs):
@@ -1032,13 +1095,13 @@ class DiscourseClient(object):
Returns:
"""
kwargs['name'] = name
kwargs["name"] = name
if bool(enabled):
kwargs['enabled'] = 'true'
kwargs["enabled"] = "true"
else:
kwargs['enabled'] = 'false'
kwargs['stylesheet'] = stylesheet
kwargs = {'site_customization': kwargs}
kwargs["enabled"] = "false"
kwargs["stylesheet"] = stylesheet
kwargs = {"site_customization": kwargs}
return self._post("/admin/site_customizations", json=True, **kwargs)
def trust_level_lock(self, user_id, locked, **kwargs):
@@ -1054,10 +1117,10 @@ class DiscourseClient(object):
"""
if bool(locked):
kwargs['locked'] = 'true'
kwargs["locked"] = "true"
else:
kwargs['locked'] = 'false'
return self._put('/admin/users/{}/trust_level_lock'.format(user_id), **kwargs)
kwargs["locked"] = "false"
return self._put("/admin/users/{}/trust_level_lock".format(user_id), **kwargs)
def block(self, user_id, **kwargs):
"""
@@ -1070,7 +1133,7 @@ class DiscourseClient(object):
Returns:
"""
return self._put('/admin/users/{}/block'.format(user_id), **kwargs)
return self._put("/admin/users/{}/block".format(user_id), **kwargs)
def upload_image(self, image, type, synchronous, **kwargs):
"""
@@ -1086,13 +1149,13 @@ class DiscourseClient(object):
Returns:
"""
kwargs['type'] = type
kwargs["type"] = type
if bool(synchronous):
kwargs['synchronous'] = 'true'
kwargs["synchronous"] = "true"
else:
kwargs['synchronous'] = 'false'
files = {'file': open(image, 'rb')}
return self._post('/uploads.json', files=files, **kwargs)
kwargs["synchronous"] = "false"
files = {"file": open(image, "rb")}
return self._post("/uploads.json", files=files, **kwargs)
def user_actions(self, username, filter, offset=0, **kwargs):
"""
@@ -1106,10 +1169,10 @@ class DiscourseClient(object):
Returns:
"""
kwargs['username'] = username
kwargs['filter'] = filter
kwargs['offset'] = offset
return self._get('/user_actions.json', **kwargs)['user_actions']
kwargs["username"] = username
kwargs["filter"] = filter
kwargs["offset"] = offset
return self._get("/user_actions.json", **kwargs)["user_actions"]
def tag_group(self, name, tag_names, parent_tag_name=None, **kwargs):
"""
@@ -1124,10 +1187,10 @@ class DiscourseClient(object):
Returns:
"""
kwargs['name'] = name
kwargs['tag_names'] = tag_names
kwargs['parent_tag_name'] = parent_tag_name
return self._post('/tag_groups', json=True, **kwargs)['tag_group']
kwargs["name"] = name
kwargs["tag_names"] = tag_names
kwargs["parent_tag_name"] = parent_tag_name
return self._post("/tag_groups", json=True, **kwargs)["tag_group"]
def _get(self, path, **kwargs):
"""
@@ -1153,6 +1216,7 @@ class DiscourseClient(object):
"""
if not json:
return self._request(PUT, path, data=kwargs)
else:
return self._request(PUT, path, json=kwargs)
@@ -1168,6 +1232,7 @@ class DiscourseClient(object):
"""
if not json:
return self._request(POST, path, files=files, data=kwargs)
else:
return self._request(POST, path, files=files, json=kwargs)
@@ -1195,25 +1260,33 @@ class DiscourseClient(object):
Returns:
"""
params['api_key'] = self.api_key
if 'api_username' not in params:
params['api_username'] = self.api_username
params["api_key"] = self.api_key
if "api_username" not in params:
params["api_username"] = self.api_username
url = self.host + path
headers = {'Accept': 'application/json; charset=utf-8'}
headers = {"Accept": "application/json; charset=utf-8"}
response = requests.request(
verb, url, allow_redirects=False, params=params, files=files, data=data, json=json, headers=headers,
timeout=self.timeout)
verb,
url,
allow_redirects=False,
params=params,
files=files,
data=data,
json=json,
headers=headers,
timeout=self.timeout,
)
log.debug('response %s: %s', response.status_code, repr(response.text))
log.debug("response %s: %s", response.status_code, repr(response.text))
if not response.ok:
try:
msg = u','.join(response.json()['errors'])
msg = u",".join(response.json()["errors"])
except (ValueError, TypeError, KeyError):
if response.reason:
msg = response.reason
else:
msg = u'{0}: {1}'.format(response.status_code, response.text)
msg = u"{0}: {1}".format(response.status_code, response.text)
if 400 <= response.status_code < 500:
raise DiscourseClientError(msg, response=response)
@@ -1222,27 +1295,32 @@ class DiscourseClient(object):
if response.status_code == 302:
raise DiscourseError(
'Unexpected Redirect, invalid api key or host?', response=response)
"Unexpected Redirect, invalid api key or host?", response=response
)
json_content = 'application/json; charset=utf-8'
content_type = response.headers['content-type']
json_content = "application/json; charset=utf-8"
content_type = response.headers["content-type"]
if content_type != json_content:
# some calls return empty html documents
if not response.content.strip():
return None
raise DiscourseError('Invalid Response, expecting "{0}" got "{1}"'.format(
json_content, content_type), response=response)
raise DiscourseError(
'Invalid Response, expecting "{0}" got "{1}"'.format(
json_content, content_type
),
response=response,
)
try:
decoded = response.json()
except ValueError:
raise DiscourseError('failed to decode response', response=response)
raise DiscourseError("failed to decode response", response=response)
if 'errors' in decoded:
message = decoded.get('message')
if "errors" in decoded:
message = decoded.get("message")
if not message:
message = u','.join(decoded['errors'])
message = u",".join(decoded["errors"])
raise DiscourseError(message, response=response)
return decoded
pydiscourse/main.py
+20 -16
View File
@@ -12,30 +12,32 @@ from pydiscourse.client import DiscourseClient, DiscourseError
class DiscourseCmd(cmd.Cmd):
prompt = 'discourse>'
prompt = "discourse>"
output = sys.stdout
def __init__(self, client):
cmd.Cmd.__init__(self)
self.client = client
self.prompt = '%s>' % self.client.host
self.prompt = "%s>" % self.client.host
def __getattr__(self, attr):
if attr.startswith('do_'):
if attr.startswith("do_"):
method = getattr(self.client, attr[3:])
def wrapper(arg):
args = arg.split()
kwargs = dict(a.split('=') for a in args if '=' in a)
args = [a for a in args if '=' not in a]
kwargs = dict(a.split("=") for a in args if "=" in a)
args = [a for a in args if "=" not in a]
try:
return method(*args, **kwargs)
except DiscourseError as e:
print(e, e.response.text)
return e.response
return wrapper
elif attr.startswith('help_'):
elif attr.startswith("help_"):
method = getattr(self.client, attr[5:])
def wrapper():
@@ -47,24 +49,26 @@ class DiscourseCmd(cmd.Cmd):
def postcmd(self, result, line):
try:
json.dump(result, self.output, sort_keys=True, indent=4, separators=(',', ': '))
json.dump(
result, self.output, sort_keys=True, indent=4, separators=(",", ": ")
)
except TypeError:
self.output.write(result.text)
def main():
op = optparse.OptionParser()
op.add_option('--host', default='http://localhost:4000')
op.add_option('--api-user', default='system')
op.add_option('-v', '--verbose', action='store_true')
op.add_option("--host", default="http://localhost:4000")
op.add_option("--api-user", default="system")
op.add_option("-v", "--verbose", action="store_true")
options, args = op.parse_args()
if not options.host.startswith('http'):
op.error('host must include protocol, eg http://')
if not options.host.startswith("http"):
op.error("host must include protocol, eg http://")
api_key = os.environ.get('DISCOURSE_API_KEY')
api_key = os.environ.get("DISCOURSE_API_KEY")
if not api_key:
op.error('please set DISCOURSE_API_KEY')
op.error("please set DISCOURSE_API_KEY")
client = DiscourseClient(options.host, options.api_user, api_key)
@@ -74,12 +78,12 @@ def main():
c = DiscourseCmd(client)
if args:
line = ' '.join(args)
line = " ".join(args)
result = c.onecmd(line)
c.postcmd(result, line)
else:
c.cmdloop()
if __name__ == '__main__':
if __name__ == "__main__":
main()
pydiscourse/sso.py
+23 -19
View File
@@ -45,34 +45,36 @@ def sso_validate(payload, signature, secret):
return value: The nonce used by discourse to validate the redirect URL
"""
if None in [payload, signature]:
raise DiscourseError('No SSO payload or signature.')
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..')
decoded = b64decode(payload.encode("utf-8")).decode("utf-8")
if "nonce" not in decoded:
raise DiscourseError("Invalid payload..")
h = hmac.new(secret.encode('utf-8'), payload.encode('utf-8'), digestmod=hashlib.sha256)
h = hmac.new(
secret.encode("utf-8"), payload.encode("utf-8"), digestmod=hashlib.sha256
)
this_signature = h.hexdigest()
if this_signature != signature:
raise DiscourseError('Payload does not match signature.')
raise DiscourseError("Payload does not match signature.")
# Discourse returns querystring encoded value. We only need `nonce`
qs = parse_qs(decoded)
return qs['nonce'][0]
return qs["nonce"][0]
def sso_payload(secret, **kwargs):
return_payload = b64encode(urlencode(kwargs).encode('utf-8'))
h = hmac.new(secret.encode('utf-8'), return_payload, digestmod=hashlib.sha256)
query_string = urlencode({'sso': return_payload, 'sig': h.hexdigest()})
return_payload = b64encode(urlencode(kwargs).encode("utf-8"))
h = hmac.new(secret.encode("utf-8"), return_payload, digestmod=hashlib.sha256)
query_string = urlencode({"sso": return_payload, "sig": h.hexdigest()})
return query_string
@@ -86,11 +88,13 @@ def sso_redirect_url(nonce, secret, email, external_id, username, **kwargs):
return value: URL to redirect users back to discourse, now logged in as user_username
"""
kwargs.update({
'nonce': nonce,
'email': email,
'external_id': external_id,
'username': username
})
kwargs.update(
{
"nonce": nonce,
"email": email,
"external_id": external_id,
"username": username,
}
)
return '/session/sso_login?%s' % sso_payload(secret, **kwargs)
return "/session/sso_login?%s" % sso_payload(secret, **kwargs)
tests/test_client.py
+70 -53
View File
@@ -5,18 +5,25 @@ import mock
from pydiscourse import client
import sys
if sys.version_info < (3,):
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'})
request.return_value = mock.MagicMock(
headers={"content-type": "application/json; charset=utf-8"}
)
class ClientBaseTestCase(unittest.TestCase):
@@ -25,9 +32,9 @@ class ClientBaseTestCase(unittest.TestCase):
"""
def setUp(self):
self.host = 'http://testhost'
self.api_username = 'testuser'
self.api_key = 'testkey'
self.host = "http://testhost"
self.api_username = "testuser"
self.api_key = "testkey"
self.client = client.DiscourseClient(self.host, self.api_username, self.api_key)
@@ -39,28 +46,27 @@ class ClientBaseTestCase(unittest.TestCase):
self.assertEqual(args[0], verb)
self.assertEqual(args[1], self.host + url)
kwargs = kwargs['params']
self.assertEqual(kwargs.pop('api_username'), self.api_username)
self.assertEqual(kwargs.pop('api_key'), self.api_key)
kwargs = kwargs["params"]
self.assertEqual(kwargs.pop("api_username"), self.api_username)
self.assertEqual(kwargs.pop("api_key"), self.api_key)
if verb == 'GET':
if verb == "GET":
self.assertEqual(kwargs, params)
class TestClientRequests(ClientBaseTestCase):
"""
Tests for common request handling
"""
@mock.patch('pydiscourse.client.requests')
@mock.patch("pydiscourse.client.requests")
def test_empty_content_http_ok(self, mocked_requests):
"""Empty content should not raise error
Critical to test against *bytestrings* rather than unicode
"""
mocked_response = mock.MagicMock()
mocked_response.content = b(' ')
mocked_response.content = b(" ")
mocked_response.status_code = 200
mocked_response.headers = {"content-type": "text/plain; charset=utf-8"}
@@ -69,126 +75,137 @@ class TestClientRequests(ClientBaseTestCase):
mocked_requests.request = mock.MagicMock()
mocked_requests.request.return_value = mocked_response
resp = self.client._request('GET', '/users/admin/1/unsuspend', {})
resp = self.client._request("GET", "/users/admin/1/unsuspend", {})
self.assertIsNone(resp)
@mock.patch('requests.request')
@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')
self.client.user("someuser")
self.assertRequestCalled(request, "GET", "/users/someuser.json")
def test_create_user(self, request):
prepare_response(request)
self.client.create_user('Test User', 'testuser', 'test@example.com', 'notapassword')
self.client.create_user(
"Test User", "testuser", "test@example.com", "notapassword"
)
self.assertEqual(request.call_count, 2)
# XXX incomplete
# XXX incomplete
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)
email = "test@example.com"
self.client.update_email("someuser", email)
self.assertRequestCalled(
request, "PUT", "/users/someuser/preferences/email", email=email
)
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')
self.client.update_user("someuser", a="a", b="b")
self.assertRequestCalled(request, "PUT", "/users/someuser", a="a", b="b")
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')
self.client.update_username("someuser", "newname")
self.assertRequestCalled(
request, "PUT", "/users/someuser/preferences/username", username="newname"
)
def test_by_external_id(self, request):
prepare_response(request)
self.client.by_external_id(123)
self.assertRequestCalled(request, 'GET',
'/users/by-external/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")
self.assertRequestCalled(
request, "PUT", "/admin/users/123/suspend", duration=1, reason="Testing"
)
def test_unsuspend_user(self, request):
prepare_response(request)
self.client.unsuspend(123)
self.assertRequestCalled(request, 'PUT', '/admin/users/123/unsuspend')
self.assertRequestCalled(request, "PUT", "/admin/users/123/unsuspend")
def test_user_bagdes(self, request):
prepare_response(request)
self.client.user_badges('username')
self.assertRequestCalled(request, 'GET', '/user-badges/{}.json'.format('username'))
self.client.user_badges("username")
self.assertRequestCalled(
request, "GET", "/user-badges/{}.json".format("username")
)
@mock.patch('requests.request')
@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')
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')
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')
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')
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'])
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'
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)
self.assertRequestCalled(
request, "POST", "/t/22/invite.json", email=email, topic_id=22
)
@mock.patch('requests.request')
@mock.patch("requests.request")
class MiscellaneousTests(ClientBaseTestCase):
def test_search(self, request):
prepare_response(request)
self.client.search('needle')
self.assertRequestCalled(request, 'GET', '/search.json', term='needle')
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'])
self.assertRequestCalled(request, "GET", "/categories.json")
self.assertEqual(r, request().json()["category_list"]["categories"])
def test_users(self, request):
prepare_response(request)
self.client.users()
self.assertRequestCalled(request, 'GET', '/admin/users/list/active.json')
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')
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)
self.client.grant_badge_to("username", 1)
self.assertRequestCalled(
request, "POST", "/user_badges", username="username", badge_id=1
)
tests/test_sso.py
+35 -24
View File
@@ -19,56 +19,67 @@ from pydiscourse.exceptions import DiscourseError
class SSOTestCase(unittest.TestCase):
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'
self.secret = "d836444a9e4084d5b224a60c208dce14"
self.nonce = "cb68251eefb5211e58c00ff1395f0c0b"
self.payload = "bm9uY2U9Y2I2ODI1MWVlZmI1MjExZTU4YzAwZmYxMzk1ZjBjMGI%3D%0A"
self.signature = "2828aa29899722b35a2f191d34ef9b3ce695e0e6eeec47deb46d588d70c7cb56"
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'
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)
with self.assertRaises(DiscourseError):
sso.sso_validate('', self.signature, self.secret)
sso.sso_validate("", self.signature, self.secret)
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)
sso.sso_validate(self.payload, "notavalidsignature", self.secret)
def test_valid_nonce(self):
nonce = sso.sso_validate(self.payload, self.signature, self.secret)
self.assertEqual(nonce, self.nonce)
def test_valid_redirect_url(self):
url = sso.sso_redirect_url(self.nonce, self.secret, self.email, self.external_id, self.username, name='sam')
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])
self.assertIn("/session/sso_login", 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], self.secret)
payload = params["sso"][0]
sso.sso_validate(payload, params["sig"][0], self.secret)
# check the params have all the data we expect
payload = b64decode(payload.encode('utf-8')).decode('utf-8')
payload = b64decode(payload.encode("utf-8")).decode("utf-8")
payload = unquote(payload)
payload = dict((p.split('=') for p in payload.split('&')))
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
})
self.assertEqual(
payload,
{
"username": self.username,
"nonce": self.nonce,
"external_id": self.external_id,
"name": self.name,
"email": self.email,
},
)