| # Copyright 2016 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Module that contains unittests for auth module.""" |
| |
| import time |
| from unittest import mock |
| |
| from chromite.lib import auth |
| from chromite.lib import cros_test_lib |
| |
| |
| class AuthTest(cros_test_lib.RunCommandTestCase): |
| """Test cases for methods in auth.""" |
| |
| def setUp(self): |
| self.PatchObject(time, "sleep") |
| self.PatchObject(auth, "GetLuciAuth", return_value="luci-auth") |
| self.PatchObject( |
| auth, "GetLuciGitCreds", return_value="git-credential-luci" |
| ) |
| |
| def testLoginFailed(self): |
| """Test Login failing.""" |
| self.rc.AddCmdResult(["luci-auth", "login"], stderr="", returncode=1) |
| self.assertRaises(auth.AccessTokenError, auth.Login) |
| |
| def testLoginPassed(self): |
| """Test Login working.""" |
| self.rc.AddCmdResult(["luci-auth", "login"], stdout="") |
| self.assertIsNone(auth.Login()) |
| |
| def testTokenFailed(self): |
| """Test Token failing.""" |
| self.rc.AddCmdResult(["luci-auth", "token"], stderr="", returncode=1) |
| self.assertRaises(auth.AccessTokenError, auth.Token) |
| |
| def testTokenPassed(self): |
| """Test Token working.""" |
| self.rc.AddCmdResult(["luci-auth", "token"], stdout="token") |
| self.assertEqual(auth.Token(), "token") |
| |
| def testTokenAndLoginIfNeed(self): |
| """Test TokenAndLoginIfNeed.""" |
| # pylint: disable=protected-access |
| sc_json = "/tmp/service_account.json" |
| |
| mock_login = self.PatchObject(auth, "Login", return_value=None) |
| mock_token = self.PatchObject(auth, "Token", return_value="token") |
| auth._TokenAndLoginIfNeed(service_account_json=sc_json) |
| mock_token.assert_called_once_with(service_account_json=sc_json) |
| mock_login.assert_not_called() |
| |
| mock_login.reset_mock() |
| mock_token.reset_mock() |
| auth._TokenAndLoginIfNeed( |
| service_account_json=sc_json, force_token_renew=True |
| ) |
| mock_token.assert_called_once_with(service_account_json=sc_json) |
| mock_login.assert_called_once_with(service_account_json=sc_json) |
| |
| mock_login.reset_mock() |
| mock_token = self.PatchObject( |
| auth, "Token", side_effect=auth.AccessTokenError() |
| ) |
| self.assertRaises( |
| auth.AccessTokenError, |
| auth._TokenAndLoginIfNeed, |
| service_account_json=sc_json, |
| ) |
| mock_token.assert_called_once_with(service_account_json=sc_json) |
| mock_login.assert_called_once_with(service_account_json=sc_json) |
| |
| mock_login.reset_mock() |
| mock_token.reset_mock() |
| self.assertRaises( |
| auth.AccessTokenError, |
| auth._TokenAndLoginIfNeed, |
| service_account_json=sc_json, |
| force_token_renew=True, |
| ) |
| mock_token.assert_called_once_with(service_account_json=sc_json) |
| mock_login.assert_called_once_with(service_account_json=sc_json) |
| |
| def testGetAccessToken(self): |
| """Test GetAccessToken.""" |
| mock_login = self.PatchObject(auth, "Login", return_value=None) |
| mock_token = self.PatchObject( |
| auth, "Token", side_effect=auth.AccessTokenError() |
| ) |
| auth.GetAccessToken() |
| self.assertEqual(mock_login.call_count, auth.RETRY_GET_ACCESS_TOKEN + 1) |
| self.assertEqual(mock_token.call_count, auth.RETRY_GET_ACCESS_TOKEN + 1) |
| |
| mock_login.reset_mock() |
| mock_token.reset_mock() |
| auth.GetAccessToken(force_token_renew=True) |
| self.assertEqual(mock_login.call_count, auth.RETRY_GET_ACCESS_TOKEN + 1) |
| self.assertEqual(mock_token.call_count, auth.RETRY_GET_ACCESS_TOKEN + 1) |
| |
| def testGitCredsFailed(self): |
| """Test git-credential-luci failing.""" |
| self.rc.AddCmdResult( |
| ["git-credential-luci", "get"], stderr="", returncode=1 |
| ) |
| self.assertRaises(auth.AccessTokenError, auth.GitCreds) |
| |
| def testGitCredsPassed(self): |
| """Test git-credential-luci working.""" |
| stdout = "\n".join( |
| [ |
| "user=some-luci-user", |
| "password=some-git-password", |
| ] |
| ) |
| self.rc.AddCmdResult(["git-credential-luci", "get"], stdout=stdout) |
| self.assertEqual(auth.GitCreds(), "some-git-password") |
| |
| def testGitCredsNoPassword(self): |
| """Test git-credential-luci returning unknown output.""" |
| stdout = "\n".join( |
| [ |
| "unknown stdout format, line #1", |
| "unknown stdout format, line #2", |
| ] |
| ) |
| self.rc.AddCmdResult(["git-credential-luci", "get"], stdout=stdout) |
| self.assertRaises(auth.AccessTokenError, auth.GitCreds) |
| |
| |
| class AuthorizedHttp(cros_test_lib.MockTestCase): |
| """Test cases for AuthorizedHttp.""" |
| |
| def setUp(self): |
| self.mock_http = mock.Mock() |
| self.mock_resp = mock.Mock() |
| self.mock_get_token = self.PatchObject( |
| auth, "GetAccessToken", return_value="token" |
| ) |
| self.account_json = "service_account_json" |
| |
| def testAuthorize(self): |
| self.mock_resp.status = 200 |
| self.mock_http.request.return_value = self.mock_resp, "content" |
| |
| auth_http = auth.AuthorizedHttp( |
| auth.GetAccessToken, |
| self.mock_http, |
| service_account_json=self.account_json, |
| ) |
| auth_http.request("url", "GET", body={}, headers={}) |
| auth_http.request("url", "PUT", body={}, headers={}) |
| auth_http.request("url", "POST", body={}, headers={}) |
| |
| self.assertEqual(1, self.mock_get_token.call_count) |
| self.assertEqual(3, self.mock_http.request.call_count) |
| self.mock_get_token.assert_called_with( |
| service_account_json=self.account_json |
| ) |
| |
| def testAuthorize2(self): |
| self.mock_resp.status = 401 |
| self.mock_http.request.return_value = self.mock_resp, "content" |
| |
| auth_http = auth.AuthorizedHttp( |
| auth.GetAccessToken, |
| self.mock_http, |
| service_account_json=self.account_json, |
| ) |
| auth_http.request("url", "GET", body={}, headers={}) |
| auth_http.request("url", "PUT", body={}, headers={}) |
| auth_http.request("url", "POST", body={}, headers={}) |
| |
| self.assertEqual(4, self.mock_get_token.call_count) |
| self.assertEqual(6, self.mock_http.request.call_count) |
| self.mock_get_token.assert_called_with( |
| service_account_json=self.account_json, force_token_renew=True |
| ) |
| |
| def testAuthorize3(self): |
| self.mock_resp.status = 500 |
| self.mock_http.request.return_value = self.mock_resp, "content" |
| |
| auth_http = auth.AuthorizedHttp(auth.GetAccessToken, self.mock_http) |
| auth_http.request("url", "GET", body={}, headers={}) |
| auth_http.request("url", "PUT", body={}, headers={}) |
| auth_http.request("url", "POST", body={}, headers={}) |
| |
| self.assertEqual(1, self.mock_get_token.call_count) |
| self.mock_get_token.assert_called_with() |
