"""Utilities functions for setting up test fixtures."""
import tempfile
from contextlib import contextmanager, ExitStack
from pathlib import Path
from unittest import mock
import shutil
import json
import urllib.parse
from uuid import uuid4
import pandas as pd
from iblutil.io.parquet import str2np
from iblutil.io.params import set_hidden
import one.params
from one.alf.cache import QC_TYPE
from one.converters import session_record2path
[docs]
def set_up_env() -> tempfile.TemporaryDirectory:
"""Create a temporary directory and copy cache fixtures over.
Returns
-------
tempfile.TemporaryDirectory
The temporary directory containing the test ONE caches
"""
fixture = Path(__file__).parent.joinpath('fixtures')
tempdir = tempfile.TemporaryDirectory()
# Copy cache files to temporary directory
for cache_file in ('sessions', 'datasets'):
filename = shutil.copy(fixture / f'{cache_file}.pqt', tempdir.name)
assert Path(filename).exists()
# Copy cached rest responses
setup_rest_cache(tempdir.name)
return tempdir
[docs]
def setup_rest_cache(cache_dir):
"""Copy REST cache fixtures to the .one parameter directory.
Parameters
----------
cache_dir : str, pathlib.Path
The location of the ONE cache directory (e.g. ~/Downloads/ONE/alyx.example.com)
"""
fixture = Path(__file__).parent.joinpath('fixtures')
rest_cache_dir = Path(cache_dir).joinpath('.rest')
# Ensure empty
shutil.rmtree(rest_cache_dir, ignore_errors=True)
rest_cache_dir.mkdir(parents=True, exist_ok=True)
set_hidden(rest_cache_dir, True)
# Copy over fixtures
for file in fixture.joinpath('rest_responses').glob('*'):
filename = shutil.copy(file, rest_cache_dir)
assert Path(filename).exists()
[docs]
def create_file_tree(one):
"""Touch all the files in the datasets table.
Parameters
----------
one : one.api.One
An instance of One containing cache tables to use.
"""
# Create dset files from cache
for (eid, _), rel_path in one._cache.datasets['rel_path'].items():
session_path = session_record2path(one._cache.sessions.loc[eid])
filepath = Path(one.cache_dir).joinpath(session_path, rel_path)
filepath.parent.mkdir(exist_ok=True, parents=True)
filepath.touch()
[docs]
def setup_test_params(token=False, cache_dir=None):
"""Copy cache parameter fixture to .one directory.
Parameters
----------
token : bool
If true, save a token file so that client doesn't hit auth endpoint
cache_dir : str, pathlib.Path
The cache_dir to save
"""
params_dir = Path(one.params.get_params_dir())
params_dir.mkdir(exist_ok=True)
fixture = Path(__file__).parent.joinpath('fixtures')
test_pars = '.test.alyx.internationalbrainlab.org'
if not next(params_dir.glob(test_pars), None):
filename = shutil.copy(fixture / 'params' / test_pars, params_dir / test_pars)
assert Path(filename).exists()
# Add to cache map
map_file = params_dir / '.caches'
if not map_file.exists():
shutil.copy(fixture / 'params' / '.caches', map_file)
assert Path(filename).exists()
with open(map_file, 'r+') as f:
data = json.load(f)
data['CLIENT_MAP'][test_pars[1:]] = str(cache_dir or '')
f.seek(0)
json.dump(data, f)
f.truncate()
# Add token to file so db not hit
if token:
pars = one.params.get(client=test_pars[1:])
if not getattr(pars, 'TOKEN', False):
one.params.save(pars.set('TOKEN', {'token': 'T0k3N'}), test_pars[1:])
[docs]
def revisions_datasets_table(collections=('', 'alf/probe00', 'alf/probe01'),
revisions=('', '2020-01-08', '2021-07-06'),
object='spikes',
attributes=('times', 'waveforems'),
touch_path=None):
"""Return a datasets cache DataFrame containing datasets with revision folders.
As there are no revised datasets on the test databases, this function acts as a fixture for
testing the filtering of datasets by a revision.
Parameters
----------
collections : tuple
A list of collections
revisions : tuple
A list of revisions
object : str
An ALF object
attributes : tuple
A list of ALF attributes
touch_path : pathlib.Path, str
If provided, files are created in this directory.
Returns
-------
pd.DataFrame
A datasets cache table containing datasets made from the input names.
"""
rel_path = []
for attr in attributes:
for collec in collections:
for rev in (f'#{x}#' if x else '' for x in revisions):
rel_path.append('/'.join(x for x in (collec, rev, f'{object}.{attr}.npy') if x))
d = {
'rel_path': rel_path,
'file_size': 0,
'hash': None,
'exists': True,
'qc': 'NOT_SET',
'eid': uuid4(),
'id': (uuid4() for _ in rel_path)
}
if touch_path:
for p in rel_path:
path = Path(touch_path).joinpath(p)
path.parent.mkdir(parents=True, exist_ok=True)
path.touch()
return pd.DataFrame(data=d).astype({'qc': QC_TYPE}).set_index(['eid', 'id'])
[docs]
def create_schema_cache(param_dir=None):
"""Save REST cache file for docs/ endpoint.
Ensures the database isn't hit when the rest_schemas property is accessed.
Parameters
----------
param_dir : str, pathlib.Path
The location of the parameter directory. If None, the default one is used.
"""
actions = dict.fromkeys(['list', 'read', 'create', 'update', 'partial_update', 'delete'])
endpoints = ['cache', 'dataset-types', 'datasets', 'downloads', 'insertions', 'sessions']
path_parts = ('.rest', 'test.alyx.internationalbrainlab.org', 'https')
rest_cache_dir = Path(param_dir or one.params.get_params_dir()).joinpath(*path_parts)
with open(rest_cache_dir / '1baff95c2d0e31059720a3716ad5b5a34b61a207', 'r') as f:
json.dump({k: actions for k in endpoints}, f)
[docs]
def get_file(root: str, str_id: str) -> str:
"""A stub function for `iblutil.io.params.getfile`.
Allows the injection of a different param dir.
Parameters
----------
root : str, pathlib.Path
The root directory of the new parameters
str_id : str
The parameter string identifier
Returns
-------
str
The parameter file path
"""
parts = ['.' + p if not p.startswith('.') else p for p in Path(str_id).parts]
pfile = Path(root, *parts).as_posix()
return pfile
[docs]
def caches_str2int(caches):
"""Convert str ids to int ids for cache tables.
Parameters
----------
caches : Bunch
A bunch of cache tables (from One._cache)
"""
for table in ('sessions', 'datasets'):
index = caches[table].index
names = (index.name,) if index.name else index.names
cache = caches[table].reset_index()
for name in names:
for i, col in enumerate(str2np(cache.pop(name).values).T):
cache[f'{name}_{i}'] = col
int_cols = cache.filter(regex=r'_\d{1}$').columns.sort_values().tolist()
caches[table] = cache.set_index(int_cols)
[docs]
@contextmanager
def mock_http_download(content=b'mock file contents', tag=True):
"""Stand in for the HTTP transfer so download tests need no data server access.
The test database's file records point at the private FlatIron tree. Only the bytes on the
wire are simulated; URL validation, the hash check and UUID stripping still run.
The mismatch handler is stubbed out too: a fabricated file never matches the recorded hash,
and the handler would flag the file record on Alyx, writing to the shared test database.
Parameters
----------
content : bytes
The contents of each downloaded file.
tag : bool
If false, `_tag_mismatched_file_record` is not stubbed out.
Yields
------
unittest.mock.MagicMock, None
The patched `_tag_mismatched_file_record`, or None if `tag` is false.
"""
from iblutil.io import hashfile
def download(url, chunks=None, *, clobber=False, silent=False, username='', password='',
target_dir='', return_md5=False, headers=None):
name = urllib.parse.urlsplit(url).path.rsplit('/', 1)[-1]
target = Path(target_dir or '')
target.mkdir(parents=True, exist_ok=True)
file = target.joinpath(name)
file.write_bytes(content)
return (file, hashfile.md5(file)) if return_md5 else file
with ExitStack() as stack:
stack.enter_context(mock.patch('one.webclient.http_download_file', new=download))
if tag:
tag = stack.enter_context(mock.patch('one.api.OneAlyx._tag_mismatched_file_record'))
yield tag or None
[docs]
def rest_schemes():
"""Load the Alyx REST schema fixture, so that an AlyxClient need not fetch it.
Returns
-------
one.webclient.RestSchemeOpenApi
The REST endpoints and their parameters.
"""
from one.webclient import RestSchemeOpenApi
fixture = Path(__file__).parent.joinpath('fixtures', 'rest_responses', 'openapiv3.json')
with open(fixture, 'r') as f:
return RestSchemeOpenApi(json.load(f))