def authorize(self, username, password, credentials, *args, **kwargs):
        """Authorize an account for Codebase.

        Codebase usees HTTP Basic Auth with an API username (consisting of the
        Codebase team's domain and the account username) and an API key (for
        the password) for API calls, and a standard username/password for
        Subversion repository access. We need to store all of this.

        Args:
            username (unicode):
                The username to authorize.

            password (unicode):
                The API token used as a password.

            credentials (dict):
                Additional credentials from the authentication form.

            *args (tuple):
                Extra unused positional arguments.

            **kwargs (dict):
                Extra unused keyword arguments.

        Raises:
            reviewboard.hostingsvcs.errors.AuthorizationError:
                The credentials provided were not valid.
        """
        self.account.data.update(
            {
                "domain": credentials["domain"],
                "api_key": encrypt_password(credentials["api_key"]),
                "password": encrypt_password(password),
            }
        )

        # Test the account to make sure the credentials are fine. Note that
        # we can only really sanity-check the API token, domain, and username
        # from here. There's no way good way to check the actual password,
        # which we only use for Subversion repositories.
        #
        # This will raise a suitable error message if authorization fails.
        try:
            self.client.api_get_public_keys(username)
        except AuthorizationError:
            raise AuthorizationError(
                ugettext("One or more of the credentials provided were not " "accepted by Codebase.")
            )

        self.account.save()

    def _test_get_file(self, tool_name, revision, base_commit_id,
                       expected_revision):
        def _http_get(service, url, *args, **kwargs):
            self.assertEqual(
                url,
                'https://bitbucket.org/api/1.0/repositories/'
                'myuser/myrepo/raw/%s/path'
                % expected_revision)
            return b'My data', {}

        account = self._get_hosting_account()
        service = account.service
        repository = Repository(hosting_account=account,
                                tool=Tool.objects.get(name=tool_name))
        repository.extra_data = {
            'bitbucket_repo_name': 'myrepo',
        }

        account.data['password'] = encrypt_password('abc123')

        self.spy_on(service.client.http_get, call_fake=_http_get)

        result = service.get_file(repository, 'path', revision,
                                  base_commit_id)
        self.assertTrue(service.client.http_get.called)
        self.assertEqual(result, 'My data')

    def _test_get_file_exists(self, tool_name, revision, base_commit_id,
                              expected_revision, expected_found,
                              expected_http_called=True):
        def _http_get(service, url, *args, **kwargs):
            self.assertEqual(
                url,
                'https://bitbucket.org/api/1.0/repositories/'
                'myuser/myrepo/raw/%s/path'
                % expected_revision)

            if expected_found:
                return b'{}', {}
            else:
                error = HTTPError(url, 404, 'Not Found', {}, None)
                error.read = lambda: error.msg
                raise error

        account = self._get_hosting_account()
        service = account.service
        repository = Repository(hosting_account=account,
                                tool=Tool.objects.get(name=tool_name))
        repository.extra_data = {
            'bitbucket_repo_name': 'myrepo',
        }

        account.data['password'] = encrypt_password('abc123')

        self.spy_on(service.client.http_get, call_fake=_http_get)

        result = service.get_file_exists(repository, 'path', revision,
                                         base_commit_id)
        self.assertEqual(service.client.http_get.called, expected_http_called)
        self.assertEqual(result, expected_found)

    def save(self, **kwargs):
        """Save the authentication form.

        Args:
            **kwargs (dict):
                Keyword arguments to pass to
                :py:meth`:HostingServiceAuthForm.save()
                <reviewboard.hostingsvcs.forms.HostingServiceAuthForm.save>`.

        Returns:
            reviewboard.hostingsvcs.models.HostingServiceAccount:
            The hosting service account.

        Raises:
            reviewboard.hostingsvcs.errors.AuthorizationError:
                Information needed to authorize was missing, or authorization
                failed.
        """
        hosting_account = super(GerritAuthForm, self).save(save=False,
                                                           **kwargs)

        if not hosting_account.data:
            hosting_account.data = {}

        hosting_account.data['gerrit_http_password'] = encrypt_password(
            self.get_credentials()['password'])
        hosting_account.save()

        return hosting_account

    def authorize(self, username, password, hosting_url, *args, **kwargs):
        """Authorize the Review Board Gateway repository.

        Review Board Gateway uses HTTP Basic Auth, so this will store the
        provided password, encrypted, for use in later API requests.

        Similar to GitLab's API, Review Board Gateway will return a private
        token on session authentication.
        """
        try:
            response = self.client.http_post(url='%s/session' % hosting_url,
                                             username=username,
                                             password=password)
        except HTTPError as e:
            if e.code == 401:
                raise AuthorizationError(
                    ugettext('The username or password is incorrect.'))
            elif e.code == 404:
                raise HostingServiceError(
                    ugettext('A Review Board Gateway server was not found at '
                             'the provided URL.'))
            else:
                logger.exception('Failed authorization at %s/session: %s',
                                 hosting_url, e)

            raise

        self.account.data['private_token'] = \
            encrypt_password(response.json['private_token'])
        self.account.save()

    def test_other_user_check_repository(self):
        """Testing Bitbucket other-user check_repository"""
        def _http_get(service, url, *args, **kwargs):
            self.assertEqual(
                url,
                'https://bitbucket.org/api/2.0/repositories/someuser/myrepo'
                '?fields=scm')
            return (
                json.dumps({
                    'scm': 'git',
                }),
                {})

        account = self._get_hosting_account()
        service = account.service

        account.data['password'] = encrypt_password('abc123')

        self.spy_on(service.client.http_get, call_fake=_http_get)

        service.check_repository(bitbucket_other_user_username='someuser',
                                 bitbucket_other_user_repo_name='myrepo',
                                 plan='other-user',
                                 tool_name='Git')
        self.assertTrue(service.client.http_get.called)

    def authorize(self, username, password, hosting_url, *args, **kwargs):
        """Authorizes the GitLab repository.

        GitLab uses HTTP Basic Auth for the API, so this will store the
        provided password, encrypted, for use in later API requests.
        """
        if self._is_email(username):
            login_key = "email"
        else:
            login_key = "login"

        # This will raise an exception if it fails, which the form will
        # catch.
        try:
            rsp, headers = self.client.json_post(
                url=self._build_api_url(hosting_url, "session"), fields={login_key: username, "password": password}
            )
        except HTTPError as e:
            if e.code == 404:
                raise HostingServiceError(ugettext("A GitLab server was not found at the " "provided URL."))
            elif e.code == 401:
                raise AuthorizationError(ugettext("The username or password is incorrect."))
            else:
                logging.exception("Unexpected HTTP error when linking GitLab " "account for %s: %s", username, e)
                raise HostingServiceError(ugettext("Unexpected HTTP error %s.") % e.code)
        except Exception as e:
            logging.exception("Unexpected error when linking GitLab account " "for %s: %s", username, e)
            raise HostingServiceError(ugettext('Unexpected error "%s"') % e)

        self.account.data["private_token"] = encrypt_password(rsp["private_token"])
        self.account.save()

    def authorize(self, username, password, hosting_url, *args, **kwargs):
        """Authorizes the GitLab repository.

        GitLab uses HTTP Basic Auth for the API, so this will store the
        provided password, encrypted, for use in later API requests.
        """
        if self._is_email(username):
            login_key = 'email'
        else:
            login_key = 'login'

        # This will raise an exception if it fails, which the form will
        # catch.
        try:
            rsp, headers = self.client.json_post(
                url=self._build_api_url(hosting_url, 'session'),
                fields={
                    login_key: username,
                    'password': password,
                })
        except HTTPError as e:
            if e.code == 404:
                raise HostingServiceError(
                    ugettext('A GitLab server was not found at the '
                             'provided URL.'))
            elif e.code == 401:
                raise AuthorizationError(
                    ugettext('The username or password is incorrect.'))
            else:
                raise

        self.account.data['private_token'] = \
            encrypt_password(rsp['private_token'])
        self.account.save()

    def test_get_branches(self):
        """Testing GitLab get_branches implementation"""
        branches_api_response = json.dumps([
            {
                'name': 'master',
                'commit': {
                    'id': 'ed899a2f4b50b4370feeea94676502b42383c746'
                }
            },
            {
                'name': 'branch1',
                'commit': {
                    'id': '6104942438c14ec7bd21c6cd5bd995272b3faff6'
                }
            },
            {
                'name': 'branch2',
                'commit': {
                    'id': '21b3bcabcff2ab3dc3c9caa172f783aad602c0b0'
                }
            },
            {
                'branch-name': 'branch3',
                'commit': {
                    'id': 'd5a3ff139356ce33e37e73add446f16869741b50'
                }
            }
        ])

        def _http_get(self, *args, **kwargs):
            return branches_api_response, None

        account = self._get_hosting_account(use_url=True)
        account.data['private_token'] = encrypt_password('abc123')

        service = account.service

        repository = Repository(hosting_account=account)
        repository.extra_data = {'gitlab_project_id': 123456}

        self.spy_on(service.client.http_get, call_fake=_http_get)

        branches = service.get_branches(repository)

        self.assertTrue(service.client.http_get.called)
        self.assertEqual(len(branches), 3)
        self.assertEqual(
            branches,
            [
                Branch(id='master',
                       commit='ed899a2f4b50b4370feeea94676502b42383c746',
                       default=True),
                Branch(id='branch1',
                       commit='6104942438c14ec7bd21c6cd5bd995272b3faff6',
                       default=False),
                Branch(id='branch2',
                       commit='21b3bcabcff2ab3dc3c9caa172f783aad602c0b0',
                       default=False)
            ])

    def authorize(self, username, password, hosting_url, local_site_name=None, *args, **kwargs):
        """Authorizes the Beanstalk repository.

        Beanstalk uses HTTP Basic Auth for the API, so this will store the
        provided password, encrypted, for use in later API requests.
        """
        self.account.data["password"] = encrypt_password(password)
        self.account.save()

 def test_encrypt_password(self):
     """Testing encrypt_password"""
     # The encrypted value will change every time, since the iv changes,
     # so we can't compare a direct value. Instead, we need to ensure that
     # we can decrypt what we encrypt.
     self.assertEqual(
         decrypt_password(encrypt_password(self.PLAIN_TEXT)),
         self.PLAIN_TEXT)

    def test_encrypt_password_with_custom_key(self):
        """Testing encrypt_password with custom key"""
        # The encrypted value will change every time, since the iv changes,
        # so we can't compare a direct value. Instead, we need to ensure that
        # we can decrypt what we encrypt.
        encrypted = encrypt_password(self.PLAIN_TEXT, key=self.CUSTOM_KEY)

        self.assertEqual(decrypt_password(encrypted, key=self.CUSTOM_KEY),
                         self.PLAIN_TEXT)

    def authorize(self, username, password, *args, **kwargs):
        """Authorizes the Bitbucket repository.

        Bitbucket supports HTTP Basic Auth or OAuth for the API. We use
        HTTP Basic Auth for now, and we store provided password,
        encrypted, for use in later API requests.
        """
        self.account.data["password"] = encrypt_password(password)
        self.account.save()

    def authorize(self, username, password, unfuddle_account_domain=None, *args, **kwargs):
        """Authorizes the Unfuddle repository.

        Unfuddle uses HTTP Basic Auth for the API, so this will store the
        provided password, encrypted, for use in later API requests.
        """
        # This will raise an exception if it fails, which the form will
        # catch.
        self._api_get(self._build_api_url(unfuddle_account_domain, "account/"), username=username, password=password)

        self.account.data["password"] = encrypt_password(password)
        self.account.save()

    def test_get_commits(self):
        """Testing ReviewBoardGateway get_commits implementation"""
        commits_api_response = json.dumps([
            {
                'author': 'myname',
                'id': 'bfdde95432b3af879af969bd2377dc3e55ee46e6',
                'date': '2015-02-13 22:34:01 -0700 -0700',
                'message': 'mymessage',
                'parent_id': '304c53c163aedfd0c0e0933776f09c24b87f5944',
            },
            {
                'author': 'myname',
                'id': '304c53c163aedfd0c0e0933776f09c24b87f5944',
                'date': '2015-02-13 22:32:42 -0700 -0700',
                'message': 'mymessage',
                'parent_id': 'fa1330719893098ae397356e8125c2aa45b49221',
            },
            {
                'author': 'anothername',
                'id': 'fa1330719893098ae397356e8125c2aa45b49221',
                'date': '2015-02-12 16:01:48 -0700 -0700',
                'message': 'mymessage',
                'parent_id': '',
            }
        ])

        def _http_get(self, *args, **kwargs):
            return commits_api_response, None

        account = self._get_hosting_account()
        account.data['private_token'] = encrypt_password('abc123')

        repository = Repository(hosting_account=account)
        repository.extra_data = {
            'rbgateway_repo_name': 'myrepo',
        }

        service = account.service
        self.spy_on(service.client.http_get, call_fake=_http_get)

        commits = service.get_commits(
            repository, branch='bfdde95432b3af879af969bd2377dc3e55ee46e6')

        self.assertTrue(service.client.http_get.called)

        self.assertEqual(len(commits), 3)
        self.assertEqual(commits[0].parent, commits[1].id)
        self.assertEqual(commits[1].parent, commits[2].id)
        self.assertEqual(commits[0].date, '2015-02-13 22:34:01 -0700 -0700')
        self.assertEqual(commits[1].id,
                         '304c53c163aedfd0c0e0933776f09c24b87f5944')
        self.assertEqual(commits[2].author_name, 'anothername')
        self.assertEqual(commits[2].parent, '')

    def test_check_repository_with_dot_git(self):
        """Testing Bitbucket check_repository with .git"""
        account = self._get_hosting_account()
        account.data['password'] = encrypt_password('abc123')
        service = account.service

        self.assertRaisesMessage(
            RepositoryError,
            'Please specify just the name of the repository without ".git".',
            lambda: service.check_repository(
                bitbucket_team_name='myteam',
                bitbucket_team_repo_name='myrepo.git',
                plan='team'))

    def test_check_repository(self):
        """Testing that ReviewBoardGateway can find the repository"""
        def _http_get(service, url, *args, **kwargs):
            self.assertEqual(url, 'https://example.com/repos/myrepo/path')
            return '{}', {}

        account = self._get_hosting_account(use_url=True)
        service = account.service
        self.spy_on(service.client.http_get, call_fake=_http_get)
        account.data['private_token'] = encrypt_password('abc123')

        service.check_repository(path='https://example.com/repos/myrepo/path')
        self.assertTrue(service.client.http_get.called)

    def test_get_change(self):
        """Testing BitBucket get_change"""
        commit_sha = '1c44b461cebe5874a857c51a4a13a849a4d1e52d'
        parent_sha = '44568f7d33647d286691517e6325fea5c7a21d5e'

        commits_api_response = json.dumps({
            'hash': commit_sha,
            'author': {
                'raw': 'Some User <[email protected]>',
            },
            'date': '2017-01-24T13:11:22+00:00',
            'message': 'This is a message.',
            'parents': [{'hash': parent_sha}],
        })

        diff_api_response = b'This is a test \xc7.'
        norm_diff_api_response = b'This is a test \xc7.\n'

        def _http_get(service, url, *args, **kwargs):
            if url == ('https://bitbucket.org/api/2.0/repositories/'
                       'myuser/myrepo/commit/%s?'
                       'fields=author.raw%%2Chash%%2Cdate%%2C'
                       'message%%2Cparents.hash'
                       % commit_sha):
                return commits_api_response, None
            elif url == ('https://bitbucket.org/api/2.0/repositories/'
                         'myuser/myrepo/diff/%s' % commit_sha):
                return diff_api_response, None
            else:
                self.fail('Unexpected URL %s' % url)

        account = self._get_hosting_account()
        service = account.service
        repository = Repository(hosting_account=account,
                                tool=Tool.objects.get(name='Git'))
        repository.extra_data = {
            'bitbucket_repo_name': 'myrepo',
        }

        account.data['password'] = encrypt_password('abc123')

        self.spy_on(service.client.http_get, call_fake=_http_get)

        commit = service.get_change(repository, commit_sha)
        self.assertEqual(commit.id, commit_sha)
        self.assertEqual(commit.author_name, 'Some User <[email protected]>')
        self.assertEqual(commit.message, 'This is a message.')
        self.assertEqual(commit.date, '2017-01-24T13:11:22+00:00')
        self.assertEqual(commit.parent, parent_sha)
        self.assertEqual(commit.diff, norm_diff_api_response)

    def _set_password(self, value):
        """Sets the password for the repository.

        The password will be stored as an encrypted value, prefixed with a
        tab character in order to differentiate between legacy plain-text
        passwords.
        """
        if value:
            value = '%s%s' % (self.ENCRYPTED_PASSWORD_PREFIX,
                              encrypt_password(value))
        else:
            value = ''

        self.encrypted_password = value

    def authorize(self, username, password, *args, **kwargs):
        """Authorizes the Bitbucket repository.

        Bitbucket supports HTTP Basic Auth or OAuth for the API. We use
        HTTP Basic Auth for now, and we store provided password,
        encrypted, for use in later API requests.
        """
        self.account.data['password'] = encrypt_password(password)

        try:
            self._api_get(self._build_api_url('user'))
            self.account.save()
        except Exception:
            del self.account.data['password']
            raise