Coverage for tests/test_consdbClient.py: 31%
100 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-05 09:19 +0000
1# This file is part of summit_utils.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://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 <http://www.gnu.org/licenses/>.
22import pytest
23import responses
24from requests import HTTPError
26from lsst.summit.utils import ConsDbClient, FlexibleMetadataInfo
29@pytest.fixture
30def client():
31 """Initialize client with a fake url
32 Requires mocking connection with @responses.activate decorator
33 """
34 return ConsDbClient("http://example.com/consdb")
37def test_table_name():
38 instrument = "latiss"
39 obs_type = "exposure"
40 assert (
41 ConsDbClient.compute_flexible_metadata_table_name(instrument, obs_type)
42 == "cdb_latiss.exposure_flexdata"
43 )
46@responses.activate
47def test_add_flexible_metadata_key(client):
48 instrument = "latiss"
49 obs_type = "exposure"
50 responses.post(
51 "http://example.com/consdb/flex/latiss/exposure/addkey",
52 json={
53 "message": "Key added to flexible metadata",
54 "key": "foo",
55 "instrument": "latiss",
56 "obs_type": "exposure",
57 },
58 match=[
59 responses.matchers.json_params_matcher({"key": "foo", "dtype": "bool", "doc": "bool key"}),
60 ],
61 )
62 responses.post(
63 "http://example.com/consdb/flex/latiss/exposure/addkey",
64 json={
65 "message": "Key added to flexible metadata",
66 "key": "bar",
67 "instrument": "latiss",
68 "obs_type": "exposure",
69 },
70 match=[
71 responses.matchers.json_params_matcher({"key": "bar", "dtype": "int", "doc": "int key"}),
72 ],
73 )
74 responses.post(
75 "http://example.com/consdb/flex/bad_instrument/exposure/addkey",
76 status=404,
77 json={"message": "Unknown instrument", "value": "bad_instrument", "valid": ["latiss"]},
78 )
79 responses.post(
80 "http://example.com/consdb/flex/latiss/bad_obs_type/addkey",
81 status=404,
82 json={"message": "Unknown observation type", "value": "bad_obs_type", "valid": ["exposure"]},
83 )
85 assert (
86 client.add_flexible_metadata_key(instrument, obs_type, "foo", "bool", "bool key").json()["key"]
87 == "foo"
88 )
89 assert (
90 client.add_flexible_metadata_key(instrument, obs_type, "bar", "int", "int key").json()["instrument"]
91 == "latiss"
92 )
93 with pytest.raises(HTTPError, match="404") as e:
94 client.add_flexible_metadata_key("bad_instrument", obs_type, "error", "int", "instrument error")
95 assert "Unknown instrument" in str(e.value.__notes__)
96 json_data = e.value.response.json()
97 assert json_data["message"] == "Unknown instrument"
98 assert json_data["value"] == "bad_instrument"
99 assert json_data["valid"] == ["latiss"]
100 with pytest.raises(HTTPError, match="404"):
101 client.add_flexible_metadata_key(instrument, "bad_obs_type", "error", "int", "obs_type error")
104@responses.activate
105def test_get_flexible_metadata_keys(client):
106 description = {"foo": ["bool", "a", None, None], "bar": ["float", "b", "deg", "pos.eq.ra"]}
107 responses.get(
108 "http://example.com/consdb/flex/latiss/exposure/schema",
109 json=description,
110 )
111 instrument = "latiss"
112 obs_type = "exposure"
113 assert client.get_flexible_metadata_keys(instrument, obs_type) == {
114 "foo": FlexibleMetadataInfo("bool", "a"),
115 "bar": FlexibleMetadataInfo("float", "b", "deg", "pos.eq.ra"),
116 }
119@responses.activate
120def test_get_flexible_metadata(client):
121 results = {"bool_key": True, "int_key": 42, "float_key": 3.14159, "str_key": "foo"}
122 responses.get(
123 "http://example.com/consdb/flex/latiss/exposure/obs/271828",
124 json=results,
125 )
126 responses.get(
127 "http://example.com/consdb/flex/latiss/exposure/obs/271828?k=float_key", json={"float_key": 3.14159}
128 )
129 responses.get(
130 "http://example.com/consdb/flex/latiss/exposure/obs/271828?k=int_key&k=float_key",
131 json={"float_key": 3.14159, "int_key": 42},
132 )
133 instrument = "latiss"
134 obs_type = "exposure"
135 obs_id = 271828
136 assert client.get_flexible_metadata(instrument, obs_type, obs_id) == results
137 assert client.get_flexible_metadata(instrument, obs_type, obs_id, ["float_key"]) == {
138 "float_key": results["float_key"]
139 }
140 assert client.get_flexible_metadata(instrument, obs_type, obs_id, ["int_key", "float_key"]) == {
141 "int_key": results["int_key"],
142 "float_key": results["float_key"],
143 }
146@responses.activate
147def test_insert_flexible_metadata(client):
148 instrument = "latiss"
149 obs_type = "exposure"
150 with pytest.raises(ValueError):
151 client.insert_flexible_metadata(instrument, obs_type, 271828)
152 # TODO: more POST tests
155@responses.activate
156def test_schema(client):
157 description = {"foo": ("bool", "a"), "bar": ("int", "b")}
158 responses.get(
159 "http://example.com/consdb/schema/latiss/misc_table",
160 json=description,
161 )
162 instrument = "latiss"
163 table = "misc_table"
164 assert client.schema(instrument, table) == description
167@responses.activate
168@pytest.mark.parametrize(
169 "secret, redacted",
170 [
171 ("usdf:v987wefVMPz", "us***:v9***"),
172 ("u:v", "u***:v***"),
173 ("ulysses", "ul***"),
174 (":alberta94", "***:al***"),
175 ],
176)
177def test_clean_token_url_response(secret, redacted):
178 """Test tokens URL is cleaned when an error is thrown from requests
179 Use with pytest raises assert an error'
180 assert that url does not contain tokens
181 """
182 domain = "@usdf-fake.slackers.stanford.edu/consdb"
183 complex_client = ConsDbClient(f"https://{secret}{domain}")
185 obs_type = "exposure"
186 responses.post(
187 f"https://{secret}{domain}/flex/bad_instrument/exposure/addkey",
188 status=404,
189 )
190 with pytest.raises(HTTPError, match="404") as error:
191 complex_client.add_flexible_metadata_key(
192 "bad_instrument", obs_type, "error", "int", "instrument error"
193 )
195 url = error.value.args[0].split()[-1]
196 sanitized = f"https://{redacted}{domain}/flex/bad_instrument/exposure/addkey"
197 assert url == sanitized
200def test_client(client):
201 """Test ConsDbClient is initialized properly"""
202 assert "clean_url" in str(client.session.hooks["response"])
205@responses.activate
206def test_insert_obs_id(client):
207 """An integer obs_id targets the ``.../obs/{obs_id}`` endpoint."""
208 responses.post(
209 "http://example.com/consdb/insert/latiss/exposure/obs/271828",
210 json={"message": "Data inserted"},
211 match=[
212 responses.matchers.json_params_matcher(
213 {"table": "exposure", "obs_id": 271828, "values": {"foo": 1}}
214 ),
215 ],
216 )
217 assert client.insert("latiss", "exposure", 271828, {"foo": 1}).json()["message"] == "Data inserted"
220@responses.activate
221def test_insert_by_seq_num(client):
222 """A 2-tuple targets the ``.../{day_obs}/{seq_num}`` endpoint."""
223 responses.post(
224 "http://example.com/consdb/insert/latiss/exposure/by_seq_num/20240603/123",
225 json={"message": "Data inserted"},
226 match=[
227 responses.matchers.json_params_matcher({"table": "exposure", "values": {"foo": 1}}),
228 ],
229 )
230 assert (
231 client.insert("latiss", "exposure", (20240603, 123), {"foo": 1}).json()["message"] == "Data inserted"
232 )
235@responses.activate
236def test_insert_by_seq_num_detector(client):
237 """A 3-tuple targets the ``.../{day_obs}/{seq_num}/{detector}``
238 endpoint for per-detector (ccdexposure-level) tables."""
239 responses.post(
240 "http://example.com/consdb/insert/lsstcam/ccdexposure/by_seq_num/20240603/123/94",
241 json={"message": "Data inserted"},
242 match=[
243 responses.matchers.json_params_matcher({"table": "ccdexposure", "values": {"foo": 1}}),
244 ],
245 )
246 assert (
247 client.insert("lsstcam", "ccdexposure", (20240603, 123, 94), {"foo": 1}).json()["message"]
248 == "Data inserted"
249 )
252@responses.activate
253def test_insert_allow_update(client):
254 """allow_update appends ``?u=1`` to upsert against the addressed key."""
255 responses.post(
256 "http://example.com/consdb/insert/lsstcam/ccdexposure/by_seq_num/20240603/123/94?u=1",
257 json={"message": "Data inserted"},
258 match=[responses.matchers.query_param_matcher({"u": "1"})],
259 )
260 assert (
261 client.insert("lsstcam", "ccdexposure", (20240603, 123, 94), {"foo": 1}, allow_update=True).json()[
262 "message"
263 ]
264 == "Data inserted"
265 )
268def test_insert_bad_obs_id_tuple(client):
269 """A tuple that is not length 2 or 3 is rejected before any request."""
270 with pytest.raises(AssertionError, match="obs_id tuple"):
271 client.insert("latiss", "exposure", (20240603,), {"foo": 1})
272 with pytest.raises(AssertionError, match="obs_id tuple"):
273 client.insert("latiss", "exposure", (20240603, 123, 94, 0), {"foo": 1})
276def test_insert_no_values(client):
277 """Inserting with no values raises before any request."""
278 with pytest.raises(ValueError, match="No values to insert"):
279 client.insert("latiss", "exposure", 271828, {})
282# TODO: more POST tests
283# client.insert_multiple(instrument, table, obs_dict, allow_update)
284# client.query(query)