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