#!/usr/bin/env vpython3 # coding=utf-8 # Copyright 2019 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. import base64 import json import os import socket import subprocess import sys import threading import unittest import io from io import StringIO from pathlib import Path from typing import Optional from unittest import mock import urllib.error import urllib.request 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() default_resp = mock.MagicMock( status=200, reason="OK", headers={"metadata-flavor": "Google"}, read=mock.Mock(return_value=b""), ) default_resp.__enter__.return_value = default_resp self.mock_open = mock.patch( "urllib.request.OpenerDirector.open", return_value=default_resp, ).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): self.mock_open.side_effect = urllib.error.URLError("error") self.assertFalse(self.GceAuthenticator.is_applicable()) def testIsGce_500(self): self.mock_open.side_effect = urllib.error.HTTPError( "", 500, "Server Error", {}, io.BytesIO(b"") ) 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.MagicMock( status=200, reason="OK", headers={"metadata-flavor": "Google"}, read=mock.Mock(return_value=b"who cares"), ) response.__enter__.return_value = response self.mock_open.side_effect = [ urllib.error.HTTPError( "", 500, "Server Error", {}, io.BytesIO(b"") ), response, ] self.assertTrue(self.GceAuthenticator.is_applicable()) def testIsGce_MetadataFlavorIsNotGoogle(self): response = mock.MagicMock( status=200, reason="OK", headers={}, read=mock.Mock(return_value=b"who cares"), ) response.__enter__.return_value = response self.mock_open.return_value = response self.assertFalse(self.GceAuthenticator.is_applicable()) def testIsGce_ResultIsCached(self): response = mock.MagicMock( status=200, reason="OK", headers={"metadata-flavor": "Google"}, read=mock.Mock(return_value=b"who cares"), ) response.__enter__.return_value = response self.mock_open.side_effect = [response] self.assertTrue(self.GceAuthenticator.is_applicable()) self.assertTrue(self.GceAuthenticator.is_applicable()) self.mock_open.assert_called_once() def testGetAuthHeader_Error(self): self.mock_open.side_effect = urllib.error.URLError("error") self.assertAuthenticatedToken(None) def testGetAuthHeader_500(self): self.mock_open.side_effect = urllib.error.HTTPError( "", 500, "Server Error", {}, io.BytesIO(b"") ) self.assertAuthenticatedToken(None) def testGetAuthHeader_Non200(self): self.mock_open.side_effect = urllib.error.HTTPError( "", 403, "Forbidden", {}, io.BytesIO(b"") ) self.assertAuthenticatedToken(None) def testGetAuthHeader_OK(self): response = mock.MagicMock( status=200, reason="OK", headers={}, read=mock.Mock( return_value=b'{"expires_in": 125, "token_type": "TYPE", "access_token": "TOKEN"}' ), ) response.__enter__.return_value = response self.mock_open.return_value = response gerrit_util.time_time.return_value = 0 self.assertAuthenticatedToken("TYPE TOKEN") def testGetAuthHeader_Cache(self): response = mock.MagicMock( status=200, reason="OK", headers={}, read=mock.Mock( return_value=b'{"expires_in": 125, "token_type": "TYPE", "access_token": "TOKEN"}' ), ) response.__enter__.return_value = response self.mock_open.return_value = response gerrit_util.time_time.return_value = 0 self.assertAuthenticatedToken("TYPE TOKEN") self.assertAuthenticatedToken("TYPE TOKEN") self.mock_open.assert_called_once() def testGetAuthHeader_CacheOld(self): response = mock.MagicMock( status=200, reason="OK", headers={}, read=mock.Mock( return_value=b'{"expires_in": 125, "token_type": "TYPE", "access_token": "TOKEN"}' ), ) response.__enter__.return_value = response self.mock_open.return_value = response gerrit_util.time_time.side_effect = [0, 100, 200] self.assertAuthenticatedToken("TYPE TOKEN") self.assertAuthenticatedToken("TYPE TOKEN") self.assertEqual(2, self.mock_open.call_count) 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.EnsureAccountExists") def testGerritAccountExists_Success(self, mock_ensure_account): self.assertTrue( gerrit_util.GitCredsAuthenticator.gerrit_account_exists( "example.com" ) ) @mock.patch( "gerrit_util.EnsureAccountExists", side_effect=auth.GitLoginRequiredError(), ) def testGerritAccountExists_GitLoginRequired(self, mock_ensure_account): self.assertFalse( gerrit_util.GitCredsAuthenticator.gerrit_account_exists( "example.com" ) ) @mock.patch( "gerrit_util.EnsureAccountExists", side_effect=gerrit_util.GerritError(400, "Account missing"), ) def testGerritAccountExists_400Error(self, mock_ensure_account): self.assertFalse( gerrit_util.GitCredsAuthenticator.gerrit_account_exists( "example.com" ) ) @mock.patch( "gerrit_util.EnsureAccountExists", side_effect=gerrit_util.GerritError(404, "Account missing"), ) def testGerritAccountExists_404Error(self, mock_ensure_account): self.assertFalse( gerrit_util.GitCredsAuthenticator.gerrit_account_exists( "example.com" ) ) @mock.patch( "gerrit_util.EnsureAccountExists", side_effect=gerrit_util.GerritError(500, "Internal Server Error"), ) def testGerritAccountExists_500ErrorRaises(self, mock_ensure_account): with self.assertRaises(gerrit_util.GerritError): gerrit_util.GitCredsAuthenticator.gerrit_account_exists( "example.com" ) @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_host="example.com", 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_host="example.com", 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") @mock.patch("gerrit_util.gclient_utils.IsEnvAi", return_value=True) def testSetReview_Ai( self, mockIsEnvAi, mockJsonResponse, mockCreateHttpConn ): mockJsonResponse.return_value = {} comments = {"file.py": [{"line": 1, "message": "hello"}]} gerrit_util.SetReview("chromium", 123456, comments=comments) expected_body = { "drafts": "KEEP", "comments": { "file.py": [{"line": 1, "message": "hello", "is_ai": True}] }, } mockCreateHttpConn.assert_called_once_with( "chromium", "changes/123456/revisions/current/review", reqtype="POST", body=expected_body, reauth_context=None, ) @mock.patch("gerrit_util.CreateHttpConn") @mock.patch("gerrit_util.ReadHttpJsonResponse") @mock.patch("gerrit_util.gclient_utils.IsEnvAi", return_value=False) def testSetReview_NotAi( self, mockIsEnvAi, mockJsonResponse, mockCreateHttpConn ): mockJsonResponse.return_value = {} comments = {"file.py": [{"line": 1, "message": "hello"}]} gerrit_util.SetReview("chromium", 123456, comments=comments) expected_body = { "drafts": "KEEP", "comments": {"file.py": [{"line": 1, "message": "hello"}]}, } mockCreateHttpConn.assert_called_once_with( "chromium", "changes/123456/revisions/current/review", reqtype="POST", body=expected_body, reauth_context=None, ) @mock.patch("gerrit_util.CreateHttpConn") @mock.patch("gerrit_util.ReadHttpJsonResponse") @mock.patch("gerrit_util.gclient_utils.IsEnvAi", return_value=True) def testCreateDraft_Ai( self, mockIsEnvAi, mockJsonResponse, mockCreateHttpConn ): mockJsonResponse.return_value = {} body = {"line": 1, "message": "hello"} gerrit_util.CreateDraft("chromium", 123456, body=body) expected_body = {"line": 1, "message": "hello", "is_ai": True} mockCreateHttpConn.assert_called_once_with( "chromium", "changes/123456/revisions/current/drafts", reqtype="PUT", body=expected_body, ) @mock.patch("gerrit_util.CreateHttpConn") @mock.patch("gerrit_util.ReadHttpJsonResponse") @mock.patch("gerrit_util.gclient_utils.IsEnvAi", return_value=False) def testCreateDraft_NotAi( self, mockIsEnvAi, mockJsonResponse, mockCreateHttpConn ): mockJsonResponse.return_value = {} body = {"line": 1, "message": "hello"} gerrit_util.CreateDraft("chromium", 123456, body=body) expected_body = {"line": 1, "message": "hello"} mockCreateHttpConn.assert_called_once_with( "chromium", "changes/123456/revisions/current/drafts", reqtype="PUT", body=expected_body, ) @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}, ) @mock.patch("gerrit_util.ReadHttpResponse") @mock.patch("gerrit_util.CreateHttpConn") def testGetPatchDefaults(self, mockCreateHttpConn, mockReadHttpResponse): expected = b"From commit\nformatted patch\n" mockReadHttpResponse.return_value = StringIO( base64.b64encode(expected).decode() ) result = gerrit_util.GetPatch("host", "change") self.assertEqual(expected, result) mockCreateHttpConn.assert_called_once_with( "host", "changes/change/revisions/current/patch", reqtype="GET", ) mockReadHttpResponse.assert_called_once_with( mockCreateHttpConn.return_value ) @mock.patch("gerrit_util.ReadHttpResponse") @mock.patch("gerrit_util.CreateHttpConn") def testGetPatchOptions(self, mockCreateHttpConn, mockReadHttpResponse): mockReadHttpResponse.return_value = StringIO("") gerrit_util.GetPatch( "host", "change", revision="7", path="src/file name.cc", parent=2, context=0, ) mockCreateHttpConn.assert_called_once_with( "host", ( "changes/change/revisions/7/patch" "?path=src%2Ffile+name.cc&parent=2&context=0" ), reqtype="GET", ) 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_host, "localhost") self.assertEqual(parsed.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_host, "localhost") self.assertEqual(info.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 ) class EnsureAccountExistsTest(unittest.TestCase): @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"] ) @mock.patch("gerrit_util.GetAccountDetails") def testEnsureAccountExists_CachedWithEmail( self, mock_get_account, mock_get_config_list, mock_get_config ): gerrit_util.EnsureAccountExists("example.com") mock_get_config_list.assert_called_once_with( mock.ANY, "depot-tools.hosthasaccount" ) mock_get_account.assert_not_called() @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch("scm.GIT.GetConfigList", return_value=["example.com"]) @mock.patch("gerrit_util.GetAccountDetails") def testEnsureAccountExists_CachedLegacyHostOnly( self, mock_get_account, mock_get_config_list, mock_get_config ): gerrit_util.EnsureAccountExists("example.com") mock_get_config_list.assert_called_once_with( mock.ANY, "depot-tools.hosthasaccount" ) mock_get_account.assert_not_called() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="new_user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:old_user@example.com"], ) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "new_user@example.com"}, ) def testEnsureAccountExists_AccountSwitchMiss( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): gerrit_util.EnsureAccountExists("example.com") mock_get_account.assert_called_once_with( "example.com", authenticator=None ) mock_set_config.assert_called_once_with( mock.ANY, "depot-tools.hosthasaccount", "example.com:new_user@example.com", append=True, ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testEnsureAccountExists_UncachedSuccessWithEmail( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): gerrit_util.EnsureAccountExists("example.com") mock_get_account.assert_called_once_with( "example.com", authenticator=None ) mock_set_config.assert_called_once_with( mock.ANY, "depot-tools.hosthasaccount", "example.com:user@example.com", append=True, ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value=None) @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testEnsureAccountExists_UncachedSuccessNoLocalEmail( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): gerrit_util.EnsureAccountExists("example.com") mock_get_account.assert_called_once_with( "example.com", authenticator=None ) mock_set_config.assert_called_once_with( mock.ANY, "depot-tools.hosthasaccount", "example.com", append=True, ) @mock.patch("scm.GIT.GetConfig", return_value=None) @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", side_effect=gerrit_util.GerritError(400, "Bad Request"), ) def testEnsureAccountExists_NotFound400( self, mock_get_account, mock_get_config_list, mock_get_config ): with self.assertRaises(gerrit_util.GerritError) as ctx: gerrit_util.EnsureAccountExists("example.com") self.assertEqual(ctx.exception.http_status, 400) self.assertIn( "Account does not exist on Gerrit host 'example.com'", str(ctx.exception), ) @mock.patch("scm.GIT.GetConfig", return_value=None) @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch("gerrit_util.GetAccountDetails", return_value={}) def testEnsureAccountExists_NotFound404( self, mock_get_account, mock_get_config_list, mock_get_config ): with self.assertRaises(gerrit_util.GerritError) as ctx: gerrit_util.EnsureAccountExists("example.com") self.assertEqual(ctx.exception.http_status, 404) self.assertIn( "Account not found on Gerrit host 'example.com'", str(ctx.exception) ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com", "other.com"], ) @mock.patch("gerrit_util.GetAccountDetails", return_value=None) def testEnsureAccountExists_NotFound404InvalidatesCache( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): with self.assertRaises(gerrit_util.GerritError) as ctx: gerrit_util.EnsureAccountExists("example.com", force=True) self.assertEqual(ctx.exception.http_status, 404) self.assertIn( "Account not found on Gerrit host 'example.com'", str(ctx.exception) ) mock_set_config.assert_has_calls( [ mock.call( mock.ANY, "depot-tools.hosthasaccount", None, modify_all=True, ), mock.call( mock.ANY, "depot-tools.hosthasaccount", "other.com", append=True, ), ] ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"] ) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testEnsureAccountExists_ForceBypassesCache( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): gerrit_util.EnsureAccountExists("example.com", force=True) mock_get_account.assert_called_once_with( "example.com", authenticator=None ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com", "other.com"], ) @mock.patch( "gerrit_util.GetAccountDetails", side_effect=gerrit_util.GerritError(400, "Bad Request"), ) def testEnsureAccountExists_InvalidatesCacheOnError( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): with self.assertRaises(gerrit_util.GerritError): gerrit_util.EnsureAccountExists("example.com", force=True) mock_set_config.assert_has_calls( [ mock.call( mock.ANY, "depot-tools.hosthasaccount", None, modify_all=True, ), mock.call( mock.ANY, "depot-tools.hosthasaccount", "other.com", append=True, ), ] ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=[ "example.com:user@example.com", "example.com:user@example.com.uk", "other.com", ], ) def testInvalidateAccountCache( self, mock_get_config_list, mock_get_config, mock_set_config ): gerrit_util.InvalidateAccountCache("example.com") mock_set_config.assert_has_calls( [ mock.call( mock.ANY, "depot-tools.hosthasaccount", None, modify_all=True, ), mock.call( mock.ANY, "depot-tools.hosthasaccount", "example.com:user@example.com.uk", append=True, ), mock.call( mock.ANY, "depot-tools.hosthasaccount", "other.com", append=True, ), ] ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="User@Example.COM") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"], ) def testInvalidateAccountCache_CaseInsensitive( self, mock_get_config_list, mock_get_config, mock_set_config ): gerrit_util.InvalidateAccountCache("HTTPS://EXAMPLE.COM/") mock_set_config.assert_called_once_with( mock.ANY, "depot-tools.hosthasaccount", None, modify_all=True ) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=[ "host1.com:user1@example.com", "host2.com:user2@example.com", "example.com:user@example.com", ], ) def testInvalidateAccountCache_MultipleValuesUsesModifyAll( self, mock_get_config_list, mock_get_config, mock_set_config ): gerrit_util.InvalidateAccountCache("example.com") mock_set_config.assert_has_calls( [ mock.call( mock.ANY, "depot-tools.hosthasaccount", None, modify_all=True, ), mock.call( mock.ANY, "depot-tools.hosthasaccount", "host1.com:user1@example.com", append=True, ), mock.call( mock.ANY, "depot-tools.hosthasaccount", "host2.com:user2@example.com", append=True, ), ] ) @mock.patch("scm.GIT.GetConfig", return_value="User@Example.COM") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"], ) @mock.patch("gerrit_util.GetAccountDetails") def testEnsureAccountExists_CaseInsensitiveHit( self, mock_get_account, mock_get_config_list, mock_get_config ): gerrit_util.EnsureAccountExists("HTTPS://EXAMPLE.COM/") mock_get_account.assert_not_called() @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:8080:user@example.com"], ) @mock.patch("gerrit_util.GetAccountDetails") def testEnsureAccountExists_BareHostWithPort( self, mock_get_account, mock_get_config_list, mock_get_config ): gerrit_util.EnsureAccountExists("example.com:8080") mock_get_account.assert_not_called() @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"], ) @mock.patch("gerrit_util.GetAccountDetails") def testEnsureAccountExists_ExplicitCwdPassed( self, mock_get_account, mock_get_config_list, mock_get_config ): gerrit_util.EnsureAccountExists("example.com", cwd="/custom/repo/path") mock_get_config.assert_called_once_with( "/custom/repo/path", "user.email" ) mock_get_config_list.assert_called_once_with( "/custom/repo/path", "depot-tools.hosthasaccount" ) mock_get_account.assert_not_called() @mock.patch( "scm.GIT.SetConfig", side_effect=subprocess2.CalledProcessError( 1, ["git", "config"], None, None, None ), ) @mock.patch("scm.GIT.GetConfig", return_value=None) @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"_account_id": 12345}, ) def testEnsureAccountExists_SetConfigFailureGraceful( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): gerrit_util.EnsureAccountExists("example.com") mock_get_account.assert_called_once() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfigList", return_value=[]) def testInvalidateAccountCache_EmptyConfig( self, mock_get_config_list, mock_set_config ): gerrit_util.InvalidateAccountCache("example.com") mock_set_config.assert_not_called() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", side_effect=Exception("git error")) @mock.patch("scm.GIT.GetConfigList", return_value=["other.com"]) def testInvalidateAccountCache_NoMatchingHost( self, mock_get_config_list, mock_get_config, mock_set_config ): gerrit_util.InvalidateAccountCache("example.com") mock_set_config.assert_not_called() @mock.patch( "scm.GIT.GetConfigList", side_effect=Exception("fatal git error") ) def testInvalidateAccountCache_ExceptionSwallowed( self, mock_get_config_list ): gerrit_util.InvalidateAccountCache("example.com") @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", side_effect=Exception("config error")) @mock.patch("scm.GIT.GetConfigList", side_effect=Exception("config error")) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testEnsureAccountExists_GetConfigExceptionsSwallowed( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): gerrit_util.EnsureAccountExists("example.com") mock_get_account.assert_called_once() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"] ) @mock.patch( "gerrit_util.GetAccountDetails", side_effect=gerrit_util.GerritError(500, "Internal Server Error"), ) def testEnsureAccountExists_Non400GerritErrorDoesNotInvalidate( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): with self.assertRaises(gerrit_util.GerritError) as ctx: gerrit_util.EnsureAccountExists("example.com", force=True) self.assertEqual(ctx.exception.http_status, 500) mock_set_config.assert_not_called() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"] ) @mock.patch( "gerrit_util.GetAccountDetails", side_effect=auth.GitLoginRequiredError(), ) def testEnsureAccountExists_AuthExceptionDoesNotInvalidate( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): with self.assertRaises(auth.GitLoginRequiredError): gerrit_util.EnsureAccountExists("example.com", force=True) mock_set_config.assert_not_called() @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch( "scm.GIT.GetConfigList", return_value=["example.com:user@example.com"] ) @mock.patch("gerrit_util.GetAccountDetails") def testAsyncEnsureAccountExists_CachedFastReturn( self, mock_get_account, mock_get_config_list, mock_get_config ): join_fn = gerrit_util.AsyncEnsureAccountExists("example.com") join_fn() mock_get_account.assert_not_called() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testAsyncEnsureAccountExists_UncachedSuccess( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): join_fn = gerrit_util.AsyncEnsureAccountExists("example.com") join_fn() mock_get_account.assert_called_once() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testAsyncEnsureAccountExists_MultipleJoinCallsIdempotent( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): join_fn = gerrit_util.AsyncEnsureAccountExists("example.com") join_fn() join_fn() mock_get_account.assert_called_once() @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testAsyncEnsureAccountExists_ConcurrentJoinThreadSafe( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): join_fn = gerrit_util.AsyncEnsureAccountExists("example.com") barrier = threading.Barrier(4) errors = [] def _caller(): try: barrier.wait() join_fn() except Exception as e: errors.append(e) threads = [threading.Thread(target=_caller) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() self.assertEqual(errors, []) mock_get_account.assert_called_once() @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", side_effect=gerrit_util.GerritError(400, "Bad Request"), ) def testAsyncEnsureAccountExists_RaisesExceptionOnJoin( self, mock_get_account, mock_get_config_list, mock_get_config ): join_fn = gerrit_util.AsyncEnsureAccountExists("example.com") with self.assertRaises(gerrit_util.GerritError): join_fn() @mock.patch("scm.GIT.GetConfig", return_value="user@example.com") @mock.patch("scm.GIT.GetConfigList", return_value=[]) @mock.patch( "gerrit_util.GetAccountDetails", side_effect=SystemExit(1), ) def testAsyncEnsureAccountExists_SystemExitPropagates( self, mock_get_account, mock_get_config_list, mock_get_config ): join_fn = gerrit_util.AsyncEnsureAccountExists("example.com") with self.assertRaises(SystemExit) as ctx: join_fn() self.assertEqual(ctx.exception.code, 1) @mock.patch("scm.GIT.SetConfig") @mock.patch("scm.GIT.GetConfig", side_effect=Exception("config error")) @mock.patch("scm.GIT.GetConfigList", side_effect=Exception("config error")) @mock.patch( "gerrit_util.GetAccountDetails", return_value={"email": "user@example.com"}, ) def testAsyncEnsureAccountExists_GetConfigExceptionsSwallowed( self, mock_get_account, mock_get_config_list, mock_get_config, mock_set_config, ): join_fn = gerrit_util.AsyncEnsureAccountExists("example.com") join_fn() mock_get_account.assert_called_once() if __name__ == "__main__": unittest.main()