TestSmartWorker.test_main() - Code Metrics - Inspection of "Merge pull request #117 from MarcCote/rename_smard..." - SMART-Lab/smartdispatch - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 0b9edb...ab2b1a )

by Mathieu

created 2016-04-12 21:10 UTC

TestSmartWorker.test_main() F

↳ Parent: TestSmartWorker

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	13
dl	0
loc	73
rs	2.3081

How to fix Long Method Complexity

import os
import unittest
import tempfile
import time
import shutil

import smartdispatch
from smartdispatch import utils
from smartdispatch.filelock import open_with_lock
from smartdispatch.command_manager import CommandManager

from subprocess import Popen, call, PIPE

from nose.tools import assert_true, assert_equal


class TestSmartWorker(unittest.TestCase):

    def setUp(self):
        self.base_worker_script = os.path.join(os.path.dirname(smartdispatch.__file__), 'workers', 'base_worker.py')
        self.commands = ["echo 1", "echo 2", "echo 3", "echo 4"]
        self._commands_dir = tempfile.mkdtemp()
        self.logs_dir = tempfile.mkdtemp()

        self.command_manager = CommandManager(os.path.join(self._commands_dir, "commands.txt"))
        self.command_manager.set_commands_to_run(self.commands)

        self.commands_uid = map(utils.generate_uid_from_string, self.commands)

    def tearDown(self):
        shutil.rmtree(self._commands_dir)
        shutil.rmtree(self.logs_dir)

    def test_main(self):
        command = ['python2', self.base_worker_script, self.command_manager._commands_filename, self.logs_dir]
        assert_equal(call(command), 0)
        # Simulate a resume, i.e. re-run the command, the output/error should be concatenated.
        self.command_manager.set_commands_to_run(self.commands)
        assert_equal(call(command), 0)

        # Check output logs
        filenames = os.listdir(self.logs_dir)
        outlogs = [os.path.join(self.logs_dir, filename) for filename in filenames if filename.endswith(".out")]
        for log_filename in outlogs:
            with open(log_filename) as logfile:
                # From log's filename (i.e. uid) retrieve executed command associated with this log
                uid = os.path.splitext(os.path.basename(log_filename))[0]
                executed_command = self.commands[self.commands_uid.index(uid)]

                # Since the command was run twice.
                for i in range(2):
                    # First line is the datetime of the executed command in comment.
                    line = logfile.readline().strip()

                    if i == 0:
                        assert_true("Started" in line)
                    else:
                        assert_true("Resumed" in line)

                    assert_true(line.startswith("## SMART-DISPATCH"))
                    assert_true(time.strftime("%Y-%m-%d %H:%M:") in line)  # Don't check seconds.

                    # Second line is the executed command in comment.
                    line = logfile.readline().strip()
                    assert_true(executed_command in line)

                    # Next should be the command's output
                    line = logfile.readline().strip()
                    assert_equal(line, executed_command[-1])  # We know those are 'echo' of a digit

                    # Empty line
                    assert_equal(logfile.readline().strip(), "")

                # Log should be empty now
                assert_equal("", logfile.read())

        # Check error logs
        errlogs = [os.path.join(self.logs_dir, filename) for filename in filenames if filename.endswith(".err")]
        for log_filename in errlogs:
            with open(log_filename) as logfile:
                # From log's filename (i.e. uid) retrieve executed command associated with this log
                uid = os.path.splitext(os.path.basename(log_filename))[0]
                executed_command = self.commands[self.commands_uid.index(uid)]

                # Since the command was run twice.
                for i in range(2):
                    # First line is the datetime of the executed command in comment.
                    line = logfile.readline().strip()

                    if i == 0:
                        assert_true("Started" in line)
                    else:
                        assert_true("Resumed" in line)

                    assert_true(line.startswith("## SMART-DISPATCH"))
                    assert_true(time.strftime("%Y-%m-%d %H:%M:") in line)  # Don't check seconds.

                    # Second line is the executed command in comment.
                    line = logfile.readline().strip()
                    assert_true(executed_command in line)

                    # Empty line
                    assert_equal(logfile.readline().strip(), "")

                # Log should be empty now
                assert_equal("", logfile.read())

    def test_lock(self):
        command = ['python2', self.base_worker_script, self.command_manager._commands_filename, self.logs_dir]

        # Lock the commands file before running 'base_worker.py'
        with open_with_lock(self.command_manager._commands_filename, 'r+'):
            process = Popen(command, stdout=PIPE, stderr=PIPE)
            time.sleep(1)

        stdout, stderr = process.communicate()
        assert_equal(stdout, "")
        assert_true("write-lock" in stderr, msg="Forcing a race condition, try increasing sleeping time above.")
        assert_true("Traceback" not in stderr)  # Check that there are no errors.


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

SMART-Lab / smartdispatch

Push — master ( 0b9edb...ab2b1a )

TestSmartWorker.test_main() F

Complexity

Size

Duplication

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like