depot_tools/tests/gerrit_util_test.py

#!/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 httplib2
from io import StringIO
import json
import os
import socket
import sys
import unittest
from unittest import mock

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import gerrit_util
import metrics
import subprocess2


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.path.exists', return_value=True).start()
    mock.patch(
        'subprocess2.check_output',
        side_effect=[
            subprocess2.CalledProcessError(1, ['cmd'], 'cwd', 'out', 'err')],
    ).start()
    self.addCleanup(mock.patch.stopall)
    self.maxDiff = None

  def testGetNewPasswordUrl(self):
    auth = gerrit_util.CookiesAuthenticator()
    self.assertEqual('https://chromium.googlesource.com/new-password',
                     auth.get_new_password_url('chromium.googlesource.com'))
    self.assertEqual(
        'https://chrome-internal.googlesource.com/new-password',
        auth.get_new_password_url('chrome-internal-review.googlesource.com'))

  def testGetNewPasswordMessage(self):
    auth = gerrit_util.CookiesAuthenticator()
    self.assertIn(
        'https://chromium.googlesource.com/new-password',
        auth.get_new_password_message('chromium-review.googlesource.com'))
    self.assertIn(
        'https://chrome-internal.googlesource.com/new-password',
        auth.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())

    subprocess2.check_output.side_effect = [b'http.cookiefile']
    self.assertEqual(
        'http.cookiefile',
        gerrit_util.CookiesAuthenticator().get_gitcookies_path())
    subprocess2.check_output.assert_called_with(
        ['git', 'config', '--path', 'http.cookiefile'])

    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):
    auth = gerrit_util.CookiesAuthenticator()
    self.assertEqual(auth.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')

    auth = gerrit_util.CookiesAuthenticator()
    self.assertEqual(
        expected_chromium_header,
        auth.get_auth_header('chromium.googlesource.com'))
    self.assertEqual(
        expected_chromium_header,
        auth.get_auth_header('chromium-review.googlesource.com'))
    self.assertEqual(
        'Bearer example-bearer-token',
        auth.get_auth_header('some-review.example.com'))

  def testGetAuthEmail(self):
    auth = gerrit_util.CookiesAuthenticator()
    self.assertEqual(
        'user@chromium.org',
        auth.get_auth_email('chromium.googlesource.com'))
    self.assertEqual(
        'user@chromium.org',
        auth.get_auth_email('chromium-review.googlesource.com'))
    self.assertIsNone(auth.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 testIsGce_EnvVarSkip(self, *_mocks):
    os.getenv.return_value = '1'
    self.assertFalse(self.GceAuthenticator.is_gce())
    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_gce())

  def testIsGce_500(self):
    httplib2.Http().request.return_value = (mock.Mock(status=500), None)
    self.assertFalse(self.GceAuthenticator.is_gce())
    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_gce())

  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_gce())
    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_gce())
    self.assertTrue(self.GceAuthenticator.is_gce())
    httplib2.Http().request.assert_called_once()

  def testGetAuthHeader_Error(self):
    httplib2.Http().request.side_effect = httplib2.HttpLib2Error
    self.assertIsNone(self.GceAuthenticator().get_auth_header(''))

  def testGetAuthHeader_500(self):
    httplib2.Http().request.return_value = (mock.Mock(status=500), None)
    self.assertIsNone(self.GceAuthenticator().get_auth_header(''))

  def testGetAuthHeader_Non200(self):
    httplib2.Http().request.return_value = (mock.Mock(status=403), None)
    self.assertIsNone(self.GceAuthenticator().get_auth_header(''))

  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.assertEqual('TYPE TOKEN', self.GceAuthenticator().get_auth_header(''))

  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.assertEqual('TYPE TOKEN', self.GceAuthenticator().get_auth_header(''))
    self.assertEqual('TYPE TOKEN', self.GceAuthenticator().get_auth_header(''))
    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.assertEqual('TYPE TOKEN', self.GceAuthenticator().get_auth_header(''))
    self.assertEqual('TYPE TOKEN', self.GceAuthenticator().get_auth_header(''))
    self.assertEqual(2, len(httplib2.Http().request.mock_calls))


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.Authenticator')
  def testCreateHttpConn_Basic(self, mockAuth):
    mockAuth.get().get_auth_header.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/foo/bar',
        'method': 'GET',
        'headers': {},
        'body': None,
    }, conn.req_params)

  @mock.patch('gerrit_util.Authenticator')
  def testCreateHttpConn_Authenticated(self, mockAuth):
    mockAuth.get().get_auth_header.return_value = 'Bearer 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.Authenticator')
  def testCreateHttpConn_Body(self, mockAuth):
    mockAuth.get().get_auth_header.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/foo/bar',
        'method': 'GET',
        'headers': {'Content-Type': 'application/json'},
        'body': '{"d": {"k": "v"}, "l": [1, 2, 3]}',
    }, conn.req_params)

  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_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.assertIsNone(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'))

if __name__ == '__main__':
  unittest.main()