Source

xnet-hacks / tests / test_server.py

import py
from xnet import server
from xnet.xjob import XJobSpec, XResult
from xnet.helpers import json_encode
from xnet.xjob import XJobStorage, XResultStorage, XJobInfo
import json

def pytest_funcarg__app(request):
    app = server.app
    app.xjobstorage = XJobStorage()
    app.resultstorage = XResultStorage()
    return app.test_client()

def test_submitxjob(app):
    xjob = XJobSpec(args=["hello"])
    rv = app.post("/", data=json_encode(xjob))
    jobid = json.loads(rv.data)['xjobid']
    assert int(jobid) >= 0
    return jobid

def test_pullxjob(app):
    xjobid = test_submitxjob(app)
    r = app.post("/pull")
    res = json.loads(r.data)
    assert res['xjobid'] == xjobid
    return xjobid

def test_listxjobs(app):
    xjob = XJobSpec(args=["hello"])
    rv = app.get("/", data=json_encode(xjob))
    data = json.loads(rv.data)
    assert 'queued' in data
    assert 'pulled' in data
    assert 'active' in data
    assert 'finished' in data
    return data

def test_registerxjob(app):
    xjobid = test_pullxjob(app)
    info = XJobInfo(xjobid)
    r = app.post("/registerxjob", data=json_encode(info))
    assert r.data == "registered job with id %s" % xjobid
    jobs = test_listxjobs(app)
    entry = jobs.get('active').get(str(xjobid))
    assert entry.get('xjobid') == xjobid
    return xjobid

def test_result_submit_and_get(app):
    xjobid = test_registerxjob(app)
    result = XResult(xjobid, "out", "err", 0)
    rv = app.post("/result", data=json_encode(result))
    assert rv.data == "Thanks."
    r = app.get("/getresult/%s" % xjobid)
    result2 = XResult.from_json(json.loads(r.data)['result'])
    assert result2.out == result.out
    assert result2.err == result.err
    assert result2.retcode == result.retcode
    assert result2.xjobid == result.xjobid

@py.test.mark.xfail(reason='unfinished')
def test_result_without_job(app):
    r = app.get('/getresult', data=json_encode({'xjobid': 1}))
    assert r.status == 404
    #XXX: finish
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.