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]