From 58f65b2a8f491965f8705ef6817e9e91e69b21e6 Mon Sep 17 00:00:00 2001 From: wangyinz Date: Wed, 12 Aug 2026 12:33:52 -0500 Subject: [PATCH 1/2] Make CSS arrival interval updates safe --- .../mspasspy/preprocessing/css30/dbarrival.py | 82 +++-- .../test_dbarrival_interval_contract.py | 319 ++++++++++++++++++ 2 files changed, 369 insertions(+), 32 deletions(-) create mode 100644 python/tests/preprocessing/test_dbarrival_interval_contract.py diff --git a/python/mspasspy/preprocessing/css30/dbarrival.py b/python/mspasspy/preprocessing/css30/dbarrival.py index cb6c07000a..790d74c549 100644 --- a/python/mspasspy/preprocessing/css30/dbarrival.py +++ b/python/mspasspy/preprocessing/css30/dbarrival.py @@ -654,11 +654,14 @@ def set_netcode_time_interval( arrival collection is large. :return: number of documents updated. + :raises ValueError: if sta or net is not a nonempty string. """ basemessage = "set_netcode_time_interval: " - if sta == None or net == None: - print(basemessage + "you must specify sta and net as required parameters") + if not isinstance(sta, str) or not sta: + raise ValueError(basemessage + "sta must be a nonempty string") + if not isinstance(net, str) or not net: + raise ValueError(basemessage + "net must be a nonempty string") dbarr = db[collection] query = {"sta": sta} if starttime == None or endtime == None: @@ -686,6 +689,7 @@ def set_netcode_time_interval( tse = starttime.timestamp tee = endtime.timestamp query[time_filter_key] = {"$gte": tse, "$lte": tee} + count = 0 n = dbarr.count_documents(query) if n == 0: print( @@ -694,26 +698,26 @@ def set_netcode_time_interval( + collection ) print(query) - else: - count = 0 if use_immortal_cursor: curs = dbarr.find(query, no_cursor_timeout=True) else: curs = dbarr.find(query) - for doc in curs: - if "net" in doc: - print( - basemessage + "WARNING found document with net code set to ", - doc["net"], - ) - # this check is required for robustness when time filter is off - if time_filter_key in doc: - print("Problem document time=", UTCDateTime(doc[time_filter_key])) - print("Setting net in this document to requested net code=", net) - oid = doc["_id"] - updaterec = {"net": net} - dbarr.update_one({"_id": oid}, {"$set": updaterec}) - count += 1 + for doc in curs: + if "net" in doc: + print( + basemessage + "WARNING found document with net code set to ", + doc["net"], + ) + # this check is required for robustness when time filter is off + if time_filter_key in doc: + print("Problem document time=", UTCDateTime(doc[time_filter_key])) + print("Setting net in this document to requested net code=", net) + oid = doc["_id"] + updaterec = {"net": net} + update_query = dict(query) + update_query["_id"] = oid + dbarr.update_one(update_query, {"$set": updaterec}) + count += 1 return count @@ -933,11 +937,11 @@ def set_arrival_by_time_interval( arrival collection is large. :param verbose: if true prints a bunch a few messages. Silent (default) otherwise :return: count of number of documents updated. + :raises ValueError: if sta or a matching site record's net is not a + nonempty string. """ - if sta == None: - raise MsPASSError( - "Missing required parameter sta=station code to repair", "Fatal" - ) + if not isinstance(sta, str) or not sta: + raise ValueError("set_arrival_by_time_interval: sta must be a nonempty string") dbarr = db.arrival dbsite = db.site query = {"sta": sta} @@ -945,9 +949,16 @@ def set_arrival_by_time_interval( curs = dbsite.find(query, no_cursor_timeout=True).sort("starttime", 1) else: curs = dbsite.find(query).sort("starttime", 1) + site_docs = list(curs) + for doc in site_docs: + net = doc.get("net") + if not isinstance(net, str) or not net: + raise ValueError( + "set_arrival_by_time_interval: site net must be a nonempty string" + ) # First make sure we don't have any overlapping time periods n = 0 - for doc in curs: + for doc in site_docs: if n == 0: lastend = doc["endtime"] lastnet = doc["net"] @@ -968,12 +979,13 @@ def set_arrival_by_time_interval( lastend = doc["endtime"] lastnet = doc["net"] n += 1 - curs.rewind() - for doc in curs: + nupdates = 0 + for doc in site_docs: net = doc["net"] - arquerry = dict() - arquerry["sta"] = sta - arquerry["time"] = {"$gte": doc["starttime"], "$lte": doc["endtime"]} + arrival_query = { + "sta": sta, + "time": {"$gte": doc["starttime"], "$lte": doc["endtime"]}, + } if verbose: print( "Setting net=", @@ -982,13 +994,19 @@ def set_arrival_by_time_interval( UTCDateTime(doc["starttime"]), UTCDateTime(doc["endtime"]), ) - nset = dbarr.count_documents(arquerry) + nset = dbarr.count_documents(arrival_query) print("Setting net code to ", net, " in ", nset, " documents") - arcursor = dbarr.find(query) + if use_immortal_cursor: + arcursor = dbarr.find(arrival_query, no_cursor_timeout=True) + else: + arcursor = dbarr.find(arrival_query) for ardoc in arcursor: # print(ardoc['sta'],UTCDateTime(ardoc['time'])) - id = ardoc["_id"] - dbarr.update_one({"_id": id}, {"$set": {"net": net}}) + update_query = dict(arrival_query) + update_query["_id"] = ardoc["_id"] + dbarr.update_one(update_query, {"$set": {"net": net}}) + nupdates += 1 + return nupdates def force_net(db, sta=None, net=None): diff --git a/python/tests/preprocessing/test_dbarrival_interval_contract.py b/python/tests/preprocessing/test_dbarrival_interval_contract.py new file mode 100644 index 0000000000..2b9910d652 --- /dev/null +++ b/python/tests/preprocessing/test_dbarrival_interval_contract.py @@ -0,0 +1,319 @@ +from copy import deepcopy + +import pytest +from obspy import UTCDateTime + +from mspasspy.ccore.utility import MsPASSError +from mspasspy.preprocessing.css30.dbarrival import ( + set_arrival_by_time_interval, + set_netcode_time_interval, +) + + +def _matches(document, query): + for key, expected in query.items(): + if key not in document: + return False + actual = document[key] + if isinstance(expected, dict): + if "$gte" in expected and actual < expected["$gte"]: + return False + if "$lte" in expected and actual > expected["$lte"]: + return False + elif actual != expected: + return False + return True + + +class _Cursor(list): + def sort(self, key, direction): + assert direction == 1 + list.sort(self, key=lambda document: document[key]) + return self + + def rewind(self): + return self + + +class _Collection: + def __init__(self, documents): + self.documents = deepcopy(documents) + self.find_calls = [] + self.count_calls = [] + self.update_calls = [] + + def find(self, query, **kwargs): + self.find_calls.append((deepcopy(query), deepcopy(kwargs))) + return _Cursor( + deepcopy(document) + for document in self.documents + if _matches(document, query) + ) + + def count_documents(self, query): + self.count_calls.append(deepcopy(query)) + return sum(_matches(document, query) for document in self.documents) + + def update_one(self, query, update): + self.update_calls.append((deepcopy(query), deepcopy(update))) + for document in self.documents: + if _matches(document, query): + document.update(update["$set"]) + return + raise AssertionError(f"update query matched no document: {query}") + + +class _Database: + def __init__(self, arrivals, sites=()): + self.arrival = _Collection(arrivals) + self.site = _Collection(sites) + + def __getitem__(self, collection): + assert collection == "arrival" + return self.arrival + + +class _NoDatabaseAccess: + def __getitem__(self, collection): + raise AssertionError(f"unexpected collection access: {collection}") + + @property + def arrival(self): + raise AssertionError("unexpected arrival access") + + @property + def site(self): + raise AssertionError("unexpected site access") + + +@pytest.mark.parametrize("use_immortal_cursor", [False, True]) +def test_site_intervals_update_only_matching_arrivals(use_immortal_cursor): + sites = [ + {"_id": "site-a", "sta": "AAA", "net": "A", "starttime": 0.0, "endtime": 10.0}, + {"_id": "site-b", "sta": "AAA", "net": "B", "starttime": 20.0, "endtime": 30.0}, + ] + arrivals = [ + {"_id": "a0", "sta": "AAA", "time": 0.0}, + {"_id": "a10", "sta": "AAA", "time": 10.0}, + {"_id": "gap", "sta": "AAA", "time": 15.0}, + {"_id": "b20", "sta": "AAA", "time": 20.0}, + {"_id": "b30", "sta": "AAA", "time": 30.0}, + {"_id": "other", "sta": "BBB", "time": 5.0}, + ] + database = _Database(arrivals, sites) + + count = set_arrival_by_time_interval( + database, + sta="AAA", + allowed_overlap=0.0, + use_immortal_cursor=use_immortal_cursor, + ) + + assert type(count) is int + assert count == 4 + expected_intervals = [ + {"sta": "AAA", "time": {"$gte": 0.0, "$lte": 10.0}}, + {"sta": "AAA", "time": {"$gte": 20.0, "$lte": 30.0}}, + ] + cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} + assert database.site.find_calls == [({"sta": "AAA"}, cursor_options)] + assert database.arrival.find_calls == [ + (expected_intervals[0], cursor_options), + (expected_intervals[1], cursor_options), + ] + assert database.arrival.update_calls == [ + ({**expected_intervals[0], "_id": "a0"}, {"$set": {"net": "A"}}), + ({**expected_intervals[0], "_id": "a10"}, {"$set": {"net": "A"}}), + ({**expected_intervals[1], "_id": "b20"}, {"$set": {"net": "B"}}), + ({**expected_intervals[1], "_id": "b30"}, {"$set": {"net": "B"}}), + ] + assert { + document["_id"]: document.get("net") for document in database.arrival.documents + } == { + "a0": "A", + "a10": "A", + "gap": None, + "b20": "B", + "b30": "B", + "other": None, + } + + +@pytest.mark.parametrize("use_immortal_cursor", [False, True]) +def test_forced_interval_updates_share_the_read_predicate(use_immortal_cursor): + arrivals = [ + {"_id": "left", "sta": "AAA", "time": 0.0}, + {"_id": "right", "sta": "AAA", "time": 10.0}, + {"_id": "outside", "sta": "AAA", "time": 11.0}, + {"_id": "other", "sta": "BBB", "time": 5.0}, + ] + database = _Database(arrivals) + + count = set_netcode_time_interval( + database, + sta="AAA", + net="XX", + starttime=UTCDateTime(0.0), + endtime=UTCDateTime(10.0), + use_immortal_cursor=use_immortal_cursor, + ) + + assert type(count) is int + assert count == 2 + interval = {"sta": "AAA", "time": {"$gte": 0.0, "$lte": 10.0}} + cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} + assert database.arrival.count_calls == [interval] + assert database.arrival.find_calls == [(interval, cursor_options)] + assert database.arrival.update_calls == [ + ({**interval, "_id": "left"}, {"$set": {"net": "XX"}}), + ({**interval, "_id": "right"}, {"$set": {"net": "XX"}}), + ] + assert { + document["_id"]: document.get("net") for document in database.arrival.documents + } == { + "left": "XX", + "right": "XX", + "outside": None, + "other": None, + } + + +@pytest.mark.parametrize("use_immortal_cursor", [False, True]) +def test_forced_interval_empty_result_returns_zero(use_immortal_cursor): + database = _Database([]) + + count = set_netcode_time_interval( + database, + sta="AAA", + net="XX", + use_immortal_cursor=use_immortal_cursor, + ) + + assert type(count) is int + assert count == 0 + query = {"sta": "AAA"} + cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} + assert database.arrival.count_calls == [query] + assert database.arrival.find_calls == [(query, cursor_options)] + assert database.arrival.update_calls == [] + + +@pytest.mark.parametrize("use_immortal_cursor", [False, True]) +def test_site_interval_empty_result_returns_zero(use_immortal_cursor): + sites = [ + {"_id": "site", "sta": "AAA", "net": "A", "starttime": 0.0, "endtime": 10.0} + ] + database = _Database([], sites) + + count = set_arrival_by_time_interval( + database, + sta="AAA", + allowed_overlap=0.0, + use_immortal_cursor=use_immortal_cursor, + ) + + assert type(count) is int + assert count == 0 + interval = {"sta": "AAA", "time": {"$gte": 0.0, "$lte": 10.0}} + cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} + assert database.site.find_calls == [({"sta": "AAA"}, cursor_options)] + assert database.arrival.find_calls == [(interval, cursor_options)] + assert database.arrival.update_calls == [] + + +@pytest.mark.parametrize( + "function, kwargs", + [ + (set_arrival_by_time_interval, {}), + (set_arrival_by_time_interval, {"sta": ""}), + (set_arrival_by_time_interval, {"sta": 3}), + (set_netcode_time_interval, {"sta": None, "net": "XX"}), + (set_netcode_time_interval, {"sta": "", "net": "XX"}), + (set_netcode_time_interval, {"sta": 3, "net": "XX"}), + (set_netcode_time_interval, {"sta": "AAA", "net": None}), + (set_netcode_time_interval, {"sta": "AAA", "net": ""}), + (set_netcode_time_interval, {"sta": "AAA", "net": 3}), + ], +) +def test_required_station_and_network_fail_before_database_access(function, kwargs): + with pytest.raises(ValueError): + function(_NoDatabaseAccess(), **kwargs) + + +@pytest.mark.parametrize("bad_net", [None, "", 7]) +def test_site_interval_rejects_invalid_network_before_arrival_access(bad_net): + database = _Database( + [{"_id": "arrival", "sta": "AAA", "time": 5.0}], + [ + { + "_id": "site", + "sta": "AAA", + "net": bad_net, + "starttime": 0.0, + "endtime": 10.0, + } + ], + ) + arrivals_before = deepcopy(database.arrival.documents) + + with pytest.raises(ValueError, match="site net must be a nonempty string"): + set_arrival_by_time_interval(database, sta="AAA") + + assert database.site.find_calls == [({"sta": "AAA"}, {})] + assert database.arrival.find_calls == [] + assert database.arrival.update_calls == [] + assert database.arrival.documents == arrivals_before + + +@pytest.mark.parametrize("use_immortal_cursor", [False, True]) +def test_overlapping_site_intervals_fail_before_arrival_access(use_immortal_cursor): + sites = [ + {"_id": "site-a", "sta": "AAA", "net": "A", "starttime": 0.0, "endtime": 10.0}, + {"_id": "site-b", "sta": "AAA", "net": "B", "starttime": 5.0, "endtime": 15.0}, + ] + arrivals = [{"_id": "arrival", "sta": "AAA", "time": 7.0}] + database = _Database(arrivals, sites) + arrivals_before = deepcopy(database.arrival.documents) + + with pytest.raises(MsPASSError, match="Overlapping time intervals"): + set_arrival_by_time_interval( + database, + sta="AAA", + allowed_overlap=0.0, + use_immortal_cursor=use_immortal_cursor, + ) + + cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} + assert database.site.find_calls == [({"sta": "AAA"}, cursor_options)] + assert database.arrival.find_calls == [] + assert database.arrival.count_calls == [] + assert database.arrival.update_calls == [] + assert database.arrival.documents == arrivals_before + + +@pytest.mark.parametrize( + "starttime,endtime", + [ + (UTCDateTime(0.0), None), + (None, UTCDateTime(10.0)), + (0.0, UTCDateTime(10.0)), + (UTCDateTime(0.0), 10.0), + ], +) +def test_invalid_forced_interval_fails_before_query_or_write(starttime, endtime): + database = _Database([{"_id": "arrival", "sta": "AAA", "time": 5.0}]) + arrivals_before = deepcopy(database.arrival.documents) + + with pytest.raises(MsPASSError): + set_netcode_time_interval( + database, + sta="AAA", + net="XX", + starttime=starttime, + endtime=endtime, + ) + + assert database.arrival.count_calls == [] + assert database.arrival.find_calls == [] + assert database.arrival.update_calls == [] + assert database.arrival.documents == arrivals_before From 41e99188989d7c9bafd5abf7cc3762d24c7baca0 Mon Sep 17 00:00:00 2001 From: wangyinz Date: Sun, 23 Aug 2026 09:24:44 -0500 Subject: [PATCH 2/2] Close CSS arrival update cursors --- .../mspasspy/preprocessing/css30/dbarrival.py | 59 +++++++++++-------- .../test_dbarrival_interval_contract.py | 58 +++++++++++++++++- 2 files changed, 91 insertions(+), 26 deletions(-) diff --git a/python/mspasspy/preprocessing/css30/dbarrival.py b/python/mspasspy/preprocessing/css30/dbarrival.py index 790d74c549..a03074df8d 100644 --- a/python/mspasspy/preprocessing/css30/dbarrival.py +++ b/python/mspasspy/preprocessing/css30/dbarrival.py @@ -702,22 +702,25 @@ def set_netcode_time_interval( curs = dbarr.find(query, no_cursor_timeout=True) else: curs = dbarr.find(query) - for doc in curs: - if "net" in doc: - print( - basemessage + "WARNING found document with net code set to ", - doc["net"], - ) - # this check is required for robustness when time filter is off - if time_filter_key in doc: - print("Problem document time=", UTCDateTime(doc[time_filter_key])) - print("Setting net in this document to requested net code=", net) - oid = doc["_id"] - updaterec = {"net": net} - update_query = dict(query) - update_query["_id"] = oid - dbarr.update_one(update_query, {"$set": updaterec}) - count += 1 + try: + for doc in curs: + if "net" in doc: + print( + basemessage + "WARNING found document with net code set to ", + doc["net"], + ) + # this check is required for robustness when time filter is off + if time_filter_key in doc: + print("Problem document time=", UTCDateTime(doc[time_filter_key])) + print("Setting net in this document to requested net code=", net) + oid = doc["_id"] + updaterec = {"net": net} + update_query = dict(query) + update_query["_id"] = oid + dbarr.update_one(update_query, {"$set": updaterec}) + count += 1 + finally: + curs.close() return count @@ -946,10 +949,13 @@ def set_arrival_by_time_interval( dbsite = db.site query = {"sta": sta} if use_immortal_cursor: - curs = dbsite.find(query, no_cursor_timeout=True).sort("starttime", 1) + curs = dbsite.find(query, no_cursor_timeout=True) else: - curs = dbsite.find(query).sort("starttime", 1) - site_docs = list(curs) + curs = dbsite.find(query) + try: + site_docs = list(curs.sort("starttime", 1)) + finally: + curs.close() for doc in site_docs: net = doc.get("net") if not isinstance(net, str) or not net: @@ -1000,12 +1006,15 @@ def set_arrival_by_time_interval( arcursor = dbarr.find(arrival_query, no_cursor_timeout=True) else: arcursor = dbarr.find(arrival_query) - for ardoc in arcursor: - # print(ardoc['sta'],UTCDateTime(ardoc['time'])) - update_query = dict(arrival_query) - update_query["_id"] = ardoc["_id"] - dbarr.update_one(update_query, {"$set": {"net": net}}) - nupdates += 1 + try: + for ardoc in arcursor: + # print(ardoc['sta'],UTCDateTime(ardoc['time'])) + update_query = dict(arrival_query) + update_query["_id"] = ardoc["_id"] + dbarr.update_one(update_query, {"$set": {"net": net}}) + nupdates += 1 + finally: + arcursor.close() return nupdates diff --git a/python/tests/preprocessing/test_dbarrival_interval_contract.py b/python/tests/preprocessing/test_dbarrival_interval_contract.py index 2b9910d652..8469950605 100644 --- a/python/tests/preprocessing/test_dbarrival_interval_contract.py +++ b/python/tests/preprocessing/test_dbarrival_interval_contract.py @@ -26,6 +26,10 @@ def _matches(document, query): class _Cursor(list): + def __init__(self, documents): + super().__init__(documents) + self.close_calls = 0 + def sort(self, key, direction): assert direction == 1 list.sort(self, key=lambda document: document[key]) @@ -34,21 +38,27 @@ def sort(self, key, direction): def rewind(self): return self + def close(self): + self.close_calls += 1 + class _Collection: def __init__(self, documents): self.documents = deepcopy(documents) self.find_calls = [] + self.cursors = [] self.count_calls = [] self.update_calls = [] def find(self, query, **kwargs): self.find_calls.append((deepcopy(query), deepcopy(kwargs))) - return _Cursor( + cursor = _Cursor( deepcopy(document) for document in self.documents if _matches(document, query) ) + self.cursors.append(cursor) + return cursor def count_documents(self, query): self.count_calls.append(deepcopy(query)) @@ -121,6 +131,8 @@ def test_site_intervals_update_only_matching_arrivals(use_immortal_cursor): (expected_intervals[0], cursor_options), (expected_intervals[1], cursor_options), ] + assert [cursor.close_calls for cursor in database.site.cursors] == [1] + assert [cursor.close_calls for cursor in database.arrival.cursors] == [1, 1] assert database.arrival.update_calls == [ ({**expected_intervals[0], "_id": "a0"}, {"$set": {"net": "A"}}), ({**expected_intervals[0], "_id": "a10"}, {"$set": {"net": "A"}}), @@ -164,6 +176,7 @@ def test_forced_interval_updates_share_the_read_predicate(use_immortal_cursor): cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} assert database.arrival.count_calls == [interval] assert database.arrival.find_calls == [(interval, cursor_options)] + assert [cursor.close_calls for cursor in database.arrival.cursors] == [1] assert database.arrival.update_calls == [ ({**interval, "_id": "left"}, {"$set": {"net": "XX"}}), ({**interval, "_id": "right"}, {"$set": {"net": "XX"}}), @@ -195,6 +208,7 @@ def test_forced_interval_empty_result_returns_zero(use_immortal_cursor): cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} assert database.arrival.count_calls == [query] assert database.arrival.find_calls == [(query, cursor_options)] + assert [cursor.close_calls for cursor in database.arrival.cursors] == [1] assert database.arrival.update_calls == [] @@ -218,6 +232,8 @@ def test_site_interval_empty_result_returns_zero(use_immortal_cursor): cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} assert database.site.find_calls == [({"sta": "AAA"}, cursor_options)] assert database.arrival.find_calls == [(interval, cursor_options)] + assert [cursor.close_calls for cursor in database.site.cursors] == [1] + assert [cursor.close_calls for cursor in database.arrival.cursors] == [1] assert database.arrival.update_calls == [] @@ -260,6 +276,7 @@ def test_site_interval_rejects_invalid_network_before_arrival_access(bad_net): set_arrival_by_time_interval(database, sta="AAA") assert database.site.find_calls == [({"sta": "AAA"}, {})] + assert [cursor.close_calls for cursor in database.site.cursors] == [1] assert database.arrival.find_calls == [] assert database.arrival.update_calls == [] assert database.arrival.documents == arrivals_before @@ -285,12 +302,51 @@ def test_overlapping_site_intervals_fail_before_arrival_access(use_immortal_curs cursor_options = {"no_cursor_timeout": True} if use_immortal_cursor else {} assert database.site.find_calls == [({"sta": "AAA"}, cursor_options)] + assert [cursor.close_calls for cursor in database.site.cursors] == [1] assert database.arrival.find_calls == [] assert database.arrival.count_calls == [] assert database.arrival.update_calls == [] assert database.arrival.documents == arrivals_before +@pytest.mark.parametrize("use_immortal_cursor", [False, True]) +@pytest.mark.parametrize( + "function, sites", + [ + (set_netcode_time_interval, ()), + ( + set_arrival_by_time_interval, + ( + { + "_id": "site", + "sta": "AAA", + "net": "XX", + "starttime": 0.0, + "endtime": 10.0, + }, + ), + ), + ], +) +def test_arrival_cursor_closes_when_update_fails(function, sites, use_immortal_cursor): + database = _Database([{"_id": "arrival", "sta": "AAA", "time": 5.0}], sites) + + def fail_update(query, update): + raise RuntimeError("update failed") + + database.arrival.update_one = fail_update + kwargs = {"sta": "AAA", "use_immortal_cursor": use_immortal_cursor} + if function is set_netcode_time_interval: + kwargs["net"] = "XX" + + with pytest.raises(RuntimeError, match="update failed"): + function(database, **kwargs) + + if function is set_arrival_by_time_interval: + assert [cursor.close_calls for cursor in database.site.cursors] == [1] + assert [cursor.close_calls for cursor in database.arrival.cursors] == [1] + + @pytest.mark.parametrize( "starttime,endtime", [