Source

bu-main / tests / test_execute.py

Full commit

from py.test import raises, mark, config

import os

from bu.errors import TargetNotFound, UndefinedReference, CircularDependency
from bu.parsing import Parser
from bu.execute import ExecutionQueue, ScriptExecutionGraph
from bu.actions import sh_action, py_action


def pytest_funcarg__parser(request):
    return Parser()


def pytest_funcarg__queue(request):
    def create_queue(s, name):
        p = request.getfuncargvalue('parser')
        script = p.parse_string(s)
        return ExecutionQueue(script.targets.get(name))
    return create_queue


def pytest_funcarg__graph(request):
    def create_graph(s, name):
        p = request.getfuncargvalue('parser')
        script = p.parse_string(s)
        return ScriptExecutionGraph(script, name)
    return create_graph


def test_execution_queue(queue):
    q = queue('foo:\n moo', 'foo')
    assert len(q) == 1


def test_execution_queue_missing_target(queue):
    raises(ValueError, queue, 'foo:\n moo', 'goo')


def test_execution_graph_missing_target(graph):
    raises(TargetNotFound, graph, 'foo:\n moo', 'goo')


def test_execution_graph_execute(graph):
    g = graph('foo:\n echo hello', 'foo')
    g.execute()
    g.join()
    assert g.completed == g.nodes


def test_execution_graph_circular_self(graph):
    raises(CircularDependency, graph, 'foo deps=foo:\n echo hello', 'foo')


def test_execution_graph_circular(graph):
    raises(CircularDependency, graph, 'foo deps=goo:\n echo hello\ngoo deps=foo:\n echo hello', 'foo')


def test_execution_queue_node(queue):
    q = queue('foo:\n moo', 'foo')
    node = q.get()
    assert node.action.name == 'sh'


def test_execute_execnet(queue):
    q = queue('foo:\n echo 1', 'foo')
    node = q.get()
    out, code = node()
    assert out == '1'
    assert code == 0


@mark.skipif('1') # XXX Fix this test to be proper
def test_execute_execnet_ssh(queue):
    q = queue('foo ssh=pseudoscience.co.uk:\n echo 1', 'foo')
    node = q.get()
    out, code = node()
    assert out == '1'
    assert code == 0


def test_execute_virtualenv(queue):
    q = queue('foo ve=ve:\n which python', 'foo')
    node = q.get()
    out, code = node()
    assert 've/bin/python' in out
    assert code == 0


def test_execute_python_virtualenv(queue):
    q = queue('foo ve=ve:\n !py\n import sys; print sys.prefix', 'foo')
    node = q.get()
    out, code = node()
    assert out.endswith('ve')
    assert code == 0


def test_reference(queue):
    q = queue('foo:\n boo\nnoo:\n @foo\n', 'noo')
    assert len(q) == 1
    assert list(q)[0].action.raw_content == 'boo\n'


def test_circular_reference(queue):
    q = queue('foo:\n @foo', 'foo')
    assert len(q) == 0


def test_missing_reference(queue):
    raises(UndefinedReference, queue, 'foo:\n @noo', 'foo')


def test_iter_queue(queue):
    q = queue('foo:\n moo', 'foo')
    items = list(q)
    assert len(items) == 1
    assert items[0].action.name == 'sh'


def test_sh_action():
    resp, code = sh_action.execute('name', 'echo 1', {})
    assert resp == '1\n'
    assert code == 0


def test_sh_action_ve():
    resp, code = sh_action.execute('name', 'which python', {'ve':'ve'})
    assert 've' in resp
    assert code == 0


def test_sh_action_ve_bootstrap():
    content = sh_action.bootstrap_ve('ve', '')
    assert content == 'source ve/bin/activate\n'


def test_py_action():
    resp, code = py_action.execute('name', 'print 1', {})
    assert resp == '1\n'
    assert code == 0


def test_py_action_error():
    resp, code = py_action.execute('name', 'print a', {})
    assert 'NameError' in resp
    assert code == 1


def test_py_action_ve():
    resp, code = py_action.execute('name',
        'import sys; print sys.prefix', {'ve':'ve'})
    assert 've' in resp
    assert code == 0


def test_py_action_ve_bootstrap():
    content = py_action.bootstrap_ve('ve', '')
    assert content == ('execfile("ve/bin/activate_this.py", '
                       'dict(__file__="ve/bin/activate_this.py"))\n')


def test_sh_action_env():
    """Put TEST VAR in environment and make sure its available in execute
    """
    os.environ['TEST_VAR'] = 'testvar'
    out, code = sh_action.execute('name',
        'echo $TEST_VAR', {})
    assert out == 'testvar\n'


def test_sh_action_env_namespace():
    """Put TEST_VAR in environment, and make sure it doesnt override namespace
    """
    os.environ['TEST_VAR'] = 'testvar'
    out, code = sh_action.execute('name',
        'echo $TEST_VAR', {'TEST_VAR': 'nstest'})
    assert out == 'nstest\n'