nose -> py.test

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+addopts = --cov tomahawk --cov-report html

tests/internal2/test_capture.py

@@ -0,0 +1,16 @@
+import logging
+
+def test_ng(capsys):
+    logging.basicConfig(
+        level = logging.DEBUG)
+    logging.debug("hiho")
+    out, err = capsys.readouterr()
+    assert 1 == 2
+
+def test_ok(capsys):
+    logging.basicConfig(
+        filename = '/tmp/log',
+        level = logging.DEBUG)
+    logging.debug("hiho")
+    out, err = capsys.readouterr()
+    assert 1 == 2

tests/internal2/test_command.py

@@ -1,21 +1,140 @@
 import argparse
-#import pytest
+import re
 import utils
+
 utils.append_home_to_path(__file__)
 
 from tomahawk.command import CommandMain
-import tomahawk.expect
+from tomahawk.constants import TimeoutError
+from tomahawk.expect import CommandWithExpect
 
 def test_00_run(monkeypatch):
+    EXPECTED = {
+        'command': 'uptime',
+        'command_output': "0:40  up 1 day,  8:19, 4 users, load averages: 0.00 0.50 1.00",
+        'exit_status': 0,
+    }
+    stdout = utils.StdoutCapture()
+    stdout.start()
+
+    def mock_parse_args(self):
+        return utils.create_command_namespace(command = [ EXPECTED['command'] ])
+    monkeypatch.setattr(argparse.ArgumentParser, 'parse_args', mock_parse_args)
+
+    def mock_execute(self):
+        return EXPECTED['exit_status'], EXPECTED['command_output']
+    monkeypatch.setattr(CommandWithExpect, 'execute', mock_execute)
+
+    main = CommandMain('tomahawk')
+    status = main.run()
+    o = stdout.stop().captured_value()
+
+    assert status == 0
+    s = \
+"""tomahawk@localhost %% %(command)s
+%(command_output)s
+
+""" % EXPECTED
+    assert o == s
+
+def test_01_run_error(monkeypatch):
+    EXPECTED = {
+        'command': 'command_not_found',
+        'exit_status': 127,
+    }
+    stderr = utils.StderrCapture()
+    stderr.start()
+
+    def mock_parse_args(self):
+        return utils.create_command_namespace(command = [ EXPECTED['command'] ])
+    monkeypatch.setattr(argparse.ArgumentParser, 'parse_args', mock_parse_args)
+
+    def mock_execute(self):
+        return EXPECTED['exit_status'], "/bin/sh: command_not_found: command not found"
+    monkeypatch.setattr(CommandWithExpect, 'execute', mock_execute)
+
+    main = CommandMain('tomahawk')
+    status = main.run()
+    err = stderr.stop().captured_value()
+
+    assert status == 1
+    assert re.search(r'failed on host', err)
+
+def test_02_run_timeout(monkeypatch):
+    EXPECTED = {
+        'command': 'sleep 3',
+        'command_output': "/bin/sh: command_not_found: command not found",
+    }
+    stderr = utils.StderrCapture()
+    stderr.start()
+
+    def mock_parse_args(self):
+        return utils.create_command_namespace(
+            command = [ EXPECTED['command'] ], timeout = 1
+        )
+    monkeypatch.setattr(argparse.ArgumentParser, 'parse_args', mock_parse_args)
+
+    def mock_execute(self):
+        raise TimeoutError()
+    monkeypatch.setattr(CommandWithExpect, 'execute', mock_execute)
+
+    main = CommandMain('tomahawk')
+    status = main.run()
+    err = stderr.stop().captured_value()
+    assert status == 1
+    assert re.search(r'timed out on host', err)
+
+def test_03_run_option_host_files(monkeypatch):
+    EXPECTED = {
+        'command': 'echo "hello world"',
+        'command_output': "hello world",
+    }
+    stdout = utils.StdoutCapture()
+    stdout.start()
+
     def mock_parse_args(self):
-        return utils.create_argparse_namespace(command = [ 'uptime' ])
+        return utils.create_command_namespace(
+            command = [ EXPECTED['command'] ],
+            hosts = 'localhost,localhost',
+        )
     monkeypatch.setattr(argparse.ArgumentParser, 'parse_args', mock_parse_args)
 
     def mock_execute(self):
