diff options
| author | schneefux <schneefux+commit@schneefux.xyz> | 2017-02-25 17:15:48 +0100 |
|---|---|---|
| committer | schneefux <schneefux+commit@schneefux.xyz> | 2017-02-25 17:15:48 +0100 |
| commit | 997ed7580a47285c8f515f4ff68cc0786ec27e44 (patch) | |
| tree | 2bd5231ca61b77c1e220d892fb95d5c28f97ffb7 /test_joblib.py | |
| download | joblib-997ed7580a47285c8f515f4ff68cc0786ec27e44.tar.gz joblib-997ed7580a47285c8f515f4ff68cc0786ec27e44.zip | |
first implementation, fixes #1
Diffstat (limited to 'test_joblib.py')
| -rw-r--r-- | test_joblib.py | 67 |
1 files changed, 67 insertions, 0 deletions
diff --git a/test_joblib.py b/test_joblib.py new file mode 100644 index 0000000..35ae90e --- /dev/null +++ b/test_joblib.py @@ -0,0 +1,67 @@ +#!/usr/bin/python3 + +import os +import asyncio +import asyncpg +import pytest +import joblib + +class TestJoblib: + # async fixtures are not available yet, so we use a workaround + async def queue_helper(self, q): + await q.connect( + host=os.environ["POSTGRESQL_HOST"], + port=os.environ["POSTGRESQL_PORT"], + user=os.environ["POSTGRESQL_USER"], + password=os.environ["POSTGRESQL_PASSWORD"], + database=os.environ["POSTGRESQL_DB"] + ) + async with q._pool.acquire() as con: + await con.execute("DROP TABLE IF EXISTS jobs") + + # clean db table + await q.setup() + + @pytest.fixture + def queue(self, event_loop): + queue = joblib.JobQueue() + event_loop.run_until_complete(self.queue_helper(queue)) + return queue + + @pytest.fixture + def payload(self): + return { + "key": "value", + "dict": { + "foo": 1, + "bar": "baz" + } + } + + @pytest.mark.asyncio + async def test_request_and_acquire(self, queue, payload): + # request should succeed + await queue.request(jobtype="testing", payload=payload) + # acquire should return same payload + assert payload == (await queue.acquire(jobtype="testing"))[1] + # there should not be another job + assert None == (await queue.acquire(jobtype="testing"))[1] + + @pytest.mark.asyncio + async def test_cleanup(self, queue, payload): + await queue.request(jobtype="testing", payload=payload) + # mark job as processing + jobid_1, payload_1 = await queue.acquire(jobtype="testing") + assert payload_1 == payload + await queue.cleanup() + # same job should be available again + jobid_2, payload_2 = await queue.acquire(jobtype="testing") + assert jobid_1 == jobid_2 and payload_1 == payload_2 + + @pytest.mark.asyncio + async def test_finish(self, queue, payload): + await queue.request(jobtype="testing", payload=payload) + jobid, _ = await queue.acquire(jobtype="testing") + await queue.finish(jobid) + # job should not be available again + assert None == (await queue.acquire(jobtype="testing"))[1] |
