Allura/allura/tests/functional/test_rest.py

#       Licensed to the Apache Software Foundation (ASF) under one
#       or more contributor license agreements.  See the NOTICE file
#       distributed with this work for additional information
#       regarding copyright ownership.  The ASF licenses this file
#       to you under the Apache License, Version 2.0 (the
#       "License"); you may not use this file except in compliance
#       with the License.  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#       Unless required by applicable law or agreed to in writing,
#       software distributed under the License is distributed on an
#       "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#       KIND, either express or implied.  See the License for the
#       specific language governing permissions and limitations
#       under the License.


import json

import tg
from bson import ObjectId
from tg import app_globals as g
import mock
from alluratest.tools import assert_equal, assert_in, assert_not_in
from ming.odm import ThreadLocalODMSession
from tg import config

from allura.tests import decorators as td
from alluratest.controller import TestRestApiBase
from allura.lib import helpers as h
from allura.lib.exceptions import Invalid
from allura import model as M


@with_nose_compatibility
class TestRestHome(TestRestApiBase):

    def _patch_token(self, OAuthAccessToken):
        at = OAuthAccessToken.return_value
        at.__ming__ = mock.MagicMock()
        at.api_key = 'foo'

    @mock.patch('allura.controllers.rest.M.OAuthAccessToken')
    @mock.patch('allura.controllers.rest.request')
    def test_bearer_token_non_bearer(self, request, OAuthAccessToken):
        request.headers = {}
        request.params = {'access_token': 'foo'}
        request.scheme = 'https'
        request.path = '/rest/p/test/wiki'
        self._patch_token(OAuthAccessToken)
        access_token = OAuthAccessToken.query.get.return_value
        access_token.is_bearer = False
        r = self.api_post('/rest/p/test/wiki', access_token='foo', status=401)
        OAuthAccessToken.query.get.assert_called_once_with(api_key='foo')

    @mock.patch('allura.controllers.rest.M.OAuthAccessToken')
    @mock.patch('allura.controllers.rest.request')
    def test_bearer_token_invalid(self, request, OAuthAccessToken):
        request.headers = {}
        request.params = {'access_token': 'foo'}
        request.scheme = 'https'
        request.path = '/rest/p/test/wiki'
        self._patch_token(OAuthAccessToken)
        OAuthAccessToken.query.get.return_value = None
        r = self.api_post('/rest/p/test/wiki', access_token='foo', status=401)

    @mock.patch('allura.controllers.rest.request')
    @td.with_wiki
    def test_bearer_token_valid(self, request):
        user = M.User.by_username('test-admin')
        consumer_token = M.OAuthConsumerToken(
            name='foo',
            description='foo app',
        )
        request_token = M.OAuthRequestToken(
            consumer_token_id=consumer_token._id,
            user_id=user._id,
            callback='manual',
            validation_pin=h.nonce(20),
            is_bearer=True,
        )
        access_token = M.OAuthAccessToken(
            consumer_token_id=consumer_token._id,
            request_token_id=request_token._id,
            user_id=user._id,
            is_bearer=True,
        )
        ThreadLocalODMSession.flush_all()
        request.headers = {}
        request.params = {'access_token': access_token.api_key}
        request.scheme = 'https'
        request.path = '/rest/p/test/wiki'
        r = self.api_post('/rest/p/test/wiki', access_token='foo')
        assert r.status_int == 200

    @mock.patch('allura.controllers.rest.M.OAuthAccessToken')
    @mock.patch('allura.controllers.rest.request')
    def test_bearer_token_non_bearer_via_headers(self, request, OAuthAccessToken):
        request.headers = {
            'Authorization': 'Bearer foo'
        }
        request.scheme = 'https'
        request.path = '/rest/p/test/wiki'
        self._patch_token(OAuthAccessToken)
        access_token = OAuthAccessToken.query.get.return_value
        access_token.is_bearer = False
        r = self.api_post('/rest/p/test/wiki', access_token='foo', status=401)
        OAuthAccessToken.query.get.assert_called_once_with(api_key='foo')

    @mock.patch('allura.controllers.rest.M.OAuthAccessToken')
    @mock.patch('allura.controllers.rest.request')
    def test_bearer_token_invalid_via_headers(self, request, OAuthAccessToken):
        request.headers = {
            'Authorization': 'Bearer foo'
        }
        request.scheme = 'https'
        request.path = '/rest/p/test/wiki'
        self._patch_token(OAuthAccessToken)
        OAuthAccessToken.query.get.return_value = None
        r = self.api_post('/rest/p/test/wiki', access_token='foo', status=401)

    @mock.patch('allura.controllers.rest.request')
    @td.with_wiki
    @mock.patch.dict(config, debug=False)
    def test_bearer_token_valid_via_headers(self, request):
        user = M.User.by_username('test-admin')
        consumer_token = M.OAuthConsumerToken(
            name='foo',
            description='foo app',
        )
        request_token = M.OAuthRequestToken(
            consumer_token_id=consumer_token._id,
            user_id=user._id,
            callback='manual',
            validation_pin=h.nonce(20),
            is_bearer=True,
        )
        access_token = M.OAuthAccessToken(
            consumer_token_id=consumer_token._id,
            request_token_id=request_token._id,
            user_id=user._id,
            is_bearer=True,
        )
        ThreadLocalODMSession.flush_all()
        token = access_token.api_key
        request.headers = {
            'Authorization': f'Bearer {token}'
        }
        request.scheme = 'https'
        request.path = '/rest/p/test/wiki'
        r = self.api_post('/rest/p/test/wiki', access_token='foo', status=200)
        # reverse proxy situation
        request.scheme = 'http'
        request.environ['paste.testing'] = False
        request.environ['HTTP_X_FORWARDED_PROTOx'] = 'https'
        r = self.api_post('/rest/p/test/wiki', access_token='foo', status=200)

    def test_bad_path(self):
        r = self.api_post('/rest/1/test/wiki/', status=404)
        r = self.api_post('/rest/p/1223/wiki/', status=404)
        r = self.api_post('/rest/p/test/12wiki/', status=404)

    def test_no_api(self):
        r = self.api_post('/rest/p/test/admin/', status=404)

    @td.with_wiki
    def test_project_ping(self):
        r = self.api_get('/rest/p/test/wiki/Home/')
        assert r.status_int == 200
        assert r.json['title'] == 'Home', r.json

    @td.with_tool('test/sub1', 'Wiki', 'wiki')
    def test_subproject_ping(self):
        r = self.api_get('/rest/p/test/sub1/wiki/Home/')
        assert r.status_int == 200
        assert r.json['title'] == 'Home', r.json

    def test_project_code(self):
        r = self.api_get('/rest/p/test/')
        assert r.status_int == 200

    def test_project_data(self):
        r = self.api_get('/rest/p/test/')
        assert r.json['shortname'] == 'test'
        assert r.json['name'] == 'Test Project'
        assert len(r.json['developers']) == 1
        admin_dev = r.json['developers'][0]
        assert admin_dev['username'] == 'test-admin'
        assert admin_dev['name'] == 'Test Admin'
        assert admin_dev['url'] == 'http://localhost/u/test-admin/'

    @td.with_tool('test', 'Tickets', 'bugs')
    @td.with_tool('test', 'Tickets', 'private-bugs')
    def test_project_data_tools(self):
        # Deny anonymous to see 'private-bugs' tool
        role = M.ProjectRole.by_name('*anonymous')._id
        read_permission = M.ACE.allow(role, 'read')
        app = M.Project.query.get(
            shortname='test').app_instance('private-bugs')
        if read_permission in app.config.acl:
            app.config.acl.remove(read_permission)

        # admin sees both 'Tickets' tools
        r = self.api_get('/rest/p/test/')
        assert r.json['shortname'] == 'test'
        tool_mounts = [t['mount_point'] for t in r.json['tools']]
        assert 'bugs' in tool_mounts
        assert 'private-bugs' in tool_mounts

        # anonymous sees only non-private tool
        r = self.app.get('/rest/p/test/',
                         extra_environ={'username': '*anonymous'})
        assert r.json['shortname'] == 'test'
        tool_mounts = [t['mount_point'] for t in r.json['tools']]
        assert 'bugs' in tool_mounts
        assert 'private-bugs' not in tool_mounts

    def test_neighborhood_has_access_no_params(self):
        r = self.api_get('/rest/p/has_access', status=404)
        r = self.api_get('/rest/p/has_access?user=test-admin', status=404)
        r = self.api_get('/rest/p/has_access?perm=read', status=404)

    def test_neighborhood_has_access_unknown_params(self):
        """Unknown user and/or permission always False for has_access API"""
        r = self.api_get(
            '/rest/p/has_access?user=babadook&perm=read',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == False
        r = self.api_get(
            '/rest/p/has_access?user=test-admin&perm=jump',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == False

    def test_neighborhood_has_access_not_admin(self):
        """
        User which has no 'admin' permission on neighborhood can't use
        has_access API
        """
        self.api_get(
            '/rest/p/has_access?user=test-admin&perm=admin',
            user='test-user',
            status=403)

    def test_neighborhood_has_access(self):
        r = self.api_get(
            '/rest/p/has_access?user=root&perm=update',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == True
        r = self.api_get(
            '/rest/p/has_access?user=test-user&perm=update',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == False

    def test_neighborhood(self):
        self.api_get('/rest/p/', status=404)

    def test_neighborhood_tools(self):
        r = self.api_get('/rest/p/wiki/Home/')
        assert r.status_int == 200
        assert r.json['title'] == 'Home'

        r = self.api_get('/rest/p/admin/installable_tools', status=403)

        r = self.api_get('/rest/p/admin/installable_tools', user='root')
        assert r.status_int == 200
        assert [t for t in r.json['tools'] if t['tool_label'] == 'Wiki'], r.json

    def test_project_has_access_no_params(self):
        self.api_get('/rest/p/test/has_access', status=404)
        self.api_get('/rest/p/test/has_access?user=root', status=404)
        self.api_get('/rest/p/test/has_access?perm=read', status=404)

    def test_project_has_access_unknown_params(self):
        """Unknown user and/or permission always False for has_access API"""
        r = self.api_get(
            '/rest/p/test/has_access?user=babadook&perm=read',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == False
        r = self.api_get(
            '/rest/p/test/has_access?user=test-admin&perm=jump',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == False

    def test_project_has_access_not_admin(self):
        """
        User which has no 'admin' permission on neighborhood can't use
        has_access API
        """
        self.api_get(
            '/rest/p/test/has_access?user=test-admin&perm=admin',
            user='test-user',
            status=403)

    def test_project_has_access(self):
        r = self.api_get(
            '/rest/p/test/has_access?user=test-admin&perm=update',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == True
        r = self.api_get(
            '/rest/p/test/has_access?user=test-user&perm=update',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == False

    def test_subproject_has_access(self):
        r = self.api_get(
            '/rest/p/test/sub1/has_access?user=test-admin&perm=update',
            user='root')
        assert r.status_int == 200
        assert r.json['result'] == True

    def test_unicode(self):
        self.app.post(
            h.urlquote('/wiki/tést/update'),
            params={
                'title': 'tést'.encode(),
                'text': 'sometext',
                'labels': '',
                })
        r = self.api_get(h.urlquote('/rest/p/test/wiki/tést/'))
        assert r.status_int == 200
        assert r.json['title'] == 'tést', r.json

    @td.with_wiki
    def test_deny_access(self):
        wiki = M.Project.query.get(shortname='test').app_instance('wiki')
        anon_read_perm = M.ACE.allow(
            M.ProjectRole.by_name('*anonymous')._id, 'read')
        auth_read_perm = M.ACE.allow(
            M.ProjectRole.by_name('*authenticated')._id, 'read')
        acl = wiki.config.acl
        if anon_read_perm in acl:
            acl.remove(anon_read_perm)
        if auth_read_perm in acl:
            acl.remove(auth_read_perm)
        self.app.get('/rest/p/test/wiki/Home/',
                     extra_environ={'username': '*anonymous'},
                     status=401)
        self.app.get('/rest/p/test/wiki/Home/',
                     extra_environ={'username': 'test-user-0'},
                     status=403)

    def test_index(self):
        eps = {
            'site_stats': {
                'foo_24hr': lambda: 42,
                'bar_24hr': lambda: 84,
                'qux_24hr': lambda: 0,
            },
        }
        with mock.patch.dict(g.entry_points, eps):
            response = self.app.get('/rest/')
            assert response.json == {
                'site_stats': {
                    'foo_24hr': 42,
                    'bar_24hr': 84,
                    'qux_24hr': 0,
                },
            }

    def test_name_validation(self):
        r = self.api_get('/rest/p/test/')
        assert r.status_int == 200
        with mock.patch('allura.lib.plugin.ProjectRegistrationProvider') as Provider:
            Provider.get().shortname_validator.to_python.side_effect = Invalid(
                'name', 'value', {})
            r = self.api_get('/rest/p/test/', status=404)

    @td.with_wiki
    def test_cors_POST_req_blocked_by_csrf(self):
        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True'}

        # regular login to get a session cookie set up
        r = self.app.get('/auth/')
        encoded = self.app.antispam_field_names(r.form)
        r.form[encoded['username']] = 'test-admin'
        r.form[encoded['password']] = 'foo'
        r.form.submit()

        # simulate CORS ajax request withCredentials (cookie headers)
        # make sure we don't allow the cookies to authorize the request (else could be a CSRF attack vector)
        assert self.app.cookies['allura']
        self.app.post('/rest/p/test/wiki/NewPage', headers={'Origin': 'http://bad.com/'},
                      status=401)

    @mock.patch('allura.lib.plugin.ThemeProvider._get_site_notification')
    def test_notification(self, _get_site_notification):
        user = M.User.by_username('test-admin')
        note = M.SiteNotification()
        cookie = f'{note._id}-1-False'
        g.theme._get_site_notification = mock.Mock(return_value=(note, cookie))

        r = self.app.get('/rest/notification?url=test_url&cookie=test_cookie&tool_name=test_tool')
        g.theme._get_site_notification.assert_called_once_with(
            url='test_url',
            user=user,
            tool_name='test_tool',
            site_notification_cookie_value='test_cookie'
        )

        assert r.status_int == 200
        assert r.json['cookie'] == cookie
        assert r.json['notification'] == note.__json__()

        g.theme._get_site_notification = mock.Mock(return_value=None)
        r = self.app.get('/rest/notification')
        assert r.json == {}


@with_nose_compatibility
class TestRestNbhdAddProject(TestRestApiBase):

    def setUp(self):
        super().setup_method(method)
        # create some troves we'll need
        M.TroveCategory(fullname="Root", trove_cat_id=1, trove_parent_id=0)
        M.TroveCategory(fullname="License", trove_cat_id=2, trove_parent_id=1)
        M.TroveCategory(fullname="Apache License V2.0", fullpath="License :: Apache License V2.0",
                                     trove_cat_id=4, trove_parent_id=2)
        M.TroveCategory(fullname="Public Domain", fullpath="License :: Public Domain",
                                      trove_cat_id=5, trove_parent_id=2)
        p_nbhd = M.Neighborhood.query.get(url_prefix='/p/')
        p_nbhd.features['private_projects'] = False

    def test_add_project_disallowed(self):
        self.api_post('/rest/p/add_project', user='*anonymous', status=401)
        self.api_post('/rest/adobe/add_project', user='test-user', status=403)
        self.api_post('/rest/p/add_project', user='test-admin', status=403)  # not a nbhd admin

    def test_add_project_missing_data(self):
        project_data = {
            "shortname": "my-new-proj",
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=400)
        assert 'Required' in r.json['error']

    def test_add_project_bad_data(self):
        project_data = {
            "shortname": "my-new-proj",
            "admin": 'test-admin',
            "name": "asdf",
            "trove_licenses": ["asdfasdf"]
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=400)
        assert 'is not a valid trove category' in r.json['error']

        # local icon paths are not allowed through web (have to use icon_url)
        project_data = {
            "shortname": "my-new-proj",
            "admin": 'test-admin',
            "name": "asdf",
            "icon": "/local/file/path.png"
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=400)
        assert 'Unrecognized keys in mapping' in r.json['error']

    def test_add_project_name_error(self):
        project_data = {
            "shortname": "test",
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=400)
        assert 'This project name is taken.' == r.json['error']

        project_data = {
            "shortname": "/?xyz",
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=400)
        assert 'Please use' in r.json['error']

    def test_add_project_create_error(self):
        project_data = {
            "shortname": "private-project",
            "name": "private-project",
            "admin": 'test-admin',
            "private": True
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=400)
        assert "You can't create private projects in the Projects neighborhood" == r.json['error']

    def test_add_project(self):
        project_data = {
            "shortname": "my-new-proj",
            "admin": "test-admin",
            "name": "My New Project",
            "summary": "My summary.",
            "description": "My description.",
            "external_homepage": "http://wwww.example.com/test",
            "labels": ["label1", "label2"],
            "trove_licenses": [
                "Apache License V2.0",
                "Public Domain"],
            "awards": [],
            "tool_data": {
                "allura": {
                    "grouping_threshold": 5
                }
            },
            # any image url can work here, just don't want to rely on any other websites:
            "icon_url": "https://allura.apache.org/theme/img/logo_white.png",
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=201)
        assert r.json == {
            'status': 'success',
            'html_url': 'http://localhost/p/my-new-proj/',
            'url': 'http://localhost/rest/p/my-new-proj/',
        }
        p = M.Project.query.get(shortname='my-new-proj')
        assert p.name == 'My New Project'
        assert len(p.trove_license) == 2
        assert isinstance(p.trove_license[0], ObjectId), p.trove_license[0]
        # these are sensitive fields only admins should be able to set:
        assert p.get_tool_data('allura', 'grouping_threshold') == 5
        assert p.admins()[0].username == 'test-admin'

    def test_add_project_automatic_shortname(self):
        # no shortname given, and name "Test" would conflict with existing "test" project
        project_data = {
            "admin": "test-admin",
            "name": "Test",
        }
        r = self.api_post('/rest/p/add_project',
                          params=json.dumps(project_data),
                          user='root',
                          status=201)
        assert r.json == {
            'status': 'success',
            'html_url': 'http://localhost/p/test1/',
            'url': 'http://localhost/rest/p/test1/',
        }


@with_nose_compatibility
class TestDoap(TestRestApiBase):
    validate_skip = True
    ns = '{http://usefulinc.com/ns/doap#}'
    ns_sf = '{http://sourceforge.net/api/sfelements.rdf#}'
    foaf = '{http://xmlns.com/foaf/0.1/}'
    dc = '{http://dublincore.org/documents/dcmi-namespace/}'
    rdf = '{http://www.w3.org/1999/02/22-rdf-syntax-ns#}'

    def test_project_data(self):
        project = M.Project.query.get(shortname='test')
        project.summary = 'A Summary'
        project.short_description = 'A Short Description'
        ThreadLocalODMSession.flush_all()
        r = self.app.get('/rest/p/test?doap')
        assert r.content_type == 'application/rdf+xml'
        p = r.xml.find(self.ns + 'Project')
        assert p.attrib[self.rdf + 'about'] == 'http://localhost/rest/p/test?doap#'
        assert p.find(self.ns + 'name').text == 'test'
        assert p.find(self.dc + 'title').text == 'Test Project'
        assert p.find(self.ns_sf + 'private').text == '0'
        assert p.find(self.ns + 'shortdesc').text == 'A Summary'
        assert p.find(self.ns + 'description').text == 'A Short Description'
        assert p.find(self.ns + 'created').text == project._id.generation_time.strftime('%Y-%m-%d')

        maintainers = p.findall(self.ns + 'maintainer')
        assert len(maintainers) == 1
        user = maintainers[0].find(self.foaf + 'Person')
        assert user.find(self.foaf + 'name').text == 'Test Admin'
        assert user.find(self.foaf + 'nick').text == 'test-admin'
        assert (list(user.find(self.foaf + 'homepage').items())[0][1] ==
                     'http://localhost/u/test-admin/')

    @td.with_tool('test', 'Tickets', 'bugs')
    @td.with_tool('test', 'Tickets', 'private-bugs')
    def test_project_data_tools(self):
        # Deny anonymous to see 'private-bugs' tool
        role = M.ProjectRole.by_name('*anonymous')._id
        read_permission = M.ACE.allow(role, 'read')
        app = M.Project.query.get(
            shortname='test').app_instance('private-bugs')
        if read_permission in app.config.acl:
            app.config.acl.remove(read_permission)

        # admin sees both 'Tickets' tools
        r = self.app.get('/rest/p/test?doap')
        p = r.xml.find(self.ns + 'Project')
        tools = p.findall(self.ns_sf + 'feature')
        tools = [(t.find(self.ns_sf + 'Feature').find(self.ns + 'name').text,
                  list(t.find(self.ns_sf + 'Feature').find(self.foaf + 'page').items())[0][1])
                 for t in tools]
        assert ('Tickets', 'http://localhost/p/test/bugs/') in tools
        assert ('Tickets', 'http://localhost/p/test/private-bugs/') in tools

        # anonymous sees only non-private tool
        r = self.app.get('/rest/p/test?doap',
                         extra_environ={'username': '*anonymous'})
        p = r.xml.find(self.ns + 'Project')
        tools = p.findall(self.ns_sf + 'feature')
        tools = [(t.find(self.ns_sf + 'Feature').find(self.ns + 'name').text,
                  list(t.find(self.ns_sf + 'Feature').find(self.foaf + 'page').items())[0][1])
                 for t in tools]
        assert ('Tickets', 'http://localhost/p/test/bugs/') in tools
        assert ('Tickets', 'http://localhost/p/test/private-bugs/') not in tools


@with_nose_compatibility
class TestUserProfile(TestRestApiBase):
    @td.with_user_project('test-admin')
    def test_profile_data(self):
        r = self.app.get('/rest/u/test-admin/profile/')
        assert r.content_type == 'application/json'
        json = r.json
        assert json['username'] == 'test-admin'
        assert json['name'] == 'Test Admin'
        assert 'availability' in json
        assert 'joined' in json
        assert 'localization' in json
        assert 'projects' in json
        assert 'sex' in json
        assert 'skills' in json
        assert 'skypeaccount' in json
        assert 'socialnetworks' in json
        assert 'telnumbers' in json
        assert 'webpages' in json