datestex/tools/file_abstract.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------------------------------------------
# Author : Ulrich Carmesin
# Source : gitea.ucarmesin.de
# ---------------------------------------------------------------------------------------------------------
import os
import basic.program
import tools.config_tool
import basic.constants as B
import basic.toolHandling
import tools.file_tool
import tools.path_tool


class FileFcts():
    """
    this is an abstract class
    """
    def __init__(self):
        pass

    def setComp(self, job, comp=None):
        self.job = job
        self.comp = comp

    def getMsg(self):
        if self.comp is not None:
            return self.comp.m
        if self.job is not None and hasattr(self.job, "m"):
            return self.job.m
        return None

    def load_file(self, path):
        """
        this function parses the text and translates it to dict
        :param text:
        :return:
        """
        raise Exception(B.EXCEPT_NOT_IMPLEMENT)

    def dump_file(self, data, path):
        """
        this function parses the text and translates it to dict
        :param text:
        :return:
        """
        raise Exception(B.EXCEPT_NOT_IMPLEMENT)


    def file2dict(self):
        pass

    def reset_TData(self, job):
        pass

    # Funktionen
    #=== init_testcase ===
    def removeFiles(self):
        tools.file_tool.removeFiles(self.comp.m, "envpath", "pattern", self.comp.conf["conn"])

    def copyFiles(self):
        fileList = []
        srcpath = ""
        envpath = ""
        pattern = ""
        tools.file_tool.copyFiles(self.job, fileList, srcpath, envpath, pattern)

    def readEnvFiles(self, job):
        envpath = ""
        pattern = ""
        fileList = tools.file_tool.getFiles(self.comp.m, job, envpath, pattern, self.comp.conf["conn"])


    # === execute_testcase ===
    def create_request(self, job, tdata, step):
        mapping = ""
        schema = ""
        archivpath = ""
        filename = step.args["filename"]
        txt = ""
        for o in self.comp.conf[B.SUBJECT_ARTS][B.TOPIC_NODE_FILE]:
            if o["name"] != filename:
                continue
            mapping = o["mapping"]
            schema = o["schema"]
            archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/request", self.comp), filename) # ergebnisse/comp/request )
        #txt = self.createDict()
        tools.file_tool.write_tile_text(self.comp.m, job, archivpath, txt)

    def send_request(self, job, step):
        archivpath = ""
        filename = step.args["filename"]
        technique = step.args["technique"]
        archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/request", self.comp), filename)
        if technique == "cli":
            for o in self.comp.conf[B.SUBJECT_ARTS][B.TOPIC_NODE_FILE]:
                if o["name"] != filename:
                    continue
                envpath = o["envpath"]
                envpath = tools.path_tool.compose_path(job, envpath, self.comp)
            fct = basic.toolHandling.getCliTool(job, self.comp)
            fct.copy(self.job, archivpath, envpath)
        elif technique == "api":
            txt = tools.file_tool.read_file_text(job, archivpath, self.comp.m)
            fct = basic.toolHandling.getApiTool(job, self.comp)
            response = fct.send(self.job, self.comp, txt)
            archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/response", self.comp), filename)

"""
get_response:
- channel (sync/async)
  ... implement
- archivpath ( ergebnisse/comp/response )
- envpath ( ./log) / envconn ( = in Request empfangen )

=== collect_testcase ===
- envpath
- pattern
> readfiles
"""