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()