Coverage for tests / test_panda_auth_utils.py: 39%

1# This file is part of ctrl_bps_panda. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <https://www.gnu.org/licenses/>. 

27 

28"""Unit tests for PanDA authentication utilities.""" 

29 

30import base64 

31import json 

32import os 

33import unittest 

34from datetime import UTC, datetime, timedelta 

35from unittest import mock 

36 

37from lsst.ctrl.bps.panda import __version__ as version 

38from lsst.ctrl.bps.panda.panda_auth_utils import ( 

39 TokenExpiredError, 

40 panda_auth_refresh, 

41 panda_auth_status, 

42) 

43 

44 

45def make_fake_jwt(exp_offset_days): 

46 """Return a fake id_token that expires in N days.""" 

47 payload = {"exp": int((datetime.now(UTC) + timedelta(days=exp_offset_days)).timestamp())} 

48 b64_payload = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=") 

49 return f"header.{b64_payload}.sig" 

50 

51 

52def fake_token_file(exp_days=3, refresh_token="fake_refresh"): 

53 """Generate fake token file data""" 

54 token = make_fake_jwt(exp_days) 

55 return json.dumps({"id_token": token, "refresh_token": refresh_token}) 

56 

57 

58def fetch_page_side_effect(url): 

59 """Simulate OpenIdConnect_Utils.fetch_page behavior in tests.""" 

60 if url.endswith("auth_config.json"): 

61 return True, { 

62 "client_secret": "secret", 

63 "audience": "https://iam.example.com", 

64 "client_id": "cid", 

65 "oidc_config_url": "https://oidc.example.org/.well-known/openid-configuration", 

66 "vo": "fake_vo", 

67 "no_verify": "True", 

68 "robot_ids": "NONE", 

69 } 

70 elif url.endswith("openid-configuration"): 

71 return True, {"token_endpoint": "https://oidc.example.org/token"} 

72 return False, {} 

73 

74 

75class VersionTestCase(unittest.TestCase): 

76 """Test versioning.""" 

77 

78 def test_version(self): 

79 # Check that version is defined. 

80 self.assertIsNotNone(version) 

81 

82 

83class TestPandaAuthUtils(unittest.TestCase): 

84 """Simple test of auth utilities.""" 

85 

86 def setUp(self): 

87 self.test_env = { 

88 "PANDA_CONFIG_ROOT": "/fake/token", 

89 "PANDA_URL_SSL": "https://fake.server.com:8443/server/panda", 

90 "PANDA_URL": "https://fake.server.com:8443/server/panda", 

91 "PANDACACHE_URL": "https://fake.server.com:8443/server/panda", 

92 "PANDAMON_URL": "https://fake.monitor.com:8443/", 

93 "PANDA_AUTH": "oidc", 

94 "PANDA_VERIFY_HOST": "off", 

95 "PANDA_AUTH_VO": "fake_vo", 

96 "PANDA_BEHIND_REAL_LB": "true", 

97 "PANDA_SYS": "/fake/pandasys", 

98 "IDDS_CONFIG": "/fake/pandasys/etc/idds/idds.cfg.client.template", 

99 } 

100 

101 def testPandaAuthStatusWrongEnviron(self): 

102 unwanted = { 

103 "PANDA_AUTH", 

104 "PANDA_VERIFY_HOST", 

105 "PANDA_AUTH_VO", 

106 "PANDA_URL_SSL", 

107 "PANDA_URL", 

108 } 

109 test_environ = {key: val for key, val in os.environ.items() if key not in unwanted} 

110 with mock.patch.dict(os.environ, test_environ, clear=True): 

111 with self.assertRaises(OSError): 

112 panda_auth_status() 

113 

114 @mock.patch("builtins.print") 

115 @mock.patch("os.path.exists", return_value=True) 

116 @mock.patch("pandaclient.openidc_utils.OpenIdConnect_Utils") 

117 def test_expired_token(self, mock_oidc, mock_exists, mock_print): 

118 mock_oidc.return_value.get_token_path.return_value = "/fake/token.json" 

119 

120 with mock.patch.dict("os.environ", self.test_env): 

121 with mock.patch("builtins.open", mock.mock_open(read_data=fake_token_file(exp_days=-1))): 

122 with self.assertRaises(TokenExpiredError): 

123 panda_auth_refresh(days=4) 

124 

125 @mock.patch("builtins.print") 

126 @mock.patch("lsst.ctrl.bps.panda.panda_auth_utils.panda_auth_status") 

127 @mock.patch("os.path.exists", return_value=True) 

128 @mock.patch("lsst.ctrl.bps.panda.panda_auth_utils.OpenIdConnect_Utils") 

129 def test_successful_refresh(self, mock_oidc, mock_exists, mock_status, mock_print): 

130 fake_openid = mock_oidc.return_value 

131 fake_openid.get_token_path.return_value = "/fake/token.json" 

132 fake_openid.auth_config_url = "https://fake.server/auth_config.json" 

133 

134 fake_openid.fetch_page.side_effect = fetch_page_side_effect 

135 

136 fake_openid.refresh_token.return_value = (True, {"access_token": "new_token"}) 

137 

138 mock_status.return_value = {"exp": int((datetime.now(UTC) + timedelta(seconds=3600)).timestamp())} 

139 

140 with mock.patch.dict("os.environ", self.test_env): 

141 token_json = fake_token_file(exp_days=2) 

142 with mock.patch("builtins.open", mock.mock_open(read_data=token_json)): 

143 panda_auth_refresh(days=4) 

144 

145 fake_openid.refresh_token.assert_called_once() 

146 found = any("Success to refresh token" in str(c[0][0]) for c in mock_print.call_args_list) 

147 assert found 

148 

149 

150if __name__ == "__main__": 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 unittest.main()