Commit 77e75f61 authored by mancini's avatar mancini

test for slurm_control list jobs command

parent 6aa152a1
from setuptools import setup
import unittest
def test_suite():
test_loader = unittest.TestLoader()
test_suite = test_loader.discover('tests', pattern='test_*.py')
return test_suite
setup(
name="slurm-executor-plugin",
......@@ -7,6 +15,7 @@ setup(
url='https://git.astron.nl/eosc/slurmexecutorplugin/',
package_dir = {'': 'lib'},
packages = ['slurm_cli', 'slurm_executor'],
test_suite = 'setup.test_suite',
entry_points = {
'airflow.plugins': [
'slurm = slurm_executor.slurm:SlurmExecutorPlugin'
......
import slurm_cli.slurm_control as slurm_control
from slurm_cli.jobs import SlurmJobStatus
import unittest
from unittest.mock import patch, MagicMock
from subprocess import CompletedProcess
import logging
def get_mocked_call_output(status_code, stdout_string, stderr_string):
mocked_run_return_value = MagicMock(CompletedProcess)
mocked_run_return_value.stdout = stdout_string
mocked_run_return_value.stderr = stderr_string
mocked_run_return_value.returncode = status_code
return mocked_run_return_value
class TestSlurmControl(unittest.TestCase):
@patch('slurm_cli.slurm_control.run_process')
def test_get_job_statuses_job_queue_empty(self, call_mock):
output_string = ''
call_mock.return_value = get_mocked_call_output(0, output_string, '')
jobs = slurm_control.get_jobs_status()
self.assertEqual(len(jobs), 0)
@patch('slurm_cli.slurm_control.run_process')
def test_get_job_statuses_job_queue_filled(self, call_mock):
output_string = '123;test_job;CD;COMPLETED;None\n124;test_job;F;FAILED;error 1\n'
expected_parsed_results = [
SlurmJobStatus(job_id='123', job_name='test_job', status_code='CD', status='COMPLETED', reason='None'),
SlurmJobStatus(job_id='124', job_name='test_job', status_code='F', status='FAILED', reason='error 1')
]
call_mock.return_value = get_mocked_call_output(0, output_string, '')
jobs = slurm_control.get_jobs_status()
self.assertEqual(len(jobs), 2)
for job_status in expected_parsed_results:
self.assertIn(job_status.job_id, jobs)
self.assertEqual(jobs[job_status.job_id], job_status)
@patch('slurm_cli.slurm_control.run_process')
def test_get_job_statuses_command_non_zero_exit_code(self, call_mock):
output_string = '123;test_job;CD;COMPLETED;None\n124;test_job;F;FAILED;error 1\n'
call_mock.return_value = get_mocked_call_output(1, output_string, 'stuff')
with self.assertRaises(slurm_control.SlurmCallError):
_ = slurm_control.get_jobs_status()
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s-%(name)s-%(levelname)s %(message)s')
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment