Source

xnet-hacks / tests / test_slave.py

Full commit
import sys
import py
from xnet.xjob import XJobSpec, XResult
from xnet.helpers import json_encode
from xnet import slave
from mocksupport import MockConnPost, MockConnGet
        
def test_pullxjob_nojob():
    conn = MockConnPost("/pull", json_encode({'xjobid':'no'}))
    r = slave.pullxjob(conn)
    assert r is None

def test_pullxjob():
    xjobspec = XJobSpec(args="123".split())
    data = dict(xjobid=1, spec=xjobspec)
    conn = MockConnPost("/pull", json_encode(data))
    r = slave.pullxjob(conn)
    assert r is not None
    assert r.xjobid == 1
    assert r == xjobspec

def test_xjobfinished():
    xresult = XResult(1, out="1", err="2", retcode=0)
    conn = MockConnPost("/result", "ok")
    r = slave.xjobfinished(conn, xresult)
    assert conn._body == json_encode(xresult)
    assert r == "ok"


# needs to be rewritten
# class TestXJobExecution:
#     def test_basic(self):
#         pm = slave.ProcessManager()
#         xjobspec = XJobSpec(args=[sys.executable, '--version'])
#         xjobspec.xjobid = 1
#         popen = pm.start(xjobspec)
#         popen2 = pm.getpopen(1)
#         assert popen is popen2
#         out, err = popen.communicate()
#         retcode = pm.wait(1)
#         py.test.raises(ValueError, pm.getpopen, 1)
#         assert out or err
        #popen = slave.start_process(xjobspec)