Belle II Software  release-05-01-25
test_clustercontroldrmaa.py
1 #!/usr/bin/env python3
2 # -*- coding: utf-8 -*-
3 
4 import unittest
5 import sys
6 import tempfile
7 import os
8 from enum import Enum
9 from unittest.mock import MagicMock
10 from clustercontroldrmaa import Cluster
11 from validationscript import Script
12 
13 
14 class TestClusterControlDrmaa(unittest.TestCase):
15  """
16  Test for for the DRMAA-backend of the clustercontrol
17  """
18  class SessionMock:
19  """
20  Class to mock a DRMAA session
21  """
22 
23  class JobState(Enum):
24  """
25  Possible DRMAA Job States
26  """
27 
28  DONE = 1
29 
30  FAILED = 2
31 
32  RUNNING = 3
33 
34 
35  jobStatusReturn = JobState.RUNNING
36 
37  def __enter__(self):
38  """ to support python with syntax"""
39  return self
40 
41  def __exit__(self, exc, value, tb):
42  """ to support python with syntax"""
43  pass
44 
45  def createJobTemplate(self):
46  """fake creating job template"""
47  return MagicMock()
48 
49  def runJob(self, jt):
50  """fake job running"""
51  return MagicMock()
52 
53  def jobStatus(self, jt):
54  """fake job status terun"""
55  return self.jobStatusReturn
56 
57  def deleteJobTemplate(self, jt):
58  """fake job deletion"""
59  return MagicMock()
60 
61  def test_no_drmaa(self):
62  """
63  Overwrite drmaa module and see what happens ...
64  """
65  sys.modules['drmaa'] = None
66  self.assertFalse(Cluster.is_supported())
67 
69  """
70  Test to submit and monitor a regular job
71  """
72 
73  with tempfile.TemporaryDirectory() as td:
74  # switch to temp folder as the cluster controller will create
75  # some helper files
76  os.chdir(str(td))
77 
78  drmaa_mock = MagicMock()
79  drmaa_mock.Session = self.SessionMock # MagicMock(return_value=session_mock)
80  drmaa_mock.JobState = self.SessionMock.JobState
81 
82  sys.modules['drmaa'] = drmaa_mock
83 
84  # this just imports the drmaa module which should always work with our
85  # mock
86  self.assertTrue(Cluster.is_supported())
87 
88  job = Script(path="myscript1.py", package="test_package", log=None)
89 
90  cc = Cluster()
91  cc.execute(job)
92 
93  # check if job id has been set
94  self.assertTrue(job.job_id)
95 
96  # check on job, not finished yet
97  self.assertFalse(cc.is_job_finished(job)[0])
98 
99  # check on job, finished !
100  # change behaviour of jobStatus
101  self.SessionMock.jobStatusReturn = self.SessionMock.JobState.DONE
102 
103  self.assertTrue(cc.is_job_finished(job)[0])
104 
105  cc.terminate(job)
106 
107 
108 if __name__ == "__main__":
109  unittest.main()
test_clustercontroldrmaa.TestClusterControlDrmaa.test_no_drmaa
def test_no_drmaa(self)
Definition: test_clustercontroldrmaa.py:61
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.jobStatusReturn
jobStatusReturn
the job state this session mock will return
Definition: test_clustercontroldrmaa.py:35
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.__enter__
def __enter__(self)
Definition: test_clustercontroldrmaa.py:37
test_clustercontroldrmaa.TestClusterControlDrmaa
Definition: test_clustercontroldrmaa.py:14
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.runJob
def runJob(self, jt)
Definition: test_clustercontroldrmaa.py:49
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.deleteJobTemplate
def deleteJobTemplate(self, jt)
Definition: test_clustercontroldrmaa.py:57
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.__exit__
def __exit__(self, exc, value, tb)
Definition: test_clustercontroldrmaa.py:41
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.jobStatus
def jobStatus(self, jt)
Definition: test_clustercontroldrmaa.py:53
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock
Definition: test_clustercontroldrmaa.py:18
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.JobState
Definition: test_clustercontroldrmaa.py:23
clustercontroldrmaa.Cluster
Definition: clustercontroldrmaa.py:11
test_clustercontroldrmaa.TestClusterControlDrmaa.test_submit_and_monitor_job
def test_submit_and_monitor_job(self)
Definition: test_clustercontroldrmaa.py:68
test_clustercontroldrmaa.TestClusterControlDrmaa.SessionMock.createJobTemplate
def createJobTemplate(self)
Definition: test_clustercontroldrmaa.py:45