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: int) -> str: 

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

47 

48 Parameters 

49 ---------- 

50 exp_offset_days : `int` 

51 Number of days to use for expiry. 

52 """ 

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

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

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

56 

57 

58def fake_token_file(exp_days: int = 3, refresh_token: str = "fake_refresh") -> str: 

59 """Generate fake token file data. 

60 

61 Parameters 

62 ---------- 

63 exp_days : `int`, optional 

64 Number of days to use for expiry. 

65 refresh_token : `str`, optional 

66 Fake token to use. 

67 """ 

68 token = make_fake_jwt(exp_days) 

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

70 

71 

72def fetch_page_side_effect(url: str) -> tuple[bool, dict[str, str]]: 

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

74 

75 Parameters 

76 ---------- 

77 url : `str` 

78 URL to fetch. 

79 """ 

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

81 return True, { 

82 "client_secret": "secret", 

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

84 "client_id": "cid", 

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

86 "vo": "fake_vo", 

87 "no_verify": "True", 

88 "robot_ids": "NONE", 

89 } 

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

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

92 return False, {} 

93 

94 

95class VersionTestCase(unittest.TestCase): 

96 """Test versioning.""" 

97 

98 def test_version(self): 

99 # Check that version is defined. 

100 self.assertIsNotNone(version) 

101 

102 

103class TestPandaAuthUtils(unittest.TestCase): 

104 """Simple test of auth utilities.""" 

105 

106 def setUp(self): 

107 self.test_env = { 

108 "PANDA_CONFIG_ROOT": "/fake/token", 

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

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

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

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

113 "PANDA_AUTH": "oidc", 

114 "PANDA_VERIFY_HOST": "off", 

115 "PANDA_AUTH_VO": "fake_vo", 

116 "PANDA_BEHIND_REAL_LB": "true", 

117 "PANDA_SYS": "/fake/pandasys", 

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

119 } 

120 

121 def testPandaAuthStatusWrongEnviron(self): 

122 unwanted = { 

123 "PANDA_AUTH", 

124 "PANDA_VERIFY_HOST", 

125 "PANDA_AUTH_VO", 

126 "PANDA_URL_SSL", 

127 "PANDA_URL", 

128 } 

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

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

131 with self.assertRaises(OSError): 

132 panda_auth_status() 

133 

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

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

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

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

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

139 

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

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

142 with self.assertRaises(TokenExpiredError): 

143 panda_auth_refresh(days=4) 

144 

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

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

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

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

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

150 fake_openid = mock_oidc.return_value 

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

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

153 

154 fake_openid.fetch_page.side_effect = fetch_page_side_effect 

155 

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

157 

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

159 

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

161 token_json = fake_token_file(exp_days=2) 

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

163 panda_auth_refresh(days=4) 

164 

165 fake_openid.refresh_token.assert_called_once() 

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

167 assert found 

168 

169 

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

171 unittest.main()