mirror of
https://github.com/qgis/QGIS.git
synced 2025-10-31 00:06:02 -04:00
209 lines
7.0 KiB
Python
209 lines
7.0 KiB
Python
# -*- 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()
|