test_api.py

# -*- coding: utf-8 -*-
from __future__ import (
    absolute_import, division, print_function, unicode_literals)

import base64
import datetime
import json
from builtins import str

import mock
from allauth.socialaccount.models import SocialAccount
from django.contrib.auth.models import User
from django.core.urlresolvers import reverse
from django.http import QueryDict
from django.test import TestCase
from django.utils import six
from django_dynamic_fixture import get
from rest_framework import status
from rest_framework.test import APIClient

from readthedocs.builds.models import Build, BuildCommandResult, Version
from readthedocs.integrations.models import Integration
from readthedocs.oauth.models import RemoteOrganization, RemoteRepository
from readthedocs.projects.models import Feature, Project, APIProject
from readthedocs.restapi.views.integrations import GitHubWebhookView
from readthedocs.restapi.views.task_views import get_status_data

super_auth = base64.b64encode(b'super:test').decode('utf-8')
eric_auth = base64.b64encode(b'eric:test').decode('utf-8')


class APIBuildTests(TestCase):
    fixtures = ['eric.json', 'test_data.json']

    def test_make_build(self):
        """Test that a superuser can use the API."""
        client = APIClient()
        client.login(username='super', password='test')
        resp = client.post(
            '/api/v2/build/',
            {
                'project': 1,
                'version': 1,
                'success': True,
                'output': 'Test Output',
                'error': 'Test Error',
                'state': 'cloning',
            },
            format='json',
        )
        self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
        build = resp.data
        self.assertEqual(build['state_display'], 'Cloning')

        resp = client.get('/api/v2/build/%s/' % build['id'])
        self.assertEqual(resp.status_code, 200)
        build = resp.data
        self.assertEqual(build['output'], 'Test Output')
        self.assertEqual(build['state_display'], 'Cloning')

    def test_response_building(self):
        """
        The ``view docs`` attr should return a link
        to the dashboard.
        """
        client = APIClient()
        client.login(username='super', password='test')
        project = get(
            Project,
            language='en',
            main_language_project=None,
        )
        version = get(
            Version,
            project=project,
            built=False,
            uploaded=False,
        )
        build = get(
            Build,
            project=project,
            version=version,
            state='cloning',
            exit_code=0,
        )
        resp = client.get('/api/v2/build/{build}/'.format(build=build.pk))
        self.assertEqual(resp.status_code, 200)

        dashboard_url = reverse(
            'project_version_detail',
            kwargs={
                'project_slug': project.slug,
                'version_slug': version.slug,
            },
        )
        build = resp.data
        self.assertEqual(build['state'], 'cloning')
        self.assertEqual(build['error'], '')
        self.assertEqual(build['exit_code'], 0)
        self.assertEqual(build['success'], True)
        self.assertEqual(build['docs_url'], dashboard_url)

    def test_response_finished_and_success(self):
        """
        The ``view docs`` attr should return a link
        to the docs.
        """
        client = APIClient()
        client.login(username='super', password='test')
        project = get(
            Project,
            language='en',
            main_language_project=None,
        )
        version = get(
            Version,
            project=project,
            built=True,
            uploaded=True,
        )
        build = get(
            Build,
            project=project,
            version=version,
            state='finished',
            exit_code=0,
        )
        resp = client.get('/api/v2/build/{build}/'.format(build=build.pk))
        self.assertEqual(resp.status_code, 200)
        build = resp.data
        docs_url = 'http://readthedocs.org/docs/{project}/en/{version}/'.format(
            project=project.slug,
            version=version.slug,
        )
        self.assertEqual(build['state'], 'finished')
        self.assertEqual(build['error'], '')
        self.assertEqual(build['exit_code'], 0)
        self.assertEqual(build['success'], True)
        self.assertEqual(build['docs_url'], docs_url)

    def test_response_finished_and_fail(self):
        """
        The ``view docs`` attr should return a link
        to the dashboard.
        """
        client = APIClient()
        client.login(username='super', password='test')
        project = get(
            Project,
            language='en',
            main_language_project=None,
        )
        version = get(
            Version,
            project=project,
            built=False,
            uploaded=False,
        )
        build = get(
            Build,
            project=project,
            version=version,
            state='finished',
            success=False,
            exit_code=1,
        )

        resp = client.get('/api/v2/build/{build}/'.format(build=build.pk))
        self.assertEqual(resp.status_code, 200)

        dashboard_url = reverse(
            'project_version_detail',
            kwargs={
                'project_slug': project.slug,
                'version_slug': version.slug,
            },
        )
        build = resp.data
        self.assertEqual(build['state'], 'finished')
        self.assertEqual(build['error'], '')
        self.assertEqual(build['exit_code'], 1)
        self.assertEqual(build['success'], False)
        self.assertEqual(build['docs_url'], dashboard_url)

    def test_make_build_without_permission(self):
        """Ensure anonymous/non-staff users cannot write the build endpoint."""
        client = APIClient()

        def _try_post():
            resp = client.post(
                '/api/v2/build/',
                {
                    'project': 1,
                    'version': 1,
                    'success': True,
                    'output': 'Test Output',
                    'error': 'Test Error',
                },
                format='json',
            )
            self.assertEqual(resp.status_code, 403)

        _try_post()

        api_user = get(User, staff=False, password='test')
        assert api_user.is_staff is False
        client.force_authenticate(user=api_user)
        _try_post()

    def test_update_build_without_permission(self):
        """Ensure anonymous/non-staff users cannot update build endpoints."""
        client = APIClient()
        api_user = get(User, staff=False, password='test')
        client.force_authenticate(user=api_user)
        build = get(Build, project_id=1, version_id=1, state='cloning')
        resp = client.put(
            '/api/v2/build/{0}/'.format(build.pk),
            {
                'project': 1,
                'version': 1,
                'state': 'finished',
            },
            format='json',
        )
        self.assertEqual(resp.status_code, 403)

    def test_make_build_protected_fields(self):
        """
        Ensure build api view delegates correct serializer.

        Super users should be able to read/write the `builder` property, but we
        don't expose this to end users via the API
        """
        build = get(Build, project_id=1, version_id=1, builder='foo')
        client = APIClient()

        api_user = get(User, staff=False, password='test')
        client.force_authenticate(user=api_user)
        resp = client.get('/api/v2/build/{0}/'.format(build.pk), format='json')
        self.assertEqual(resp.status_code, 200)

        client.force_authenticate(user=User.objects.get(username='super'))
        resp = client.get('/api/v2/build/{0}/'.format(build.pk), format='json')
        self.assertEqual(resp.status_code, 200)
        self.assertIn('builder', resp.data)

    def test_make_build_commands(self):
        """Create build and build commands."""
        client = APIClient()
        client.login(username='super', password='test')
        resp = client.post(
            '/api/v2/build/',
            {
                'project': 1,
                'version': 1,
                'success': True,
            },
            format='json',
        )
        self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
        build = resp.data
        now = datetime.datetime.utcnow()
        resp = client.post(
            '/api/v2/command/',
            {
                'build': build['id'],
                'command': 'echo test',
                'description': 'foo',
                'exit_code': 0,
                'start_time': str(now - datetime.timedelta(seconds=5)),
                'end_time': str(now),
            },
            format='json',
        )
        self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
        resp = client.get('/api/v2/build/%s/' % build['id'])
        self.assertEqual(resp.status_code, 200)
        build = resp.data
        self.assertEqual(len(build['commands']), 1)
        self.assertEqual(build['commands'][0]['run_time'], 5)
        self.assertEqual(build['commands'][0]['description'], 'foo')

    def test_get_raw_log_success(self):
        build = get(Build, project_id=1, version_id=1, builder='foo')
        get(
            BuildCommandResult,
            build=build,
            command='python setup.py install',
            output='Installing dependencies...'
        )
        get(
            BuildCommandResult,
            build=build,
            command='git checkout master',
            output='Switched to branch "master"'
        )
        client = APIClient()

        api_user = get(User, user='test', password='test')
        client.force_authenticate(user=api_user)
        resp = client.get('/api/v2/build/{0}.txt'.format(build.pk))
        self.assertEqual(resp.status_code, 200)

        self.assertIn('Read the Docs build information', resp.content.decode())
        self.assertIn('Build id: {}'.format(build.id), resp.content.decode())
        self.assertIn('Project: {}'.format(build.project.slug), resp.content.decode())
        self.assertIn('Version: {}'.format(build.version.slug), resp.content.decode())
        self.assertIn('Commit: {}'.format(build.commit), resp.content.decode())
        self.assertIn('Date: ', resp.content.decode())
        self.assertIn('State: finished', resp.content.decode())
        self.assertIn('Success: True', resp.content.decode())
        self.assertIn('[rtd-command-info]', resp.content.decode())
        self.assertIn(
            'python setup.py install\nInstalling dependencies...',
            resp.content.decode()
        )
        self.assertIn(
            'git checkout master\nSwitched to branch "master"',
            resp.content.decode()
        )

    def test_get_raw_log_building(self):
        build = get(
            Build, project_id=1, version_id=1,
            builder='foo', success=False,
            exit_code=1, state='building',
        )
        get(
            BuildCommandResult,
            build=build,
            command='python setup.py install',
            output='Installing dependencies...',
            exit_code=1,
        )
        get(
            BuildCommandResult,
            build=build,
            command='git checkout master',
            output='Switched to branch "master"'
        )
        client = APIClient()

        api_user = get(User, user='test', password='test')
        client.force_authenticate(user=api_user)
        resp = client.get('/api/v2/build/{0}.txt'.format(build.pk))
        self.assertEqual(resp.status_code, 200)

        self.assertIn('Read the Docs build information', resp.content.decode())
        self.assertIn('Build id: {}'.format(build.id), resp.content.decode())
        self.assertIn('Project: {}'.format(build.project.slug), resp.content.decode())
        self.assertIn('Version: {}'.format(build.version.slug), resp.content.decode())
        self.assertIn('Commit: {}'.format(build.commit), resp.content.decode())
        self.assertIn('Date: ', resp.content.decode())
        self.assertIn('State: building', resp.content.decode())
        self.assertIn('Success: Unknow', resp.content.decode())
        self.assertIn('[rtd-command-info]', resp.content.decode())
        self.assertIn(
            'python setup.py install\nInstalling dependencies...',
            resp.content.decode()
        )
        self.assertIn(
            'git checkout master\nSwitched to branch "master"',
            resp.content.decode()
        )

    def test_get_raw_log_failure(self):
        build = get(
            Build, project_id=1, version_id=1,
            builder='foo', success=False, exit_code=1
        )
        get(
            BuildCommandResult,
            build=build,
            command='python setup.py install',
            output='Installing dependencies...',
            exit_code=1,
        )
        get(
            BuildCommandResult,
            build=build,
            command='git checkout master',
            output='Switched to branch "master"'
        )
        client = APIClient()

        api_user = get(User, user='test', password='test')
        client.force_authenticate(user=api_user)
        resp = client.get('/api/v2/build/{0}.txt'.format(build.pk))
        self.assertEqual(resp.status_code, 200)

        self.assertIn('Read the Docs build information', resp.content.decode())
        self.assertIn('Build id: {}'.format(build.id), resp.content.decode())
        self.assertIn('Project: {}'.format(build.project.slug), resp.content.decode())
        self.assertIn('Version: {}'.format(build.version.slug), resp.content.decode())
        self.assertIn('Commit: {}'.format(build.commit), resp.content.decode())
        self.assertIn('Date: ', resp.content.decode())
        self.assertIn('State: finished', resp.content.decode())
        self.assertIn('Success: False', resp.content.decode())
        self.assertIn('[rtd-command-info]', resp.content.decode())
        self.assertIn(
            'python setup.py install\nInstalling dependencies...',
            resp.content.decode()
        )
        self.assertIn(
            'git checkout master\nSwitched to branch "master"',
            resp.content.decode()
        )

    def test_get_invalid_raw_log(self):
        client = APIClient()

        api_user = get(User, user='test', password='test')
        client.force_authenticate(user=api_user)
        resp = client.get('/api/v2/build/{0}.txt'.format(404))
        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)


