Add explicit tests for ohlcv regex

This commit is contained in:
Matthias 2021-11-28 14:33:46 +01:00
parent 107e124f60
commit c096c7f5cb
4 changed files with 19 additions and 4 deletions

View File

@ -30,8 +30,7 @@ class HDF5DataHandler(IDataHandler):
""" """
_tmp = [ _tmp = [
re.search( re.search(
r'^([a-zA-Z_]+(\:[a-zA-Z]{2,}(\-[0-9]{2,})?)?)\-(\d+\S)\-?([a-zA-Z_]*)?(?=.h5)', cls._OHLCV_REGEX, p.name
p.name
) for p in datadir.glob("*.h5") ) for p in datadir.glob("*.h5")
] ]
return [(match[1].replace('_', '/'), match[2], match[3]) for match in _tmp return [(match[1].replace('_', '/'), match[2], match[3]) for match in _tmp

View File

@ -23,6 +23,8 @@ logger = logging.getLogger(__name__)
class IDataHandler(ABC): class IDataHandler(ABC):
_OHLCV_REGEX = r'^([a-zA-Z_]+)\-(\d+\S+)(?=\.)'
def __init__(self, datadir: Path) -> None: def __init__(self, datadir: Path) -> None:
self._datadir = datadir self._datadir = datadir

View File

@ -31,8 +31,7 @@ class JsonDataHandler(IDataHandler):
""" """
_tmp = [ _tmp = [
re.search( re.search(
r'^([a-zA-Z_]+(\:[a-zA-Z]{2,}(\-[0-9]{2,})?)?)\-(\d+\S)\-?([a-zA-Z_]*)?(?=.json)', cls._OHLCV_REGEX, p.name
p.name
) for p in datadir.glob(f"*.{cls._get_file_extension()}")] ) for p in datadir.glob(f"*.{cls._get_file_extension()}")]
return [(match[1].replace('_', '/'), match[2], match[3]) for match in _tmp return [(match[1].replace('_', '/'), match[2], match[3]) for match in _tmp
if match and len(match.groups()) > 1] if match and len(match.groups()) > 1]

View File

@ -1,6 +1,7 @@
# pragma pylint: disable=missing-docstring, protected-access, C0103 # pragma pylint: disable=missing-docstring, protected-access, C0103
import json import json
import re
import uuid import uuid
from pathlib import Path from pathlib import Path
from shutil import copyfile from shutil import copyfile
@ -669,6 +670,20 @@ def test_datahandler_ohlcv_get_pairs(testdatadir):
# assert set(pairs) == {'UNITTEST/BTC'} # assert set(pairs) == {'UNITTEST/BTC'}
@pytest.mark.parametrize('filename,pair,timeframe', [
('XMR_BTC-5m.json', 'XMR_BTC', '5m'),
('XMR_USDT-1h.h5', 'XMR_USDT', '1h'),
('BTC_USDT-2h.jsongz', 'BTC_USDT', '2h'),
])
def test_datahandler_ohlcv_regex(filename, pair, timeframe):
regex = JsonDataHandler._OHLCV_REGEX
match = re.search(regex, filename)
assert len(match.groups()) > 1
assert match[1] == pair
assert match[2] == timeframe
def test_datahandler_ohlcv_get_available_data(testdatadir): def test_datahandler_ohlcv_get_available_data(testdatadir):
paircombs = JsonDataHandler.ohlcv_get_available_data(testdatadir) paircombs = JsonDataHandler.ohlcv_get_available_data(testdatadir)
# Convert to set to avoid failures due to sorting # Convert to set to avoid failures due to sorting