1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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]
|
