diff --git a/tests/test_rtutil.py b/tests/test_rtutil.py index 042634c270bd5cff11c698b22ebd94a018bb4967..0ca15fe9af646d972c5571275663395d9032adce 100644 --- a/tests/test_rtutil.py +++ b/tests/test_rtutil.py @@ -14,6 +14,8 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +import contextlib + import pytest from . import testutil @@ -46,6 +48,14 @@ def new_client(): TICKET_DATA = testutil.RTClient.TICKET_DATA.copy() return RTClient() +def new_cache(database=':memory:'): + db = rtutil.RTLinkCache.setup(database) + if db is None: + print("NOTE: did not set up database cache at {}".format(database)) + return contextlib.nullcontext(db) + else: + return contextlib.closing(db) + @pytest.mark.parametrize('ticket_id,attachment_id,expected', EXPECTED_URLS) def test_url(rt, ticket_id, attachment_id, expected): if expected is not None: @@ -125,3 +135,47 @@ def test_uncommon_server_url_parsing(): client = testutil.RTClient(url + 'REST/1.0/') rt = rtutil.RT(client) assert rt.url(1).startswith(url) + +def test_shared_cache(new_client): + ticket_id, _, expected = EXPECTED_URLS[0] + expected = DEFAULT_RT_URL + expected + with new_cache() as cachedb: + rt1 = rtutil.RT(new_client, cachedb) + assert rt1.url(ticket_id) == expected + new_client.TICKET_DATA.clear() + rt2 = rtutil.RT(new_client, cachedb) + assert rt2.url(ticket_id) == expected + assert not rt2.exists(ticket_id + 1) + assert rt1 is not rt2 + +def test_no_shared_cache(new_client): + with new_cache() as cache1, new_cache() as cache2: + rt1 = rtutil.RT(new_client, cache1) + rt2 = rtutil.RT(new_client, cache2) + assert rt1.exists(1) + new_client.TICKET_DATA.clear() + assert not rt2.exists(1) + assert rt1.exists(1) + +def test_read_only_cache(new_client, tmp_path): + db_path = tmp_path / 'test.db' + ticket_id, _, expected = EXPECTED_URLS[0] + expected = DEFAULT_RT_URL + expected + with new_cache(db_path) as cache1: + rt1 = rtutil.RT(new_client, cache1) + assert rt1.url(ticket_id) == expected + new_client.TICKET_DATA.clear() + db_path.chmod(0o400) + with new_cache(db_path) as cache2: + rt2 = rtutil.RT(new_client, cache2) + assert rt2.url(ticket_id) == expected + assert rt2.url(ticket_id + 1) is None + +def test_results_not_found_only_in_transient_cache(new_client): + with new_cache() as cache: + rt1 = rtutil.RT(new_client, cache) + rt2 = rtutil.RT(new_client, cache) + assert not rt1.exists(9) + new_client.TICKET_DATA['9'] = [('99', '(Unnamed)', 'text/plain', '0b')] + assert not rt1.exists(9) + assert rt2.exists(9)