Commit a739a50c authored by Daniel Scheffler's avatar Daniel Scheffler
Browse files

Some PEP8 editing for the tests.

Former-commit-id: 7172c224
Former-commit-id: 1e23a6a5
parent a033d60a
......@@ -16,83 +16,76 @@ from gms_preprocessing import __file__
from gms_preprocessing.algorithms.cloud_masking import FMASK_Runner_Landsat, FMASK_Runner_Sentinel2, FmaskWarning
from gms_preprocessing.misc.exception_handler import ignore_warning
rootpath_testdata = os.path.join(os.path.dirname(__file__), '../tests/data/')
testdata = dict(
Landsat5_collections_data = os.path.join(rootpath_testdata, 'archive_data',
'Landsat-5', 'TM', 'LT05_L1TP_149031_20101204_20161011_01_T1.tar.gz'),
Landsat5_precollections_data = os.path.join(rootpath_testdata, 'archive_data',
'Landsat-5', 'TM', 'LT51850252011113KIS01.tar.gz'),
Landsat7_collections_data = os.path.join(rootpath_testdata, 'archive_data',
'Landsat-7', 'ETM+', 'LE07_L1TP_108035_20161229_20170219_01_T1.tar.gz'),
Landsat7_precollections_data = os.path.join(rootpath_testdata, 'archive_data',
'Landsat-7', 'ETM+', 'LE71920232001238EDC00.tar.gz'),
Landsat8_collections_data = os.path.join(rootpath_testdata, 'archive_data',
'Landsat-8', 'OLI_TIRS', 'LC08_L1TP_171033_20170421_20170501_01_T1.tar.gz'),
Landsat8_precollections_data = os.path.join(rootpath_testdata, 'archive_data',
'Landsat-8', 'OLI_TIRS', 'LC80010702013141LGN01.tar.gz'),
Sentinel2A_old_style_data = os.path.join(rootpath_testdata, 'archive_data',
'Sentinel-2A', 'MSI', 'S2A_OPER_PRD_MSIL1C_PDMC_20161206T101533_R022_V20161205T101402_20161205T101402.zip'),
Sentinel2A_new_style_data = os.path.join(rootpath_testdata, 'archive_data',
'Sentinel-2A', 'MSI', 'S2A_MSIL1C_20161231T103432_N0204_R108_T32TLT_20161231T103428.zip'),
Landsat5_collections_data=os.path.join(rootpath_testdata, 'archive_data', 'Landsat-5', 'TM',
'LT05_L1TP_149031_20101204_20161011_01_T1.tar.gz'),
Landsat5_precollections_data=os.path.join(rootpath_testdata, 'archive_data', 'Landsat-5', 'TM',
'LT51850252011113KIS01.tar.gz'),
Landsat7_collections_data=os.path.join(rootpath_testdata, 'archive_data', 'Landsat-7', 'ETM+',
'LE07_L1TP_108035_20161229_20170219_01_T1.tar.gz'),
Landsat7_precollections_data=os.path.join(rootpath_testdata, 'archive_data', 'Landsat-7', 'ETM+',
'LE71920232001238EDC00.tar.gz'),
Landsat8_collections_data=os.path.join(rootpath_testdata, 'archive_data', 'Landsat-8', 'OLI_TIRS',
'LC08_L1TP_171033_20170421_20170501_01_T1.tar.gz'),
Landsat8_precollections_data=os.path.join(rootpath_testdata, 'archive_data',
'Landsat-8', 'OLI_TIRS', 'LC80010702013141LGN01.tar.gz'),
Sentinel2A_old_style_data=os.path.join(rootpath_testdata, 'archive_data', 'Sentinel-2A', 'MSI',
'S2A_OPER_PRD_MSIL1C_PDMC_20161206T101533_R022_V20161205T101402_'
'20161205T101402.zip'),
Sentinel2A_new_style_data=os.path.join(rootpath_testdata, 'archive_data', 'Sentinel-2A', 'MSI',
'S2A_MSIL1C_20161231T103432_N0204_R108_T32TLT_20161231T103428.zip'),
)
class Test_FMASK_Runner_Landsat(unittest.TestCase):
@ignore_warning(FmaskWarning)
def test_Landsat5_collections_data(self):
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Landsat(testdata['Landsat5_collections_data'], 'Landsat-5')
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
@ignore_warning(FmaskWarning)
def test_Landsat5_precollections_data(self):
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Landsat(testdata['Landsat5_precollections_data'], 'Landsat-5')
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
@ignore_warning(FmaskWarning)
def test_Landsat7_collections_data(self):
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Landsat(testdata['Landsat7_collections_data'], 'Landsat-7')
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
@ignore_warning(FmaskWarning)
def test_Landsat7_precollections_data(self):
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Landsat(testdata['Landsat7_precollections_data'], 'Landsat-7')
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
@ignore_warning(FmaskWarning)
def test_Landsat8_collections_data(self):
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Landsat(testdata['Landsat8_collections_data'], 'Landsat-8')
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
@ignore_warning(FmaskWarning)
def test_Landsat8_precollections_data(self):
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Landsat(testdata['Landsat8_precollections_data'], 'Landsat-8')
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
class Test_FMASK_Runner_Sentinel2(unittest.TestCase):
@ignore_warning(FmaskWarning)
def test_Sentinel2A_old_style(self):
# FIXME scene ID or granule ID must be provided in case of multiple files within source archive
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Sentinel2(testdata['Sentinel2A_old_style_data'], 'Sentinel-2A', scene_ID=None)
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
@ignore_warning(FmaskWarning)
def test_Sentinel2A_new_style(self):
#os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
# os.environ['RIOS_DFLT_DRIVER'] = 'VRT'
FMR = FMASK_Runner_Sentinel2(testdata['Sentinel2A_new_style_data'], 'Sentinel-2A')
self.assertIsInstance(FMR.calc_cloudMask(), GeoArray)
......@@ -34,11 +34,7 @@ that can be found in the directory "tests/logs".
Program edited in July 2017.
"""
###################################################################################
__author__ = 'Daniel Scheffler' # edited by Jessica Palka.
# Imports from the python standard library.
# Import python standard libraries.
import itertools
import logging
import os
......@@ -57,6 +53,7 @@ from gms_preprocessing.algorithms.L2B_P import L2B_object
from gms_preprocessing.algorithms.L2C_P import L2C_object
from gms_preprocessing.misc.database_tools import get_info_from_postgreSQLdb
__author__ = 'Daniel Scheffler' # edited by Jessica Palka.
# Rootpath of the gms_preprocessing-repository.
gmsRepo_rootpath = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
......@@ -64,7 +61,7 @@ gmsRepo_rootpath = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')
# Defining the configurations needed to start a job containing the different dataset scenes.
# TODO Change the job-configurations for selected datasets.
job_config_kwargs = dict(is_test = True)
job_config_kwargs = dict(is_test=True)
##########################
......@@ -80,7 +77,7 @@ class BaseTestCases:
in this test-script.
"""
class TestAll(unittest.TestCase):
PC = None # default
PC = None # default
@classmethod
def tearDownClass(cls):
......@@ -102,35 +99,43 @@ class BaseTestCases:
def test_L1A_processing(self):
self.L1A_newObjects = self.PC.L1A_processing()
self.assertIsInstance(self.L1A_newObjects, list)
self.assertNotEqual(len(self.L1A_newObjects), 0, msg='L1A_processing did not output an L1A object.')
self.assertIsInstance(self.L1A_newObjects[0], L1A_object)
def test_L1B_processing(self):
self.L1B_newObjects = self.PC.L1B_processing()
self.assertIsInstance(self.L1B_newObjects, list)
self.assertNotEqual(len(self.L1B_newObjects), 0, msg='L1B_processing did not output an L1B object.')
self.assertIsInstance(self.L1B_newObjects[0], L1B_object)
def test_L1C_processing(self):
self.L1C_newObjects = self.PC.L1C_processing()
self.assertIsInstance(self.L1C_newObjects, list)
self.assertNotEqual(len(self.L1C_newObjects), 0, msg='L1C_processing did not output an L1C object.')
self.assertIsInstance(self.L1C_newObjects[0], L1C_object)
def test_L2A_processing(self):
self.L2A_newObjects = self.PC.L2A_processing()
self.assertIsInstance(self.L2A_newObjects, list)
self.assertNotEqual(len(self.L2A_newObjects), 0, msg='L2A_processing did not output an L2A object.')
self.assertIsInstance(self.L2A_newObjects[0], L2A_object)
def test_L2B_processing(self):
self.L2B_newObjects = self.PC.L2B_processing()
self.assertIsInstance(self.L2B_newObjects, list)
self.assertNotEqual(len(self.L2B_newObjects), 0, msg='L2B_processing did not output an L2B object.')
self.assertIsInstance(self.L2B_newObjects[0], L2B_object)
def test_L2C_processing(self):
self.L2C_newObjects = self.PC.L2C_processing()
self.assertIsInstance(self.L2C_newObjects, list)
self.assertNotEqual(len(self.L2C_newObjects), 0, msg='L2C_processing did not output an L2C object.')
self.assertIsInstance(self.L2C_newObjects[0], L2C_object)
# Setting the job.status manually.
# if self.L2C_newObjects:
# self.PC.job.status = "finished" # FIXME after updating the job.status-attribute for the level-processes, delete the code that is commented out.
# self.PC.job.status = "finished"
# FIXME after updating the job.status-attribute for the level-processes, delete the code that is commented
# FIXME out.
###################################################################################
......@@ -246,16 +251,17 @@ class Test_MultipleDatasetsInOneJob(BaseTestCases.TestAll):
# information are sorted according to the testdata.
# 3-letter code:
# UKR-Ukraine, KGZ-Kyrgyztan, POL-Poland, AUT-Austria, JPN-Japan, BOL-Bolivia, TUR-Turkey, DEU-Germany, CHE-Switzerland.
testdata_scenes = {'Landsat5_PreCollectionData' : list(['UKR', 'City region, forest', 'Sparsely', 'Zone 34/35']),
# 'Landsat5_CollectionData' : list(['KGZ', 'Snowy Mountains', 'Yes', 'None']),
'Landsat7_SLC_on_PreCollectionData' : list(['POL', 'City region, lakes', 'Yes', 'None']),
'Landsat7_SLC_off_PreCollectionData': list(['AUT', 'Stripes (partly), Mountains', 'None', 'None']),
# 'Landsat7_SLC_off_CollectionData' : list(['JPN', 'Stripes (completly), Mountains', 'Yes', 'Zone 53/54']),
'Landsat8_PreCollectionData' : list(['BOL', 'Forest', 'Yes', 'None']),
'Landsat8_CollectionData' : list(['TUR', 'Snowy Mountains', 'Yes', 'None']),
'Sentinel2A_PreCollectionData' : list(['DEU', 'Potsdam', 'Sparsely', 'None']),
'Sentinel2A_CollectionData' : list(['CHE', 'City region, on the Rhine', 'Yes', 'None'])
}
testdata_scenes = \
{'Landsat5_PreCollectionData': list(['UKR', 'City region, forest', 'Sparsely', 'Zone 34/35']),
# 'Landsat5_CollectionData': list(['KGZ', 'Snowy Mountains', 'Yes', 'None']),
'Landsat7_SLC_on_PreCollectionData': list(['POL', 'City region, lakes', 'Yes', 'None']),
'Landsat7_SLC_off_PreCollectionData': list(['AUT', 'Stripes (partly), Mountains', 'None', 'None']),
# 'Landsat7_SLC_off_CollectionData': list(['JPN', 'Stripes (completly), Mountains', 'Yes', 'Zone 53/54']),
'Landsat8_PreCollectionData': list(['BOL', 'Forest', 'Yes', 'None']),
'Landsat8_CollectionData': list(['TUR', 'Snowy Mountains', 'Yes', 'None']),
'Sentinel2A_PreCollectionData': list(['DEU', 'Potsdam', 'Sparsely', 'None']),
'Sentinel2A_CollectionData': list(['CHE', 'City region, on the Rhine', 'Yes', 'None'])
}
# The key of the dictionary is the key-value to parametrize the testclasses so that each testclass is executed
# automatically.
......@@ -284,8 +290,8 @@ if __name__ == '__main__':
summary_failures.append(testResult.failures)
summary_skipped.append(testResult.skipped)
# jobstatus.append(eval("Test_"+items).PC.job.status) # FIXME: If the job.status-issue is fixed, the commented out section can be nullified.
# FIXME: If the job.status-issue is fixed, the commented out section can be nullified.
# jobstatus.append(eval("Test_"+items).PC.job.status)
# Part 3: Summarizing the testresults of each testsuite and outputting the results in an orderly fashion on the
# console and in a textfile.
......@@ -302,8 +308,8 @@ if __name__ == '__main__':
test_log_path = os.path.join(gmsRepo_rootpath, 'tests', 'data', 'logs', time.strftime('%Y%m%d_%H%M%S_log.txt'))
# Creating a logging system for the testresults.
# Source: The "GMS_logger"-function in the "gms_preprocessing" --> "misc" --> "logging.py"-script was used and slightly
# altered to meet the needs of the current problem.
# Source: The "GMS_logger"-function in the "gms_preprocessing" --> "misc" --> "logging.py"-script was used and
# slightly altered to meet the needs of the current problem.
logger = logging.getLogger("log_Test")
logger.setLevel(logging.INFO)
......@@ -335,7 +341,8 @@ if __name__ == '__main__':
# Adding a table displaying the characteristic features of each dataset.
logger.info(pandas.DataFrame.from_items(testdata_scenes.items(),
orient='index', columns=['Country', 'Characteristic', 'Clouds', 'Overlap_area']))
orient='index',
columns=['Country', 'Characteristic', 'Clouds', 'Overlap_area']))
logger.info("\nThe jobID used in Test_" + testdata[-1] + " contains the datasets: "
"\n-Landsat5_PreCollectionData,\n-Landsat7_SLC_off_PreCollectionData and "
"\n-Sentinel2A_CollectionData.")
......@@ -346,13 +353,13 @@ if __name__ == '__main__':
results = ["Run", "Success", "Errors", "Failures", "Skips"]
testdata_index = ["Test_" + item for item in testdata]
logger.info(pandas.DataFrame(summary_testResults, columns= results, index=testdata_index))
logger.info(pandas.DataFrame(summary_testResults, columns=results, index=testdata_index))
# If errors, failures or skips (there is yet nothing to skip in the code) occurres, the respective message will
# be printed.
logger.info("\n***************************************************************************************")
if list(itertools.chain(*summary_errors)) or list(itertools.chain(*summary_failures)) or \
list(itertools.chain(*summary_skipped)):
list(itertools.chain(*summary_skipped)):
logger.info("--> ERRORS/FAILURES/SKIPS:")
logger.info("\n---------------------------------------------------------------------------------------")
......
......@@ -22,6 +22,7 @@ from gms_preprocessing.algorithms.L2B_P import SpectralResampler as SR
testdata = os.path.join(os.path.dirname(__file__),
'../tests/data/hy_spec_data/Bavaria_farmland_LMU_Hyspex_subset.bsq')
class Test_SpectralResampler(unittest.TestCase):
"""Tests class for gms_preprocessing.algorithms.L2B_P.SpectralResampler1D"""
......
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