class APITests(TestCase):
    fixtures = ['eric.json', 'test_data.json']

    def test_make_project(self):
        """Test that a superuser can use the API."""
        post_data = {
            'name': 'awesome-project',
            'repo': 'https://github.com/ericholscher/django-kong.git',
        }
        resp = self.client.post(
            '/api/v1/project/',
            data=json.dumps(post_data),
            content_type='application/json',
            HTTP_AUTHORIZATION='Basic %s' % super_auth,
        )
        self.assertEqual(resp.status_code, 201)
        self.assertEqual(resp['location'], '/api/v1/project/24/')
        resp = self.client.get(
            '/api/v1/project/24/',
            data={'format': 'json'},
            HTTP_AUTHORIZATION='Basic %s' % eric_auth,
        )
        self.assertEqual(resp.status_code, 200)
        obj = json.loads(resp.content)
        self.assertEqual(obj['slug'], 'awesome-project')

    def test_user_doesnt_get_full_api_return(self):
        user_normal = get(User, is_staff=False)
        user_admin = get(User, is_staff=True)
        project = get(Project, main_language_project=None, conf_py_file='foo')
        client = APIClient()

        client.force_authenticate(user=user_normal)
        resp = client.get('/api/v2/project/%s/' % (project.pk))
        self.assertEqual(resp.status_code, 200)
        self.assertNotIn('conf_py_file', resp.data)

        client.force_authenticate(user=user_admin)
        resp = client.get('/api/v2/project/%s/' % (project.pk))
        self.assertEqual(resp.status_code, 200)
        self.assertIn('conf_py_file', resp.data)
        self.assertEqual(resp.data['conf_py_file'], 'foo')

    def test_invalid_make_project(self):
        """Test that the authentication is turned on."""
        post_data = {
            'user': '/api/v1/user/2/',
            'name': 'awesome-project-2',
            'repo': 'https://github.com/ericholscher/django-bob.git',
        }
        resp = self.client.post(
            '/api/v1/project/',
            data=json.dumps(post_data),
            content_type='application/json',
            HTTP_AUTHORIZATION='Basic %s' %
            base64.b64encode(b'tester:notapass').decode('utf-8'),
        )
        self.assertEqual(resp.status_code, 401)

    def test_make_project_dishonest_user(self):
        """Test that you can't create a project for another user."""
        # represents dishonest data input, authentication happens for user 2
        post_data = {
            'users': ['/api/v1/user/1/'],
            'name': 'awesome-project-2',
            'repo': 'https://github.com/ericholscher/django-bob.git',
        }
        resp = self.client.post(
            '/api/v1/project/',
            data=json.dumps(post_data),
            content_type='application/json',
            HTTP_AUTHORIZATION='Basic %s' %
            base64.b64encode(b'tester:test').decode('utf-8'),
        )
        self.assertEqual(resp.status_code, 401)

    def test_ensure_get_unauth(self):
        """Test that GET requests work without authenticating."""
        resp = self.client.get('/api/v1/project/', data={'format': 'json'})
        self.assertEqual(resp.status_code, 200)

    def test_project_features(self):
        user = get(User, is_staff=True)
        project = get(Project, main_language_project=None)
        # One explicit, one implicit feature
        feature1 = get(Feature, projects=[project])
        feature2 = get(Feature, projects=[], default_true=True)
        feature3 = get(Feature, projects=[], default_true=False)  # noqa
        client = APIClient()

        client.force_authenticate(user=user)
        resp = client.get('/api/v2/project/%s/' % (project.pk))
        self.assertEqual(resp.status_code, 200)
        self.assertIn('features', resp.data)
        six.assertCountEqual(
            self,
            resp.data['features'],
            [feature1.feature_id, feature2.feature_id],
        )

    def test_project_features_multiple_projects(self):
        user = get(User, is_staff=True)
        project1 = get(Project, main_language_project=None)
        project2 = get(Project, main_language_project=None)
        feature = get(Feature, projects=[project1, project2], default_true=True)
        client = APIClient()

        client.force_authenticate(user=user)
        resp = client.get('/api/v2/project/%s/' % (project1.pk))
        self.assertEqual(resp.status_code, 200)
        self.assertIn('features', resp.data)
        self.assertEqual(resp.data['features'], [feature.feature_id])

    def test_project_pagination(self):
        for _ in range(100):
            get(Project)

        resp = self.client.get('/api/v2/project/')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(len(resp.data['results']), 100)  # page_size
        self.assertIn('?page=2', resp.data['next'])

    def test_remote_repository_pagination(self):
        account = get(SocialAccount, provider='github')
        user = get(User, socialaccount_set=[account])
        for _ in range(20):
            get(RemoteRepository, users=[user], account=account)

        client = APIClient()
        client.force_authenticate(user=user)

        resp = client.get('/api/v2/remote/repo/')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(len(resp.data['results']), 15)  # page_size
        self.assertIn('?page=2', resp.data['next'])

    def test_remote_organization_pagination(self):
        account = get(SocialAccount, provider='github')
        user = get(User, socialaccount_set=[account])
        for _ in range(30):
            get(RemoteOrganization, users=[user], account=account)

        client = APIClient()
        client.force_authenticate(user=user)

        resp = client.get('/api/v2/remote/org/')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(len(resp.data['results']), 25)  # page_size
        self.assertIn('?page=2', resp.data['next'])

    def test_init_api_project(self):
        project_data = {
            'name': 'Test Project',
            'slug': 'test-project',
            'show_advertising': True,
        }

        api_project = APIProject(**project_data)
        self.assertEqual(api_project.slug, 'test-project')
        self.assertEqual(api_project.features, [])
        self.assertFalse(api_project.ad_free)
        self.assertTrue(api_project.show_advertising)

        project_data['features'] = ['test-feature']
        project_data['show_advertising'] = False
        api_project = APIProject(**project_data)
        self.assertEqual(api_project.features, ['test-feature'])
        self.assertTrue(api_project.ad_free)
        self.assertFalse(api_project.show_advertising)


class APIImportTests(TestCase):

    """Import API endpoint tests."""

    fixtures = ['eric.json', 'test_data.json']

    def test_permissions(self):
        """Ensure user repositories aren't leaked to other users."""
        client = APIClient()

        account_a = get(SocialAccount, provider='github')
        account_b = get(SocialAccount, provider='github')
        account_c = get(SocialAccount, provider='github')
        user_a = get(User, password='test', socialaccount_set=[account_a])
        user_b = get(User, password='test', socialaccount_set=[account_b])
        user_c = get(User, password='test', socialaccount_set=[account_c])
        org_a = get(RemoteOrganization, users=[user_a], account=account_a)
        repo_a = get(
            RemoteRepository,
            users=[user_a],
            organization=org_a,
            account=account_a,
        )
        repo_b = get(
            RemoteRepository,
            users=[user_b],
            organization=None,
            account=account_b,
        )

        client.force_authenticate(user=user_a)
        resp = client.get('/api/v2/remote/repo/', format='json')
        self.assertEqual(resp.status_code, status.HTTP_200_OK)
        repos = resp.data['results']
        self.assertEqual(repos[0]['id'], repo_a.id)
        self.assertEqual(repos[0]['organization']['id'], org_a.id)
        self.assertEqual(len(repos), 1)

        resp = client.get('/api/v2/remote/org/', format='json')
        self.assertEqual(resp.status_code, status.HTTP_200_OK)
        orgs = resp.data['results']
        self.assertEqual(orgs[0]['id'], org_a.id)
        self.assertEqual(len(orgs), 1)

        client.force_authenticate(user=user_b)
        resp = client.get('/api/v2/remote/repo/', format='json')
        self.assertEqual(resp.status_code, status.HTTP_200_OK)
        repos = resp.data['results']
        self.assertEqual(repos[0]['id'], repo_b.id)
        self.assertEqual(repos[0]['organization'], None)
        self.assertEqual(len(repos), 1)

        client.force_authenticate(user=user_c)
        resp = client.get('/api/v2/remote/repo/', format='json')
        self.assertEqual(resp.status_code, status.HTTP_200_OK)
        repos = resp.data['results']
        self.assertEqual(len(repos), 0)


