Tests: expand temporary directory helpers for more directories (#29735)

* create helpers for testing

* document it

* fix other tests
old-commit-hash: 405d09714c
beeps
Justin Newberry 2 years ago committed by GitHub
parent aca5580900
commit 1cc555c37b
  1. 42
      selfdrive/test/helpers.py
  2. 4
      tools/lib/tests/test_caching.py

@ -1,6 +1,7 @@
import os import os
import time import time
import tempfile import tempfile
from typing import List, Union
from unittest import mock from unittest import mock
from functools import wraps from functools import wraps
@ -72,17 +73,44 @@ def with_processes(processes, init_time=0, ignore_stopped=None):
return wrapper return wrapper
def temporary_mock_dir(mock_path): def temporary_mock_dir(mock_paths_in: Union[List[str], str], kwarg: Union[str, None] = None, generator = tempfile.TemporaryDirectory):
"""
mock_paths_in: string or string list representing the full path of the variable you want to mock.
kwarg: str or None representing the kwarg that gets passed into the test function, in case the test needs access to the temporary directory.
generator: a context to use to generate the temporary directory
"""
def wrapper(func): def wrapper(func):
@wraps(func) @wraps(func)
def wrap(*args, **kwargs): def wrap(*args, **kwargs):
with tempfile.TemporaryDirectory() as temp_dir: mock_paths = mock_paths_in if isinstance(mock_paths_in, list) else [mock_paths_in]
cache_dir_patch = mock.patch(mock_path, temp_dir) with generator() as temp_dir:
cache_dir_patch.start() mocks = []
func(*args, **kwargs, temp_dir=temp_dir) for mock_path in mock_paths:
cache_dir_patch.stop() mocks.append(mock.patch(mock_path, str(temp_dir)))
[mock.start() for mock in mocks]
try:
if kwarg is not None:
kwargs[kwarg] = temp_dir
func(*args, **kwargs)
finally:
[mock.stop() for mock in mocks]
return wrap return wrap
return wrapper return wrapper
def string_context(context):
class StringContext:
def __enter__(self):
return context
def __exit__(self, *args):
pass
return StringContext
temporary_dir = temporary_mock_dir([], "temp_dir")
temporary_cache_dir = temporary_mock_dir("openpilot.tools.lib.url_file.CACHE_DIR") temporary_cache_dir = temporary_mock_dir("openpilot.tools.lib.url_file.CACHE_DIR")
temporary_swaglog_dir = temporary_mock_dir("openpilot.system.swaglog.SWAGLOG_DIR") temporary_swaglog_dir = temporary_mock_dir("openpilot.system.swaglog.SWAGLOG_DIR", "temp_dir")
temporary_laikad_downloads_dir = temporary_mock_dir("openpilot.selfdrive.locationd.laikad.DOWNLOADS_CACHE_FOLDER")
temporary_swaglog_ipc = temporary_mock_dir(["openpilot.system.swaglog.SWAGLOG_IPC", "system.logmessaged.SWAGLOG_IPC"],
generator=string_context("/tmp/test_swaglog_ipc"))

@ -32,7 +32,7 @@ class TestFileDownload(unittest.TestCase):
self.assertEqual(response_cached, response_downloaded) self.assertEqual(response_cached, response_downloaded)
@temporary_cache_dir @temporary_cache_dir
def test_small_file(self, temp_dir): def test_small_file(self):
# Make sure we don't force cache # Make sure we don't force cache
os.environ["FILEREADER_CACHE"] = "0" os.environ["FILEREADER_CACHE"] = "0"
small_file_url = "https://raw.githubusercontent.com/commaai/openpilot/master/docs/SAFETY.md" small_file_url = "https://raw.githubusercontent.com/commaai/openpilot/master/docs/SAFETY.md"
@ -53,7 +53,7 @@ class TestFileDownload(unittest.TestCase):
self.compare_loads(small_file_url, 100 * i, 100) self.compare_loads(small_file_url, 100 * i, 100)
@temporary_cache_dir @temporary_cache_dir
def test_large_file(self, temp_dir): def test_large_file(self):
large_file_url = "https://commadataci.blob.core.windows.net/openpilotci/0375fdf7b1ce594d/2019-06-13--08-32-25/3/qlog.bz2" large_file_url = "https://commadataci.blob.core.windows.net/openpilotci/0375fdf7b1ce594d/2019-06-13--08-32-25/3/qlog.bz2"
# Load the end 100 bytes of both files # Load the end 100 bytes of both files
file_large = URLFile(large_file_url) file_large = URLFile(large_file_url)

Loading…
Cancel
Save