Skip to content

Make HTTP cache tests deterministic #181

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mariuszlipinski wants to merge 1 commit into from
Closed

Make HTTP cache tests deterministic #181

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 44 additions & 45 deletions tests/test_cache.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,66 +1,65 @@
"""
This is just a very simple unit test, that checks if the HttpCache is able to download a file, save it in the
specified cache directory and serves it from there.
"""
"""Deterministic unit tests for HTTP caching and request pacing."""

import logging
import os
import sys
import tempfile
import time
import unittest
from unittest.mock import Mock, call, patch

from xbrl.cache import HttpCache
from xbrl.helper.connection_manager import ConnectionManager


class CacheHelperTest(unittest.TestCase):
def test_cache_file(self):
"""
Unit test for CacheHelper.cache_file
:return:
"""
logging.basicConfig(stream=sys.stdout, level=logging.INFO)
def test_cache_file_downloads_then_serves_cached_content(self):
with tempfile.TemporaryDirectory(prefix="py-xbrl-cache-test-") as tmp_cache_dir:
cache_dir: str = os.path.join(tmp_cache_dir, "")
delay: int = 5000
cache: HttpCache = HttpCache(cache_dir, delay)
cache = HttpCache(cache_dir, delay=0)
test_url = "https://example.test/xml/note.xml"
expected_path = os.path.join(cache_dir, "example.test", "xml", "note.xml")
response_content = b"<note><body>cached response</body></note>"
response = Mock(status_code=200, content=response_content)

test_url: str = "https://www.w3schools.com/xml/note.xml"
expected_path: str = os.path.join(cache_dir, "www.w3schools.com", "xml", "note.xml")
with patch.object(cache.connection_manager, "download", return_value=response) as download:
self.assertEqual(cache.cache_file(test_url), expected_path)
self.assertEqual(download.call_count, 1)
self.assertEqual(cache.cache_file(test_url), expected_path)
self.assertEqual(download.call_count, 1)

# if the testing file already exists delete if first
if os.path.isfile(expected_path):
os.remove(expected_path)
self.assertTrue(cache.purge_file(test_url))
self.assertEqual(cache.cache_file(test_url), expected_path)
self.assertEqual(download.call_count, 2)

# on the first execution the file will be downloaded from the internet, no delay for first download
time_stamp: float = time.time()
self.assertEqual(cache.cache_file(test_url), expected_path)
time_delta = time.time() - time_stamp
self.assertLess(time_delta, delay / 1000)
logging.info(f"Time delta for first download: {time_delta}ms")

# delete the file and download it again to check if the delay for the second download is working
self.assertEqual(download.call_args_list, [call(test_url, headers={}), call(test_url, headers={})])
self.assertTrue(os.path.isfile(expected_path))
with open(expected_path, "rb") as cached_file:
self.assertEqual(cached_file.read(), response_content)
self.assertTrue(cache.purge_file(test_url))
self.assertFalse(os.path.isfile(expected_path))

time_stamp: float = time.time()
self.assertEqual(cache.cache_file(test_url), expected_path)
time_delta = time.time() - time_stamp
self.assertGreaterEqual(time_delta, delay / 1000)
logging.info(f"Time delta for second download: {time_delta}ms")

# now that the file is cached on the hard drive, the file path should be returned immediately
time_stamp = time.time()
self.assertEqual(cache.cache_file(test_url), expected_path)
time_delta = time.time() - time_stamp
self.assertLess(time_delta, delay / 1000)
logging.info(f"Time delta for third download: {time_delta}ms")
class ConnectionManagerTest(unittest.TestCase):
def test_rate_limit_delay_is_applied_before_follow_up_download(self):
manager = ConnectionManager(delay=5000, logs=False)
manager.next_try_systime_ms = 1000
test_url = "https://example.test/xml/note.xml"
response = Mock(status_code=200)

# test if the file was downloaded
self.assertTrue(os.path.isfile(expected_path))
# delete the file
self.assertTrue(cache.purge_file(test_url))
# test if the file was deleted
self.assertFalse(os.path.isfile(expected_path))
with (
patch.object(manager, "_get_systime_ms", side_effect=[1000, 1000, 1000, 1000]),
patch.object(manager._session, "get", return_value=response) as request,
patch("xbrl.helper.connection_manager.time.sleep") as sleep,
):
self.assertIs(manager.download(test_url, headers={}), response)
self.assertIs(manager.download(test_url, headers={}), response)

self.assertEqual(sleep.call_args_list, [call(0.0), call(5.0)])
self.assertEqual(
request.call_args_list,
[
call(test_url, headers={}, allow_redirects=True, verify=True),
call(test_url, headers={}, allow_redirects=True, verify=True),
],
)


if __name__ == "__main__":
Expand Down
Loading