Allow admins to end polls (closes #38)

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': {

mattermost_api.py

@@ -8,25 +8,66 @@
 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}
         url = settings.MATTERMOST_URL + '/api/v4/users/' + 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:

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.

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

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