TestClusterControlDrmaa Class Reference

Inheritance diagram for TestClusterControlDrmaa:

Classes
class	SessionMock

Public Member Functions
def	test_no_drmaa (self)
def	test_submit_and_monitor_job (self)

Detailed Description

Test for for the DRMAA-backend of the clustercontrol

Definition at line 21 of file test_clustercontroldrmaa.py.

Member Function Documentation

test_no_drmaa()

def test_no_drmaa

(

self

)

Overwrite drmaa module and see what happens ...

Definition at line 69 of file test_clustercontroldrmaa.py.

    def test_no_drmaa(self):
        """
        Overwrite drmaa module and see what happens ...
        """
        sys.modules["drmaa"] = None
        self.assertFalse(Cluster.is_supported())
 

test_submit_and_monitor_job()

def test_submit_and_monitor_job

(

self

)

Test to submit and monitor a regular job

Definition at line 76 of file test_clustercontroldrmaa.py.

    def test_submit_and_monitor_job(self):
        """
        Test to submit and monitor a regular job
        """
 
        with tempfile.TemporaryDirectory() as td:
            # switch to temp folder as the cluster controller will create
            # some helper files
            os.chdir(str(td))
 
            drmaa_mock = MagicMock()
            drmaa_mock.Session = (
                self.SessionMock
            )  # MagicMock(return_value=session_mock)
            drmaa_mock.JobState = self.SessionMock.JobState
 
            sys.modules["drmaa"] = drmaa_mock
 
            # this just imports the drmaa module which should always work with our
            # mock
            self.assertTrue(Cluster.is_supported())
 
            job = Script(path="myscript1.py", package="test_package", log=None)
 
            cc = Cluster()
            cc.execute(job)
 
            # check if job id has been set
            self.assertTrue(job.job_id)
 
            # check on job, not finished yet
            self.assertFalse(cc.is_job_finished(job)[0])
 
            # check on job, finished !
            # change behaviour of jobStatus
            self.SessionMock.jobStatusReturn = self.SessionMock.JobState.DONE
 
            self.assertTrue(cc.is_job_finished(job)[0])
 
            cc.terminate(job)
 
 

---

The documentation for this class was generated from the following file:

• validation/tests/test_clustercontroldrmaa.py