@mock.patch('readthedocs.core.views.hooks.trigger_build')
class IntegrationsTests(TestCase):

    """Integration for webhooks, etc."""

    fixtures = ['eric.json', 'test_data.json']

    def setUp(self):
        self.project = get(Project)
        self.version = get(Version, verbose_name='master', active=True, project=self.project)
        self.version_tag = get(Version, verbose_name='v1.0', active=True, project=self.project)

    def test_github_webhook_for_branches(self, trigger_build):
        """GitHub webhook API."""
        client = APIClient()

        client.post(
            '/api/v2/webhook/github/{0}/'.format(self.project.slug),
            {'ref': 'master'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=self.version, project=self.project)])

        client.post(
            '/api/v2/webhook/github/{0}/'.format(self.project.slug),
            {'ref': 'non-existent'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=mock.ANY, project=self.project)])

        client.post(
            '/api/v2/webhook/github/{0}/'.format(self.project.slug),
            {'ref': 'refs/heads/master'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=self.version, project=self.project)])

    def test_github_webhook_for_tags(self, trigger_build):
        """GitHub webhook API."""
        client = APIClient()

        client.post(
            '/api/v2/webhook/github/{0}/'.format(self.project.slug),
            {'ref': 'v1.0'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=self.version_tag, project=self.project)])

        client.post(
            '/api/v2/webhook/github/{0}/'.format(self.project.slug),
            {'ref': 'refs/heads/non-existent'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=mock.ANY, project=self.project)])

        client.post(
            '/api/v2/webhook/github/{0}/'.format(self.project.slug),
            {'ref': 'refs/tags/v1.0'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=self.version_tag, project=self.project)])

    def test_github_parse_ref(self, trigger_build):
        wh = GitHubWebhookView()

        self.assertEqual(wh._normalize_ref('refs/heads/master'), 'master')
        self.assertEqual(wh._normalize_ref('refs/heads/v0.1'), 'v0.1')
        self.assertEqual(wh._normalize_ref('refs/tags/v0.1'), 'v0.1')
        self.assertEqual(wh._normalize_ref('refs/tags/tag'), 'tag')
        self.assertEqual(wh._normalize_ref('refs/heads/stable/2018'), 'stable/2018')
        self.assertEqual(wh._normalize_ref('refs/tags/tag/v0.1'), 'tag/v0.1')

    def test_github_invalid_webhook(self, trigger_build):
        """GitHub webhook unhandled event."""
        client = APIClient()
        resp = client.post(
            '/api/v2/webhook/github/{0}/'.format(self.project.slug),
            {'foo': 'bar'},
            format='json',
            HTTP_X_GITHUB_EVENT='pull_request',
        )
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.data['detail'], 'Unhandled webhook event')

    def test_gitlab_webhook(self, trigger_build):
        """GitLab webhook API."""
        client = APIClient()
        client.post(
            '/api/v2/webhook/gitlab/{0}/'.format(self.project.slug),
            {'object_kind': 'push', 'ref': 'master'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=mock.ANY, project=self.project)])
        client.post(
            '/api/v2/webhook/gitlab/{0}/'.format(self.project.slug),
            {'object_kind': 'push', 'ref': 'non-existent'},
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=mock.ANY, project=self.project)])

    def test_gitlab_invalid_webhook(self, trigger_build):
        """GitLab webhook unhandled event."""
        client = APIClient()
        resp = client.post(
            '/api/v2/webhook/gitlab/{0}/'.format(self.project.slug),
            {'object_kind': 'pull_request'},
            format='json',
        )
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.data['detail'], 'Unhandled webhook event')

    def test_bitbucket_webhook(self, trigger_build):
        """Bitbucket webhook API."""
        client = APIClient()
        client.post(
            '/api/v2/webhook/bitbucket/{0}/'.format(self.project.slug),
            {
                'push': {
                    'changes': [{
                        'new': {
                            'name': 'master',
                        },
                    }],
                },
            },
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=mock.ANY, project=self.project)])
        client.post(
            '/api/v2/webhook/bitbucket/{0}/'.format(self.project.slug),
            {
                'push': {
                    'changes': [
                        {
                            'new': {'name': 'non-existent'},
                        },
                    ],
                },
            },
            format='json',
        )
        trigger_build.assert_has_calls(
            [mock.call(force=True, version=mock.ANY, project=self.project)])

        trigger_build_call_count = trigger_build.call_count
        client.post(
            '/api/v2/webhook/bitbucket/{0}/'.format(self.project.slug),
            {
                'push': {
                    'changes': [
                        {
                            'new': None,
                        },
                    ],
                },
            },
            format='json',
        )
        self.assertEqual(trigger_build_call_count, trigger_build.call_count)

    def test_bitbucket_invalid_webhook(self, trigger_build):
        """Bitbucket webhook unhandled event."""
        client = APIClient()
        resp = client.post(
            '/api/v2/webhook/bitbucket/{0}/'.format(self.project.slug),
            {'foo': 'bar'}, format='json', HTTP_X_EVENT_KEY='pull_request')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.data['detail'], 'Unhandled webhook event')

    def test_generic_api_fails_without_auth(self, trigger_build):
        client = APIClient()
        resp = client.post(
            '/api/v2/webhook/generic/{0}/'.format(self.project.slug),
            {},
            format='json',
        )
        self.assertEqual(resp.status_code, 403)
        self.assertEqual(
            resp.data['detail'],
            'Authentication credentials were not provided.',
        )

    def test_generic_api_respects_token_auth(self, trigger_build):
        client = APIClient()
        integration = Integration.objects.create(
            project=self.project,
            integration_type=Integration.API_WEBHOOK,
        )
        self.assertIsNotNone(integration.token)
        resp = client.post(
            '/api/v2/webhook/{0}/{1}/'.format(
                self.project.slug,
                integration.pk,
            ),
            {'token': integration.token},
            format='json',
        )
        self.assertEqual(resp.status_code, 200)
        self.assertTrue(resp.data['build_triggered'])
        # Test nonexistent branch
        resp = client.post(
            '/api/v2/webhook/{0}/{1}/'.format(
                self.project.slug,
                integration.pk,
            ),
            {'token': integration.token, 'branches': 'nonexistent'},
            format='json',
        )
        self.assertEqual(resp.status_code, 200)
        self.assertFalse(resp.data['build_triggered'])

    def test_generic_api_respects_basic_auth(self, trigger_build):
        client = APIClient()
        user = get(User)
        self.project.users.add(user)
        client.force_authenticate(user=user)
        resp = client.post(
            '/api/v2/webhook/generic/{0}/'.format(self.project.slug),
            {},
            format='json',
        )
        self.assertEqual(resp.status_code, 200)
        self.assertTrue(resp.data['build_triggered'])

    def test_generic_api_falls_back_to_token_auth(self, trigger_build):
        client = APIClient()
        user = get(User)
        client.force_authenticate(user=user)
        integration = Integration.objects.create(
            project=self.project, integration_type=Integration.API_WEBHOOK)
        self.assertIsNotNone(integration.token)
        resp = client.post(
            '/api/v2/webhook/{0}/{1}/'.format(
                self.project.slug,
                integration.pk,
            ),
            {'token': integration.token},
            format='json',
        )
        self.assertEqual(resp.status_code, 200)
        self.assertTrue(resp.data['build_triggered'])


