"""
generalTestCase.py
==================
generalTestCase offers the GeneralTestCaseSub class. The class wants to represent each test case which can be described as dependencies between jobs. Jobs in these
context are atomical executables to be called from the command line or single python functions.
As first test case type it implements two methods to get respectively a job list for the computation part
and for the post processing of the test case. Plans are to implement these methods for every test case type so PAVAYO does not have to care about the type of
a test case.
GeneralTestCase aims to replace ParallelTestCase in the midterm.
"""
import collections
from . import testcase_generated as supermod
from . import testSuiteComponents as testSuiteComponents
from .abstractTestCase import AbstractTestCase
from ...jobManagement.jobs.jobList import JobList as JobManagementJobList
[docs]class GeneralTestCaseSub(supermod.GeneralTestCase, collections.Sequence, AbstractTestCase):
"""Class to represent a general test case. You can iterate over the job lists of a instance.
All possible parameters to the constructor of this class can be found in the XSD file used to generate
testcase_generated.py.
"""
[docs] def getMeshingJobList(self, options, executableDict, resourcesDict=None):
"""Collects all job objects from the jobs in the current test case instance and creates a job list of them.
:param options: options instance to extract the executables and the queue to use
:param executableDict: dictionary of executables
:type options: ArgumentParser
:type executableDict: ExecutableResources
:return: job list of all update-restart jobs
:rtype: JobManagementJobList
"""
return self.__getJoblist(options, executableDict, testSuiteComponents.JobListSub.MESHING, resourcesDict=resourcesDict)
[docs] def getRestartJobList(self, options, executableDict, resourcesDict=None):
"""Collects all job objects from the jobs in the current test case instance and creates a job list of them.
:param options: options instance to extract the executables and the queue to use
:param executableDict: dictionary of executables
:type options: ArgumentParser
:type executableDict: ExecutableResources
:return: job list of all update-restart jobs
:rtype: JobManagementJobList
"""
return self.__getJoblist(options, executableDict, testSuiteComponents.JobListSub.RESTART, resourcesDict=resourcesDict)
[docs] def getComputationJobList(self, options, executableDict, resourcesDict=None):
"""Collects all job objects from the jobs in the current test case instance and creates a job list of them.
:param options: options instance to extract the executables and the queue to use
:param executableDict: dictionary of executables
:type options: ArgumentParser
:type executableDict: ExecutableResources
:return: job list of all computation jobs
:rtype: JobManagementJobList
"""
return self.__getJoblist(options, executableDict, testSuiteComponents.JobListSub.COMPUTATION, resourcesDict=resourcesDict)
[docs] def getPostprocessingJobList(self, options, executableDict, resourcesDict=None):
"""Returns a job representing the post script.
:param options: options instance to extract the executables and the queue to use
:param executableDict: dictionary of executables
:type options: ArgumentParser
:type executableDict: ExecutableResources
:return: job list of all post-processing jobs
:rtype: JobManagementJobList
"""
return self.__getJoblist(options, executableDict, testSuiteComponents.JobListSub.POSTPROCESSING, resourcesDict=resourcesDict)
def __getJoblist(self, options, executableDict, step, parents=None, resourcesDict=None):
"""Creates the job list for the defined step.
:param options: a options instance to extract the executables and the queue to use
:param executableDict: dictionary of executables
:param step: the step for which the job list is wanted
:param parents: job ids of parent jobs
:type options: OptionParser instance
:type executableDict: ExecutableResources
:type step: str
:type parents: list or None
:return: the job list representing the given step
:rtype: jobList or None
"""
if [jobList for jobList in self if jobList.step == step]:
myJobList = JobManagementJobList(name=self.name + "_JOBLIST", verbosity=options.verbose,
retries=0 if step != testSuiteComponents.JobListSub.COMPUTATION else options.retriesComputation,
deactivateJobGrouping=options.deactivateClusterJobGrouping)
selectedSimDict = self.findSubsetInJobList("", "jobList", "job", options.selectiveCompute, step=step) if options.selectiveCompute else None
jobs = self.createJobListDictionary(options, executableDict, "", "jobList", "job", self._getWorkingDirForJob,
step=step, selectDict=selectedSimDict, resourcesDict=resourcesDict)
self.setDependenciesInJobList(myJobList, jobs, parents=parents)
return myJobList
[docs] def append(self, jobListObject):
"""Appends a job list to this test case
:param jobListObject: a test case job list to append
:type jobListObject: TestcaseJoblist instance
"""
self.jobList.append(jobListObject)
[docs] def insert(self, index, value):
"""Inserts a job list to this test case
:param index: place to insert in
:param value: a test case job list to append
:type index: integer
:type value: TestcaseJoblist instance
"""
self.jobList.insert(index, value)
[docs] def getUsedExecutables(self, compute=False, restart=False, meshing=False, postprocessing=False):
"""Returns the version of a software the test case was computed with.
:param compute: add executables of compute step to set?
:param restart: add executables of restart step to set?
:param meshing: add executables of meshing step to set?
:param postprocessing: add executables of postprocessing step to set?
:type compute: bool
:type restart: bool
:type meshing: bool
:type postprocessing: bool
:return: the used executables
:rtype: set(str)
"""
usedExecutables = []
for jobList in self.jobList:
if (compute and jobList.step == testSuiteComponents.JobListSub.COMPUTATION) or \
(restart and jobList.step == testSuiteComponents.JobListSub.RESTART) or \
(meshing and jobList.step == testSuiteComponents.JobListSub.MESHING) or \
(postprocessing and jobList.step == testSuiteComponents.JobListSub.POSTPROCESSING):
for job in jobList:
if job.executable not in usedExecutables:
usedExecutables.append(job.executable)
return set(usedExecutables)
def __len__(self):
"""Returns the number of operation point lists in the test case.
:return: number of operation point lists
:rtype: int
"""
return len(self.jobList)
def __getitem__(self, index):
"""Returns a operation point list at a given index.
:return: operation point list at given index
:rtype: operation point list instance
"""
return self.jobList[index]
