Testskripter
/
datestex
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Zwischenstand

master
Ulrich Carmesin 3 years ago
parent
2726a1f518
commit
ad6337f098
13 changed files with 391 additions and 248 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      check_environment.py
  2. 30
      components/catalog.py
  3. 246
      components/component.py
  4. 18
      components/maintain.py
  5. 15
      components/report.py
  6. 8
      components/sysmonitor.py
  7. 10
      components/testa/CONFIG.yml
  8. 2
      components/testa/Testa.py
  9. 201
      components/testrun.py
  10. 2
      finish_testcase.py
  11. 2
      init_testset.py
  12. 82
      ulrich/componentHandling.py
  13. 19
      ulrich/program.py

2
check_environment.py
View File

@ -19,7 +19,7 @@ if __name__ == '__main__':
exit(x.m.rc * (-1) + 3)
# now in theory the program is runnable
x.m.setMsg("# job initialized")
cm = ulrich.componentHandling.ComponentManager()
cm = ulrich.componentHandling
cm.initComponents()
comps = cm.getComponents(PROGRAM_NAME)
x.m.setMsg("# Components initialized with these relevant components " + str(comps))

30
components/catalog.py
View File

@ -0,0 +1,30 @@
#
# ------------------------------------
"""
a lot of test-results are created dynamically - so these dont belog to the testdata.
In order to get an overview over the testcases these results should be imported into a database.
"""
from datetime import datetime
import ulrich.message
import ulrich.program
import inspect
import components.component
class Catalog(components.component.CompData):
def init_Catalog(self):
"""
initialize special tables in the catalog-database
:return:
"""
pass
def check_Instance(self):
"""
checks system-instances and writes it into the parameter-file
"""
job = ulrich.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())

246
components/component.py
View File

