Coverage for tests/test_sasquatchDatastore.py: 98%

1# This file is part of analysis_tools. 

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 program is free software: you can redistribute it and/or modify 

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

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

12# (at your option) any later version. 

13# 

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

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

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

17# GNU General Public License for more details. 

18# 

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

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

21 

22"""Tests for SasquatchDatastore and SasquatchDispatcher. 

23 

24These catch (1) a put that no longer hands the bundle to the dispatcher, 

25(2) a broken or unresponsive Sasquatch proxy failing or hanging the put, 

26and with it the pipeline task, and (3) explicit timestamp versions being 

27misparsed, or accepted when malformed. 

28""" 

29 

30import datetime 

31import os 

32import unittest 

33from unittest.mock import MagicMock, patch 

34 

35import astropy.units as u 

36import requests 

37 

38import lsst.daf.butler.tests as butlerTests 

39from lsst.analysis.tools.interfaces import MetricMeasurementBundle 

40from lsst.analysis.tools.interfaces.datastore import SasquatchDispatcher, SasquatchDispatchFailure 

41from lsst.analysis.tools.interfaces.datastore._dispatcher import DEFAULT_TIMEOUT 

42from lsst.daf.butler import CollectionType, Config 

43from lsst.daf.butler.tests.utils import makeTestTempDir, removeTestTempDir 

44from lsst.verify import Measurement 

45 

46TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

47CONFIG_FILE = os.path.join(TESTDIR, "config", "butler-sasquatch.yaml") 

48 

49# Where the dispatcher looks up its HTTP session factory, for patching. 

50HTTP_CLIENT = "lsst.analysis.tools.interfaces.datastore._dispatcher.http_client" 

51 

52 

53def _fakeResponse(status: int = 200, json: object = None, jsonError: Exception | None = None) -> MagicMock: 

54 """Make a stand-in for a `requests.Response`. 

55 

56 Parameters 

57 ---------- 

58 status : `int`, optional 

59 HTTP status code; ``raise_for_status`` raises for 4xx and 5xx. 

60 json : `object`, optional 

61 Value returned by ``json()``. 

62 jsonError : `Exception`, optional 

63 Exception raised by ``json()`` in place of returning ``json``. 

64 

65 Returns 

66 ------- 

67 response : `unittest.mock.MagicMock` 

68 The fake response. 

69 """ 

70 response = MagicMock() 

71 response.status_code = status 

72 if status >= 400: 

73 response.raise_for_status.side_effect = requests.HTTPError(f"HTTP {status}", response=response) 

74 if jsonError is not None: 

75 response.json.side_effect = jsonError 

76 else: 

77 response.json.return_value = json 

78 return response 

79 

80 

81def _fakeSession(getResponse: MagicMock | None = None, postResponse: MagicMock | None = None) -> MagicMock: 

82 """Make a stand-in for a `requests.Session` returning canned responses. 

83 

84 Parameters 

85 ---------- 

86 getResponse : `unittest.mock.MagicMock`, optional 

87 Response returned by ``get``; a bare 200 if not given. 

88 postResponse : `unittest.mock.MagicMock`, optional 

89 Response returned by ``post``; a bare 200 if not given. 

90 

91 Returns 

92 ------- 

93 session : `unittest.mock.MagicMock` 

94 The fake session. 

95 """ 

96 session = MagicMock() 

97 session.get.return_value = getResponse if getResponse is not None else _fakeResponse() 

98 session.post.return_value = postResponse if postResponse is not None else _fakeResponse() 

99 return session 

100 

101 

102class SasquatchDatastoreTest(unittest.TestCase): 

103 def setUp(self): 

104 self.root = makeTestTempDir(TESTDIR) 

105 

106 config = Config() 

107 config["datastore", "cls"] = "lsst.analysis.tools.interfaces.datastore.SasquatchDatastore" 

108 config["datastore", "restProxyUrl"] = "https://example.com/sasquatch-rest-proxy" 

109 config["datastore", "timeout"] = 5.5 

110 

111 dataIds = { 

112 "instrument": ["DummyCam"], 

113 "physical_filter": ["d-r"], 

114 "visit": [42, 43, 44], 

115 "detector": [1, 2, 3], 

116 } 

117 self.butler = butlerTests.makeTestRepo(self.root, dataIds, config=config) 

118 