-        return 0, "0:40  up 1 day,  8:19, 4 users, load averages: 0.00 0.50 1.00"
-    monkeypatch.setattr(tomahawk.expect.CommandWithExpect, 'execute', mock_execute)
+        return 0, EXPECTED['command_output']
+    monkeypatch.setattr(CommandWithExpect, 'execute', mock_execute)
 
     main = CommandMain('tomahawk')
     status = main.run()
+    out = stdout.stop().captured_value()
+    assert status == 0
+    assert re.search(r'hello world', out)
+
+def test_04_run_option_continue_on_error(monkeypatch):
+    EXPECTED = {
+        'command': 'failure_command',
+        'command_output': "hello world",
+    }
+    stderr = utils.StderrCapture()
+    stderr.start()
+
+    def mock_parse_args(self):
+        return utils.create_command_namespace(
+            command = [ EXPECTED['command'] ],
+            continue_on_error = True,
+            hosts = 'localhost,localhost',
+        )
+    monkeypatch.setattr(argparse.ArgumentParser, 'parse_args', mock_parse_args)
+
+    def mock_execute(self):
+        return 127, EXPECTED['command_output']
+    monkeypatch.setattr(CommandWithExpect, 'execute', mock_execute)
+
+    main = CommandMain('tomahawk')
+    status = main.run()
+    err = stderr.stop().captured_value()
     assert status == 1
+    assert len(err.split('\n')) == 4
+
+def test_05_execute_option_paralle_continue_on_error():
+    pass
 

tests/internal2/utils.py

@@ -17,13 +17,13 @@ def get_bin_dir(file):
 def append_home_to_path(file):
     sys.path.insert(0, get_home_dir(file))
 
-def create_argparse_namespace(**kwargs):
+def create_command_namespace(**kwargs):
     defaults = {
         'command': [ '' ], 'continue_on_error': None,
-        'debug': True, 'deep_debug': True,
+        'debug': False, 'deep_debug': False,
         'delay': 0, 'expect_delay': 0.1,
-        'hosts': 'localhost', 'profile': False,
-        'timeout': 10
+        'output_file': None, 'hosts': 'localhost', 'profile': False,
+        'ssh_user': 'tomahawk',  'timeout': 10
     }
     for k, v in defaults.iteritems():
         kwargs.setdefault(k, v)
@@ -36,6 +36,36 @@ def create_argparse_namespace(**kwargs):
     DEFAULT_EXPECT_DELAY
 )
 
+class StdoutCapture(object):
+    def __init__(self):
+        self.captured = cStringIO.StringIO()
+
+    def start(self):
+        sys.stdout = self.captured
+        return self
+
+    def stop(self):
+        sys.stdout = sys.__stdout__
+        return self
+
+    def captured_value(self):
+        return self.captured.getvalue()
+
+    def close(self):
+        self.captured.close()
+
+class StderrCapture(StdoutCapture):
+    def __init__(self):
+        super(StderrCapture, self).__init__()
+
+    def start(self):
+        sys.stderr = self.captured
+        return self
+
+    def stop(self):
+        sys.stderr = sys.__stderr__
+        return self
+
 class MockPexpect(object):
     def __init__(
         self, command, args = [], timeout = 30, maxread = 2000,

tomahawk/base.py

@@ -40,7 +40,11 @@ def __init__(self, script_path):
         self.script_path = script_path
         self.arg_parser = self.create_argument_parser(script_path)
         self.options = self.arg_parser.parse_args()
-        self.log = create_logger(self.options.debug or self.options.deep_debug, self.options.deep_debug)
+        self.log = create_logger(
+            None,
+            self.options.debug or self.options.deep_debug,
+            self.options.deep_debug
+        )
 
     def run(self):
         try:

tomahawk/log.py

@@ -1,15 +1,19 @@
 import logging
+import sys
 
-def create_logger(debug_enabled=False, deep_debug_enabled=False):
+def create_logger(file_path = None, debug_enabled=False, deep_debug_enabled=False):
     level = logging.INFO
     if debug_enabled:
         level = logging.DEBUG
     if deep_debug_enabled:
         format = "[%(levelname)s] %(filename)s:%(lineno)d %(message)s"
     else:
         format = "[%(levelname)s] %(message)s"
-    logging.basicConfig(
-        level = level,
-        format = format
-    )
+
+    kwargs = { 'level': level, 'format': format }
+    if file_path:
+        kwargs['filename'] = file_path
+    else:
+        kwargs['stream'] = sys.stdout
+    logging.basicConfig(**kwargs)
     return logging