@ -2,14 +2,33 @@
# ---------------------------------------------------------------------
from datetime import datetime
import components.sysmonitor
import components.testrun
import components.report
import components.maintain
import components.catalog
import ulrich.message
import ulrich.program
import inspect
"""
A component represents an application of the system-under-test
As the representation it has to knowlegde of the url, which other components depends on this component.
During a test-run the component must be checked, prepared, appfiles has to be collected, etc. For this doing there are some standard-methods implemented.
"""
class Component:
import threading
class CompData:
def __init__(self):
self.name = ""
self.m = None
self.conf = None
class Component(components.sysmonitor.SystemMonitor, components.testrun.Testrun, components.report.Report,
components.maintain.Maintainer, components.catalog, threading.Thread):
"""
A component represents an application of the system-under-test or a data-artifact which is created from the system-under-test.
As the representation it has to knowlegde of the url, which other components depends on this component.
During a test-run the component must be checked, prepared, appfiles has to be collected, etc. For this doing there are some standard-methods implemented.
"""
def init(self):
"""
The range of the test-system is defined by the dependancy graph which starts at the application.and indate each component.
@ -24,209 +43,24 @@ class Component:
verify = job.getDebugLevel(self.name)
self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() " + str(self.name))
def prepare_system(self):
"""
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:
"""
pass
def init_Testset(self):
"""
main methode for system-preparation
:param self:
:return:
"""
job = ulrich.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.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 reset_TsData(self):
job = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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 db-content "+ self.name)
if "lob" in self.conf["artifact"]:
self.m.logInfo("lob is deleted with db "+ self.name)
if "file" in self.conf["artifact"]:
self.m.logInfo("rm files in "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(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_TsData(self, testdata):
job = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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 "testdata" in self.conf:
if self.conf["testdata"] == "db":
self.m.logInfo("insert db-content " + self.name)
self.m.setMsg("data loaded for " + self.name + " is OK")
self.m.logDebug(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 read_TsData(self):
job = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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)
if "lob" in self.conf["artifact"]:
self.m.logInfo("check lob if is deleted with db "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(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 init_Testcase(self):
"""
main methode for system-preparation
:param self:
:return:
"""
job = ulrich.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 reset_TcData(self):
def run(self):
job = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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 db-content "+ self.name)
if "lob" in self.conf["artifact"]:
self.m.logInfo("lob is deleted with db "+ self.name)
if "file" in self.conf["artifact"]:
self.m.logInfo("rm files in "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
if job.program == "check_environment":
self.check_Instance()
elif job.program == "init_testset":
self.prepare_system("testset")
elif job.program == "init_testcase":
self.prepare_system("testcase")
elif job.program == "test_system":
self.test_System("test")
elif job.program == "test_system":
self.test_System("test")
elif job.program == "finish_testcase":
self.finish_Test("test")
elif job.program == "finish_testset":
self.finish_Test("test")
def load_TcData(self):
job = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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 "testdata" in self.conf:
if self.conf["testdata"] == "db":
self.m.logInfo("insert db-content " + self.name)
self.m.setMsg("data loaded for " + self.name + " is OK")
self.m.logDebug(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 read_TcData(self):
job = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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)
if "lob" in self.conf["artifact"]:
self.m.logInfo("check lob if is deleted with db "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(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 test_system(self):
"""
:return:
"""
pass
def spiele_ein(self):
pass
def erzeuge_Anfrage(self):
pass
def sende_Anfrage(self, testdata):
pass
def lese_Antwort(self):
pass
def finish_Testcase(self):
"""
initialization-routine for finish-step
:return:
"""
job = ulrich.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):
"""
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):
"""
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 = ulrich.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 "db" in self.conf["artifact"]:
self.m.logInfo("select db-content "+ self.name)
if "lob" in self.conf["artifact"]:
pass # after selection get file from db
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):
"""
transforms the result from subfolder {tcorigin}.
the result is written as utf-8-readable parts in subfolder {tcparts}
:return:
"""
job = ulrich.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 "db" in self.conf["artifact"]:
pass # stored in table
if "lob" in self.conf["artifact"]:
self.m.logInfo("move file from db "+ 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):
"""
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 = ulrich.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):
"""
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:
"""
pass
def collect_TcResult(self):
"""

18
components/maintain.py
View File

@ -0,0 +1,18 @@
#
# --------------------------------
"""
outside the testrun there are some tasks in order maintain the tests, for example:
# testrunner to control or execute the testrun
# set the target-results
# generate testcases
# partial cleaning of the storage
"""
from datetime import datetime
import ulrich.message
import ulrich.program
import inspect
import components.component
class Maintainer(components.component.CompData):
pass

15
components/report.py
View File

@ -0,0 +1,15 @@
#
# -------------------------------------------------
"""
reporting-applications
"""
from datetime import datetime
import ulrich.message
import ulrich.program
import inspect
import components.component
class Report(components.component.CompData):
pass

8
components/systemInstance.py → components/sysmonitor.py
View File

@ -5,8 +5,14 @@ When the test system is installed - especially in a continous pipeline - you don
installed and if each instance is running. In order to know these fundamental points it should be checked.
The result could be inserted into a database so you can get an overview over all test environments.
"""
from datetime import datetime
import ulrich.message
import ulrich.program
import inspect
import components.component
class SystemMonitor:
class SystemMonitor(components.component.CompData):
def check_Instance(self):
"""

10
components/testa/CONFIG.yml
View File

@ -22,11 +22,19 @@ conf:
check_environment: "todo"
init_testcase: "todo"
finish_testcase: "todo"
artifact:
system: # look at sysmonitor
script: "{dompath}/bin/sysctl.sh" # for checking
checks: "status,version,start,stop,restart,available"
artifact: # look at testrun
log: # log|db|file|lob
path: "{dompath}/log/debug.log"
rotate: jmx
reset: testcase
oldfile: "{dompath}/log/debug_{timepattern}.log"
catalog: # TODO
fields:
- field1:
type: "varchar(20)"
domain: product # for which stakeholder it is important and a view should be implemented

2
components/testa/Testa.py
View File

@ -1,7 +1,7 @@
import components.component
class Testa(components.component.Component):
def __init__(self):
print('init Testa')

201
components/testrun.py
View File

@ -0,0 +1,201 @@
# 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 ulrich.message
import ulrich.program
import inspect
import components.component
class Testrun(components.component.CompData):
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 = ulrich.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.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())
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 = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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 db-content "+ self.name)
if "lob" in self.conf["artifact"]:
self.m.logInfo("lob is deleted with db "+ self.name)
if "file" in self.conf["artifact"]:
self.m.logInfo("rm files in "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(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 = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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 "testdata" in self.conf:
if self.conf["testdata"] == "db":
self.m.logInfo("insert db-content " + self.name)
self.m.setMsg("data loaded for " + self.name + " is OK")
self.m.logDebug(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 read_TData(self, granularity):
"""
:param granularity:
:return:
"""
job = ulrich.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(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)
if "lob" in self.conf["artifact"]:
self.m.logInfo("check lob if is deleted with db "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(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 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 = ulrich.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 = ulrich.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 "db" in self.conf["artifact"]:
self.m.logInfo("select db-content "+ self.name)
if "lob" in self.conf["artifact"]:
pass # after selection get file from db
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 = ulrich.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 "db" in self.conf["artifact"]:
pass # stored in table
if "lob" in self.conf["artifact"]:
self.m.logInfo("move file from db "+ 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 = ulrich.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, granularity):
"""
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:
"""
pass

2
finish_testcase.py
View File

@ -22,7 +22,7 @@ if __name__ == '__main__':
# now in theory the program is runnable
cm = ComponentManager()
cm.initComponents()
comps = cm.getComponents(PROGRAM_NAME)
comps = getComponents(PROGRAM_NAME)
print(" relevant components for this job: " + str(comps))
tdata = utils.tdata_tool.getTestdata()
x.stopJob()

2
init_testset.py
View File

@ -22,7 +22,7 @@ if __name__ == '__main__':
# now in theory the program is runnable
cm = ComponentManager()
cm.initComponents()
comps = cm.getComponents(PROGRAM_NAME)
comps = getComponents(PROGRAM_NAME)
print(" relevant components for this job: " + str(comps))
tdata = utils.tdata_tool.getTestdata()
x.stopJob()

82
ulrich/componentHandling.py
View File

@ -16,17 +16,33 @@ import ulrich.message
import components.component
import importlib
import copy
components = {}
PARAM_NOSUBNODE = [ "artifact", "components", "instance" ]
comps = {}
PARAM_NOSUBNODE = ["artifact", "components", "instance"]
def getComponents(mainfct):
job = ulrich.program.Job.getInstance()
verify = -2 + job.getDebugLevel("job_tool")
job.debug(verify, "getComponents " + mainfct)
out = []
for c in comps:
job.debug(verify, "getComponents " + c + ": " + str(comps[c].conf))
print("getComponents " + c + ": " + str(comps[c].conf))
if mainfct in comps[c].conf["function"]:
out.append(c)
return out
class ComponentManager:
__instance = None
"""
initializes the Manager with all necessary components
"""
def __init__(self):
job = ulrich.program.Job.getInstance()
job.m.logDebug("applicationscomponente -- " + str(type(job.par)) )
job.m.logDebug("applicationscomponente -- " + str(type(job.par)))
self.components = {}
def initComponents(self):
@ -34,37 +50,24 @@ class ComponentManager:
# afterwards set components from parameter-file
job = ulrich.program.Job.getInstance()
anw = job.par.application
job.m.logDebug("applicationscomponente -- " + str(type(job.par)) )
job.m.logDebug("applicationscomponente -- " + str(type(job.par)))
if not job.conf.confs["applicationen"].get(anw):
job.m.setFatal("application " + job.par.application + " is not configured")
return
for k in job.conf.confs["applicationen"].get(anw):
job.m.logDebug("applicationscomponente -- " + k + ":" )
job.m.logDebug("applicationscomponente -- " + k + ":")
self.createComponents(k, 0, "")
def setComponents(self):
# set components from parameter-file
job = ulrich.program.Job.getInstance()
job.m.logDebug("applicationscomponente -- " + str(type(job.par)) )
job.m.logDebug("applicationscomponente -- " + str(type(job.par)))
def getComponent(self, compobjname):
job = ulrich.program.Job.getInstance()
verify=-2+job.getDebugLevel("job_tool")
verify = -2 + job.getDebugLevel("job_tool")
job.debug(verify, "getComponents " + compobjname)
return components[compobjname]
def getComponents(self, mainfct):
job = ulrich.program.Job.getInstance()
verify=-2+job.getDebugLevel("job_tool")
job.debug(verify, "getComponents " + mainfct)
out = []
for c in components:
job.debug(verify, "getComponents " + c + ": " + str(components[c].conf))
print("getComponents " + c + ": " + str(components[c].conf))
if mainfct in components[c].conf["function"]:
out.append(c)
return out
return comps[compobjname]
@staticmethod
def getInstance():
@ -81,11 +84,11 @@ class ComponentManager:
* business-knowledge of the component, especially of their interfaces resp. artifacts
:param componentName: Name of the component
:param nr:
:param add:
:param suffix:
:return:
"""
job = ulrich.program.Job.getInstance()
verify=job.getDebugLevel("job_tool")
verify = job.getDebugLevel("job_tool")
componentName = componentName.lower()
print("createComponent " + componentName)
confs = utils.config_tool.getConfig("comp", componentName)
@ -118,16 +121,16 @@ class ComponentManager:
print(vars(c))
print(vars(c.m))
if readedPar is not None:
print("createComponent 5 a " + name + " : "+ str(readedPar))
print("createComponent 5 a " + name + " : " + str(readedPar))
for k in readedPar["components"][name].keys():
c.conf[k] = readedPar["components"][name][k]
components[name] = c
comps[name] = c
self.createComponent(c, nr, suffix)
else:
i = 1
print("createComponent 3 b " + componentName)
for cn in conns:
name = componentName+ "_0"+str(i)
name = componentName + "_0" + str(i)
c = class_()
c.name = name
c.conf = confs["conf"]
@ -139,11 +142,11 @@ class ComponentManager:
if readedPar is not None:
for k in readedPar["components"][name].keys():
c.conf[k] = readedPar["components"][name][k]
components[name] = c
comps[name] = c
self.createComponent(c, i, suffix)
i = i + 1
print("createComponent 9 " + componentName)
print(components)
print(comps)
def createComponent(self, comp, nr, suffix):
job = ulrich.program.Job.getInstance()
@ -154,30 +157,39 @@ class ComponentManager:
continue
self.createComponents(c, nr, suffix)
def getComponentFolder(comp):
return comp.lower()
def getComponentModul(comp):
return comp[0:1].upper() + comp[1:].lower()
def getComponentClass(comp):
return comp[0:1].upper() + comp[1:].lower()
def getComponentPath(comp):
return "components." + getComponentFolder(comp) + "." + getComponentModul(comp)
def getComponentDict():
job = ulrich.program.Job.getInstance()
verify=-2+job.getDebugLevel("job_tool")
verify = -2 + job.getDebugLevel("job_tool")
job.debug(verify, "getComponents ")
out = {}
for c in components:
out[components[c].name] = {}
print("getCompDict " + components[c].name)
for k in components[c].confs.keys():
for c in comps:
out[comps[c].name] = {}
print("getCompDict " + comps[c].name)
for k in comps[c].confs.keys():
print("getCompDict " + k)
if isParameterSubnode(k): # "artifact" in k or "components" in k or "instance" in k:
if isParameterSubnode(k): # "artifact" in k or "components" in k or "instance" in k:
print("getCompDict -b- " + k)
out[components[c].name][k] = copy.deepcopy(components[c].confs[k])
out[comps[c].name][k] = copy.deepcopy(comps[c].confs[k])
return out
def isParameterSubnode(key):
for k in PARAM_NOSUBNODE:
if key in k:

19
ulrich/program.py
View File

@ -39,12 +39,31 @@ jobdef = {
"pfiletarget" : "tcparfile",
"basedir": "tcbase",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt" }
"test_system": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource": "tsparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt"}
"finish_testcase": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource": "tsparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt"}
"finish_testset": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource": "tsparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt"}
}
def setGlobal():
pass
class Job:
__instance = None

Write Preview
Loading…
Cancel
Save