119 butlerTests.addDatasetType( 

120 self.butler, "Metrics", {"instrument", "visit", "detector"}, "MetricMeasurementBundle" 

121 ) 

122 self.butler.registry.registerCollection("run1", CollectionType.RUN) 

123 

124 def tearDown(self): 

125 removeTestTempDir(self.root) 

126 

127 def test_put(self): 

128 """Simple test for put method.""" 

129 m = Measurement("nopackage.fancyMetric", 42.2 * u.s) 

130 bundle = MetricMeasurementBundle({"m": [m]}) 

131 

132 # Patch dispatcher method to check parameters. 

133 with patch.object(SasquatchDispatcher, "dispatchRef") as mock_method: 

134 self.butler.put(bundle, "Metrics", run="run1", instrument="DummyCam", visit=42, detector=2) 

135 

136 mock_method.assert_called() 

137 self.assertIs(mock_method.call_args[0][0], bundle) 

138 

139 def testTimeoutConfiguration(self) -> None: 

140 # The configured timeout must reach the dispatcher, and a dispatcher 

141 # made without one must still get DEFAULT_TIMEOUT rather than none. 

142 self.assertEqual(self.butler._datastore.timeout, 5.5) 

143 self.assertEqual(self.butler._datastore._dispatcher.timeout, 5.5) 

144 self.assertEqual(SasquatchDispatcher("http://test.local", "na").timeout, DEFAULT_TIMEOUT) 

145 

146 def testRequestsUseTimeout(self) -> None: 

147 # An HTTP call made without the timeout lets a proxy that accepts 

148 # connections but never answers hang the put indefinitely. 

149 dispatcher = SasquatchDispatcher("http://test.local", "na", timeout=12.5) 

150 session = _fakeSession(getResponse=_fakeResponse(json={"data": [{"cluster_id": "abc"}]})) 

151 

152 with patch(HTTP_CLIENT) as mockHttpClient: 

153 mockHttpClient.return_value.__enter__.return_value = session 

154 self.assertEqual(dispatcher.clusterId, "abc") 

155 self.assertTrue(dispatcher._create_topic("some.metric")) 

156 

157 self.assertEqual(session.get.call_args.kwargs["timeout"], 12.5) 

158 self.assertEqual(session.post.call_args.kwargs["timeout"], 12.5) 

159 

160 def testClusterIdFailuresRaiseDispatchFailure(self) -> None: 

161 # Each bad response must surface as SasquatchDispatchFailure, the 

162 # failure the datastore expects. Parsing the body before checking 

163 # the status lets the error document through as a bare KeyError. 

164 dispatcher = SasquatchDispatcher("http://test.local", "na") 

165 jsonError = requests.exceptions.JSONDecodeError("Expecting value", "<html>", 0) 

166 badResponses = { 

167 "5xx with an error document": _fakeResponse(503, json={"error_code": 50301}), 

168 "2xx with a non-JSON body": _fakeResponse(200, jsonError=jsonError), 

169 } 

170 for description, response in badResponses.items(): 

171 with self.subTest(description): 

172 with patch(HTTP_CLIENT) as mockHttpClient: 

173 mockHttpClient.return_value.__enter__.return_value = _fakeSession(getResponse=response) 

174 with self.assertRaises(SasquatchDispatchFailure): 

175 _ = dispatcher.clusterId 

176 

177 def test_explicit_timestamp_version(self): 

178 dispatcher = SasquatchDispatcher("http://test.local", "na") 

179 bundle = MetricMeasurementBundle() 

180 bundle.timestamp_version = "explicit_timestamp" 

181 # verify this raises with no specified time 

182 with self.assertRaises(ValueError): 

183 dispatcher._handleTimes({}, bundle, "localRun") 

184 # verify this raise with a date that can't be parsed 

185 bundle.timestamp_version = "explicit_timestamp:123233" 

186 with self.assertRaises(ValueError): 

187 dispatcher._handleTimes({}, bundle, "localRun") 

188 # verify that a correct time gets parsed 

189 bundle.timestamp_version = "explicit_timestamp:20230728T165102Z" 

190 meta = {} 

191 dispatcher._handleTimes(meta, bundle, "localRun") 

192 dt = datetime.datetime(2023, 7, 28, 16, 51, 2, tzinfo=datetime.UTC) 

193 self.assertEqual(meta["timestamp"], dt.timestamp()) 

194 

195 

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

197 unittest.main()