mirror of
https://chromium.googlesource.com/chromium/tools/depot_tools.git
synced 2026-01-11 02:31:29 +00:00
90eb3fe8ba
This CL adds a new --allow-conflicts flag to the `git cl cherry-pick` command. This allows users to create cherry-pick CLs on Gerrit even if they result in merge conflicts. The created CL will be in a WIP state, allowing the user to resolve conflicts in the Gerrit UI or locally. Additionally, this CL: - Updates `CMDcherry_pick` to print a regex-friendly warning if the created cherry-pick contains conflicts: "Warning: Change <URL> contains merge conflicts" (needed for the Chrome Cherry Picker) - Fixes some existing test failures in `gerrit_util_test.py` related to authentication mocking in the current environment. - Adds unit tests for the new flag in both `git_cl_test.py` and `gerrit_util_test.py`. Bug: 439992429 Change-Id: I306e21329688a31a63c9996f1405f5ef5ad07108 Reviewed-on: https://chromium-review.googlesource.com/c/chromium/tools/depot_tools/+/7319362 Reviewed-by: Gavin Mak <gavinmak@google.com> Commit-Queue: Gennady Tsitovich <gtsitovich@google.com>
1069 lines
43 KiB
Python
Executable File
1069 lines
43 KiB
Python
Executable File
#!/usr/bin/env vpython3
|
|
# coding=utf-8
|
|
# Copyright (c) 2019 The Chromium Authors. All rights reserved.
|
|
# Use of this source code is governed by a BSD-style license that can be
|
|
# found in the LICENSE file.
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import unittest
|
|
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from unittest import mock
|
|
|
|
import httplib2
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import scm_mock
|
|
|
|
import gerrit_util
|
|
import auth
|
|
import metrics
|
|
import scm
|
|
import subprocess2
|
|
|
|
RUN_SUBPROC_TESTS = 'RUN_SUBPROC_TESTS' in os.environ
|
|
|
|
|
|
def makeConn(host: str) -> gerrit_util.HttpConn:
|
|
"""Makes an empty gerrit_util.HttpConn for the given host."""
|
|
return gerrit_util.HttpConn(
|
|
req_uri='???',
|
|
req_method='GET',
|
|
req_host=host,
|
|
req_headers={},
|
|
req_body=None,
|
|
)
|
|
|
|
|
|
class CookiesAuthenticatorTest(unittest.TestCase):
|
|
_GITCOOKIES = '\n'.join([
|
|
'\t'.join([
|
|
'chromium.googlesource.com',
|
|
'FALSE',
|
|
'/',
|
|
'TRUE',
|
|
'2147483647',
|
|
'o',
|
|
'git-user.chromium.org=1/chromium-secret',
|
|
]),
|
|
'\t'.join([
|
|
'chromium-review.googlesource.com',
|
|
'FALSE',
|
|
'/',
|
|
'TRUE',
|
|
'2147483647',
|
|
'o',
|
|
'git-user.chromium.org=1/chromium-secret',
|
|
]),
|
|
'\t'.join([
|
|
'.example.com',
|
|
'FALSE',
|
|
'/',
|
|
'TRUE',
|
|
'2147483647',
|
|
'o',
|
|
'example-bearer-token',
|
|
]),
|
|
'\t'.join([
|
|
'another-path.example.com',
|
|
'FALSE',
|
|
'/foo',
|
|
'TRUE',
|
|
'2147483647',
|
|
'o',
|
|
'git-example.com=1/another-path-secret',
|
|
]),
|
|
'\t'.join([
|
|
'another-key.example.com',
|
|
'FALSE',
|
|
'/',
|
|
'TRUE',
|
|
'2147483647',
|
|
'not-o',
|
|
'git-example.com=1/another-key-secret',
|
|
]),
|
|
'#' + '\t'.join([
|
|
'chromium-review.googlesource.com',
|
|
'FALSE',
|
|
'/',
|
|
'TRUE',
|
|
'2147483647',
|
|
'o',
|
|
'git-invalid-user.chromium.org=1/invalid-chromium-secret',
|
|
]),
|
|
'Some unrelated line\t that should not be here',
|
|
])
|
|
|
|
def setUp(self):
|
|
mock.patch('gclient_utils.FileRead',
|
|
return_value=self._GITCOOKIES).start()
|
|
mock.patch('os.getenv', return_value={}).start()
|
|
mock.patch('os.environ', {'HOME': '$HOME'}).start()
|
|
mock.patch('os.getcwd', return_value='/fame/cwd').start()
|
|
mock.patch('os.path.exists', return_value=True).start()
|
|
mock.patch(
|
|
'git_common.run',
|
|
side_effect=[
|
|
subprocess2.CalledProcessError(1, ['cmd'], 'cwd', 'out', 'err')
|
|
],
|
|
).start()
|
|
scm_mock.GIT(self)
|
|
|
|
self.addCleanup(mock.patch.stopall)
|
|
self.maxDiff = None
|
|
|
|
def assertAuthenticatedConnAuth(
|
|
self, cookieAuth: gerrit_util.CookiesAuthenticator, host: str,
|
|
expected: str):
|
|
conn = makeConn(host)
|
|
cookieAuth.authenticate(conn)
|
|
self.assertEqual(conn.req_headers['Authorization'], expected)
|
|
|
|
def testGetNewPasswordUrl(self):
|
|
cookieAuth = gerrit_util.CookiesAuthenticator()
|
|
self.assertEqual(
|
|
'https://chromium.googlesource.com/new-password',
|
|
cookieAuth.get_new_password_url('chromium.googlesource.com'))
|
|
self.assertEqual(
|
|
'https://chrome-internal.googlesource.com/new-password',
|
|
cookieAuth.get_new_password_url(
|
|
'chrome-internal-review.googlesource.com'))
|
|
|
|
def testGetNewPasswordMessage(self):
|
|
cookieAuth = gerrit_util.CookiesAuthenticator()
|
|
self.assertIn(
|
|
'https://chromium.googlesource.com/new-password',
|
|
cookieAuth._get_new_password_message(
|
|
'chromium-review.googlesource.com'))
|
|
self.assertIn(
|
|
'https://chrome-internal.googlesource.com/new-password',
|
|
cookieAuth._get_new_password_message(
|
|
'chrome-internal.googlesource.com'))
|
|
|
|
def testGetGitcookiesPath(self):
|
|
self.assertEqual(
|
|
os.path.expanduser(os.path.join('~', '.gitcookies')),
|
|
gerrit_util.CookiesAuthenticator().get_gitcookies_path())
|
|
|
|
scm.GIT.SetConfig(os.getcwd(), 'http.cookiefile', '/some/path')
|
|
self.assertEqual(
|
|
'/some/path',
|
|
gerrit_util.CookiesAuthenticator().get_gitcookies_path())
|
|
|
|
os.getenv.return_value = 'git-cookies-path'
|
|
self.assertEqual(
|
|
'git-cookies-path',
|
|
gerrit_util.CookiesAuthenticator().get_gitcookies_path())
|
|
os.getenv.assert_called_with('GIT_COOKIES_PATH')
|
|
|
|
def testGitcookies(self):
|
|
cookieAuth = gerrit_util.CookiesAuthenticator()
|
|
self.assertEqual(
|
|
cookieAuth.gitcookies, {
|
|
'chromium.googlesource.com':
|
|
('git-user.chromium.org', '1/chromium-secret'),
|
|
'chromium-review.googlesource.com':
|
|
('git-user.chromium.org', '1/chromium-secret'),
|
|
'.example.com': ('', 'example-bearer-token'),
|
|
})
|
|
|
|
def testGetAuthHeader(self):
|
|
expected_chromium_header = (
|
|
'Basic Z2l0LXVzZXIuY2hyb21pdW0ub3JnOjEvY2hyb21pdW0tc2VjcmV0')
|
|
|
|
cookieAuth = gerrit_util.CookiesAuthenticator()
|
|
self.assertAuthenticatedConnAuth(cookieAuth,
|
|
'chromium.googlesource.com',
|
|
expected_chromium_header)
|
|
self.assertAuthenticatedConnAuth(cookieAuth,
|
|
'chromium-review.googlesource.com',
|
|
expected_chromium_header)
|
|
self.assertAuthenticatedConnAuth(cookieAuth, 'some-review.example.com',
|
|
'Bearer example-bearer-token')
|
|
|
|
def testGetAuthEmail(self):
|
|
cookieAuth = gerrit_util.CookiesAuthenticator()
|
|
self.assertEqual('user@chromium.org',
|
|
cookieAuth.get_auth_email('chromium.googlesource.com'))
|
|
self.assertEqual(
|
|
'user@chromium.org',
|
|
cookieAuth.get_auth_email('chromium-review.googlesource.com'))
|
|
self.assertIsNone(cookieAuth.get_auth_email('some-review.example.com'))
|
|
|
|
|
|
class GceAuthenticatorTest(unittest.TestCase):
|
|
def setUp(self):
|
|
super(GceAuthenticatorTest, self).setUp()
|
|
mock.patch('httplib2.Http').start()
|
|
mock.patch('os.getenv', return_value=None).start()
|
|
mock.patch('gerrit_util.time_sleep').start()
|
|
mock.patch('gerrit_util.time_time').start()
|
|
self.addCleanup(mock.patch.stopall)
|
|
|
|
# GceAuthenticator has class variables that cache the results. Build a
|
|
# new class for every test to avoid inter-test dependencies.
|
|
class GceAuthenticator(gerrit_util.GceAuthenticator):
|
|
pass
|
|
|
|
self.GceAuthenticator = GceAuthenticator
|
|
|
|
def assertAuthenticatedToken(self, token: Optional[str]):
|
|
conn = makeConn('some.example.com')
|
|
self.GceAuthenticator().authenticate(conn)
|
|
if token is None:
|
|
self.assertNotIn('Authorization', conn.req_headers)
|
|
else:
|
|
self.assertEqual(conn.req_headers['Authorization'], token)
|
|
|
|
def testIsGce_EnvVarSkip(self, *_mocks):
|
|
os.getenv.return_value = '1'
|
|
self.assertFalse(self.GceAuthenticator.is_applicable())
|
|
os.getenv.assert_called_once_with('SKIP_GCE_AUTH_FOR_GIT')
|
|
|
|
def testIsGce_Error(self):
|
|
httplib2.Http().request.side_effect = httplib2.HttpLib2Error
|
|
self.assertFalse(self.GceAuthenticator.is_applicable())
|
|
|
|
def testIsGce_500(self):
|
|
httplib2.Http().request.return_value = (mock.Mock(status=500), None)
|
|
self.assertFalse(self.GceAuthenticator.is_applicable())
|
|
last_call = gerrit_util.time_sleep.mock_calls[-1]
|
|
self.assertLessEqual(last_call, mock.call(43.0))
|
|
|
|
def testIsGce_FailsThenSucceeds(self):
|
|
response = mock.Mock(status=200)
|
|
response.get.return_value = 'Google'
|
|
httplib2.Http().request.side_effect = [
|
|
(mock.Mock(status=500), None),
|
|
(response, 'who cares'),
|
|
]
|
|
self.assertTrue(self.GceAuthenticator.is_applicable())
|
|
|
|
def testIsGce_MetadataFlavorIsNotGoogle(self):
|
|
response = mock.Mock(status=200)
|
|
response.get.return_value = None
|
|
httplib2.Http().request.return_value = (response, 'who cares')
|
|
self.assertFalse(self.GceAuthenticator.is_applicable())
|
|
response.get.assert_called_once_with('metadata-flavor')
|
|
|
|
def testIsGce_ResultIsCached(self):
|
|
response = mock.Mock(status=200)
|
|
response.get.return_value = 'Google'
|
|
httplib2.Http().request.side_effect = [(response, 'who cares')]
|
|
self.assertTrue(self.GceAuthenticator.is_applicable())
|
|
self.assertTrue(self.GceAuthenticator.is_applicable())
|
|
httplib2.Http().request.assert_called_once()
|
|
|
|
def testGetAuthHeader_Error(self):
|
|
httplib2.Http().request.side_effect = httplib2.HttpLib2Error
|
|
self.assertAuthenticatedToken(None)
|
|
|
|
def testGetAuthHeader_500(self):
|
|
httplib2.Http().request.return_value = (mock.Mock(status=500), None)
|
|
self.assertAuthenticatedToken(None)
|
|
|
|
def testGetAuthHeader_Non200(self):
|
|
httplib2.Http().request.return_value = (mock.Mock(status=403), None)
|
|
self.assertAuthenticatedToken(None)
|
|
|
|
def testGetAuthHeader_OK(self):
|
|
httplib2.Http().request.return_value = (
|
|
mock.Mock(status=200),
|
|
'{"expires_in": 125, "token_type": "TYPE", "access_token": "TOKEN"}'
|
|
)
|
|
gerrit_util.time_time.return_value = 0
|
|
self.assertAuthenticatedToken('TYPE TOKEN')
|
|
|
|
def testGetAuthHeader_Cache(self):
|
|
httplib2.Http().request.return_value = (
|
|
mock.Mock(status=200),
|
|
'{"expires_in": 125, "token_type": "TYPE", "access_token": "TOKEN"}'
|
|
)
|
|
gerrit_util.time_time.return_value = 0
|
|
self.assertAuthenticatedToken('TYPE TOKEN')
|
|
self.assertAuthenticatedToken('TYPE TOKEN')
|
|
httplib2.Http().request.assert_called_once()
|
|
|
|
def testGetAuthHeader_CacheOld(self):
|
|
httplib2.Http().request.return_value = (
|
|
mock.Mock(status=200),
|
|
'{"expires_in": 125, "token_type": "TYPE", "access_token": "TOKEN"}'
|
|
)
|
|
gerrit_util.time_time.side_effect = [0, 100, 200]
|
|
self.assertAuthenticatedToken('TYPE TOKEN')
|
|
self.assertAuthenticatedToken('TYPE TOKEN')
|
|
self.assertEqual(2, len(httplib2.Http().request.mock_calls))
|
|
|
|
def testAttemptAuthenticateWithReAuth(self):
|
|
authn = self.GceAuthenticator()
|
|
with mock.patch.object(authn, 'authenticate') as mock_authenticate:
|
|
conn = makeConn('some.example.com')
|
|
context = auth.ReAuthContext(host='some.example.com',
|
|
project='some/project')
|
|
self.assertTrue(
|
|
authn.attempt_authenticate_with_reauth(conn, context))
|
|
mock_authenticate.assert_called_once_with(conn)
|
|
|
|
|
|
class LuciContextAuthenticatorTest(unittest.TestCase):
|
|
|
|
@mock.patch('gerrit_util.LuciContextAuthenticator.authenticate')
|
|
def testAttemptAuthenticateWithReAuth(self, mock_authenticate):
|
|
luci_context_auth = gerrit_util.LuciContextAuthenticator()
|
|
conn = makeConn('some.example.com')
|
|
context = auth.ReAuthContext(host='some.example.com',
|
|
project='some/project')
|
|
self.assertTrue(
|
|
luci_context_auth.attempt_authenticate_with_reauth(conn, context))
|
|
mock_authenticate.assert_called_once_with(conn)
|
|
|
|
|
|
class GitCredsAuthenticatorTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
super(GitCredsAuthenticatorTest, self).setUp()
|
|
self.authenticator = gerrit_util.ChainedAuthenticator(
|
|
[gerrit_util.GitCredsAuthenticator()])
|
|
|
|
@mock.patch('gerrit_util.GitCredsAuthenticator.gerrit_account_exists',
|
|
return_value=True)
|
|
@mock.patch('gerrit_util.GitCredsAuthenticator._is_usehttppath_set',
|
|
return_value=True)
|
|
@mock.patch('auth.GerritAuthenticator.get_access_token',
|
|
return_value="Bearer abcd")
|
|
def testEnsureAuthenticated(self, mock_get_access_token, _set, _exists):
|
|
bypassable, err_msg = self.authenticator.ensure_authenticated(
|
|
gerrit_host='chromium-review.googlesource.com',
|
|
git_host="chromium.googlesource.com")
|
|
mock_get_access_token.assert_called_once()
|
|
self.assertTrue(bypassable, msg=err_msg)
|
|
self.assertEqual(err_msg, "")
|
|
|
|
@mock.patch('gerrit_util.GitCredsAuthenticator._is_usehttppath_set',
|
|
return_value=True)
|
|
@mock.patch('gerrit_util.GitCredsAuthenticator.gerrit_account_exists',
|
|
return_value=True)
|
|
@mock.patch('auth.GerritAuthenticator.get_authorization_header',
|
|
return_value="BearerReAuth xyz")
|
|
def testEnsureAuthenticatedWithReAuth(self, mock_get_authorization_header,
|
|
_exists, _set):
|
|
reauth_context = auth.ReAuthContext(
|
|
host="chromium-review.googlesource.com", project="chromium/src")
|
|
gerrit_host = "chromium-review.googlesource.com"
|
|
git_host = "chromium.googlesource.com"
|
|
|
|
bypassable, err_msg = self.authenticator.ensure_authenticated(
|
|
gerrit_host=gerrit_host,
|
|
git_host=git_host,
|
|
reauth_context=reauth_context)
|
|
|
|
mock_get_authorization_header.assert_called_once_with(reauth_context)
|
|
self.assertTrue(bypassable, msg=err_msg)
|
|
self.assertEqual(err_msg, "")
|
|
|
|
@mock.patch('gerrit_util.GitCredsAuthenticator._is_usehttppath_set',
|
|
return_value=False)
|
|
@mock.patch('gerrit_util.GitCredsAuthenticator.gerrit_account_exists',
|
|
return_value=True)
|
|
@mock.patch('auth.GerritAuthenticator.get_authorization_header',
|
|
return_value="BearerReAuth xyz")
|
|
def testEnsureAuthenticatedMissingUseHttpPath(self, mock_get_header,
|
|
_exists, _set):
|
|
reauth_context = auth.ReAuthContext(
|
|
host="chromium-review.googlesource.com", project="chromium/src")
|
|
gerrit_host = "chromium-review.googlesource.com"
|
|
git_host = "chromium.googlesource.com"
|
|
|
|
bypassable, err_msg = self.authenticator.ensure_authenticated(
|
|
gerrit_host=gerrit_host,
|
|
git_host=git_host,
|
|
reauth_context=reauth_context)
|
|
|
|
self.assertFalse(bypassable)
|
|
self.assertRegex(err_msg, "You have not set credential.useHttpPath")
|
|
|
|
@mock.patch('auth.GerritAuthenticator.get_authorization_header',
|
|
side_effect=auth.GitReAuthRequiredError())
|
|
@mock.patch('gerrit_util.GitCredsAuthenticator.gerrit_account_exists',
|
|
return_value=True)
|
|
def testEnsureAuthenticatedMissingReAuth(self, _exists,
|
|
mock_get_authorization_header):
|
|
gerrit_host = "chromium-review.googlesource.com"
|
|
git_host = "chromium.googlesource.com"
|
|
reauth_context = auth.ReAuthContext(
|
|
host="chromium-review.googlesource.com", project="chromium/src")
|
|
|
|
bypassable, err_msg = self.authenticator.ensure_authenticated(
|
|
gerrit_host=gerrit_host,
|
|
git_host=git_host,
|
|
reauth_context=reauth_context)
|
|
|
|
mock_get_authorization_header.assert_called_once_with(reauth_context)
|
|
self.assertFalse(bypassable)
|
|
self.assertRegex(err_msg, "You have not done ReAuth")
|
|
|
|
class GerritUtilTest(unittest.TestCase):
|
|
def setUp(self):
|
|
super(GerritUtilTest, self).setUp()
|
|
mock.patch('gerrit_util.LOGGER').start()
|
|
mock.patch('gerrit_util.time_sleep').start()
|
|
mock.patch('metrics.collector').start()
|
|
mock.patch('metrics_utils.extract_http_metrics',
|
|
return_value='http_metrics').start()
|
|
self.addCleanup(mock.patch.stopall)
|
|
|
|
def testQueryString(self):
|
|
self.assertEqual('', gerrit_util._QueryString([]))
|
|
self.assertEqual('first%20param%2B',
|
|
gerrit_util._QueryString([], 'first param+'))
|
|
self.assertEqual(
|
|
'key:val+foo:bar',
|
|
gerrit_util._QueryString([('key', 'val'), ('foo', 'bar')]))
|
|
self.assertEqual(
|
|
'first%20param%2B+key:val+foo:bar',
|
|
gerrit_util._QueryString([('key', 'val'), ('foo', 'bar')],
|
|
'first param+'))
|
|
|
|
@mock.patch('gerrit_util.CookiesAuthenticator._get_auth_for_host')
|
|
@mock.patch('gerrit_util._Authenticator.get')
|
|
def testCreateHttpConn_Basic(self, mockAuth, cookieAuth):
|
|
mockAuth.return_value = gerrit_util.CookiesAuthenticator()
|
|
cookieAuth.return_value = None
|
|
|
|
conn = gerrit_util.CreateHttpConn('host.example.com', 'foo/bar')
|
|
self.assertEqual('host.example.com', conn.req_host)
|
|
self.assertEqual(
|
|
{
|
|
'uri': 'https://host.example.com/a/foo/bar',
|
|
'method': 'GET',
|
|
'headers': {},
|
|
'body': None,
|
|
}, conn.req_params)
|
|
|
|
@mock.patch('gerrit_util.CookiesAuthenticator._get_auth_for_host')
|
|
@mock.patch('gerrit_util._Authenticator.get')
|
|
def testCreateHttpConn_Authenticated(self, mockAuth, cookieAuth):
|
|
mockAuth.return_value = gerrit_util.CookiesAuthenticator()
|
|
cookieAuth.return_value = (None, 'token')
|
|
|
|
conn = gerrit_util.CreateHttpConn('host.example.com',
|
|
'foo/bar',
|
|
headers={'header': 'value'})
|
|
self.assertEqual('host.example.com', conn.req_host)
|
|
self.assertEqual(
|
|
{
|
|
'uri': 'https://host.example.com/a/foo/bar',
|
|
'method': 'GET',
|
|
'headers': {
|
|
'Authorization': 'Bearer token',
|
|
'header': 'value'
|
|
},
|
|
'body': None,
|
|
}, conn.req_params)
|
|
|
|
@mock.patch('gerrit_util.CookiesAuthenticator._get_auth_for_host')
|
|
@mock.patch('gerrit_util._Authenticator')
|
|
def testCreateHttpConn_Body(self, mockAuth, cookieAuth):
|
|
mockAuth.return_value = gerrit_util.CookiesAuthenticator()
|
|
cookieAuth.return_value = None
|
|
|
|
conn = gerrit_util.CreateHttpConn('host.example.com',
|
|
'foo/bar',
|
|
body={
|
|
'l': [1, 2, 3],
|
|
'd': {
|
|
'k': 'v'
|
|
}
|
|
})
|
|
self.assertEqual('host.example.com', conn.req_host)
|
|
self.assertEqual(
|
|
{
|
|
'uri': 'https://host.example.com/a/foo/bar',
|
|
'method': 'GET',
|
|
'headers': {
|
|
'Content-Type': 'application/json'
|
|
},
|
|
'body': '{"d": {"k": "v"}, "l": [1, 2, 3]}',
|
|
}, conn.req_params)
|
|
|
|
@mock.patch('auth.GerritAuthenticator.get_authorization_header')
|
|
@mock.patch('gerrit_util._Authenticator.get')
|
|
def testCreateHttpConn_ReAuth(self, mockAuth, getAuthHeader):
|
|
mockAuth.return_value = gerrit_util.GitCredsAuthenticator()
|
|
getAuthHeader.return_value = "BearerReAuth cafe"
|
|
reauth_context = auth.ReAuthContext(host="chromium",
|
|
project="infra/infra")
|
|
gerrit_util.CreateHttpConn('host.example.com',
|
|
'foo/bar',
|
|
reauth_context=reauth_context)
|
|
getAuthHeader.assert_called_with(reauth_context)
|
|
|
|
@mock.patch('auth.GerritAuthenticator.get_access_token')
|
|
@mock.patch('auth.GerritAuthenticator.get_authorization_header')
|
|
@mock.patch('gerrit_util._Authenticator.get')
|
|
def testCreateHttpConn_ReAuthOptional(self, mockAuth, getAuthHeader,
|
|
getAccessToken):
|
|
mockAuth.return_value = gerrit_util.GitCredsAuthenticator()
|
|
getAuthHeader.side_effect = auth.GitReAuthRequiredError()
|
|
getAccessToken.return_value = "decafe"
|
|
reauth_context = auth.ReAuthContext(host="chromium",
|
|
project="infra/infra")
|
|
gerrit_util.CreateHttpConn('host.example.com',
|
|
'foo/bar',
|
|
reauth_context=reauth_context,
|
|
reauth_is_optional=True)
|
|
getAuthHeader.assert_called_with(reauth_context)
|
|
getAccessToken.assert_called()
|
|
|
|
@mock.patch('auth.GerritAuthenticator.get_access_token')
|
|
@mock.patch('auth.GerritAuthenticator.get_authorization_header')
|
|
@mock.patch('gerrit_util._Authenticator.get')
|
|
def testCreateHttpConn_ReAuthRequired(self, mockAuth, getAuthHeader,
|
|
getAccessToken):
|
|
mockAuth.return_value = gerrit_util.GitCredsAuthenticator()
|
|
getAuthHeader.side_effect = auth.GitReAuthRequiredError()
|
|
getAccessToken.return_value = "decafe"
|
|
reauth_context = auth.ReAuthContext(host="chromium",
|
|
project="infra/infra")
|
|
with self.assertRaises(auth.GitReAuthRequiredError):
|
|
gerrit_util.CreateHttpConn('host.example.com',
|
|
'foo/bar',
|
|
reauth_context=reauth_context,
|
|
reauth_is_optional=False)
|
|
getAuthHeader.assert_called_with(reauth_context)
|
|
getAccessToken.assert_not_called()
|
|
|
|
def testReadHttpResponse_200(self):
|
|
conn = mock.Mock()
|
|
conn.req_params = {'uri': 'uri', 'method': 'method'}
|
|
conn.request.return_value = (mock.Mock(status=200),
|
|
b'content\xe2\x9c\x94')
|
|
|
|
content = gerrit_util.ReadHttpResponse(conn)
|
|
self.assertEqual('content✔', content.getvalue())
|
|
metrics.collector.add_repeated.assert_called_once_with(
|
|
'http_requests', 'http_metrics')
|
|
|
|
def testReadHttpResponse_AuthenticationIssue(self):
|
|
for status in (302, 401, 403):
|
|
response = mock.Mock(status=status)
|
|
response.get.return_value = None
|
|
conn = mock.Mock(req_params={'uri': 'uri', 'method': 'method'})
|
|
conn.request.return_value = (response, b'')
|
|
|
|
with mock.patch('sys.stdout', StringIO()):
|
|
with self.assertRaises(gerrit_util.GerritError) as cm:
|
|
gerrit_util.ReadHttpResponse(conn)
|
|
|
|
self.assertEqual(status, cm.exception.http_status)
|
|
self.assertIn('Your Gerrit credentials might be misconfigured',
|
|
sys.stdout.getvalue())
|
|
|
|
def testReadHttpResponse_ClientError(self):
|
|
conn = mock.Mock(req_params={'uri': 'uri', 'method': 'method'})
|
|
conn.request.return_value = (mock.Mock(status=404), b'')
|
|
|
|
with self.assertRaises(gerrit_util.GerritError) as cm:
|
|
gerrit_util.ReadHttpResponse(conn)
|
|
|
|
self.assertEqual(404, cm.exception.http_status)
|
|
|
|
def readHttpResponse_ServerErrorHelper(self, status):
|
|
conn = mock.Mock(req_params={'uri': 'uri', 'method': 'method'})
|
|
conn.request.return_value = (mock.Mock(status=status), b'')
|
|
|
|
with self.assertRaises(gerrit_util.GerritError) as cm:
|
|
gerrit_util.ReadHttpResponse(conn)
|
|
|
|
self.assertEqual(status, cm.exception.http_status)
|
|
self.assertEqual(gerrit_util.TRY_LIMIT, len(conn.request.mock_calls))
|
|
last_call = gerrit_util.time_sleep.mock_calls[-1]
|
|
self.assertLessEqual(last_call, mock.call(422.0))
|
|
|
|
def testReadHttpResponse_ServerError(self):
|
|
self.readHttpResponse_ServerErrorHelper(status=404)
|
|
self.readHttpResponse_ServerErrorHelper(status=409)
|
|
self.readHttpResponse_ServerErrorHelper(status=429)
|
|
self.readHttpResponse_ServerErrorHelper(status=500)
|
|
|
|
def testReadHttpResponse_ServerErrorAndSuccess(self):
|
|
conn = mock.Mock(req_params={'uri': 'uri', 'method': 'method'})
|
|
conn.request.side_effect = [
|
|
(mock.Mock(status=500), b''),
|
|
(mock.Mock(status=200), b'content\xe2\x9c\x94'),
|
|
]
|
|
|
|
self.assertEqual('content✔',
|
|
gerrit_util.ReadHttpResponse(conn).getvalue())
|
|
self.assertEqual(2, len(conn.request.mock_calls))
|
|
gerrit_util.time_sleep.assert_called_once_with(12.0)
|
|
|
|
def testReadHttpResponse_TimeoutAndSuccess(self):
|
|
conn = mock.Mock(req_params={'uri': 'uri', 'method': 'method'})
|
|
conn.request.side_effect = [
|
|
socket.timeout('timeout'),
|
|
(mock.Mock(status=200), b'content\xe2\x9c\x94'),
|
|
]
|
|
|
|
self.assertEqual('content✔',
|
|
gerrit_util.ReadHttpResponse(conn).getvalue())
|
|
self.assertEqual(2, len(conn.request.mock_calls))
|
|
gerrit_util.time_sleep.assert_called_once_with(12.0)
|
|
|
|
def testReadHttpResponse_SetMaxTries(self):
|
|
conn = mock.Mock(req_params={'uri': 'uri', 'method': 'method'})
|
|
conn.request.side_effect = [
|
|
(mock.Mock(status=409), b'error!'),
|
|
(mock.Mock(status=409), b'error!'),
|
|
(mock.Mock(status=409), b'error!'),
|
|
]
|
|
|
|
self.assertRaises(gerrit_util.GerritError,
|
|
gerrit_util.ReadHttpResponse,
|
|
conn,
|
|
max_tries=2)
|
|
self.assertEqual(2, len(conn.request.mock_calls))
|
|
gerrit_util.time_sleep.assert_called_once_with(12.0)
|
|
|
|
def testReadHttpResponse_Expected404(self):
|
|
conn = mock.Mock()
|
|
conn.req_params = {'uri': 'uri', 'method': 'method'}
|
|
conn.request.return_value = (mock.Mock(status=404),
|
|
b'content\xe2\x9c\x94')
|
|
|
|
content = gerrit_util.ReadHttpResponse(conn, (404, ))
|
|
self.assertEqual('', content.getvalue())
|
|
|
|
@mock.patch('gerrit_util.ReadHttpResponse')
|
|
def testReadHttpJsonResponse_NotJSON(self, mockReadHttpResponse):
|
|
mockReadHttpResponse.return_value = StringIO('not json')
|
|
with self.assertRaises(gerrit_util.GerritError) as cm:
|
|
gerrit_util.ReadHttpJsonResponse(None)
|
|
self.assertEqual(cm.exception.http_status, 200)
|
|
self.assertEqual(cm.exception.message,
|
|
'(200) Unexpected json output: not json')
|
|
|
|
@mock.patch('gerrit_util.ReadHttpResponse')
|
|
def testReadHttpJsonResponse_EmptyValue(self, mockReadHttpResponse):
|
|
mockReadHttpResponse.return_value = StringIO(')]}\'')
|
|
self.assertEqual(gerrit_util.ReadHttpJsonResponse(None), {})
|
|
|
|
@mock.patch('gerrit_util.ReadHttpResponse')
|
|
def testReadHttpJsonResponse_JSON(self, mockReadHttpResponse):
|
|
expected_value = {'foo': 'bar', 'baz': [1, '2', 3]}
|
|
mockReadHttpResponse.return_value = StringIO(')]}\'\n' +
|
|
json.dumps(expected_value))
|
|
self.assertEqual(expected_value, gerrit_util.ReadHttpJsonResponse(None))
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
def testQueryChanges(self, mockJsonResponse, mockCreateHttpConn):
|
|
gerrit_util.QueryChanges('host', [('key', 'val'), ('foo', 'bar baz')],
|
|
'first param',
|
|
limit=500,
|
|
o_params=['PARAM_A', 'PARAM_B'],
|
|
start='start')
|
|
mockCreateHttpConn.assert_called_once_with(
|
|
'host', ('changes/?q=first%20param+key:val+foo:bar+baz'
|
|
'&start=start'
|
|
'&n=500'
|
|
'&o=PARAM_A'
|
|
'&o=PARAM_B'),
|
|
timeout=30.0)
|
|
|
|
def testQueryChanges_NoParams(self):
|
|
self.assertRaises(RuntimeError, gerrit_util.QueryChanges, 'host', [])
|
|
|
|
@mock.patch('gerrit_util.QueryChanges')
|
|
def testGenerateAllChanges(self, mockQueryChanges):
|
|
mockQueryChanges.side_effect = [
|
|
# First results page
|
|
[
|
|
{
|
|
'_number': '4'
|
|
},
|
|
{
|
|
'_number': '3'
|
|
},
|
|
{
|
|
'_number': '2',
|
|
'_more_changes': True
|
|
},
|
|
],
|
|
# Second results page, there are new changes, so second page
|
|
# includes some results from the first page.
|
|
[
|
|
{
|
|
'_number': '2'
|
|
},
|
|
{
|
|
'_number': '1'
|
|
},
|
|
],
|
|
# GenerateAllChanges queries again from the start to get any new
|
|
# changes (5 in this case).
|
|
[
|
|
{
|
|
'_number': '5'
|
|
},
|
|
{
|
|
'_number': '4'
|
|
},
|
|
{
|
|
'_number': '3',
|
|
'_more_changes': True
|
|
},
|
|
],
|
|
]
|
|
|
|
changes = list(gerrit_util.GenerateAllChanges('host', 'params'))
|
|
self.assertEqual([
|
|
{
|
|
'_number': '4'
|
|
},
|
|
{
|
|
'_number': '3'
|
|
},
|
|
{
|
|
'_number': '2',
|
|
'_more_changes': True
|
|
},
|
|
{
|
|
'_number': '1'
|
|
},
|
|
{
|
|
'_number': '5'
|
|
},
|
|
], changes)
|
|
self.assertEqual([
|
|
mock.call('host', 'params', None, 500, None, 0),
|
|
mock.call('host', 'params', None, 500, None, 3),
|
|
mock.call('host', 'params', None, 500, None, 0),
|
|
], mockQueryChanges.mock_calls)
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
def testIsCodeOwnersEnabledOnRepo_Disabled(self, mockJsonResponse,
|
|
mockCreateHttpConn):
|
|
mockJsonResponse.return_value = {'status': {'disabled': True}}
|
|
self.assertFalse(gerrit_util.IsCodeOwnersEnabledOnRepo('host', 'repo'))
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
def testIsCodeOwnersEnabledOnRepo_Enabled(self, mockJsonResponse,
|
|
mockCreateHttpConn):
|
|
mockJsonResponse.return_value = {'status': {}}
|
|
self.assertTrue(gerrit_util.IsCodeOwnersEnabledOnRepo('host', 'repo'))
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
@mock.patch('gerrit_util.GetChangeDetail')
|
|
def testSetReview_ReAuth(self, mockGetChangeDetail, mockJsonResponse,
|
|
mockCreateHttpConn):
|
|
mockJsonResponse.return_value = {"labels": {"Code-Review": 1}}
|
|
mockGetChangeDetail.return_value = {
|
|
"project": "infra/infra",
|
|
}
|
|
|
|
gerrit_util.SetReview('chromium', 123456, labels={'Code-Review': 1})
|
|
|
|
# Check `project` in reauth_context is backfilled.
|
|
mockGetChangeDetail.assert_called_with('chromium', 123456)
|
|
httpConnKwargs = mockCreateHttpConn.call_args[1]
|
|
self.assertIn('reauth_context', httpConnKwargs)
|
|
self.assertEqual(
|
|
auth.ReAuthContext(host='chromium', project='infra/infra'),
|
|
httpConnKwargs['reauth_context'])
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
@mock.patch('gerrit_util.GetChangeDetail')
|
|
def testSetReview_ReAuth_WithProject(self, mockGetChangeDetail,
|
|
mockJsonResponse, mockCreateHttpConn):
|
|
mockJsonResponse.return_value = {"labels": {"Code-Review": 1}}
|
|
|
|
gerrit_util.SetReview('chromium',
|
|
123456,
|
|
labels={'Code-Review': 1},
|
|
project="infra/experimental")
|
|
|
|
# Check reauth_context uses the given project.
|
|
mockGetChangeDetail.assert_not_called()
|
|
httpConnKwargs = mockCreateHttpConn.call_args[1]
|
|
self.assertIn('reauth_context', httpConnKwargs)
|
|
self.assertEqual(
|
|
auth.ReAuthContext(host='chromium', project='infra/experimental'),
|
|
httpConnKwargs['reauth_context'])
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
@mock.patch('gerrit_util.GetChangeDetail')
|
|
def testSetReview_ReAuthNotNeededForCQLabel(self, mockGetChangeDetail,
|
|
mockJsonResponse,
|
|
mockCreateHttpConn):
|
|
mockJsonResponse.return_value = {
|
|
"ready": True,
|
|
"labels": {
|
|
"Commit-Queue": 1
|
|
},
|
|
}
|
|
|
|
gerrit_util.SetReview('chromium',
|
|
123456,
|
|
msg="test",
|
|
labels={"Commit-Queue": 1})
|
|
|
|
# ReAuth not needed.
|
|
mockGetChangeDetail.assert_not_called()
|
|
httpConnKwargs = mockCreateHttpConn.call_args[1]
|
|
self.assertIsNone(httpConnKwargs.get('reauth_context', None))
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
@mock.patch('gerrit_util.GetChangeDetail')
|
|
def testSetReview_ReAuthNotNeededWithoutLabels(self, mockGetChangeDetail,
|
|
mockJsonResponse,
|
|
mockCreateHttpConn):
|
|
mockJsonResponse.return_value = {"ready": True}
|
|
|
|
gerrit_util.SetReview('chromium', 123456, msg="test")
|
|
|
|
# ReAuth not needed.
|
|
mockGetChangeDetail.assert_not_called()
|
|
httpConnKwargs = mockCreateHttpConn.call_args[1]
|
|
self.assertIsNone(httpConnKwargs.get('reauth_context', None))
|
|
|
|
@mock.patch('gerrit_util.CreateHttpConn')
|
|
@mock.patch('gerrit_util.ReadHttpJsonResponse')
|
|
def testCherryPickWithConflicts(self, mockJsonResponse, mockCreateHttpConn):
|
|
mockJsonResponse.return_value = {'_number': 1}
|
|
gerrit_util.CherryPick('host',
|
|
'change',
|
|
'destination',
|
|
allow_conflicts=True)
|
|
mockCreateHttpConn.assert_called_once_with(
|
|
'host',
|
|
'changes/change/revisions/current/cherrypick',
|
|
reqtype='POST',
|
|
body={
|
|
'destination': 'destination',
|
|
'allow_conflicts': True
|
|
})
|
|
|
|
class SSOAuthenticatorTest(unittest.TestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls._original_timeout_secs = gerrit_util.SSOAuthenticator._timeout_secs
|
|
return super().setUpClass()
|
|
|
|
def setUp(self) -> None:
|
|
gerrit_util.SSOAuthenticator._sso_info = None
|
|
gerrit_util.SSOAuthenticator._testing_load_expired_cookies = True
|
|
gerrit_util.SSOAuthenticator._timeout_secs = self._original_timeout_secs
|
|
self.sso = gerrit_util.SSOAuthenticator()
|
|
return super().setUp()
|
|
|
|
def tearDown(self) -> None:
|
|
gerrit_util.SSOAuthenticator._sso_info = None
|
|
gerrit_util.SSOAuthenticator._testing_load_expired_cookies = False
|
|
gerrit_util.SSOAuthenticator._timeout_secs = self._original_timeout_secs
|
|
return super().tearDown()
|
|
|
|
@property
|
|
def _input_dir(self) -> Path:
|
|
base = Path(__file__).absolute().with_suffix('.inputs')
|
|
# Here _testMethodName would be a string like "testCmdAssemblyFound"
|
|
return base / self._testMethodName
|
|
|
|
@mock.patch('gerrit_util.ssoHelper.find_cmd',
|
|
return_value='/fake/git-remote-sso')
|
|
def testCmdAssemblyFound(self, _):
|
|
self.assertEqual(self.sso._resolve_sso_cmd(),
|
|
('/fake/git-remote-sso', '-print_config',
|
|
'sso://*.git.corp.google.com'))
|
|
with mock.patch('scm.GIT.GetConfig') as p:
|
|
p.side_effect = ['firefly@google.com']
|
|
self.assertTrue(self.sso.is_applicable())
|
|
|
|
@mock.patch('gerrit_util.ssoHelper.find_cmd', return_value=None)
|
|
def testCmdAssemblyNotFound(self, _):
|
|
self.assertEqual(self.sso._resolve_sso_cmd(), ())
|
|
self.assertFalse(self.sso.is_applicable())
|
|
|
|
def testParseConfigOK(self):
|
|
test_config = {
|
|
'somekey': 'a value with = in it',
|
|
'novalue': '',
|
|
'http.proxy': 'localhost:12345',
|
|
'http.cookiefile': str(self._input_dir / 'cookiefile.txt'),
|
|
'include.path': str(self._input_dir / 'gitconfig'),
|
|
}
|
|
parsed = self.sso._parse_config(test_config)
|
|
self.assertDictEqual(parsed.headers, {
|
|
'Authorization': 'Basic REALLY_COOL_TOKEN',
|
|
})
|
|
self.assertEqual(parsed.proxy.proxy_host, b'localhost')
|
|
self.assertEqual(parsed.proxy.proxy_port, 12345)
|
|
|
|
c = parsed.cookies._cookies
|
|
self.assertEqual(c['login.example.com']['/']['SSO'].value,
|
|
'TUVFUE1PUlAK')
|
|
self.assertEqual(c['.example.com']['/']['__CoolProxy'].value,
|
|
'QkxFRVBCTE9SUAo=')
|
|
|
|
@unittest.skipUnless(RUN_SUBPROC_TESTS, 'subprocess tests are flakey')
|
|
def testLaunchHelperOK(self):
|
|
gerrit_util.SSOAuthenticator._sso_cmd = ('python3',
|
|
str(self._input_dir /
|
|
'git-remote-sso.py'))
|
|
|
|
info = self.sso._get_sso_info()
|
|
self.assertDictEqual(info.headers, {
|
|
'Authorization': 'Basic REALLY_COOL_TOKEN',
|
|
})
|
|
self.assertEqual(info.proxy.proxy_host, b'localhost')
|
|
self.assertEqual(info.proxy.proxy_port, 12345)
|
|
c = info.cookies._cookies
|
|
self.assertEqual(c['login.example.com']['/']['SSO'].value,
|
|
'TUVFUE1PUlAK')
|
|
self.assertEqual(c['.example.com']['/']['__CoolProxy'].value,
|
|
'QkxFRVBCTE9SUAo=')
|
|
|
|
@unittest.skipUnless(RUN_SUBPROC_TESTS, 'subprocess tests are flakey')
|
|
def testLaunchHelperFailQuick(self):
|
|
gerrit_util.SSOAuthenticator._sso_cmd = ('python3',
|
|
str(self._input_dir /
|
|
'git-remote-sso.py'))
|
|
|
|
with self.assertRaisesRegex(SystemExit, "SSO Failure Message!!!"):
|
|
self.sso._get_sso_info()
|
|
|
|
@unittest.skipUnless(RUN_SUBPROC_TESTS, 'subprocess tests are flakey')
|
|
def testLaunchHelperFailSlow(self):
|
|
gerrit_util.SSOAuthenticator._timeout_secs = 0.2
|
|
gerrit_util.SSOAuthenticator._sso_cmd = ('python3',
|
|
str(self._input_dir /
|
|
'git-remote-sso.py'))
|
|
|
|
with self.assertRaises(subprocess.TimeoutExpired):
|
|
self.sso._get_sso_info()
|
|
|
|
@mock.patch('gerrit_util.SSOAuthenticator.authenticate')
|
|
def testAttemptAuthenticateWithReAuth(self, mockAuthenticate):
|
|
conn = makeConn("chromium")
|
|
reauth_context = auth.ReAuthContext(host="chromium",
|
|
project="infra/infra")
|
|
out = self.sso.attempt_authenticate_with_reauth(conn, reauth_context)
|
|
mockAuthenticate.assert_called()
|
|
self.assertTrue(out)
|
|
|
|
class SSOHelperTest(unittest.TestCase):
|
|
|
|
def setUp(self) -> None:
|
|
self.sso = gerrit_util.SSOHelper()
|
|
return super().setUp()
|
|
|
|
@mock.patch('shutil.which', return_value='/fake/git-remote-sso')
|
|
def testFindCmd(self, _):
|
|
self.assertEqual(self.sso.find_cmd(), '/fake/git-remote-sso')
|
|
|
|
@mock.patch('shutil.which', return_value=None)
|
|
def testFindCmdMissing(self, _):
|
|
self.assertEqual(self.sso.find_cmd(), '')
|
|
|
|
@mock.patch('shutil.which', return_value='/fake/git-remote-sso')
|
|
def testFindCmdCached(self, which):
|
|
self.sso.find_cmd()
|
|
self.sso.find_cmd()
|
|
self.assertEqual(which.called, 1)
|
|
|
|
|
|
class ShouldUseSSOTest(unittest.TestCase):
|
|
|
|
def setUp(self) -> None:
|
|
self.newauth = mock.patch('newauth.Enabled', return_value=True)
|
|
self.newauth.start()
|
|
self.cwd = mock.patch('os.getcwd', return_value='/fake/cwd')
|
|
self.cwd.start()
|
|
self.sso = mock.patch('gerrit_util.ssoHelper.find_cmd',
|
|
return_value='/fake/git-remote-sso')
|
|
self.sso.start()
|
|
scm_mock.GIT(self)
|
|
self.addCleanup(mock.patch.stopall)
|
|
|
|
gerrit_util.CheckShouldUseSSO.cache_clear()
|
|
return super().setUp()
|
|
|
|
def tearDown(self) -> None:
|
|
super().tearDown()
|
|
self.sso.stop()
|
|
self.newauth.stop()
|
|
|
|
@mock.patch('newauth.Enabled', return_value=False)
|
|
def testDisabled(self, _):
|
|
self.assertFalse(
|
|
gerrit_util.ShouldUseSSO('fake-host.googlesource.com',
|
|
'firefly@google.com'))
|
|
|
|
@mock.patch('gerrit_util.ssoHelper.find_cmd', return_value='')
|
|
def testMissingCommand(self, _):
|
|
self.assertFalse(
|
|
gerrit_util.ShouldUseSSO('fake-host.googlesource.com',
|
|
'firefly@google.com'))
|
|
|
|
def testBadHost(self):
|
|
self.assertFalse(
|
|
gerrit_util.ShouldUseSSO('fake-host.coreboot.org',
|
|
'firefly@google.com'))
|
|
|
|
def testEmptyEmail(self):
|
|
self.assertTrue(
|
|
gerrit_util.ShouldUseSSO('fake-host.googlesource.com', ''))
|
|
|
|
def testGoogleEmail(self):
|
|
self.assertTrue(
|
|
gerrit_util.ShouldUseSSO('fake-host.googlesource.com',
|
|
'firefly@google.com'))
|
|
|
|
def testGmail(self):
|
|
self.assertFalse(
|
|
gerrit_util.ShouldUseSSO('fake-host.googlesource.com',
|
|
'firefly@gmail.com'))
|
|
|
|
@mock.patch('gerrit_util.GetAccountEmails',
|
|
return_value=[{
|
|
'email': 'firefly@chromium.org'
|
|
}])
|
|
def testLinkedChromium(self, email):
|
|
self.assertTrue(
|
|
gerrit_util.ShouldUseSSO('fake-host.googlesource.com',
|
|
'firefly@chromium.org'))
|
|
email.assert_called_with('fake-host.googlesource.com',
|
|
'self',
|
|
authenticator=mock.ANY)
|
|
|
|
@mock.patch('gerrit_util.GetAccountEmails',
|
|
return_value=[{
|
|
'email': 'firefly@google.com'
|
|
}])
|
|
def testUnlinkedChromium(self, email):
|
|
self.assertFalse(
|
|
gerrit_util.ShouldUseSSO('fake-host.googlesource.com',
|
|
'firefly@chromium.org'))
|
|
email.assert_called_with('fake-host.googlesource.com',
|
|
'self',
|
|
authenticator=mock.ANY)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|