Commit 5954da26 authored by pm22d12's avatar pm22d12
Browse files

Add #61 - unit test Downloader DataReader

parent 0dd9b626
{
"test_Regexp":
[
{
"filepath": "data/regexp_20190625T183259_x.txt",
"regexp_date": "regexp_([0-9]{8}T[0-9]{6}).*",
"date_format": "%Y%m%dT%H%M%S",
"storage_path": "%Y/%j",
"getDate": "2019-06-25 18:32:59",
"getStorageName": "regexp_20190625T183259_x.txt",
"getRelativePath": "2019/176/regexp_20190625T183259_x.txt",
}
],
"test_DirRegexp":
[
{
"filepath": "data/dirregexp_20190625T183259_x/file.txt",
"regexp_date": ".*/dirregexp_([0-9]{8}T[0-9]{6}).*",
"date_format": "%Y%m%dT%H%M%S",
"storage_path": "%Y/%j",
"getDate": "2019-06-25 18:32:59",
"getStorageName": "file.txt",
"getRelativePath": "2019/176/dirregexp_20190625T183259_x/file.txt",
},
{
"filepath": "data/dirregexp_20190625T183259_x/file.txt",
"regexp_date": ".*/dirregexp_([0-9]{8}T[0-9]{6}).*",
"date_format": "%Y%m%dT%H%M%S",
"storage_path": "%Y/%j",
"getDate": "2019-06-25 18:32:59",
"getStorageName": "file.txt",
"getRelativePath": "2019/176/file.txt",
}
],
"test_NetCDF":
[
{
"filepath": "data/datareader/netcdf_data1.nc",
"regexp_date": "first_meas_time",
"date_format": "%Y-%m-%d %H:%M:%S.%f",
"storage_path": "%Y/%j",
"getDate": "2018-07-09 23:51:39.721869",
"getStorageName": "netcdf_data1.nc",
"getRelativePath": "2018/190/netcdf_data1.nc",
},
{
"filepath": "data/datareader/netcdf_data2.nc",
"regexp_date": "first_meas_time",
"date_format": "%Y-%m-%d %H:%M:%S.%f",
"storage_path": "%Y/%j",
"getDate": 'None',
"getStorageName": "netcdf_data2.nc",
"getRelativePath": 'data/datareader/netcdf_data2.nc',
}
],
"test_Grib":
[
{
"filepath": "data/datareader/grib_data1",
"regexp_date": "first_meas_time",
"date_format": "%Y-%m-%d %H:%M:%S.%f",
"storage_path": "%Y/%j",
"getDate": "2020-02-26 12:00:00",
"getStorageName": "grib_data1",
"getRelativePath": "2020/057/grib_data1",
},
{
"filepath": "data/datareader/grib_data2",
"regexp_date": "first_meas_time",
"date_format": "%Y-%m-%d %H:%M:%S.%f",
"storage_path": "%Y/%j",
"getDate": '2016-04-09 21:00:00',
"getStorageName": "grib_data2",
"getRelativePath": '2016/100/grib_data2',
},
{
"filepath": "data/datareader/grib_data3",
"regexp_date": "",
"date_format": "",
"storage_path": "%Y/%j",
"getDate": 'None',
"getStorageName": "grib_data3",
"getRelativePath": 'data/datareader/grib_data3',
}
]
}
import unittest
import os
import re
import ast
import shutil
from downloader.scheduler.plugins.DataReader.RegexpDataReader import RegexpDataReader
from downloader.scheduler.plugins.DataReader.DirRegexpDataReader import DirRegexpDataReader
from downloader.scheduler.plugins.DataReader.NetCDFDataReader import NetCDFDataReader
from downloader.scheduler.plugins.DataReader.GribDataReader import GribDataReader
class PluginsDataReaderTestCase(unittest.TestCase):
set_of_tests = {}
def create_RegexpDataReader(self, set_of_test):
return RegexpDataReader(
'unittest.TestCase',
set_of_test['regexp_date'],
set_of_test['date_format'],
set_of_test['storage_path']
)
def create_DirRegexpDataReader(self, set_of_test):
return DirRegexpDataReader(
'unittest.TestCase',
set_of_test['regexp_date'],
set_of_test['date_format'],
set_of_test['storage_path']
)
def create_NetCDFDataReader(self, set_of_test):
return NetCDFDataReader(
'unittest.TestCase',
set_of_test['regexp_date'],
set_of_test['date_format'],
set_of_test['storage_path']
)
def create_GribDataReader(self, set_of_test):
return GribDataReader(
'unittest.TestCase',
set_of_test['regexp_date'],
set_of_test['date_format'],
set_of_test['storage_path']
)
def run_DataReader(self, reader, set_of_test):
filepath = set_of_test['filepath']
self.assertEqual(str(reader.getDate(filepath)), set_of_test['getDate'])
self.assertEqual(reader.getStorageName(filepath), set_of_test['getStorageName'])
self.assertEqual(reader.getRelativePath(filepath), set_of_test['getRelativePath'])
def setUp(self) -> None:
if PluginsDataReaderTestCase.set_of_tests == {}:
cfg = ""
cfg_file = os.path.splitext(__file__)[0] + '.cfg'
with open(cfg_file, 'r') as f:
for line in f.readlines():
cfg += line.rstrip("\n")
PluginsDataReaderTestCase.set_of_tests = ast.literal_eval(cfg)
if self._testMethodName in PluginsDataReaderTestCase.set_of_tests.keys():
self.set_of_test = PluginsDataReaderTestCase.set_of_tests[self._testMethodName]
else:
self.skipTest(f"No set of test for {self._testMethodName} in {os.path.splitext(__file__)[0] + '.cfg'}")
# timeout in minutes
self.__timeout = 60
def test_Regexp(self) -> None:
set_of_test = self.set_of_test[0]
reader = self.create_RegexpDataReader(set_of_test)
self.run_DataReader(reader, set_of_test)
def test_DirRegexp(self) -> None:
set_of_test = self.set_of_test[0]
reader = self.create_DirRegexpDataReader(set_of_test)
reader.setContainer(True)
self.run_DataReader(reader, set_of_test)
set_of_test = self.set_of_test[1]
reader = self.create_DirRegexpDataReader(set_of_test)
reader.setContainer(False)
self.run_DataReader(reader, set_of_test)
def test_NetCDF(self) -> None:
for set_of_test in self.set_of_test:
reader = self.create_NetCDFDataReader(set_of_test)
self.run_DataReader(reader, set_of_test)
def test_Grib(self) -> None:
for set_of_test in self.set_of_test:
reader = self.create_GribDataReader(set_of_test)
self.run_DataReader(reader, set_of_test)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment