186 lines
8.9 KiB
Python
186 lines
8.9 KiB
Python
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from service.account import AccountStore, DEFAULT_TRASH_RETENTION_SECONDS
|
|
|
|
|
|
def user(user_id):
|
|
return {
|
|
"user_id": user_id,
|
|
"username": user_id,
|
|
"display_name": user_id.title(),
|
|
"email": f"{user_id}@example.test",
|
|
"groups": [],
|
|
}
|
|
|
|
|
|
class AccountStoreTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary_directory = tempfile.TemporaryDirectory()
|
|
self.store = AccountStore(Path(self.temporary_directory.name) / "account.sqlite3")
|
|
self.alice = user("alice")
|
|
self.bob = user("bob")
|
|
self.payload = {
|
|
"workflow": "analysis",
|
|
"mode": "tube",
|
|
"model": {"material": "rna", "celsius": 37, "sodium": 1, "magnesium": 0},
|
|
"strands": [{"name": "A", "sequence": "ACGU"}],
|
|
"complexes_text": "A",
|
|
"tube": {"max_size": 1},
|
|
"compute": ["mfe"],
|
|
}
|
|
|
|
def tearDown(self):
|
|
self.temporary_directory.cleanup()
|
|
|
|
def test_jobs_are_private_and_content_round_trips(self):
|
|
result = {"workflow": "analysis", "complex_results": [{"name": "A"}]}
|
|
self.store.create_job("job-1", self.alice, self.payload)
|
|
self.store.update_job("job-1", "success", result=result, elapsed_seconds=1.25)
|
|
|
|
self.assertIsNone(self.store.get_job(self.bob["user_id"], "job-1"))
|
|
item = self.store.get_job(self.alice["user_id"], "job-1")
|
|
self.assertEqual(item["payload"], self.payload)
|
|
self.assertEqual(item["result"], result)
|
|
self.assertEqual(item["elapsed_seconds"], 1.25)
|
|
self.assertEqual(self.store.list_jobs("alice")["total"], 1)
|
|
self.assertEqual(self.store.list_jobs("bob")["total"], 0)
|
|
|
|
def test_share_can_expire_and_be_deleted(self):
|
|
self.store.create_job("job-2", self.alice, self.payload, status="success", result={"ok": True})
|
|
share = self.store.create_share("alice", "job-2")
|
|
resolved = self.store.resolve_share(share["share_id"])
|
|
self.assertEqual(resolved["job_id"], "job-2")
|
|
self.assertEqual(resolved["result"], {"ok": True})
|
|
|
|
self.store.update_share("alice", share["share_id"], active=False)
|
|
self.assertIsNone(self.store.resolve_share(share["share_id"]))
|
|
self.assertIsNone(self.store.get_share_for_owner("alice", share["share_id"]))
|
|
self.assertIsNone(self.store.update_share("bob", share["share_id"], active=True))
|
|
|
|
expiring = self.store.create_share("alice", "job-2", expires_in=1)
|
|
with self.store._connect() as connection:
|
|
connection.execute(
|
|
"UPDATE shares SET expires_at=? WHERE share_id=?",
|
|
(time.time() - 1, expiring["share_id"]),
|
|
)
|
|
self.assertIsNone(self.store.resolve_share(expiring["share_id"]))
|
|
|
|
def test_create_share_reuses_existing_active_link(self):
|
|
self.store.create_job("job-share-once", self.alice, self.payload, status="success")
|
|
first = self.store.create_share("alice", "job-share-once", expires_in=604800)
|
|
second = self.store.create_share("alice", "job-share-once", expires_in=604800)
|
|
self.assertEqual(first["share_id"], second["share_id"])
|
|
self.assertEqual(len(self.store.list_shares("alice")), 1)
|
|
|
|
def test_running_job_can_be_shared_and_later_exposes_result(self):
|
|
self.store.create_job("job-running", self.alice, self.payload, status="running")
|
|
share = self.store.create_share("alice", "job-running")
|
|
|
|
resolved = self.store.resolve_share(share["share_id"])
|
|
self.assertEqual(resolved["job_id"], "job-running")
|
|
self.assertEqual(resolved["status"], "running")
|
|
self.assertIsNone(resolved["result"])
|
|
|
|
self.store.update_job("job-running", "success", result={"ok": True}, elapsed_seconds=2.5)
|
|
resolved = self.store.resolve_share(share["share_id"])
|
|
self.assertEqual(resolved["status"], "success")
|
|
self.assertEqual(resolved["result"], {"ok": True})
|
|
self.assertEqual(resolved["elapsed_seconds"], 2.5)
|
|
|
|
self.assertIsNone(self.store.update_share("bob", share["share_id"], active=False))
|
|
self.store.update_share("alice", share["share_id"], active=False)
|
|
self.assertIsNone(self.store.resolve_share(share["share_id"]))
|
|
|
|
def test_active_job_cannot_be_deleted(self):
|
|
self.store.create_job("job-3", self.alice, self.payload)
|
|
with self.assertRaises(ValueError):
|
|
self.store.delete_job("alice", "job-3")
|
|
self.store.update_job("job-3", "canceled", error={"message": "canceled"})
|
|
self.assertTrue(self.store.delete_job("alice", "job-3"))
|
|
|
|
def test_deleted_job_moves_to_trash_and_deletes_shares(self):
|
|
self.assertEqual(self.store.trash_retention_seconds(), DEFAULT_TRASH_RETENTION_SECONDS)
|
|
self.store.create_job("job-trash", self.alice, self.payload, status="success", result={"ok": True})
|
|
share = self.store.create_share("alice", "job-trash")
|
|
|
|
self.assertTrue(self.store.delete_job("alice", "job-trash"))
|
|
self.assertIsNone(self.store.get_job("alice", "job-trash"))
|
|
self.assertEqual(self.store.list_jobs("alice")["total"], 0)
|
|
self.assertIsNone(self.store.resolve_share(share["share_id"]))
|
|
self.assertIsNone(self.store.get_share_for_owner("alice", share["share_id"]))
|
|
self.assertEqual(self.store.list_shares("alice"), [])
|
|
|
|
trash = self.store.list_all_jobs({"deleted": "1"})
|
|
self.assertEqual(trash["total"], 1)
|
|
self.assertEqual(trash["items"][0]["job_id"], "job-trash")
|
|
self.assertIsNotNone(trash["items"][0]["deleted_at"])
|
|
self.assertAlmostEqual(
|
|
trash["items"][0]["purge_after"] - trash["items"][0]["deleted_at"],
|
|
DEFAULT_TRASH_RETENTION_SECONDS,
|
|
delta=1,
|
|
)
|
|
|
|
def test_initialize_removes_legacy_shares_for_trashed_jobs(self):
|
|
self.store.create_job("job-legacy", self.alice, self.payload, status="success")
|
|
share = self.store.create_share("alice", "job-legacy")
|
|
with self.store._connect() as connection:
|
|
connection.execute(
|
|
"UPDATE jobs SET deleted_at=?,purge_after=? WHERE job_id=?",
|
|
(time.time(), time.time() + 86400, "job-legacy"),
|
|
)
|
|
self.store._initialized = False
|
|
self.store.initialize()
|
|
self.assertIsNone(self.store.get_share_for_owner("alice", share["share_id"]))
|
|
self.assertEqual(self.store.list_shares("alice"), [])
|
|
|
|
def test_restore_and_permanent_delete_complete_trash_lifecycle(self):
|
|
self.store.create_job("job-live", self.alice, self.payload, status="success")
|
|
self.assertFalse(self.store.permanently_delete_job("job-live"))
|
|
|
|
self.store.create_job("job-restore", self.alice, self.payload, status="success")
|
|
self.store.delete_job("alice", "job-restore")
|
|
self.assertTrue(self.store.restore_job("job-restore"))
|
|
self.assertIsNotNone(self.store.get_job("alice", "job-restore"))
|
|
self.assertEqual(self.store.list_all_jobs({"deleted": "1"})["total"], 0)
|
|
|
|
self.store.delete_job("alice", "job-restore")
|
|
self.assertTrue(self.store.permanently_delete_job("job-restore"))
|
|
self.assertFalse(self.store.permanently_delete_job("job-restore"))
|
|
self.assertIsNone(self.store.get_job("alice", "job-restore", include_deleted=True))
|
|
|
|
def test_retention_setting_and_expired_job_purge(self):
|
|
self.assertEqual(self.store.set_trash_retention_days(3.5), int(3.5 * 86400))
|
|
self.assertEqual(self.store.trash_retention_seconds(), int(3.5 * 86400))
|
|
with self.assertRaises(ValueError):
|
|
self.store.set_trash_retention_days(0)
|
|
|
|
self.store.create_job("job-expired", self.alice, self.payload, status="success")
|
|
self.store.delete_job("alice", "job-expired")
|
|
with self.store._connect() as connection:
|
|
connection.execute(
|
|
"UPDATE jobs SET purge_after=? WHERE job_id=?",
|
|
(time.time() - 1, "job-expired"),
|
|
)
|
|
self.assertEqual(self.store.purge_expired_jobs(), 1)
|
|
self.assertIsNone(self.store.get_job("alice", "job-expired", include_deleted=True))
|
|
|
|
def test_admin_listing_includes_all_users_and_supports_filters(self):
|
|
self.store.create_job("alice-job", self.alice, self.payload, status="running")
|
|
bob_payload = {**self.payload, "workflow": "design"}
|
|
self.store.create_job("bob-job", self.bob, bob_payload, status="success")
|
|
|
|
listing = self.store.list_all_jobs({"limit": 10})
|
|
self.assertEqual(listing["total"], 2)
|
|
self.assertEqual({item["user_id"] for item in listing["items"]}, {"alice", "bob"})
|
|
self.assertTrue(all("username" in item for item in listing["items"]))
|
|
|
|
filtered = self.store.list_all_jobs({"q": self.bob["email"], "workflow": "design"})
|
|
self.assertEqual(filtered["total"], 1)
|
|
self.assertEqual(filtered["items"][0]["job_id"], "bob-job")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|