SND@LHC Software
Loading...
Searching...
No Matches
test_mongodbadapter.py
Go to the documentation of this file.
1""" This module contains regression tests for the MongoDB CDB adapter. """
2import os
3import pytest
4import datetime
5
6from ...factory import APIFactory
7
8
9# Package metadata
10__authors__ = ["Yusril Maulidan Raji", "Yitian Kong"]
11__copyright__ = "TU/e ST2019"
12__version__ = "1.0"
13__status__ = "Prototype"
14__description__ = "This file contains the test cases for Mongo DB API."
15
16
17@pytest.fixture
18def cdb_api():
19 """
20 provides access to the condition database api
21
22 :return: api
23 """
24
25 home_dir = os.getenv('FAIRSHIP')
26 factory = APIFactory()
27 db_api = factory.construct_DB_API(home_dir + "/conditionsDatabase/tests/test_mongodb/test_mongodb_config.yml")
28
29 return db_api
30
31
32@pytest.mark.parametrize("detector_id", [
33 # Test with None, int, and empty string
34 None, 1, "",
35 # Test with not exist detector id
36 "detector_id_not_exist",
37 # Test with exist detector id
38 "detector_id_exist",
39 # Test with exist detector id
40 "detector_id_exist",
41 # Test with exist sub detector
42 "detector_id_exist/sub_detector_id_exist",
43 # Test with exist sub sub detector
44 "detector_id_exist/sub_detector_id_exist/sub_sub_detector_id_exist",
45 # Test how the function deals with slashes at the beginning and end [ValueError]
46 "/detector_id_exist/",
47 # Test with multiple slashes in the middle of the path [ValueError]
48 "detector///subdetector",
49 # Test with multiple slashes in the beginning and end of the path [ValueError]
50 "//detector/subdetector//",
51 # Test with underscore delimiter [ValueError]
52 "detector_subdetector",
53 # Test with single quote character [Accepted]
54 "detector'subdetector",
55 # Test with tab special character [Accepted]
56 "detector\\tsubdetector",
57 # Test with double quote character [Accepted]
58 "detector\"subdetector",
59 # Test with whitespace delimiter [Accepted]
60 "detector subdetector"
61])
62def test_get_detector(cdb_api, detector_id):
63 """
64 test get_detector()
65 this function must be executed first because the other functions relies heavily
66 on this function
67
68 :param cdb_api: API
69 :param detector_id: String identifying the detector to retrieve (i.e. 'muonflux/straw_tubes').
70 """
71 has_correct_parameter = True
72 if type(detector_id) != str:
73 has_correct_parameter = False
74 # raise TypeError when detector_id is not an str.
75 with pytest.raises(TypeError):
76 assert cdb_api.get_detector(detector_id)
77
78 value_error_parameters = ["", "//detector/subdetector//", "detector///subdetector"]
79 if detector_id in value_error_parameters:
80 has_correct_parameter = False
81 # raise ValueError when detector_id is not valid.
82 with pytest.raises(ValueError):
83 assert cdb_api.get_detector(detector_id)
84
85 value_not_found_parameters = ["detector_id_not_exist", "detector_subdetector",
86 "detector'subdetector", "detector\\tsubdetector",
87 "detector\"subdetector", "detector subdetector"]
88 if detector_id in value_not_found_parameters:
89 has_correct_parameter = False
90 # raise TypeError when detector_id does not exist.
91 with pytest.raises(ValueError):
92 assert cdb_api.get_detector(detector_id)
93
94 if has_correct_parameter:
95 result = cdb_api.get_detector(detector_id)
96 # check if the result is dict or None
97 assert type(result) == dict or result is None, \
98 "The result type must be a dict object (if found) or None (if not found)"
99
100 # check if the result has the correct name
101 if type(result) == dict:
102 # check if the returned detector still has the same detector_id.
103 if detector_id == "/detector_id_exist/":
104 assert result["name"] == detector_id.split('/')[1], \
105 "the result has different detector name"
106 else:
107 assert result["name"] == detector_id.split('/')[-1], \
108 "the result has different detector name"
109
110 # check if the detector has the correct value
111 if result["name"] == "detector_id_exist":
112 assert result["conditions"][0]["name"] == "condition_exist", \
113 "condition name has wrong value"
114 assert result["conditions"][0]["tag"] == "tag", "condition tag has wrong value"
115 assert result["conditions"][0]["type"] == "type", "condition tag has wrong value"
116 assert result["conditions"][0]["collected_at"] == "2020,03,12,11,05,27,000000", \
117 "condition collected_at has wrong value"
118 assert result["conditions"][0]["valid_since"] == "2020,03,12,11,05,27,000000", \
119 "condition valid_since has wrong value"
120 assert result["conditions"][0]["valid_until"] == "2020,04,10,11,05,27,000000", \
121 "condition valid_until has wrong value"
122 assert result["conditions"][0]["values"] == \
123 {"name": "name detector", "value": "value detector"}, \
124 "condition values has wrong value"
125
126 # check if the sub detector has the correct value
127 if result["name"] == "sub_detector_id_exist":
128 assert result["conditions"][0]["name"] == "sub_detector_id_exist", \
129 "condition name has wrong value"
130 assert result["conditions"][0]["tag"] == "sub_tag", "condition tag has wrong value"
131 assert result["conditions"][0]["type"] == "sub_type", "condition tag has wrong value"
132 assert result["conditions"][0]["collected_at"] == "2020,04,12,11,05,27,000000", \
133 "condition collected_at has wrong value"
134 assert result["conditions"][0]["valid_since"] == "2020,04,12,11,05,27,000000", \
135 "condition valid_since has wrong value"
136 assert result["conditions"][0]["valid_until"] == "2020,05,12,11,05,27,000000", \
137 "condition valid_until has wrong value"
138 assert result["conditions"][0]["values"] == {"name": "name sub detector",
139 "value": "value sub detector"}, \
140 "condition values has wrong value"
141
142 # check if the sub sub detector has the correct value
143 if result["name"] == "sub_sub_detector_id_exist":
144 assert result["conditions"][0]["name"] == "sub_sub_detector_id_exist", \
145 "condition name has wrong value"
146 assert result["conditions"][0]["tag"] == "sub_sub_tag", "condition tag has wrong value"
147 assert result["conditions"][0]["type"] == "sub_sub_type", \
148 "condition type has wrong value"
149 assert result["conditions"][0]["collected_at"] == "2020,05,12,11,05,27,000000", \
150 "condition collected_at has wrong value"
151 assert result["conditions"][0]["valid_since"] == "2020,05,12,11,05,27,000000", \
152 "condition valid_since has wrong value"
153 assert result["conditions"][0]["valid_until"] == "2020,06,12,11,05,27,000000", \
154 "condition valid_until has wrong value"
155 assert result["conditions"][0]["values"] == \
156 {"name": "name sub sub detector", "value": "value sub sub detector"}, \
157 "condition values has wrong value"
158
159
160@pytest.mark.parametrize("detector_id", [
161 # Test with None, int, and empty string
162 None, 999, "",
163 # Test with exist detector
164 "detector_id_exist",
165 # Test with exist detector, but not exist sub detector
166 "detector_id_exist/sub_detector_id_not_exist",
167 # Test with not exist detector
168 "detector_id_not_exist",
169 # Test with a detector without condition
170 "detector_without_condition"
171])
172def test_get_conditions(cdb_api, detector_id):
173 """
174 test get_conditions()
175
176 :param cdb_api: API
177 :param detector_id: String identifying the detector for which the
178 conditions must be retrieved (i.e. 'muonflux/straw_tubes').
179 """
180 has_correct_parameter = True
181 if type(detector_id) != str:
182 has_correct_parameter = False
183 # raise TypeError when detector_id is not an str.
184 with pytest.raises(TypeError):
185 assert cdb_api.get_conditions(detector_id)
186
187 not_exist_detectors = ["", "detector_id_not_exist",
188 "detector_id_exist/sub_detector_id_not_exist"]
189 if detector_id in not_exist_detectors:
190 has_correct_parameter = False
191 # raise ValueError when detector_id does not exist.
192 with pytest.raises(ValueError):
193 assert cdb_api.get_conditions(detector_id)
194
195 if has_correct_parameter:
196 if detector_id == "detector_without_condition":
197 assert cdb_api.get_conditions(detector_id) is None, \
198 "detector with no condition must return None"
199
200 result = cdb_api.get_conditions(detector_id)
201 # check if the result is a list or None
202 assert type(result) == list or result is None, \
203 "the returned type must be list object (if found) or None (if not found)"
204
205 # check if the result has the correct value
206 if detector_id == "detector_id_exist":
207 assert result[0]["name"] == "condition_exist", "the condition value is wrong"
208
209
210@pytest.mark.parametrize("detector_id, name", [
211 # Test all parameters with None, int, and empty string
212 (None, None), (999, 999), ("", ""),
213 # Test with exist detector and exist condition
214 ("detector_id_exist", "condition_exist"),
215 # Test with exist detector, but not exist condition
216 ("detector_id_exist", "condition_not_exist"),
217 # Test with not exist detector
218 ("detector_id_not_exist", "condition_not_exist"),
219 # Test with invalid detector_id and valid name
220 (999, "condition_exist"),
221 # Test with exist detector, not exist detector, and exist condition
222 ("detector_id_exist/sub_detector_not_exist", "condition_exist")
223])
224def test_get_conditions_by_name(cdb_api, detector_id, name):
225 """
226 test get_conditions_by_name()
227
228 :param cdb_api: API
229 :param detector_id: String identifying the detector for which the
230 condition must be retrieved (i.e. 'muonflux/straw_tubes').
231 :param name: String specifying the name of the conditions to be retrieved (e.g.
232 'strawPositions').
233 """
234
235 has_correct_parameter_type = True
236 if type(detector_id) != str:
237 has_correct_parameter_type = False
238 # raise TypeError when detector_id is not an str.
239 with pytest.raises(TypeError):
240 assert cdb_api.get_conditions_by_name(detector_id, name)
241
242 if type(name) != str:
243 has_correct_parameter_type = False
244 # raise TypeError when name is not an str.
245 with pytest.raises(TypeError):
246 assert cdb_api.get_conditions_by_name(detector_id, name)
247
248 not_exist_detectors = ["", "detector_id_not_exist",
249 "detector_id_exist/sub_detector_not_exist"]
250 if detector_id in not_exist_detectors:
251 has_correct_parameter_type = False
252 # raise ValueError when detector_id does not exist.
253 with pytest.raises(ValueError):
254 assert cdb_api.get_conditions_by_name(detector_id, name)
255
256 if has_correct_parameter_type:
257 result = cdb_api.get_conditions_by_name(detector_id, name)
258 # check if the result is a list.
259 assert type(result) == list or result is None, \
260 "The result type must be a list (if found) or None (if not found)"
261 # check if the result has the correct value
262 if detector_id == "detector_id_exist" and name == "condition_exist":
263 assert result[0] != name, "the condition has a wrong value"
264
265
266@pytest.mark.parametrize("detector_id, name, start_date, end_date", [
267 # Test with None input for each parameter
268 (None, None, None, None),
269 # Test with int input for each parameter
270 (999, 999, 999, 999),
271 # Test with empty string input for each parameter
272 ("", "", "", ""),
273 # Test with wrong string format for date parameter
274 ("detector_id_exist", "condition_exist", "2020,03,16,12,07,30", None),
275 # Test with correct string format for date parameter
276 ("detector_id_exist", "condition_exist", "2020-03-16 12:07:30", None),
277 # Test with datetime obj for date parameter
278 ("detector_id_exist", "condition_exist", datetime.datetime(2020, 3, 16, 12, 7, 30), None),
279 # Test with string obj for date parameter with precise millisecond (wrong format)
280 ("detector_id_exist", "condition_exist", "2020,03,16,12,07,30,129920", None),
281 # Test with string obj for date parameter with precise millisecond (correct format)
282 ("detector_id_exist", "condition_exist", "2020-03-16 12:07:30:129920", None),
283 # Test with datetime obj for date parameter with precise millisecond
284 ("detector_id_exist", "condition_exist", datetime.datetime(2020, 3, 12, 11, 5, 27, 767563), None),
285 # Test with datetime obj for date parameter with not precise millisecond
286 ("detector_id_exist", "condition_exist", datetime.datetime(2020, 3, 12, 11, 5, 27, 555555), None),
287 # Test with datetime obj for date parameter with day value
288 ("detector_id_exist", "condition_exist", datetime.datetime(2020, 3, 13), None),
289 # Test with valid detector id, but invalid name and date
290 ("detector_id_exist", 999, 999, None),
291 # Test with not exist detector
292 ("detector_id_not_exist", "condition_exist", datetime.datetime(2020, 3, 16, 12, 7, 30), None),
293 # Test with not exist detector, but not exist sub detector
294 ("detector_id_exist/sub_detector_not_exist", "condition_exist", datetime.datetime(2020, 3, 16, 12, 7, 30), None),
295 # Test with not found data and one date parameter
296 ("detector_id_exist", "condition_exist", "2021-03-16 12:07:30", None),
297 # Test with str obj for one date parameter with year, month, date, hour, minute, and second
298 ("detector_id_exist", "condition_exist", "2020-03-16 12:07:30", None),
299 # Test with str obj for one date parameter with year, month, date, hour, and minute
300 ("detector_id_exist", "condition_exist", "2020-03-16 12:07", None),
301 # Test with str obj for one date parameter with year, month, date, and hour
302 ("detector_id_exist", "condition_exist", "2020-03-16 12", None),
303 # Test with str obj for one date parameter with year, month, and date
304 ("detector_id_exist", "condition_exist", "2020-03-16", None),
305 # Test with str obj for one date parameter with year and month
306 ("detector_id_exist", "condition_exist", "2020-03", None),
307 # Test with str obj for one date parameter with year
308 ("detector_id_exist", "condition_exist", "2020", None),
309 # Test with not found date range parameter
310 ("detector_id_exist", "condition_exist", "2021-03-16 12:07:30", "2021-04-16 12:07:30"),
311 # Test with str obj for date range parameter with year, month, date, hour, minute, and second
312 ("detector_id_exist", "condition_exist", "2020-03-16 12:07:30", "2020-04-10 10:07:30"),
313 # Test with str obj for date range parameter with year, month, date, hour, and minute
314 ("detector_id_exist", "condition_exist", "2020-03-16 12:07", "2020-04-10 10:07"),
315 # Test with str obj for date range parameter with year, month, date, and hour
316 ("detector_id_exist", "condition_exist", "2020-03-16 12", "2020-04-10 10"),
317 # Test with str obj for date range parameter with year, month, and date
318 ("detector_id_exist", "condition_exist", "2020-03-16", "2020-04-10"),
319 # Test with str obj for date range parameter with year and month
320 ("detector_id_exist", "condition_exist", "2020-03", "2020-04"),
321 # Test with str obj for date range parameter with year
322 ("detector_id_exist", "condition_exist", "2020", "2021"),
323])
324def test_get_conditions_by_name_and_validity(cdb_api, detector_id, name, start_date, end_date):
325 """
326 test get_conditions_by_name_and_validity()
327
328 :param cdb_api: API
329 :param detector_id: String identifying the detector for which the
330 condition must be retrieved (i.e. 'muonflux/straw_tubes').
331 :param name: String specifying the name of the conditions to be retrieved (e.g.
332 'strawPositions').
333 :param start_date: Timestamp specifying a date/time for which conditions must be valid.
334 :param end_date: Timestamp specifying the end of a date/time range for which conditions must be valid.
335 """
336
337 has_correct_parameter_type = True
338 if type(detector_id) != str:
339 has_correct_parameter_type = False
340 # raise TypeError when detector_id is not an str.
341 with pytest.raises(TypeError):
342 assert cdb_api.get_conditions_by_name_and_validity(detector_id, name,
343 start_date, end_date)
344
345 if type(name) != str:
346 has_correct_parameter_type = False
347 # raise TypeError when name is not an str.
348 with pytest.raises(TypeError):
349 assert cdb_api.get_conditions_by_name_and_validity(detector_id, name,
350 start_date, end_date)
351
352 if (type(start_date) != str and type(start_date) != datetime.datetime) or \
353 (type(end_date) != str and type(end_date) != datetime.datetime and end_date is not None):
354 has_correct_parameter_type = False
355 # raise TypeError when date is not an str.
356 with pytest.raises(TypeError):
357 assert cdb_api.get_conditions_by_name_and_validity(detector_id, name,
358 start_date, end_date)
359
360 if type(start_date) == str:
361 if not has_correct_format(start_date):
362 has_correct_parameter_type = False
363 # raise ValueError when the format is wrong
364 with pytest.raises(ValueError):
365 assert cdb_api.get_conditions_by_name_and_validity(detector_id, name,
366 start_date, end_date)
367
368 if type(end_date) == str:
369 if not has_correct_format(end_date):
370 has_correct_parameter_type = False
371 # raise ValueError when the format is wrong
372 with pytest.raises(ValueError):
373 assert cdb_api.get_conditions_by_name_and_validity(detector_id, name,
374 start_date, end_date)
375
376 not_exist_detectors = ["detector_id_not_exist", "detector_id_exist/sub_detector_not_exist"]
377 if detector_id in not_exist_detectors:
378 has_correct_parameter_type = False
379 # raise ValueError when detector id does not exist.
380 with pytest.raises(ValueError):
381 assert cdb_api.get_conditions_by_name_and_validity(detector_id, name,
382 start_date, end_date)
383
384 if has_correct_parameter_type:
385 result = cdb_api.get_conditions_by_name_and_validity(detector_id, name,
386 start_date, end_date)
387 # check if the result is a list or None
388 assert type(result) == list or result is None, \
389 "The result type must be a list (if found) or None (if not found)"
390
391 valid_cases = [
392 detector_id == "detector_id_exist" and name == "condition_exist" and
393 start_date == datetime.datetime(2020, 3, 16, 12, 7, 30),
394 detector_id == "detector_id_exist" and name == "condition_exist" and
395 start_date == "2020-03-16 12:07:30",
396 detector_id == "detector_id_exist" and name == "condition_exist" and
397 start_date == "2020-03-16 12:07:30:129920",
398 detector_id == "detector_id_exist" and name == "condition_exist" and
399 start_date == datetime.datetime(2020, 3, 12, 11, 5, 27, 767563),
400 detector_id == "detector_id_exist" and name == "condition_exist" and
401 start_date == datetime.datetime(2020, 3, 12, 11, 5, 27, 555555),
402 detector_id == "detector_id_exist" and name == "condition_exist" and
403 start_date == datetime.datetime(2020, 3, 13),
404 detector_id == "detector_id_exist" and name == "condition_exist" and
405 start_date == "2020-03-16 12:07:30",
406 detector_id == "detector_id_exist" and name == "condition_exist" and
407 start_date == "2020-03-16 12:07",
408 detector_id == "detector_id_exist" and name == "condition_exist" and
409 start_date == "2020-03-16 12",
410 detector_id == "detector_id_exist" and name == "condition_exist" and
411 start_date == "2020-03-16"
412 ]
413 invalid_cases = [
414 detector_id == "detector_id_exist" and name == "condition_exist" and
415 start_date == "2020",
416 detector_id == "detector_id_exist" and name == "condition_exist" and
417 start_date == "2020-03"
418 ]
419
420 # check if the value is correct
421 if True in valid_cases:
422 assert type(result) == list, "it must return a value. detector_id: " + detector_id + \
423 " name: " + name + " start_date: " + start_date + \
424 " end_date: " + end_date
425 assert result[0]["name"] == "condition_exist", "condition name value is wrong"
426 assert result[0]["tag"] == "tag", "condition tag value is wrong"
427 assert result[0]["type"] == "type", "condition type value is wrong"
428 assert result[0]["collected_at"] == "2020,03,12,11,05,27,000000", \
429 "condition collected_at value is wrong"
430 assert result[0]["valid_since"] == "2020,03,12,11,05,27,000000", \
431 "condition valid_since value is wrong"
432 assert result[0]["valid_until"] == "2020,04,10,11,05,27,000000", \
433 "condition valid_until value is wrong"
434 assert result[0]["values"] == {"name": "name detector", "value": "value detector"}, \
435 "condition values value is wrong"
436
437 if True in invalid_cases:
438 assert result is None, "it must return None"
439
440 if type(result) == list:
441 assert len(result) == 1, "the result must be a list with one data"
442
443
444@pytest.mark.parametrize("detector_id, name, tag", [
445 # Test all parameters with None, int, and empty string
446 (None, None, None), (999, 999, 999), ("", "", ""),
447 # Test with exist detector, condition, and tag
448 ("detector_id_exist", "condition_exist", "tag"),
449 # Test with not exist detector
450 ("detector_id_not_exist", "condition_exist", "tag"),
451 # Test with exist detector, but not valid name and tag
452 ("detector_id_exist", 999, 999),
453 # Test with exist detector, but not exist sub detector
454 ("detector_id_exist/sub_detector_not_exist", "condition_exist", "tag"),
455])
456def test_get_condition_by_name_and_tag(cdb_api, detector_id, name, tag):
457 """
458 test get_condition_by_name_and_tag()
459
460 :param cdb_api: API
461 :param detector_id: String identifying the detector for which the
462 condition must be retrieved (i.e. 'muonflux/straw_tubes').
463 :param name: String specifying the name of the condition to be retrieved (e.g.
464 'strawPositions').
465 :param tag: String specifying the tag of the condition to be retrieved.
466 """
467
468 has_correct_parameter_type = True
469 if type(detector_id) != str:
470 has_correct_parameter_type = False
471 # raise TypeError when detector_id is not an str.
472 with pytest.raises(TypeError):
473 assert cdb_api.get_condition_by_name_and_tag(detector_id, name, tag)
474
475 if type(name) != str:
476 has_correct_parameter_type = False
477 # raise TypeError when name is not an str.
478 with pytest.raises(TypeError):
479 assert cdb_api.get_condition_by_name_and_tag(detector_id, name, tag)
480
481 if type(tag) != str:
482 has_correct_parameter_type = False
483 # raise TypeError when tag is not an str.
484 with pytest.raises(TypeError):
485 assert cdb_api.get_condition_by_name_and_tag(detector_id, name, tag)
486
487 not_exist_detectors = ["", "detector_id_not_exist", "detector_id_exist/sub_detector_not_exist"]
488 if detector_id in not_exist_detectors:
489 has_correct_parameter_type = False
490 # raise ValueError when detector_id does not exists.
491 with pytest.raises(ValueError):
492 assert cdb_api.get_condition_by_name_and_tag(detector_id, name, tag)
493
494 if has_correct_parameter_type:
495 result = cdb_api.get_condition_by_name_and_tag(detector_id, name, tag)
496 # check if the result is dict or None
497 assert type(result) == dict or result is None, \
498 "The result type must be a dict (if found) or None (if not found)"
499
500 if detector_id == "detector_id_exist" and name == "condition_exist" and tag == "tag":
501 # check if the value is correct
502 assert type(result) == dict, "the returned value must be a dictionary"
503 assert result["name"] == "condition_exist", "condition name value is wrong"
504 assert result["tag"] == "tag", "condition tag value is wrong"
505 assert result["type"] == "type", "condition type value is wrong"
506 assert result["collected_at"] == "2020,03,12,11,05,27,000000", \
507 "condition collected_at value is wrong"
508 assert result["valid_since"] == "2020,03,12,11,05,27,000000", \
509 "condition valid_since value is wrong"
510 assert result["valid_until"] == "2020,04,10,11,05,27,000000", \
511 "condition valid_until value is wrong"
512 assert result["values"] == {"name": "name detector", "value": "value detector"}, \
513 "condition values value is wrong"
514
515
516@pytest.mark.parametrize("detector_id, tag", [
517 # Test all parameters with None, int, and string
518 (None, None), (999, 999), ("", ""),
519 # Test with exist detector and condition tag
520 ("detector_id_exist", "tag"),
521 # Test with exist detector, but not exist condition tag
522 ("detector_id_exist", "not_exist_condition"),
523 # Test with not exist detector and condition tag
524 ("detector_id_not_exist", "tag"),
525 # Test with None detector_id, but valid tag
526 (None, "tag"),
527 # Test with detector_id exists, but sub detector not exists
528 ("detector_id_exist/sub_detector_id_not_exist", "tag")
529])
530def test_get_conditions_by_tag(cdb_api, detector_id, tag):
531 """
532 test get_conditions_by_tag()
533
534 :param cdb_api: API
535 :param detector_id: String identifying the detector for which the
536 condition must be retrieved (i.e. 'muonflux/straw_tubes').
537 :param tag: String specifying the tag of the condition to be retrieved.
538 """
539
540 has_correct_parameter_type = True
541 if type(detector_id) != str:
542 has_correct_parameter_type = False
543 # raise TypeError when detector_id is not an str.
544 with pytest.raises(TypeError):
545 assert cdb_api.get_conditions_by_tag(detector_id, tag)
546
547 if type(tag) != str:
548 has_correct_parameter_type = False
549 # raise TypeError when tag is not an str.
550 with pytest.raises(TypeError):
551 assert cdb_api.get_conditions_by_tag(detector_id, tag)
552
553 # not exist detector_id.
554 not_exist_detector_id = ["", "detector_id_not_exist",
555 "detector_id_exist/sub_detector_id_not_exist"]
556 if detector_id in not_exist_detector_id:
557 has_correct_parameter_type = False
558 # raise ValueError when detector_id does not exist.
559 with pytest.raises(ValueError):
560 assert cdb_api.get_conditions_by_tag(detector_id, tag)
561
562 if has_correct_parameter_type:
563 result = cdb_api.get_conditions_by_tag(detector_id, tag)
564 # check if the result is dict or None
565 assert type(result) == list or result is None, \
566 "The result type must be a dict (if found) or None (if not found)"
567 # check value
568 if detector_id == "detector_id_exist" and tag == "tag":
569 assert type(result) == list and len(result) == 1, \
570 "the returned value must be a list with one element"
571 assert result[0]["name"] == "condition_exist", "condition name value is wrong"
572 assert result[0]["tag"] == "tag", "condition tag value is wrong"
573 assert result[0]["type"] == "type", "condition type value is wrong"
574 assert result[0]["collected_at"] == "2020,03,12,11,05,27,000000", \
575 "condition collected_at value is wrong"
576 assert result[0]["valid_since"] == "2020,03,12,11,05,27,000000", \
577 "condition valid_since value is wrong"
578 assert result[0]["valid_until"] == "2020,04,10,11,05,27,000000", \
579 "condition valid_until value is wrong"
580 assert result[0]["values"] == {"name": "name detector", "value": "value detector"}, \
581 "condition values value is wrong"
582
583
584@pytest.mark.parametrize("detector_id, name, collected_at", [
585 # Test all parameters with None, int, and empty string
586 (None, None, None), (999, 999, 999), ("", "", ""),
587 # Test with str collected_at
588 ("detector_id_exist", "condition_exist", "2020-03-12 11:05:27"),
589 # Test with str collected_at with millisecond
590 ("detector_id_exist", "condition_exist", "2020-03-12 11:05:27:767563"),
591 # Test with datetime collected_at
592 ("detector_id_exist", "condition_exist", datetime.datetime(2020, 3, 12, 11, 5, 27, 767563)),
593 # Test with valid detector id, but invalid condition name and collected_at
594 ("detector_id_exist", 999, 999),
595 # Test with not exist detector_id
596 ("detector_id_not_exist", "condition_exist", "2020-03-12 11:05:27"),
597 # Test with exist detector_id, but not exist sub detector
598 ("detector_id_exist/sub_detector_not_exist", "condition_exist", "2020-03-12 11:05:27"),
599 # Test with exist detector_id, but not exist condition name and collected_at
600 ("detector_id_exist", "condition_not_exist", "2021-03-12 11:05:27")
601])
602def test_get_condition_by_name_and_collection_date(cdb_api, detector_id, name, collected_at):
603 """
604 test get_condition_by_name_and_collection_date()
605
606 :param cdb_api: API
607 :param detector_id: String identifying the detector for which the
608 condition must be retrieved (i.e. 'muonflux/straw_tubes').
609 :param name: String specifying the name of the condition to be retrieved (e.g.
610 'strawPositions').
611 :param collected_at: Timestamp specifying the moment on which the
612 condition was collected / measured. This timestamp must be unique w.r.t.
613 the condition name.
614 """
615
616 has_correct_parameter_type = True
617 if type(detector_id) != str:
618 has_correct_parameter_type = False
619 # raise TypeError when detector_id is not an str.
620 with pytest.raises(TypeError):
621 assert cdb_api.get_condition_by_name_and_collection_date(detector_id, name, collected_at)
622
623 if type(name) != str:
624 has_correct_parameter_type = False
625 # raise TypeError when name is not an str.
626 with pytest.raises(TypeError):
627 assert cdb_api.get_condition_by_name_and_collection_date(detector_id, name, collected_at)
628
629 if type(collected_at) != str and type(collected_at) != datetime.datetime:
630 has_correct_parameter_type = False
631 # raise TypeError when date is not an str.
632 with pytest.raises(TypeError):
633 assert cdb_api.get_condition_by_name_and_collection_date(detector_id, name, collected_at)
634
635 if type(collected_at) is str:
636 # if the format is wrong, the function has to raise ValueError
637 if not has_correct_format(collected_at):
638 has_correct_parameter_type = False
639 with pytest.raises(ValueError):
640 assert cdb_api.get_condition_by_name_and_collection_date(detector_id, name, collected_at)
641
642 not_exist_detectors = ["detector_id_exist/sub_detector_not_exist", "detector_id_not_exist"]
643 if detector_id in not_exist_detectors:
644 has_correct_parameter_type = False
645 # if detector_id does not exist, raise ValueError
646 with pytest.raises(ValueError):
647 assert cdb_api.get_condition_by_name_and_collection_date(detector_id, name, collected_at)
648
649 if has_correct_parameter_type:
650 result = cdb_api.get_condition_by_name_and_collection_date(detector_id, name, collected_at)
651 # check if the result type is dict or None.
652 assert type(result) == dict or result is None, \
653 "The result type must be a dict (if found) or None (if not found)"
654
655 # check the value
656 if type(result) == dict:
657 assert type(result) == dict, "the returned value must be a dict"
658 assert result["name"] == "condition_exist", "condition name value is wrong"
659 assert result["tag"] == "tag", "condition tag value is wrong"
660 assert result["type"] == "type", "condition type value is wrong"
661 assert result["collected_at"] == "2020,03,12,11,05,27,000000", \
662 "condition collected_at value is wrong"
663 assert result["valid_since"] == "2020,03,12,11,05,27,000000", \
664 "condition valid_since value is wrong"
665 assert result["valid_until"] == "2020,04,10,11,05,27,000000", \
666 "condition valid_until value is wrong"
667 assert result["values"] == {"name": "name detector", "value": "value detector"}, \
668 "condition values value is wrong"
669
670
671@pytest.mark.parametrize("parent_id", [
672 # Test with None, int, and empty string
673 None, 1, "",
674 # Test with not exist detector
675 "detector_id_not_exist",
676 # Test with exist detector
677 "detector_id_exist",
678 # Test with exist detector and exist sub detector
679 "detector_id_exist/sub_detector_id_exist",
680 # Test with exist detector, exist sub detector, and exist sub sub detector
681 "detector_id_exist/sub_detector_id_exist/sub_sub_detector_id_exist"
682])
683def test_list_detectors(cdb_api, parent_id):
684 """
685 test list_detectors()
686
687 :param cdb_api: API
688 :param parent_id: (optional) String identifying the parent detector to
689 retrieve the (sub)detector names for (i.e. 'muonflux/straw_tubes').
690 """
691 has_correct_parameter = True
692 if type(parent_id) != str and parent_id is not None:
693 has_correct_parameter = False
694 # raise TypeError when parent_id is not an str.
695 with pytest.raises(TypeError):
696 assert cdb_api.list_detectors(parent_id)
697
698 if parent_id == "detector_id_not_exist":
699 has_correct_parameter = False
700 # raise ValueError when parent_id does not exist.
701 with pytest.raises(ValueError):
702 assert cdb_api.list_detectors(parent_id)
703
704 if has_correct_parameter:
705 result = cdb_api.list_detectors(parent_id)
706 # check if the result is a list
707 assert type(result) == list or result is None, \
708 "The result type must be a list object (if found) or None (if not found)"
709
710 # check when the parent_id is None, it should return all first level detectors
711 if parent_id is None:
712 assert sorted(result) == sorted(["detector_id_exist", "detector_without_condition"]), \
713 "when parent_id is None, it should return all conditions"
714
715 # check if there is an element that is not str and if the detector exists
716 for detector in result:
717 assert type(detector) == str, "Every element must be a str"
718 tmp_detector = None
719 try:
720 tmp_detector = cdb_api.get_detector(detector)
721 except ValueError:
722 tmp_detector = None
723
724 assert tmp_detector is not None, "Cannot find this detector: " + detector
725
726
727@pytest.mark.parametrize("name, parent_id", [
728 # Test with None, int, and empty string
729 (None, None), (999, 999), ("", ""),
730 # Test with a new detector and exist parent detector
731 ("detector_id_new1", "detector_id_exist"),
732 # Test with a new detector, exist parent detector, and exist parent sub detector
733 ("detector_id_new2", "detector_id_exist/sub_detector_id_exist"),
734 # Test with a new detector, exist parent detector, exist parent sub detector,
735 # and exist parent sub sub detector
736 ("detector_id_new3", "detector_id_exist/sub_detector_id_exist/sub_sub_detector_id_exist"),
737 # Test with a new detector and not exist parent detector
738 ("detector_id_new", "detector_id_not_exist"),
739 # Test with sub detector exists and parent detector exists
740 ("sub_detector_id_exist", "detector_id_exist"),
741 # Test with int at name parameter
742 (999, "detector1"),
743 # Test with int at parent_id parameter
744 ("detector_id_new", 999),
745 # Test with none at name parameter
746 ("detector_id_new", None),
747 # Test with None at parent_id parameter
748 (None, "detector_id_exist"),
749 # Test with new detector and empty string parent_id
750 ("detector_id_new4", ""),
751 # Test with empty string detector and exist parent_id
752 ("", "detector_id_exist"),
753 # Test with / in name and exist parent_id
754 ("detector_id_new4/", "detector_id_exist"),
755 # Test with int name and None parent_id
756 (999, None),
757 # Test with name exists and None parent_id
758 ("detector_id_exist", None),
759 # Test with sub detector not exists and parent detector exists
760 ("sub_sub_detector_id_not_exist", "detector_id_exist/sub_detector_id_not_exist"),
761])
762def test_add_detector(cdb_api, name, parent_id):
763 """
764 test add_detector()
765
766 :param cdb_api: API
767 :param name: String specifying the name for the new detector. Must
768 be unique. Must not contain a forward slash (i.e. /).
769 :param parent_id: (optional) String identifying the parent detector the
770 new detector should be added to as subdetector.
771 """
772
773 has_correct_parameter = True
774 if type(name) != str:
775 has_correct_parameter = False
776 # raise TypeError when detector_id is not a str.
777 with pytest.raises(TypeError):
778 assert cdb_api.add_detector(name, parent_id)
779
780 if type(parent_id) != str and parent_id is not None:
781 has_correct_parameter = False
782 # raise TypeError when parent_id is not a str and not None.
783 with pytest.raises(TypeError):
784 assert cdb_api.add_detector(name, parent_id)
785
786 if name == "detector_id_new4/":
787 has_correct_parameter = False
788 # raise ValueError when name has /.
789 with pytest.raises(ValueError):
790 assert cdb_api.add_detector(name, parent_id)
791
792 if name == "":
793 has_correct_parameter = False
794 # raise ValueError when name is empty.
795 with pytest.raises(ValueError):
796 assert cdb_api.add_detector(name, parent_id)
797
798 if name == "detector_id_exist":
799 has_correct_parameter = False
800 # raise ValueError when name exists.
801 with pytest.raises(ValueError):
802 assert cdb_api.add_detector(name, parent_id)
803
804 if name == "sub_sub_detector_id_not_exist":
805 has_correct_parameter = False
806 # raise ValueError when sub detector not exists and parent detector exists.
807 with pytest.raises(ValueError):
808 assert cdb_api.add_detector(name, parent_id)
809
810 if has_correct_parameter:
811 if parent_id == "detector_id_not_exist":
812 # if parent_id does not exist, then raise ValueError.
813 with pytest.raises(ValueError):
814 assert cdb_api.add_detector(name, parent_id)
815 elif name == "sub_detector_id_exist":
816 # if sub detector exists, then raise ValueError.
817 with pytest.raises(ValueError):
818 assert cdb_api.add_detector(name, parent_id)
819 else:
820 # Test if detector_id is the subdetector of the parent_id
821 cdb_api.add_detector(name, parent_id)
822 tmp_det_id = ""
823 if parent_id is None:
824 tmp_det_id = '/' + name
825 else:
826 tmp_det_id = parent_id + '/' + name
827
828 new_detector_id = cdb_api.get_detector(tmp_det_id)
829 if name == "":
830 assert new_detector_id['name'] == parent_id, \
831 "detector name should be equal to parent_id when the inserted detector id is empty"
832 else:
833 assert new_detector_id['name'] == name, \
834 "detector name should be equal to the inserted detector id"
835
836
837@pytest.mark.parametrize("detector_id, name, values, test_type, tag, collected_at, valid_since, valid_until", [
838 # Test all parameters with None
839 (None, None, None, None, None, None, None, None),
840 # Test all parameters with integer
841 (999, 999, 999, 999, 999, 999, 999, 999),
842 # Test all parameters with empty string
843 ("", "", "", "", "", "", "", ""),
844 # Test with detector exists
845 ("detector_id_exist", "strawPositions1", {"dictPar1": "dictVal1"}, "calibration", "tag-1",
846 "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
847 # Test with condition name exists, but tag not exists
848 ("detector_id_exist", "condition_exist", {"dictPar1": "dictVal1"}, "calibration", "tag-55",
849 "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
850 # Test with condition tag exists, but name exists
851 ("detector_id_exist", "condition_not_exist_but_tag_exist", {"dictPar1": "dictVal1"},
852 "calibration", "tag", "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
853 # Test with condition name and tag exist
854 ("detector_id_exist", "condition_exist", {"dictPar1": "dictVal1"}, "calibration", "tag",
855 "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
856 # Test with str collected_at, valid_since, valid_until
857 ("detector_id_exist", "strawPositions2", {"dictPar1": "dictVal1"}, "calibration", "tag-2",
858 "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
859 # Test with str collected_at, valid_since, valid_until with millisecond
860 ("detector_id_exist", "strawPositions3", {"dictPar1": "dictVal1"}, "calibration", "tag-3",
861 "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
862 # Test with datetime collected_at, valid_since, valid_until
863 ("detector_id_exist", "strawPositions4", {"dictPar1": "dictVal1"}, "calibration", "tag-4",
864 datetime.datetime(2020, 3, 12, 11, 5, 27, 767563),
865 datetime.datetime(2020, 3, 12, 11, 5, 27, 767563),
866 datetime.datetime(2020, 4, 10, 11, 5, 27, 767563)),
867 # Test with valid_since is greater than valid_until
868 ("detector_id_exist", "strawPositions5", {"dictPar1": "dictVal1"}, "calibration", "tag-5",
869 datetime.datetime(2020, 3, 12, 11, 5, 27, 767563),
870 datetime.datetime(2020, 3, 12, 11, 5, 27, 767563),
871 datetime.datetime(2020, 2, 10, 11, 5, 27, 767563)),
872 # Test with detector not exists
873 ("detector_id_not_exist", "strawPositions6", {"dictPar1": "dictVal1"}, "calibration", "tag-6",
874 "2020-03-12 11:05:27", "2020-03-12 11:05:27", "2020-04-10 11:05:27"),
875 # Test with valid parameters, except all datetime parameters are integer
876 ("detector_id_exist", "strawPositions1", {"dictPar1": "dictVal1"}, "calibration", "tag-1",
877 999, 999, 999),
878 # Test with valid parameters, except type is integer
879 ("detector_id_exist", "strawPositions1", {"dictPar1": "dictVal1"}, 999, "tag-1",
880 "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
881 # Test with sub detector not exist
882 ("detector_id_exist/sub_detector_not_exist", "strawPositions1", {"dictPar1": "dictVal1"},
883 "calibration", "tag-1", "2020-03-10 11:05:27", "2020-03-10 11:05:27", "2020-04-12 11:05:27"),
884 # Test with invalid collected_at, valid_since, and valid_until
885 ("detector_id_with_invalid_date_parameters", "strawPositions1", {"dictPar1": "dictVal1"},
886 "calibration", "tag-1", 999, 999, 999),
887 # Test with None values
888 ("detector_id_not_exist_with_values_none", "strawPositions7", None, "calibration", "tag-6",
889 "2020-03-12 11:05:27", "2020-03-12 11:05:27", "2020-04-10 11:05:27"),
890 # Test with empty string values
891 ("detector_id_not_exist_with_values_empty", "strawPositions8", "", "calibration", "tag-6",
892 "2020-03-12 11:05:27", "2020-03-12 11:05:27", "2020-04-10 11:05:27"),
893])
894def test_add_condition(cdb_api, detector_id, name, values, test_type, tag,
895 collected_at, valid_since, valid_until):
896 """
897 test add_condition()
898
899 :param cdb_api: API
900 :param detector_id: String identifying the detector to which the
901 condition will be added (i.e. 'muonflux/straw_tubes').
902 :param name: String specifying the name of the condition (e.g. 'strawPositions').
903 :param values: Dictionary containing the values of the condition.
904 :param test_type: String specifying the type of condition (e.g. 'calibration').
905 :param tag: String specifying a tag for the condition. Must be unique
906 for the same condition name.
907 :param collected_at: Timestamp specifying the date/time the condition
908 was acquired. Must be unique w.r.t. the condition name.
909 :param valid_since: Timestamp specifying the date/time as of when the
910 condition is valid.
911 :param valid_until: Timestamp specifying the date/time the
912 condition up until the condition is valid.
913 """
914
915 has_correct_parameter_type = True
916 if type(detector_id) != str or detector_id == "" or type(name) != str or name == "" or \
917 (type(test_type) != str and type(test_type) is not None) or \
918 type(tag) != str or tag == "" or values == "" or values is None:
919 has_correct_parameter_type = False
920 # raise TypeError when detector_id /name /type /tag is not an str.
921 # raise TypeError when /values is empty or None.
922 with pytest.raises(TypeError):
923 assert cdb_api.add_condition(detector_id, name, tag, values, test_type,
924 collected_at, valid_since, valid_until)
925
926 if (type(collected_at) != str and type(collected_at) != datetime.datetime and collected_at is not None) or \
927 (type(valid_since) != str and type(valid_since) != datetime.datetime and valid_since is not None) or \
928 (type(valid_until) != str and type(valid_until) != datetime.datetime and valid_until is not None):
929 has_correct_parameter_type = False
930 # raise TypeError when collected_at or valid_until has a wrong type.
931 with pytest.raises(TypeError):
932 assert cdb_api.add_condition(detector_id, name, tag, values, test_type,
933 collected_at, valid_since, valid_until)
934
935 if (type(collected_at) == str or type(valid_until) == str or type(valid_since) == str) and \
936 has_correct_parameter_type:
937 # if all parameters have correct type, but collected_at, valid_until, or
938 # valid_since's format is wrong, the function must raise ValueError
939 if not has_correct_format(collected_at) or not has_correct_format(valid_until) or \
940 not has_correct_format(valid_since):
941 has_correct_parameter_type = False
942 with pytest.raises(ValueError):
943 assert cdb_api.add_condition(detector_id, name, tag, values, test_type,
944 collected_at, valid_since, valid_until)
945
946 # when valid_since is greater than valid_until, raise ValueError
947 if (type(valid_since) == datetime.datetime and type(valid_until) == datetime.datetime) and \
948 valid_since > valid_until:
949 has_correct_parameter_type = False
950 with pytest.raises(ValueError):
951 assert cdb_api.add_condition(detector_id, name, tag, values, test_type, collected_at,
952 valid_since, valid_until)
953
954 # when valid_since is greater than valid_until, raise ValueError
955 if (type(valid_since) == str and type(valid_until) == str) and has_correct_parameter_type:
956 if valid_since == "" or valid_until == "":
957 has_correct_parameter_type = False
958 with pytest.raises(ValueError):
959 assert cdb_api.add_condition(detector_id, name, tag, values, test_type,
960 collected_at, valid_since, valid_until)
961 else:
962 datetime_valid_since = __convert_str_to_datetime(valid_since)
963 datetime_valid_until = __convert_str_to_datetime(valid_until)
964 if datetime_valid_since > datetime_valid_until:
965 has_correct_parameter_type = False
966 with pytest.raises(ValueError):
967 assert cdb_api.add_condition(detector_id, name, tag, values, test_type,
968 collected_at, valid_since, valid_until)
969
970 # when detector does not exist, it must raise ValueError
971 if detector_id == "detector_id_not_exist" or \
972 detector_id == "detector_id_exist/sub_detector_not_exist":
973 has_correct_parameter_type = False
974 with pytest.raises(ValueError):
975 assert cdb_api.add_condition(detector_id, name, tag, values, test_type,
976 collected_at, valid_since, valid_until)
977
978 # when condition name and tag exist, it must raise ValueError
979 if detector_id == "detector_id_exist" and name == "condition_exist" and tag == "tag":
980 has_correct_parameter_type = False
981 with pytest.raises(ValueError):
982 assert cdb_api.add_condition(detector_id, name, tag, values, test_type,
983 collected_at, valid_since, valid_until)
984
985 if has_correct_parameter_type:
986 # check detector value.
987 cdb_api.add_condition(detector_id, name, tag, values, test_type,
988 collected_at, valid_since, valid_until)
989 detector = cdb_api.get_detector(detector_id)
990
991 # Check if the name is still the same
992 assert detector["name"] == detector_id, "the detector has different name"
993
994 # get the new inserted condition
995 conditions = detector["conditions"]
996 new_condition = None
997 for condition in conditions:
998 if condition["tag"] == tag and condition["name"] == name:
999 new_condition = condition
1000 break
1001
1002 # Check if the tag is correct
1003 assert new_condition["tag"] == tag, "tag is not correct"
1004
1005 # Check if the values is correct
1006 assert new_condition["values"] == values, "value is not correct"
1007
1008 # Check if the type is correct
1009 assert new_condition["type"] == test_type, "type is not correct"
1010
1011 # Check if the collected_at is correct
1012 if type(collected_at) == datetime.datetime:
1013 assert __convert_str_to_datetime(new_condition["collected_at"]) \
1014 == collected_at.replace(microsecond=0), \
1015 "collected_at is not correct"
1016 elif type(collected_at) == str:
1017 assert __convert_str_to_datetime(new_condition["collected_at"]) == \
1018 __convert_str_to_datetime(collected_at).replace(microsecond=0), \
1019 "collected_at is not correct"
1020
1021 # Check if the valid_until exists
1022 if type(valid_until) == datetime.datetime:
1023 assert __convert_str_to_datetime(new_condition["valid_until"]) \
1024 == valid_until.replace(microsecond=0), \
1025 "valid_until is not correct"
1026 elif type(valid_until) == str:
1027 assert __convert_str_to_datetime(new_condition["valid_until"]) == \
1028 __convert_str_to_datetime(valid_until).replace(microsecond=0), \
1029 "valid_until is not correct"
1030
1031
1032@pytest.mark.parametrize("detector_id, name, tag, test_type, valid_since, valid_until", [
1033 # Test with None input for each parameter
1034 (None, None, None, None, None, None),
1035 # Test with int input for each parameter
1036 (999, 999, 999, 999, 999, 999),
1037 # Test with empty string input for each parameter
1038 ("", "", "", "", "", ""),
1039 # Test with str format for valid_since and valid_until
1040 ("detector_id_exist", "condition_exist", "tag", "type",
1041 "2020-04-16 12:07:30", "2020-04-17 12:07:30"),
1042 # Test with str format for valid_since and valid_until, and with millisecond
1043 ("detector_id_exist", "condition_exist", "tag", "type",
1044 "2020-04-16 12:07:30:129910", "2020-04-17 12:07:30:129910"),
1045 # Test with datetime format for valid_since and valid_until
1046 ("detector_id_exist", "condition_exist", "tag", "type",
1047 datetime.datetime(2020, 4, 16, 12, 7, 30),
1048 datetime.datetime(2020, 4, 17, 12, 7, 30)),
1049 # Test with datetime format for valid_since and valid_until, and with millisecond
1050 ("detector_id_exist", "condition_exist", "tag", "type",
1051 datetime.datetime(2020, 4, 16, 12, 7, 30, 129910),
1052 datetime.datetime(2020, 4, 17, 12, 7, 30, 129910)),
1053 # Test with str valid_since and datetime valid_until
1054 ("detector_id_exist", "condition_exist", "tag", "type", "2020-04-16 12:07:30",
1055 datetime.datetime(2020, 4, 17, 12, 7, 30)),
1056 # Test with datetime valid_since and str valid_until
1057 ("detector_id_exist", "condition_exist", "tag", "type",
1058 datetime.datetime(2020, 4, 13), "2020-04-16 12:07:30"),
1059 # Test if it has invalid month or date
1060 ("detector_id_exist", "condition_exist", "tag", "type", "2020-24-16 12:07:30",
1061 datetime.datetime(2020, 4, 17, 12, 7, 30)),
1062 # Test if valid_Since > valid_until
1063 ("detector_id_exist", "condition_exist", "tag", "type",
1064 datetime.datetime(2020, 4, 20, 12, 7, 30),
1065 datetime.datetime(2020, 4, 17, 12, 7, 30)),
1066 # Test with valid detector_id, but not valid other parameters
1067 ("detector_id_exist", 999, 999, 999, 999, 999),
1068 # Test with not exist detector_id
1069 ("detector_id_not_exist", "condition_exist", "tag", "type",
1070 datetime.datetime(2020, 4, 16, 12, 7, 30),
1071 datetime.datetime(2020, 4, 17, 12, 7, 30)),
1072 # Test with exist detector_id, but not exist sub detector
1073 ("detector_id_exist/sub_detector_id_not_exist", "condition_exist", "tag", "type",
1074 datetime.datetime(2020, 4, 16, 12, 7, 30),
1075 datetime.datetime(2020, 4, 17, 12, 7, 30)),
1076 # Test with exist detector_id and condition name does not exist
1077 ("detector_id_exist", "condition_not_exist", "tag", "type",
1078 datetime.datetime(2020, 4, 16, 12, 7, 30),
1079 datetime.datetime(2020, 4, 17, 12, 7, 30)),
1080])
1081def test_update_condition_by_name_and_tag(cdb_api, detector_id, name, tag,
1082 test_type, valid_since, valid_until):
1083 """
1084 test update_condition_by_name_and_tag()
1085
1086 :param cdb_api: API
1087 :param detector_id: String identifying the detector for which the
1088 condition must be retrieved (i.e. 'muonflux/straw_tubes').
1089 :param name: String specifying the name of the condition to be retrieved (e.g.
1090 'strawPositions').
1091 :param tag: String specifying the tag of the condition to be updated.
1092 :param test_type: (optional) String specifying the type of condition (e.g. 'calibration').
1093 :param valid_since: Timestamp specifying the date/time as of when the
1094 condition is valid. Can be of type String or datetime.
1095 :param valid_until: Timestamp specifying the date/time up until the
1096 condition is valid. Can be of type String or datetime.
1097 """
1098 has_correct_parameter = True
1099 if type(detector_id) != str or type(name) != str or type(tag) != str or \
1100 (type(test_type) != str and test_type is not None):
1101 has_correct_parameter = False
1102 # raise TypeError when detector/name/tag_id is not an str.
1103 with pytest.raises(TypeError):
1104 assert cdb_api.update_condition_by_name_and_tag(detector_id, name, tag,
1105 test_type, valid_since, valid_until)
1106
1107 if (type(valid_since) != str and type(valid_since) != datetime.datetime and valid_since is not None) or \
1108 (type(valid_until) != str and type(valid_until) != datetime.datetime and valid_until is not None):
1109 has_correct_parameter = False
1110 # raise TypeError when date is not correct format.
1111 with pytest.raises(TypeError):
1112 assert cdb_api.update_condition_by_name_and_tag(detector_id, name, tag,
1113 test_type, valid_since, valid_until)
1114
1115 if type(valid_since) == str and not has_correct_format(valid_since):
1116 has_correct_parameter = False
1117 # if the format is wrong, the function has to raise ValueError
1118 with pytest.raises(ValueError):
1119 assert cdb_api.update_condition_by_name_and_tag(detector_id, name, tag,
1120 test_type, valid_since, valid_until)
1121
1122 if type(valid_until) == str and not has_correct_format(valid_until):
1123 has_correct_parameter = False
1124 # if the format is wrong, the function has to raise ValueError
1125 with pytest.raises(ValueError):
1126 assert cdb_api.update_condition_by_name_and_tag(detector_id, name, tag,
1127 test_type, valid_since, valid_until)
1128
1129 if type(valid_since) == datetime.datetime \
1130 and type(valid_until) == datetime.datetime \
1131 and valid_since > valid_until:
1132 has_correct_parameter = False
1133 # if valid_since > valid_until, raise ValueError
1134 with pytest.raises(ValueError):
1135 assert cdb_api.update_condition_by_name_and_tag(detector_id, name, tag,
1136 test_type, valid_since, valid_until)
1137
1138 if detector_id == "detector_id_not_exist" \
1139 or detector_id == "detector_id_exist/sub_detector_id_not_exist":
1140 has_correct_parameter = False
1141 # if detector_id does not exist, raise ValueError
1142 with pytest.raises(ValueError):
1143 assert cdb_api.update_condition_by_name_and_tag(detector_id, name, tag,
1144 test_type, valid_since, valid_until)
1145
1146 if name == "condition_not_exist":
1147 has_correct_parameter = False
1148 # if condition name does not exist, raise ValueError
1149 with pytest.raises(ValueError):
1150 assert cdb_api.update_condition_by_name_and_tag(detector_id, name, tag, test_type,
1151 valid_since, valid_until)
1152
1153 if has_correct_parameter:
1154 cdb_api.update_condition_by_name_and_tag(detector_id, name, tag,
1155 test_type, valid_since, valid_until)
1156
1157 # check if the result is correct.
1158 if detector_id == "detector_id_exist":
1159 result_con = cdb_api.get_condition_by_name_and_tag(detector_id, name, tag)
1160 if type(valid_since) == str:
1161 assert __convert_str_to_datetime(result_con["valid_since"]) == \
1162 __convert_str_to_datetime(valid_since).replace(microsecond=0), \
1163 "valid_since is not updated successfully"
1164 elif type(valid_since) == datetime.datetime:
1165 assert __convert_str_to_datetime(result_con["valid_since"]) == \
1166 valid_since.replace(microsecond=0), \
1167 "valid_since is not updated successfully"
1168
1169 if type(valid_until) == str:
1170 assert __convert_str_to_datetime(result_con["valid_until"]) == \
1171 __convert_str_to_datetime(valid_until).replace(microsecond=0), \
1172 "valid_until is not updated successfully"
1173 elif type(valid_until) == datetime.datetime:
1174 assert __convert_str_to_datetime(result_con["valid_until"]) == \
1175 valid_until.replace(microsecond=0), \
1176 "valid_until is not updated successfully"
1177
1178
1179@pytest.mark.parametrize("detector_id", [
1180 # Test with None, int, and empty string
1181 "", None, 1,
1182 # Test with not exist detector id
1183 "detector_id_not_exist",
1184 # Test with exist sub sub detector
1185 "detector_id_exist/sub_detector_id_exist/sub_sub_detector_id_exist",
1186 # Test with exist sub detector
1187 "detector_id_exist/sub_detector_id_exist",
1188 # Test with exist detector id
1189 "detector_id_exist"
1190])
1191def test_remove_detector(cdb_api, detector_id):
1192 """
1193 test remove_detector().
1194 This function must be executed last because the test data is used in the other test case
1195
1196 :param cdb_api: API
1197 :param detector_id: String identifying the detector to remove (i.e. 'muonflux/straw_tubes').
1198 """
1199
1200 if type(detector_id) != str:
1201 # raise TypeError when detector_id is not an str.
1202 with pytest.raises(TypeError):
1203 assert cdb_api.remove_detector(detector_id)
1204 elif detector_id == "detector_id_not_exist" or detector_id == "":
1205 # raise ValueError since detector does not exists.
1206 with pytest.raises(ValueError):
1207 assert cdb_api.remove_detector(detector_id)
1208 else:
1209 # check if detector_id still exists or not
1210 cdb_api.remove_detector(detector_id)
1211 tmp_detector = None
1212 try:
1213 tmp_detector = cdb_api.get_detector(detector_id)
1214 except ValueError:
1215 tmp_detector = None
1216
1217 assert tmp_detector is None, "detector_id still exists: " + detector_id
1218
1219
1220def has_correct_format(time_stamp):
1221 """
1222 check if the passed time_stamp has a correct format or not
1223 allowed format: "%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d %H", "%Y-%m-%d", "%Y-%m", "%Y"
1224 :param time_stamp: time_stamp with str type
1225 :return: Boolean validation result
1226 """
1227
1228 if type(time_stamp) != str:
1229 raise TypeError("time_stamp must be a string")
1230
1231 # Accepted formats for time_stamp
1232 time_stamp_str_format = ["%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d %H",
1233 "%Y-%m-%d", "%Y-%m", "%Y"]
1234
1235 correct_format = False
1236 for time_stamp_format in time_stamp_str_format:
1237 try:
1238 datetime.datetime.strptime(time_stamp, time_stamp_format)
1239 correct_format = True
1240 break
1241 except ValueError:
1242 continue
1243
1244 return correct_format
1245
1246
1247def __convert_str_to_datetime(input_date_string):
1248 """
1249 convert string into datetime
1250
1251 :param input_date_string string that contains datetime value
1252 """
1253 # Accepted formats for time_stamp
1254 time_stamp_str_format = ["%Y,%m,%d,%H,%M,%S,%f", "%Y", "%Y-%m", "%Y-%m-%d", "%Y-%m-%d %H",
1255 "%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S"]
1256 datetime_value = None
1257 for time_stamp_format in time_stamp_str_format:
1258 try:
1259 datetime_value = datetime.datetime.strptime(input_date_string, time_stamp_format)
1260 break
1261 except ValueError:
1262 pass
1263
1264 if datetime_value is None:
1265 raise ValueError("Please pass the correct date input. This date string should "
1266 "contain only digits/:/ /-. The minimum length could be 4 digits, "
1267 "representing the year. ")
1268 else:
1269 return datetime_value
This class creates an instance of the specified database API.
Definition factory.py:15
test_get_conditions_by_name_and_validity(cdb_api, detector_id, name, start_date, end_date)
test_update_condition_by_name_and_tag(cdb_api, detector_id, name, tag, test_type, valid_since, valid_until)
test_get_condition_by_name_and_collection_date(cdb_api, detector_id, name, collected_at)
test_add_condition(cdb_api, detector_id, name, values, test_type, tag, collected_at, valid_since, valid_until)