class APIVersionTests(TestCase):
    fixtures = ['eric', 'test_data']

    def test_get_version_by_id(self):
        """
        Test the full response of ``/api/v2/version/{pk}`` is what we expects.

        Allows us to notice changes in the fields returned by the endpoint
        instead of let them pass silently.
        """
        pip = Project.objects.get(slug='pip')
        version = pip.versions.get(slug='0.8')

        data = {
            'pk': version.pk,
        }
        resp = self.client.get(
            reverse('version-detail', kwargs=data),
            content_type='application/json',
            HTTP_AUTHORIZATION='Basic {}'.format(eric_auth),
        )
        self.assertEqual(resp.status_code, 200)

        version_data = {
            'type': 'tag',
            'verbose_name': '0.8',
            'built': False,
            'id': 18,
            'active': True,
            'project': {
                'analytics_code': None,
                'canonical_url': 'http://readthedocs.org/docs/pip/en/latest/',
                'cdn_enabled': False,
                'conf_py_file': '',
                'container_image': None,
                'container_mem_limit': None,
                'container_time_limit': None,
                'default_branch': None,
                'default_version': 'latest',
                'description': '',
                'documentation_type': 'sphinx',
                'enable_epub_build': True,
                'enable_pdf_build': True,
                'features': ['allow_deprecated_webhooks'],
                'has_valid_clone': False,
                'has_valid_webhook': False,
                'id': 6,
                'install_project': False,
                'language': 'en',
                'name': 'Pip',
                'programming_language': 'words',
                'python_interpreter': 'python',
                'repo': 'https://github.com/pypa/pip',
                'repo_type': 'git',
                'requirements_file': None,
                'show_advertising': True,
                'skip': False,
                'slug': 'pip',
                'suffix': '.rst',
                'use_system_packages': False,
                'users': [1],
            },
            'downloads': {},
            'identifier': '2404a34eba4ee9c48cc8bc4055b99a48354f4950',
            'slug': '0.8',
        }

        self.assertDictEqual(
            resp.data,
            version_data,
        )

    def test_get_active_versions(self):
        """
        Test the full response of ``/api/v2/version/?project__slug=pip&active=true``
        """
        pip = Project.objects.get(slug='pip')

        data = QueryDict('', mutable=True)
        data.update({
            'project__slug': pip.slug,
            'active': 'true',
        })
        url = '{base_url}?{querystring}'.format(
            base_url=reverse('version-list'),
            querystring=data.urlencode()
        )

        resp = self.client.get(url, content_type='application/json')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.data['count'], pip.versions.filter(active=True).count())

        # Do the same thing for inactive versions
        data.update({
            'project__slug': pip.slug,
            'active': 'false',
        })
        url = '{base_url}?{querystring}'.format(
            base_url=reverse('version-list'),
            querystring=data.urlencode()
        )

        resp = self.client.get(url, content_type='application/json')
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.data['count'], pip.versions.filter(active=False).count())


class TaskViewsTests(TestCase):

    def test_get_status_data(self):
        data = get_status_data(
            'public_task_exception',
            'SUCCESS',
            {'data': 'public'},
            'Something bad happened',
        )
        self.assertEqual(data, {
            'name': 'public_task_exception',
            'data': {'data': 'public'},
            'started': True,
            'finished': True,
            'success': False,
            'error': 'Something bad happened',
        })