# -*- coding: utf-8 -*- """ *************************************************************************** TestOTBAlgorithms.py --------------------- Copyright : (C) 2013 by CS Systemes d'information Email : otb at c-s dot fr Contributors : Oscar Picas *************************************************************************** * * * This program is free software; you can redistribute it and/or modify * * it under the terms of the GNU General Public License as published by * * the Free Software Foundation; either version 2 of the License, or * * (at your option) any later version. * * * *************************************************************************** """ # This will get replaced with a git SHA1 when you do a git archive __revision__ = '$Format:%H$' import unittest import signal import os import shlex import subprocess import shelve try: import processing # NOQA except ImportError as e: raise Exception("Processing must be installed and available in PYTHONPATH") try: import otbApplication except ImportError as e: raise Exception("OTB python plugins must be installed and available in PYTHONPATH") from processing.algs.otb.OTBHelper import get_OTB_log, create_xml_descriptors from processing.algs.otb.OTBTester import MakefileParser class Alarm(Exception): pass def alarm_handler(signum, frame): raise Alarm class AlgoTestCase(unittest.TestCase): def setUp(self): self.logger = get_OTB_log() self.the_files = [os.path.join(os.path.join(os.path.abspath(os.curdir), 'description'), each) for each in os.listdir(os.path.join(os.path.abspath(os.curdir), 'description')) if '.xml' in each] def tearDown(self): self.logger = None class TestSequence(unittest.TestCase): def setUp(self): self.data = shelve.open("tests.shelve", writeback=True) def tearDown(self): self.data.close() def ut_generator(test_name, a_tuple): def test(self): logger = get_OTB_log() needs_update = False if test_name not in self.data: needs_update = True if test_name in self.data: if (self.data[test_name][0] != a_tuple[0]) or (self.data[test_name][1] != a_tuple[1]) or (self.data[test_name][2] is False): needs_update = True if needs_update: signal.signal(signal.SIGALRM, alarm_handler) signal.alarm(6 * 60) # 6 minutes black_list = [] ut_command = a_tuple[0] self.assertTrue(ut_command is not None) self.assertTrue(ut_command != "") ut_command_validation = a_tuple[1] self.assertTrue(ut_command_validation is not None) self.assertTrue(ut_command_validation != "") if ut_command.split(" ")[0] in black_list: raise Exception("Blacklisted test!") args = shlex.split(ut_command) failed = False logger.info("Running [%s]" % ut_command) p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) (pout, perr) = p.communicate() if ("ERROR" in pout or "ERROR" in perr) or ("FATAL" in pout or "FATAL" in perr) or ("CRITICAL" in pout or "CRITICAL" in perr): error_text = "Command [%s] returned [%s]" % (ut_command, pout) if "Invalid image filename" in pout or "Invalid vector data filename" in pout or "Failed to open" in pout: logger.warning(error_text) else: logger.error(error_text) self.fail(error_text) failed = True else: logger.info(pout) if (len(ut_command_validation) > 0) and not failed: new_ut_command_validation = ut_command_validation + " Execute " + ut_command logger.info("Running Unit test [%s]" % new_ut_command_validation) argz = shlex.split(new_ut_command_validation) q = subprocess.Popen(argz, stdout=subprocess.PIPE, stderr=subprocess.PIPE) (qout, qerr) = q.communicate() if not ("Test EXIT SUCCESS" in qout or "Test EXIT SUCCESS" in qerr): error_text = "Unit test [%s] returned [%s]" % (new_ut_command_validation, qout) if "Invalid image filename" in qout or "Invalid vector data filename" in qout or "Failed to open" in qout: logger.warning(error_text) else: logger.error(error_text) self.fail(error_text) else: logger.info(qout) signal.alarm(0) self.data[test_name] = [a_tuple[0], a_tuple[1], failed] else: logger.info("Passed test: %s" % test_name) return test def get_client_apps(): app_clients = [] for available_app in otbApplication.Registry.GetAvailableApplications(): app_instance = otbApplication.Registry.CreateApplication(available_app) app_instance.UpdateParameters() ct = "otbcli_" + available_app app_clients.append(ct) return app_clients def unfiltered_processing_mapping(): mkf = MakefileParser() the_tests = mkf.test_algos() for t in the_tests: test_name = 'test_std_%s' % t if the_tests[t][1] is None: skip = True else: if the_tests[t][1] == "": skip = True if not skip: test = ut_generator(test_name, the_tests[t]) setattr(TestSequence, test_name, test) suite = unittest.TestLoader().loadTestsFromTestCase(TestSequence) unittest.TextTestRunner(verbosity=2).run(suite) def test_processing_mapping(): mkf = MakefileParser() the_tests = mkf.test_algos() clients = get_client_apps() already_tested = set() for t in the_tests: test_name = 'test_%s' % t if the_tests[t][0].split(" ")[0] in clients: skip = False if the_tests[t][1] is None: skip = True else: if the_tests[t][1] == "": skip = True if not skip: runnable = the_tests[t][0].split(" ")[0] if runnable not in already_tested: test = ut_generator(test_name, the_tests[t]) setattr(TestSequence, test_name, test) already_tested.add(runnable) suite = unittest.TestLoader().loadTestsFromTestCase(TestSequence) unittest.TextTestRunner(verbosity=2).run(suite) def test_xml_generation(): create_xml_descriptors() if __name__ == '__main__': mkf = MakefileParser() the_tests = mkf.test_algos() for t in the_tests: test_name = 'test_%s' % t test = ut_generator(test_name, the_tests[t]) setattr(TestSequence, test_name, test) unittest.main()