From 325ce181ee3f5e6148aaecafc2b45da952e6058a Mon Sep 17 00:00:00 2001 From: M-Mueller Date: Mon, 12 Aug 2019 22:18:02 +0200 Subject: [PATCH] Allow admins to end polls (closes #38) --- app.py | 13 ++++++-- mattermost_api.py | 53 ++++++++++++++++++++++++++++---- settings.py.example | 4 ++- tests/test_interface.py | 40 +++++++++++++++++++++++++ tests/test_mattermost_api.py | 58 ++++++++++++++++++++++++++++++++++++ 5 files changed, 158 insertions(+), 10 deletions(-) diff --git a/app.py b/app.py index 8db9bc3..b7021c9 100644 --- a/app.py +++ b/app.py @@ -9,7 +9,7 @@ from poll import Poll, NoMoreVotesError, InvalidPollError from formatters import format_help, format_poll, format_user_vote -from mattermost_api import user_locale +from mattermost_api import user_locale, is_admin_user, is_team_admin import settings @@ -291,6 +291,7 @@ def end_poll(): """ json = request.get_json() user_id = json['user_id'] + team_id = json['team_id'] poll_id = json['context']['poll_id'] request.user_id = user_id @@ -303,8 +304,14 @@ def end_poll(): }) app.logger.info('Ending poll "%s"', poll_id) - if user_id == poll.creator_id: - # only the creator may end a poll + + # only the creator and admins may end a poll + can_end_poll = \ + user_id == poll.creator_id or \ + is_admin_user(user_id) or \ + is_team_admin(user_id=user_id, team_id=team_id) + + if can_end_poll: poll.end() return jsonify({ 'update': { diff --git a/mattermost_api.py b/mattermost_api.py index 5216ffe..69f6cf1 100644 --- a/mattermost_api.py +++ b/mattermost_api.py @@ -8,10 +8,10 @@ logger = logging.getLogger('flask.app') -def user_locale(user_id): - """Returns the locale of the user with the given user_id.""" +def get_user(user_id): + """Return the json data of the user.""" if not settings.MATTERMOST_PA_TOKEN: - return "en" + return {} try: header = {'Authorization': 'Bearer ' + settings.MATTERMOST_PA_TOKEN} @@ -19,14 +19,55 @@ def user_locale(user_id): r = requests.get(url, headers=header) if r.ok: - locale = json.loads(r.text)['locale'] - if locale: - return locale + return json.loads(r.text) except KeyError as e: logger.error(e) + return {} + + +def user_locale(user_id): + """Return the locale of the user with the given user_id.""" + if not settings.MATTERMOST_PA_TOKEN: + return "en" + + user = get_user(user_id) + if 'locale' in user: + locale = user['locale'] + if locale: + return locale + return "en" +def is_admin_user(user_id): + """Return whether the user is an admin.""" + + user = get_user(user_id) + if 'roles' in user: + return 'system_admin' in user['roles'] + + return False + + +def is_team_admin(user_id, team_id): + """Return whether the user is an admin in the given team.""" + if not settings.MATTERMOST_PA_TOKEN: + return False + + try: + header = {'Authorization': 'Bearer ' + settings.MATTERMOST_PA_TOKEN} + url = settings.MATTERMOST_URL + '/api/v4/teams/' + team_id + '/members/' + user_id + + r = requests.get(url, headers=header) + if r.ok: + roles = json.loads(r.text)['roles'] + return 'team_admin' in roles + except KeyError as e: + logger.error(e) + + return False + + def resolve_usernames(user_ids): """Resolve the list of user ids to list of user names.""" if len(user_ids) == 0: diff --git a/settings.py.example b/settings.py.example index 11d52f7..99f608d 100644 --- a/settings.py.example +++ b/settings.py.example @@ -16,8 +16,10 @@ MATTERMOST_TOKENS = None # URL of the Mattermost server MATTERMOST_URL = 'http://localhost' -# Optional URL to the image used for the bars. +# URL to the image used for the bars. # If set to None, the poll server will provide the one in img/bar.png. +# None will only work since Mattermost 5.8 with Image Proxy enabled. +# If None does not work for you, try 'https://raw.githubusercontent.com/M-Mueller/mattermost-poll/master/img/bar.png' BAR_IMG_URL = None # Private access token of some user. diff --git a/tests/test_interface.py b/tests/test_interface.py index 75d3be9..8a50695 100644 --- a/tests/test_interface.py +++ b/tests/test_interface.py @@ -189,6 +189,7 @@ def test_end(base_url, client, votes, expected): context = action_contexts[-1] data = json.dumps({ 'user_id': 'user0', + 'team_id': 'team0', 'context': context }) response = client.post(action_urls[-1], data=data, @@ -217,6 +218,7 @@ def test_end_wrong_user(base_url, client): context = action_contexts[-1] data = json.dumps({ 'user_id': 'user1', + 'team_id': 'team0', 'context': context }) response = client.post(action_urls[-1], data=data, @@ -230,6 +232,43 @@ def test_end_wrong_user(base_url, client): assert rd['ephemeral_text'] == 'You are not allowed to end this poll' +def patched_is_admin_user(user_id): + if user_id == 'user1': + return True + return False + + +def test_end_admin(mocker, base_url, client): + mocker.patch('app.is_admin_user', new=patched_is_admin_user) + + # create a new poll + data = { + 'user_id': 'user0', + 'text': 'Message' + } + response = client.post('/', data=data, base_url=base_url) + rd = json.loads(response.data.decode('utf-8')) + + actions = rd['attachments'][0]['actions'] + action_urls = [a['integration']['url'].replace(base_url, '') + for a in actions] + action_contexts = [a['integration']['context'] for a in actions] + + context = action_contexts[-1] + data = json.dumps({ + 'user_id': 'user1', + 'team_id': 'team0', + 'context': context + }) + response = client.post(action_urls[-1], data=data, + content_type='application/json', + base_url=base_url) + assert response.status_code == 200 + + rd = json.loads(response.data.decode('utf-8')) + __validate_end_response(rd, 'Message', ['Yes', 'No']) + + def test_vote_invalid_poll(base_url, client): data = json.dumps({ 'user_id': 'user0', @@ -251,6 +290,7 @@ def test_vote_invalid_poll(base_url, client): def test_end_invalid_poll(base_url, client): data = json.dumps({ 'user_id': 'user0', + 'team_id': 'team0', 'context': { 'poll_id': 'invalid123', 'vote': 0 diff --git a/tests/test_mattermost_api.py b/tests/test_mattermost_api.py index 78317fd..9929079 100644 --- a/tests/test_mattermost_api.py +++ b/tests/test_mattermost_api.py @@ -63,3 +63,61 @@ def requests_mock(url, headers): actual_locale = mattermost_api.user_locale('user1') assert actual_locale == 'en' + + +@pytest.mark.usefixtures('set_pa_token') +@pytest.mark.parametrize("user_id, admin", [ + ('user1', False), + ('user2', False), + ('user3', True), + ('user4', True), + ('invalid', False), +]) +def test_user_is_admin(mocker, user_id, admin): + def requests_mock(url, headers): + assert url == 'http://www.example.com/api/v4/users/' + user_id + assert headers['Authorization'] == 'Bearer 123abc456xyz' + if user_id == 'user1': + return Response(True, json.dumps({'roles': ['admin', 'whatever']})) + if user_id == 'user2': + return Response(True, json.dumps({'roles': ['team_admin']})) + if user_id == 'user3': + return Response(True, json.dumps({'roles': ['whatever', 'system_admin']})) + if user_id == 'user4': + return Response(True, json.dumps({'roles': ['whatever', 'team_admin', 'system_admin']})) + if user_id == 'invalid': + return Response(True, json.dumps({})) + assert False + + mocker.patch('requests.get', new=requests_mock) + + assert mattermost_api.is_admin_user(user_id) is admin + + +@pytest.mark.usefixtures('set_pa_token') +@pytest.mark.parametrize("user_id, team_admin", [ + ('user1', False), + ('user2', True), + ('user3', False), + ('user4', True), + ('invalid', False), +]) +def test_user_is_team_admin(mocker, user_id, team_admin): + def requests_mock(url, headers): + assert url == 'http://www.example.com/api/v4/teams/myteam/members/' + user_id + assert headers['Authorization'] == 'Bearer 123abc456xyz' + if user_id == 'user1': + return Response(True, json.dumps({'roles': ['admin', 'whatever']})) + if user_id == 'user2': + return Response(True, json.dumps({'roles': ['team_admin']})) + if user_id == 'user3': + return Response(True, json.dumps({'roles': ['whatever', 'system_admin']})) + if user_id == 'user4': + return Response(True, json.dumps({'roles': ['whatever', 'team_admin', 'system_admin']})) + if user_id == 'invalid': + return Response(True, json.dumps({})) + assert False + + mocker.patch('requests.get', new=requests_mock) + + assert mattermost_api.is_team_admin(user_id, 'myteam') is team_admin