Coverage for tests/test_utilsEfd.py: 76%
136 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 09:29 +0000
1# This file is part of cp_pipe.
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/>.
22import contextlib
23import glob
24import json
25import os
26import re
28import pytest
29import vcr
30import yaml
32import lsst.utils.tests
34from lsst.cp.pipe.utilsEfd import CpEfdClient
36TESTDIR = os.path.abspath(os.path.dirname(__file__))
37CASSETTE_DIR = os.path.join(TESTDIR, "data", "cassettes")
38EMPTY_EFDAUTH = os.path.join(TESTDIR, "data", "efdauth_sanitized.json")
40REDACTED = "REDACTED"
41_PASSWORD_KEY = re.compile(r"pass(word|wd)", re.IGNORECASE)
44def _redact(obj):
45 """Recursively redact password-like values in place.
47 Returns True if anything was changed.
48 """
49 changed = False
50 if isinstance(obj, dict):
51 for key, value in obj.items():
52 if isinstance(key, str) and _PASSWORD_KEY.search(key) and isinstance(value, str) and value: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 obj[key] = REDACTED
54 changed = True
55 else:
56 changed |= _redact(value)
57 elif isinstance(obj, list):
58 for item in obj:
59 changed |= _redact(item)
60 return changed
63def _scrubBody(body):
64 """Return a redacted copy of a JSON body, or None if nothing changed."""
65 if not body:
66 return None
67 try:
68 data = json.loads(body)
69 except (ValueError, UnicodeDecodeError, TypeError):
70 return None
71 if not _redact(data): 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was always true
72 return None
73 text = json.dumps(data)
74 return text.encode() if isinstance(body, bytes) else text
77def _scrubRequest(request):
78 new = _scrubBody(request.body)
79 if new is not None: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 request.body = new
81 return request
84def _scrubResponse(response):
85 body = response.get("body", {})
86 new = _scrubBody(body.get("string"))
87 if new is not None: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true
88 body["string"] = new
89 length = len(new if isinstance(new, bytes) else new.encode())
90 headers = response.get("headers", {})
91 for key in headers:
92 if key.lower() == "content-length":
93 headers[key] = [str(length)]
94 return response
97VCR_CONFIG = dict(
98 match_on=["method", "scheme", "host", "port", "path", "query", "body"],
99 decode_compressed_response=True, # so response bodies are scrubbable JSON
100 filter_headers=["authorization", "cookie"], # basic auth carries the password
101 filter_query_parameters=["p", "password"], # influx-style ?u=...&p=...
102 filter_post_data_parameters=["p", "password"],
103 before_record_request=_scrubRequest,
104 before_record_response=_scrubResponse, # e.g. credential-service JSON
105)
108# pytest-recording hooks
109@pytest.fixture(scope="module")
110def vcr_config():
111 return VCR_CONFIG
114@pytest.fixture(scope="module")
115def vcr_cassette_dir():
116 return CASSETTE_DIR
119@pytest.fixture(scope="class")
120def efdClient(request):
121 """Class-level replacement for setUpClass.
123 Playback (``--record-mode=none``, the default): EFDAUTH is set to an
124 empty-credentials file in tests/data.
125 Recording (any other mode, or --disable-recording): EFDAUTH is taken
126 from the calling environment.
127 """
128 recordMode = request.config.getoption("--record-mode") or "none"
129 disabled = request.config.getoption("--disable-recording")
130 recording = disabled or recordMode != "none"
132 with pytest.MonkeyPatch.context() as mp:
133 if recording: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 if not os.environ.get("EFDAUTH"):
135 pytest.fail("EFDAUTH must be set in the environment to record EFD cassettes.")
136 else:
137 mp.setenv("EFDAUTH", EMPTY_EFDAUTH)
139 if disabled: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 ctx = contextlib.nullcontext()
141 else:
142 cassette = os.path.join(CASSETTE_DIR, f"{request.cls.__name__}.setUpClass.yaml")
143 mode = recordMode
144 if mode == "rewrite": # pytest-recording-only mode; emulate for vcrpy 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 if os.path.exists(cassette):
146 os.remove(cassette)
147 mode = "all"
148 ctx = vcr.VCR(record_mode=mode, **VCR_CONFIG).use_cassette(cassette)
150 with ctx:
151 try:
152 request.cls.client = CpEfdClient()
153 except Exception as e:
154 if recording:
155 pytest.skip(f"Could not initialize EFD client: {e}")
156 raise # in playback, a failure here is a real regression
158 yield # EFDAUTH stays patched for all tests in the class
161@pytest.mark.usefixtures("efdClient")
162class UtilsEfdTestCase(lsst.utils.tests.TestCase):
163 """Unit test for EFD access code."""
165 @pytest.mark.vcr
166 def test_monochromator(self):
167 data = self.client.getEfdMonochromatorData(
168 dateMin="2023-12-19T00:00:00",
169 dateMax="2023-12-19T23:59:59"
170 )
172 indexDate, wavelength = self.client.parseMonochromatorStatus(
173 data,
174 "2023-12-19T14:37:19.498"
175 )
176 self.assertEqual(wavelength, 550.0)
177 self.assertEqual(indexDate, "2023-12-19T14:37:17.799")
179 @pytest.mark.vcr
180 def test_electrometer(self):
181 data = self.client.getEfdElectrometerData(
182 dateMin="2024-05-30T00:00:00",
183 dateMax="2024-05-30T05:00:00",
184 )
186 # Test single lookups:
187 for (iDate, rDate), (iVal, rVal) in zip([("2024-05-30T04:21:48.6", "2024-05-30T04:22:08"),
188 ("2024-05-30T04:21:50", "2024-05-30T04:22:10"),
189 ("2024-05-30T04:21:53", "2024-05-30T04:22:13"),
190 ("2024-05-30T04:21:58", "2024-05-30T04:22:18"),
191 ("2024-05-30T04:22:17", "2024-05-30T04:22:37")],
192 [(-1.5269e-07, -1.5244e-07),
193 (-1.5168e-07, -1.5137e-07),
194 (-1.5165e-07, -1.5205e-07),
195 (-1.5223e-07, -1.5147e-07),
196 (-1.5226e-07, -1.5558e-07)]):
197 indexDate, intensity, _ = self.client.parseElectrometerStatus(
198 data,
199 iDate
200 )
201 self.assertFloatsAlmostEqual(intensity, iVal, atol=1e-10)
203 indexDateReference, intensityReference, _ = self.client.parseElectrometerStatus(
204 data,
205 rDate
206 )
207 self.assertFloatsAlmostEqual(intensityReference, rVal, atol=1e-10)
209 # Test integrated lookups:
210 iDate = "2024-05-30T04:21:48.6"
211 iDateEnd = "2024-05-30T04:22:18"
212 rDate = "2024-05-30T04:22:08"
213 rDateEnd = "2024-05-30T04:22:38"
214 indexDate, intensity, endDate = self.client.parseElectrometerStatus(
215 data,
216 iDate,
217 dateEnd=iDateEnd,
218 doIntegrateSamples=True,
219 index=201
220 )
221 self.assertFloatsAlmostEqual(intensity, -1.52297e-7, atol=1e-10)
222 indexDateReference, intensityReference, endDateReference = self.client.parseElectrometerStatus(
223 data,
224 rDate,
225 dateEnd=rDateEnd,
226 doIntegrateSamples=True,
227 index=201
228 )
229 self.assertFloatsAlmostEqual(intensityReference, -1.532977e-07, atol=1e-10)
231 @pytest.mark.vcr
232 def test_electrometer_alternate(self):
233 # This should raise if no dates are passed:
234 with self.assertRaises(RuntimeError):
235 data = self.client.getEfdElectrometerData(
236 dataSeries='lsst.sal.Electrometer.logevent_logMessage')
238 data = self.client.getEfdElectrometerData(
239 dataSeries='lsst.sal.Electrometer.logevent_logMessage',
240 dateMin='2024-07-26T16:30:00',
241 dateMax='2024-07-26T16:45:00')
243 # Test single lookups. These should not be integrated.
244 for iDate, iVal in zip(["2024-07-26T16:38:32.228",
245 "2024-07-26T16:38:54.581",
246 "2024-07-26T16:40:56.579",
247 "2024-07-26T16:42:58.553"],
248 [-2.24234e-07,
249 -2.24388e-07,
250 -2.24105e-07,
251 -2.23784e-07]):
252 indexDate, intensity, _ = self.client.parseElectrometerStatus(
253 data,
254 iDate
255 )
256 self.assertFloatsAlmostEqual(intensity, iVal, atol=1e-10)
259class CassetteSanitizationTestCase(lsst.utils.tests.TestCase):
260 """Guard against committing credentials again."""
262 def test_no_credentials_in_cassettes(self):
263 pattern = re.compile(r'"[^"]*pass(?:word|wd)[^"]*"\s*:\s*"([^"]*)"', re.IGNORECASE)
264 for fname in glob.glob(os.path.join(CASSETTE_DIR, "*.yaml")):
265 with open(fname) as f:
266 cassette = yaml.safe_load(f)
267 for interaction in cassette.get("interactions", []):
268 req = interaction["request"]
269 headers = {k.lower() for k in req.get("headers", {})}
270 self.assertNotIn("authorization", headers, msg=fname)
271 self.assertIsNone(re.search(r"[?&]p(assword)?=", req.get("uri", "")), msg=fname)
272 body = interaction["response"]["body"].get("string") or ""
273 if isinstance(body, bytes): 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 body = body.decode(errors="replace")
275 for m in pattern.finditer(body): 275 ↛ 276line 275 didn't jump to line 276 because the loop on line 275 never started
276 self.assertIn(m.group(1), ("", REDACTED), msg=fname)