Completed
Pull Request — master (#104)
by Marc-Alexandre
45s
created

tests.TestSmartWorker.test_main()   D

Complexity

Conditions 11

Size

Total Lines 59

Duplication

Lines 0
Ratio 0 %
Metric Value
cc 11
dl 0
loc 59
rs 4.2187

How to fix   Long Method    Complexity   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

Complexity

Complex classes like tests.TestSmartWorker.test_main() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
import os
2
import unittest
3
import tempfile
4
import time
5
import shutil
6
7
from smartdispatch import utils
8
from smartdispatch.command_manager import CommandManager
9
10
from subprocess import Popen, call, PIPE
11
12
from nose.tools import assert_true, assert_equal
13
14
15
class TestSmartWorker(unittest.TestCase):
16
17
    def setUp(self):
18
        self.commands = ["echo 1", "echo 2", "echo 3", "echo 4"]
19
        self._commands_dir = tempfile.mkdtemp()
20
        self.logs_dir = tempfile.mkdtemp()
21
22
        self.command_manager = CommandManager(os.path.join(self._commands_dir, "commands.txt"))
23
        self.command_manager.set_commands_to_run(self.commands)
24
25
        self.commands_uid = map(utils.generate_uid_from_string, self.commands)
26
27
    def tearDown(self):
28
        shutil.rmtree(self._commands_dir)
29
        shutil.rmtree(self.logs_dir)
30
31
    def test_main(self):
32
        command = ["smart_worker.py", self.command_manager._commands_filename, self.logs_dir]
33
        assert_equal(call(command), 0)
34
        # Simulate a resume, i.e. re-run the command, the output/error should be concatenated.
35
        self.command_manager.set_commands_to_run(self.commands)
36
        assert_equal(call(command), 0)
37
38
        # Check output logs
39
        filenames = os.listdir(self.logs_dir)
40
        outlogs = [os.path.join(self.logs_dir, filename) for filename in filenames if filename.endswith(".out")]
41
        for log_filename in outlogs:
42
            with open(log_filename) as logfile:
43
                # From log's filename (i.e. uid) retrieve executed command associated with this log
44
                uid = os.path.splitext(os.path.basename(log_filename))[0]
45
                executed_command = self.commands[self.commands_uid.index(uid)]
46
47
                # Since the command was run twice.
48
                for _ in range(2):
49
                    # First line is the datetime of the executed command in comment.
50
                    line = logfile.readline().strip()
51
                    assert_true(time.strftime("## %Y-%m-%d %H:%M:") in line)  # Don't check seconds.
52
53
                    # Second line is the executed command in comment.
54
                    line = logfile.readline().strip()
55
                    assert_equal(line, "# " + executed_command)
56
57
                    # Next should be the command's output
58
                    line = logfile.readline().strip()
59
                    assert_equal(line, executed_command[-1])  # We know those are 'echo' of a digit
60
61
                    # Empty line
62
                    assert_equal(logfile.readline().strip(), "")
63
64
                # Log should be empty now
65
                assert_equal("", logfile.read())
66
67
        # Check error logs
68
        errlogs = [os.path.join(self.logs_dir, filename) for filename in filenames if filename.endswith(".err")]
69
        for log_filename in errlogs:
70
            with open(log_filename) as logfile:
71
                # From log's filename (i.e. uid) retrieve executed command associated with this log
72
                uid = os.path.splitext(os.path.basename(log_filename))[0]
73
                executed_command = self.commands[self.commands_uid.index(uid)]
74
75
                # Since the command was run twice.
76
                for _ in range(2):
77
                    # First line is the datetime of the executed command in comment.
78
                    line = logfile.readline().strip()
79
                    assert_true(time.strftime("## %Y-%m-%d %H:%M:") in line)  # Don't check seconds.
80
81
                    # Second line is the executed command in comment.
82
                    line = logfile.readline().strip()
83
                    assert_equal(line, "# " + executed_command)
84
85
                    # Empty line
86
                    assert_equal(logfile.readline().strip(), "")
87
88
                # Log should be empty now
89
                assert_equal("", logfile.read())
90
91
    def test_lock(self):
92
        command = ["smart_worker.py", self.command_manager._commands_filename, self.logs_dir]
93
94
        # Lock the commands file before running 'smart_worker.py'
95
        with utils.open_with_lock(self.command_manager._commands_filename, 'r+'):
96
            process = Popen(command, stdout=PIPE, stderr=PIPE)
97
            time.sleep(1)
98
99
        stdout, stderr = process.communicate()
100
        assert_equal(stdout, "")
101
        assert_true("write-lock" in stderr, msg="Forcing a race condition, try increasing sleeping time above.")
102
        assert_true("Traceback" not in stderr)  # Check that there are no errors.
103