datestex/components/testexec.py

# abstract class for testrunning-functions
# ---------------------------------------------
"""
The test runs in 3 step:
1.1 precondition of system
1.2 test the system
1.3 postconditions of system incl. comparison
   ---
The main test is the testcase but there are superior granularities like
2.a test-sequence : few test-cases depending in a businessflow
    therefore you test a testcase without resetting the system
2.b test-set : some independing test-cases
    therefore you can test postconditions that works not directly
2.c test-matrix : a set of test-sets for the whole test
The granularity could be an implicite parameter in the main-module.
   ---
the test could be in different test-levels like
3.a component/service-test
    a quick test to check single functions like in component-test, so you have the possibility to test x-product.
    In difference of developer-test you can run large test-sets
3.b system-test
3.c integration-test
3.d acceptance-test
the test-level could be a configuration of the test-center or implicite of the application-definition.
   ---
the test can check different quality-measures like
4.a functionality
4.b
"""
from datetime import datetime
import basic.message
import basic.program
import inspect
import os
import utils.db_abstract
import basic.toolHandling
import components.component
import basic.componentHandling
import utils.db_abstract
import utils.path_tool
import utils.file_tool
import utils.match_tool
import utils.tdata_tool
import basic.constants as B


class Testexecuter():
    def prepare_system(self, granularity):
        """
        In order to preparate the test system test data should be cleaned and loaded with actual data.
        This can happen on different granularities: for each testcase or for each test sequences or for a test set or
        for a whole test.
        The loaded test data should be selected for a later comparison.
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = -1+job.getDebugLevel(self.name)
        self.m.logInfo("--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
        self.reset_TData(granularity)
        self.m.setMsg("prepare_system for " + self.name +" "+ granularity + " is OK")
        self.m.logInfo("--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
        pass

    def reset_TData(self, granularity):
        """
        the testdata in the component of the testsystem will be resetted
        correponding with the configuration of the componend
        :param granularity:
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = -1+job.getDebugLevel(self.name)
        self.m.debug(verify, "--- "+str(inspect.currentframe().f_code.co_name)+"() started at "
                        + datetime.now().strftime("%Y%m%d_%H%M%S")+" for " + str(self.name).upper())
        if "log" in self.conf["artifact"]:
            self.m.logInfo("log rotate  in "+ self.name)
        if "db" in self.conf["artifact"]:
            self.m.logInfo("delete content "+ self.name)
            dbi = basic.toolHandling.getDbTool(self)
            dbi.deleteTables()
        if "lob" in self.conf["artifact"]:
            self.m.logInfo("lob is deleted with flaskdb "+ self.name)
        if "file" in self.conf["artifact"]:
            self.m.logInfo("rm files in "+ self.name)
        self.m.setMsg("resetInstance for " + self.name + " is OK")
        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())

    def load_TData(self, granularity, testdata):
        """
        the testdata will be loaded into the componend especially into databses
        or with import-functions of the component
        :param granularity:
        :param testdata:
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = -1+job.getDebugLevel(self.name)
        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
        plainname = basic.componentHandling.getPlainCompname(self.name)
        if plainname in testdata:
            print("plainname in testdata "+plainname)
            if "db" in self.conf["artifact"]:
                self.m.logInfo("delete content "+ self.name)
                dbi = basic.toolHandling.getDbTool(self)
                dbi.insertTables(testdata)
            self.m.setMsg("data loaded for " + self.name + " is OK")
        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())

    def collect_TcArtifacts(self):
        """
        collects the artifacts from the test-system.
        the result is written as original in subfolder {tsorigin}
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = -1+job.getDebugLevel(self.name)
        self.read_TData("nachher", "testcase")

    def read_TData(self, subdir, granularity):
        """

        :param granularity:
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = -1+job.getDebugLevel(self.name)
        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
        if "db" in self.conf["artifact"]:
            self.m.logInfo("select db-content "+ self.name)
            dbi = basic.toolHandling.getDbTool(self)
            dbi.selectTables()
        if "lob" in self.conf["artifact"]:
            self.m.logInfo("check lob if is deleted with flaskdb "+ self.name)
        self.m.setMsg("readInstance for " + self.name + " is OK")
        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())


    def composeFileClauses(self, pattern):
        job = basic.program.Job.getInstance()
        out = {}
        attr = utils.db_abstract.getDbAttributes(self, "null")
        while "{" in pattern:
            pre = pattern[0:pattern.index("{")]
            pat = pattern[pattern.index("{"):pattern.index("}")]
            post = pattern[pattern.index("}"):]
            pattern = pre+attr[pat]+post
        if (hasattr(job.par, B.PAR_DB_PARTITION)) and (attr[B.ATTR_DB_PARTITION] != "n"):
            parts = getattr(job.par, B.PAR_DB_PARTITION)
            a = parts.split(",")
            for p in a:
                pattern = pattern.replace(attr[B.ATTR_DB_PARTITION], p)
                out[p] = pattern
        else:
            out["ALL"] = pattern
        return out

    def composeSqlClauses(self, sql):
        job = basic.program.Job.getInstance()
        out = {}
        print("composeSqlClauses "+sql)
        table = sql[sql.upper().index(" FROM ")+6:].strip()
        print(table)
        sql_new = sql[0:sql.upper().index(" FROM ")+5]
        print(sql_new)
        attr = utils.db_abstract.getDbAttributes(self, table)
        if attr[B.ATTR_DB_TABNAME] != "":
            table = attr[B.ATTR_DB_TABNAME]
        if attr[B.ATTR_DB_SCHEMA] != "":
            table = attr[B.ATTR_DB_SCHEMA]+"."+table
        sql_new += " "+attr[B.ATTR_DB_DATABASE]+"."+table
        print(sql_new)
        if (hasattr(job.par, B.PAR_DB_WHERE)):
            sql_new += " WHERE "+getattr(job.par, B.PAR_DB_WHERE)
        if sql_new[0:6] == "SELECT":
            sql_new += " ORDER BY id"
        print(sql_new)
        sql_new = sql_new.replace('!', "\'")
        if (hasattr(job.par, B.PAR_DB_PARTITION)) and (attr[B.ATTR_DB_PARTITION] != "n"):
            parts = getattr(job.par, B.PAR_DB_PARTITION)
            a = parts.split(",")
            for p in a:
                sql_part = sql_new
                sql_part = sql_part.replace(attr[B.ATTR_DB_PARTITION], p)
                out[p] = sql_part
        else:
            out["ALL"] = sql_new
        return out

    def test_System(self, granularity):
        """

        :param granularity:
        :return:
        """
    def create_Request(self, granularity):
        pass
    def send_Request(self, granularity):
        pass
    def get_Response(self, granularity):
        pass

    def finish_Test(self, granularity):
        """
        initialization-routine for finish-step
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = -1+job.getDebugLevel(self.name)
        self.m.logInfo("--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
        self.m.logInfo("something in "+ self.name)
        self.m.setMsg("checkInstance for " + self.name + " is OK")
        self.m.logInfo("--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())

    def collect_system(self, granularity):
        """
        After the test the test data of each component should be selected again - to get the post-condition.
        In each component is configured how these data can be collected.
        In this phase the collected data have to be transformed to comparable files.
        :return:
        """
        pass

    def collect_TcArtifact(self, granularity):
        """
        collects the artifacts from the test-system.
        the result is written as original in subfolder {tcorigin}
        post: a further contact zo the test-system is not necessary
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = job.getDebugLevel(self.name)
        if "log" in self.conf["artifact"]:
            self.m.logInfo("get files in for "+ self.name + " in " + self.conf["artifact"]["log"]["path"])
        if "flaskdb" in self.conf["artifact"]:
            self.m.logInfo("select flaskdb-content "+ self.name)
        if "lob" in self.conf["artifact"]:
            pass # after selection get file from flaskdb
        if "file" in self.conf["artifact"]:
            self.m.logInfo("get files in for "+ self.name + " in " + self.conf["artifact"]["file"]["path"])

        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() " + str(self.name))

    def split_TcResult(self, granularity):
        """
        transforms the result from subfolder {tcorigin}.
        the result is written as utf-8-readable parts in subfolder {tcparts}
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = job.getDebugLevel(self.name)
        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() " + str(self.name))
        if "log" in self.conf["artifact"]:
            pass #
        if "flaskdb" in self.conf["artifact"]:
            pass # stored in table
        if "lob" in self.conf["artifact"]:
            self.m.logInfo("move file from flaskdb "+ self.name)
            self.m.logInfo("tidy files in for " + self.name + " in " + self.conf["artifact"]["lob"]["format"])
        if "file" in self.conf["artifact"]:
            self.m.logInfo("tidy files in for "+ self.name + " in " + self.conf["artifact"]["file"]["format"])

    def fix_TcResult(self, granularity):
        """
        fixes the result which is collected and transformed from the test-system.
        the result is written in comparable form in folder {tcresult}
        with the identifiable name - like in target
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = job.getDebugLevel(self.name)
        self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() " + str(self.name))

    def compare_results(self, report):
        """
        to compare the actual result with the target result has three steps:
        1 matching each element so you get a comparable pair
        2 comparing this pair so maybe you get differences
        3 rating the difference if this can be accepted
        :return:
        """
        job = basic.program.Job.getInstance()
        verify = job.getDebugLevel(self.name)
        cm = basic.componentHandling.ComponentManager.getInstance()
        data = {}
        matching = utils.match_tool.Matching()
        if "db" in self.conf["artifact"]:
            for t in self.conf["artifact"]["db"]:
                if t in ["type"]:
                    continue
                # fill each data into matching-object
                for side in utils.match_tool.MATCH_SIDES:
                    if side == utils.match_tool.MATCH_SIDE_PRESTEP:
                        if "prestep" in self.conf["artifact"]["db"][t]:
                            a = self.conf["artifact"]["db"][t]["prestep"].split(":")
                            if a[0] != self.name:
                                comp = cm.getComponent(a[0])
                            else:
                                comp = self
                            path = os.path.join(utils.path_tool.composePattern(
                                "{"+utils.match_tool.MATCH[utils.match_tool.MATCH_SIDE_POSTACTUAL]["filepattern"]+"}", comp), a[1]+".csv")
                        pass
                    elif side == utils.match_tool.MATCH_SIDE_TESTCASE:
                        if hasattr(job.par, "testcase_example"):
                            path = os.path.join(utils.path_tool.composePattern(
                                "{"+utils.match_tool.MATCH[utils.match_tool.MATCH_SIDE_POSTEXPECT]["filepattern"]+"}", self), t+".csv")
                            path.replace(getattr(job.par, "testcase"), getattr(job.par, "testcase_example"))
                    else:
                        path = os.path.join(utils.path_tool.composePattern("{" + utils.match_tool.MATCH[side]["filepattern"] + "}", self), t + ".csv")
                    filedata = utils.tdata_tool.readCsv(self.m, path, self)
                    data[side] = utils.match_tool.MATCH[side]
                    data[side]["path"] = path
                    data[side]["data"] = filedata
                # execute the matches
                for type in utils.match_tool.MATCH_TYPES:
                    matching.setData(data, type)
                    report.setPaths(job.par.testcase, self.name, t, type, matching.matchfiles["A"], matching.matchfiles["B"])
                    text = utils.match_tool.matchTree(matching)
                    report.setMatchResult(job.par.testcase, self.name, t, type, matching.cssClass, matching.diffText)
                    path = os.path.join(utils.path_tool.composePattern("{tcresult}", self),
                                t+"_"+utils.match_tool.MATCH[type]["filename"]+".html")
                    utils.file_tool.writeFileText(self.m, path, text)
                    # write text
            pass
        pass