1#!/usr/local/bin/python3.8 2from __future__ import print_function 3import shutil 4import unittest 5from ruffus.drmaa_wrapper import write_job_script_to_temp_file, read_stdout_stderr_from_files 6import ruffus.drmaa_wrapper 7import ruffus 8import sys 9 10""" 11 12 test_drmaa_wrapper_run_job_locally.py 13 14""" 15 16import os 17script_dir = os.path.abspath(os.path.dirname(__file__)) 18tempdir = os.path.relpath(os.path.abspath(os.path.splitext(__file__)[0])) 19 20# add grandparent to search path for testing 21grandparent_dir = os.path.abspath( 22 os.path.join(os.path.dirname(__file__), "..", "..")) 23sys.path.insert(0, grandparent_dir) 24 25# module name = script name without extension 26module_name = os.path.splitext(os.path.basename(__file__))[0] 27 28 29# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888 30 31# imports 32 33 34# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888 35 36 37# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888 38 39# Main logic 40 41 42# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888 43class Test_ruffus(unittest.TestCase): 44 45 class t_test_logger: 46 47 """ 48 Does nothing! 49 """ 50 51 def __init__(self): 52 self.clear() 53 54 def clear(self): 55 self.info_msg = [] 56 self.debug_msg = [] 57 self.warning_msg = [] 58 self.error_msg = [] 59 60 def info(self, message, *args, **kwargs): 61 self.info_msg.append(message) 62 63 def debug(self, message, *args, **kwargs): 64 self.debug_msg.append(message) 65 66 def warning(self, message, *args, **kwargs): 67 self.warning_msg.append(message) 68 69 def error(self, message, *args, **kwargs): 70 self.error_msg.append(message) 71 72 # ___________________________________________________________________________ 73 # 74 # setup and cleanup 75 # ___________________________________________________________________________ 76 77 def setUp(self): 78 try: 79 os.mkdir(tempdir) 80 except OSError: 81 pass 82 83 def tearDown(self): 84 shutil.rmtree(tempdir) 85 86 def test_read_stdout_stderr_from_files(self): 87 # 88 # Test missing stdout and stderr files 89 # 90 stdout_path = os.path.join(tempdir, "stdout.txt") 91 stderr_path = os.path.join(tempdir, "stderr.txt") 92 logger = Test_ruffus.t_test_logger() 93 read_stdout_stderr_from_files( 94 stdout_path, stderr_path, logger, cmd_str="test_cmd", tries=0) 95 self.assertTrue("could not open stdout" in "".join(logger.warning_msg)) 96 self.assertTrue("could not open stderr" in "".join(logger.warning_msg)) 97 logger.clear() 98 99 # 100 # Test present stdout and stderr files 101 # 102 with open(stdout_path, "w") as so: 103 so.write("STDOUT\nSTDOUT\n") 104 with open(stderr_path, "w") as se: 105 se.write("STDERR\nSTDERR\n") 106 107 stdout_msg, stderr_msg = read_stdout_stderr_from_files( 108 stdout_path, stderr_path, logger, cmd_str="test_cmd", tries=1) 109 self.assertEqual(logger.warning_msg, []) 110 self.assertEqual(stdout_msg, ["STDOUT\n", "STDOUT\n"]) 111 self.assertEqual(stderr_msg, ["STDERR\n", "STDERR\n"]) 112 113 def test_run_job(self): 114 environ = {"RUFFUS_HEEHEE": "what?"} 115 home_dir = os.path.expanduser("~") 116 sys.stderr.write(" Run echoing to screen...\n") 117 stdout, stderr = ruffus.drmaa_wrapper.run_job(cmd_str="python %s/slow_process_for_testing.py" % script_dir, 118 job_environment=environ, 119 working_directory=home_dir, 120 run_locally=True, 121 verbose=1, 122 local_echo=True) 123 sys.stderr.write(" Run silently...\n") 124 stdout, stderr = ruffus.drmaa_wrapper.run_job(cmd_str="python %s/slow_process_for_testing.py" % script_dir, 125 job_environment=environ, 126 working_directory=home_dir, 127 run_locally=True, 128 verbose=1, 129 local_echo=False) 130 self.assertEqual( 131 stdout, 132 [' Stdout 0\n', ' Stdout 1\n', ' Stdout 2\n', ' Stdout 3\n']) 133 134 stderr_fixed = [stderr[x] for x in (0, 2, 3, 4, 5)] 135 stderr_variable = stderr[1] 136 self.assertEqual( 137 stderr_fixed, 138 [' %s\n' % home_dir, 139 ' Stderr 0\n', 140 ' Stderr 1\n', 141 ' Stderr 2\n', 142 ' Stderr 3\n']) 143 self.assertTrue("'PWD': '{}'".format(home_dir) in stderr_variable) 144 self.assertTrue("'RUFFUS_HEEHEE': 'what?'" in stderr_variable) 145 146 def test_write_job_script_to_temp_file(self): 147 sys.stderr.write(" Write to temp_file...\n") 148 job_script_path, stdout_path, stderr_path = write_job_script_to_temp_file( 149 "ls", None, "job_name", "", None, None) 150 os.unlink(job_script_path) 151 job_script_path, stdout_path, stderr_path = write_job_script_to_temp_file( 152 "ls", tempdir, "job_name", "", None, None) 153 154 def test_ls(self): 155 sys.stderr.write(" ls...\n") 156 with open(os.path.join(tempdir, "temp.txt"), "w") as oo: 157 oo.write("done") 158 stdout, stderr = ruffus.drmaa_wrapper.run_job(cmd_str="ls %s" % tempdir, 159 run_locally=True, 160 verbose=1, 161 local_echo=True) 162 self.assertEqual(stdout, ['temp.txt\n']) 163 164 165if __name__ == '__main__': 166 unittest.main() 167