"""Tests for the C API through the Python binding. LLS_LIBRARY=build/liblls_api.so python3 python/test_lls.py -v """ import os import sys import tempfile import unittest from datetime import datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import lls # noqa: E402 NOON = datetime(2030, 6, 15, 12, 0) # the pretend "now" used by most tests class Base(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = os.path.join(self.tmp.name, "tasks.db") self.db = lls.Database(self.path) self.db.set_clock(NOON) self.addCleanup(self.tmp.cleanup) self.addCleanup(self.db.close) def at(self, hour, minute=0): self.db.set_clock(NOON.replace(hour=hour, minute=minute)) def statuses(self): return {t.name: t.status_label for t in self.db.tasks()} class CrudTests(Base): def test_accounts_and_tasks(self): self.db.add_account("Acrux") self.db.add_account("beta") self.db.add_task("Breeze A", account="acrux") # names are case-insensitive task = self.db.task("breeze a") self.assertEqual(task.name, "Breeze A") self.assertEqual(task.account, "Acrux") # the account's own spelling self.assertEqual(task.status, lls.Status.PROMPT) # most free first: beta has no tasks (Free), Acrux has a Prompt task (Taken) self.assertEqual([a.name for a in self.db.accounts()], ["beta", "Acrux"]) def test_errors_have_types(self): self.db.add_account("A") with self.assertRaises(lls.AlreadyExists): self.db.add_account("a") with self.assertRaises(lls.NotFound): self.db.delete_account("nope") with self.assertRaises(lls.NotFound): self.db.add_task("t", account="nope") with self.assertRaises(lls.InvalidArgument): self.db.add_account("-") with self.assertRaises(lls.InvalidArgument): self.db.add_task("t", status="sleeping") with self.assertRaises(lls.InvalidArgument): self.db.set_account_wait("A", "25:99") self.db.add_task("t") with self.assertRaises(lls.AlreadyExists): self.db.add_task("T") try: self.db.delete_task("zzz") except lls.LLSError as error: self.assertEqual(error.code, lls.E_NOT_FOUND) self.assertIn("No task named 'zzz'", str(error)) def test_rename_reassign_delete(self): self.db.add_account("A") self.db.add_account("B") self.db.add_task("t1", "A") self.db.rename_task("t1", "t2") self.db.set_task_account("t2", "B") self.assertEqual(self.db.task("t2").account, "B") self.db.set_task_account("t2", None) self.assertIsNone(self.db.task("t2").account) self.db.set_task_account("t2", "A") self.db.rename_account("A", "Alpha") # tasks follow the rename self.assertEqual(self.db.task("t2").account, "Alpha") self.db.delete_account("Alpha") # tasks become unassigned self.assertIsNone(self.db.task("t2").account) self.db.delete_task("t2") self.assertEqual(self.db.tasks(), []) def test_unicode(self): self.db.add_account("Cuenta ñandú") self.db.add_task("Tarea ✓", "Cuenta ñandú") self.db.set_task_notes("Tarea ✓", "línea uno\nlínea dos é") task = self.db.task("Tarea ✓") self.assertEqual(task.account, "Cuenta ñandú") self.assertEqual(task.notes, "línea uno\nlínea dos é") self.assertEqual(task.notes_lines, 2) class OrderingTests(Base): def test_task_order(self): for account in ("Zeta", "alpha", "Mid"): self.db.add_account(account) self.at(12) # waiting / limit reset come from the account self.db.add_task("d1", "Zeta", "done") self.db.add_task("f1", "alpha", "frozen") self.db.add_task("u1", "alpha", "uploading") self.db.add_task("p-zeta", "Zeta") self.db.add_task("p-alpha", "alpha") self.db.add_task("p-none") self.db.add_task("t1", "Zeta", "testing") self.db.add_task("lr", "Mid", "limit reset") order = [t.name for t in self.db.tasks()] self.assertEqual(order, ["lr", "t1", "p-alpha", "p-zeta", "p-none", "u1", "f1", "d1"]) def test_wait_and_limit_reset_sort_by_time_before_account(self): for account in ("A", "B", "C"): self.db.add_account(account) self.db.add_task("task-" + account, account) self.db.set_task_status("task-A", "wait 6pm") self.db.set_task_status("task-B", "wait 2pm") self.db.set_task_status("task-C", "wait 4pm") self.assertEqual([t.name for t in self.db.tasks()], ["task-B", "task-C", "task-A"]) self.at(17) # B and C are over now: Limit reset first, ordered by when they ended self.assertEqual([t.status_label for t in self.db.tasks()], ["Limit reset", "Limit reset", "Wait 18:00"]) self.assertEqual([t.name for t in self.db.tasks()], ["task-B", "task-C", "task-A"]) self.db.add_account("Z") # any change saves what time did: same order afterwards self.assertEqual([t.name for t in self.db.tasks()], ["task-B", "task-C", "task-A"]) self.assertEqual(self.db.task("task-B").stored_status, lls.Status.LIMIT_RESET) def test_accounts_most_free_first(self): for account in ("Busy1", "Taken1", "Free1", "Free2", "Busy2"): self.db.add_account(account) self.db.add_task("t", "Busy1", "testing") self.db.add_task("w1", "Busy2", "testing") self.db.add_task("w2", "Busy2", "testing") self.db.add_task("p", "Taken1") self.db.add_task("d", "Free1", "done") accounts = self.db.accounts() self.assertEqual([a.name for a in accounts], ["Free1", "Free2", "Taken1", "Busy1", "Busy2"]) self.assertEqual([a.level for a in accounts], [lls.Level.FREE, lls.Level.FREE, lls.Level.TAKEN, lls.Level.BUSY, lls.Level.BUSY]) self.assertEqual(accounts[0].tasks, ("d",)) # a Free account lists its Done task self.assertEqual(accounts[2].tasks, ("p",)) self.assertEqual(accounts[4].tasks, ("w1", "w2")) class WaitTests(Base): def setUp(self): super().setUp() self.db.add_account("A") self.db.add_account("B") self.db.add_task("a-testing", "A", "testing") self.db.add_task("a-prompt", "A") self.db.add_task("a-done", "A", "done") self.db.add_task("b-prompt", "B") def test_only_tasks_set_to_wait_wait(self): self.db.set_task_status("a-prompt", "wait 4:30pm") self.assertEqual(self.statuses(), {"a-testing": "Testing (account-waiting)", "a-prompt": "Wait 16:30", "a-done": "Done", "b-prompt": "Prompt"}) account = self.db.account("A") self.assertEqual((account.level, account.wait_state), (lls.Level.BUSY, lls.WaitState.WAITING)) self.assertEqual(self.db.task("a-prompt").until, NOON.replace(hour=16, minute=30).astimezone()) def test_prompt_and_testing_show_as_they_are_with_a_note(self): self.db.set_account_wait("A", "6pm") # nothing is set to wait testing = self.db.task("a-testing") self.assertEqual((testing.status, testing.status_label, testing.account_waiting), (lls.Status.TESTING, "Testing (account-waiting)", True)) prompt = self.db.task("a-prompt") self.assertEqual((prompt.status, prompt.status_label), (lls.Status.PROMPT, "Prompt (account-waiting)")) self.assertEqual(self.db.task("a-done").status_label, "Done") # no note on other statuses self.assertFalse(self.db.task("b-prompt").account_waiting) # another account account = self.db.account("A") self.assertEqual((account.level, account.wait_listed), (lls.Level.BUSY, True)) self.db.set_task_status("a-testing", "testing") # setting it again changes nothing self.assertEqual(self.db.task("a-testing").status_label, "Testing (account-waiting)") self.assertEqual(self.db.messages, ["Task 'a-testing' set to Testing."]) def test_the_account_stops_waiting_by_itself_and_waiting_tasks_become_limit_reset(self): self.db.add_task("w1", "A", "wait 4pm") self.db.add_task("w2", "A", "wait") # joins the running wait self.assertEqual(self.db.task("w2").status_label, "Wait 16:00") self.at(17) # the time has passed self.assertEqual(self.statuses()["w1"], "Limit reset") self.assertEqual(self.statuses()["w2"], "Limit reset") self.assertEqual(self.statuses()["a-testing"], "Testing") # no note any more self.assertEqual(self.statuses()["a-prompt"], "Prompt") account = self.db.account("A") self.assertEqual((account.wait_state, account.wait_until), (lls.WaitState.NONE, None)) self.assertEqual(account.tasks[:2], ("w1", "w2")) # Limit reset keeps it Busy self.db.add_account("Z") # a change saves what time did self.assertEqual(self.db.task("w1").stored_status, lls.Status.LIMIT_RESET) self.db.set_task_status("w1", "testing") # handled: leaves Limit reset self.assertEqual(self.db.task("w1").status_label, "Testing") self.assertEqual(self.db.task("w2").status_label, "Limit reset") def test_a_new_wait_leaves_limit_reset_tasks_alone(self): self.db.add_task("w1", "A", "wait 1pm") self.at(14) self.db.add_task("w2", "A", "wait 8pm") # the account hit its limit again self.assertEqual(self.statuses()["w1"], "Limit reset") self.assertEqual(self.statuses()["w2"], "Wait 20:00") self.assertEqual(self.db.account("A").wait_state, lls.WaitState.WAITING) def test_unwait_ends_the_wait_now(self): self.db.add_task("w1", "A", "wait 6pm") self.assertTrue(self.db.clear_account_wait("A")) self.assertIn("1 task now Limit reset", self.db.messages[0]) self.assertEqual(self.statuses()["w1"], "Limit reset") self.assertEqual(self.db.account("A").wait_state, lls.WaitState.NONE) self.assertEqual(self.db.task("a-testing").status_label, "Testing") self.assertFalse(self.db.clear_account_wait("A")) # it was not waiting any more def test_changing_the_account_wait_moves_waiting_tasks(self): self.db.add_task("w1", "A", "wait 4pm") self.db.set_account_wait("A", "7pm") self.assertEqual(self.statuses()["w1"], "Wait 19:00") def test_time_in_the_past_means_tomorrow(self): self.db.set_account_wait("A", "9am") # it is noon until = self.db.account("A").wait_until self.assertEqual((until.day, until.hour), (16, 9)) def test_set_wait_with_datetime_and_seconds(self): self.db.set_account_wait("A", NOON.replace(hour=15)) self.assertEqual(self.db.account("A").wait_until, NOON.replace(hour=15).astimezone()) self.db.set_account_wait("A", int(NOON.replace(hour=20).timestamp())) self.assertEqual(self.db.account("A").wait_until, NOON.replace(hour=20).astimezone()) def test_wait_without_time_only_while_account_is_waiting(self): with self.assertRaises(lls.InvalidArgument): self.db.set_task_status("b-prompt", "wait") with self.assertRaises(lls.InvalidArgument): self.db.add_task("new", "B", "wait") self.db.add_task("loose") with self.assertRaises(lls.InvalidArgument): self.db.set_task_status("loose", "wait") self.db.set_account_wait("B", "4pm") self.db.add_task("follower", "B", "wait") # fine now self.assertEqual(self.db.task("follower").status_label, "Wait 16:00") self.db.set_task_status("b-prompt", "wait") self.assertEqual(self.db.task("b-prompt").status_label, "Wait 16:00") self.at(17) # over: a time is required again with self.assertRaises(lls.InvalidArgument): self.db.add_task("late", "B", "wait") def test_a_task_without_an_account_keeps_its_own_wait(self): self.db.add_task("loose", None, "wait 4pm") self.assertEqual(self.db.task("loose").status_label, "Wait 16:00") self.at(17) self.assertEqual(self.db.task("loose").status_label, "Limit reset") def test_deleting_the_waiting_tasks_keeps_the_account_waiting(self): self.db.add_task("w1", "A", "wait 4pm") self.db.delete_task("w1") account = self.db.account("A") self.assertEqual((account.level, account.wait_state, account.wait_listed), (lls.Level.BUSY, lls.WaitState.WAITING, True)) self.assertEqual(self.db.task("a-prompt").status_label, "Prompt (account-waiting)") def test_account_wait_with_no_waiting_tasks(self): self.db.add_account("Empty") self.db.set_account_wait("Empty", "5pm") account = self.db.account("Empty") self.assertEqual((account.level, account.wait_listed, account.tasks), (lls.Level.BUSY, True, ())) self.at(18) # over: nothing left to show account = self.db.account("Empty") self.assertEqual((account.level, account.wait_state), (lls.Level.FREE, lls.WaitState.NONE)) def test_messages(self): self.db.set_account_wait("A", "4:30pm") self.assertEqual(len(self.db.messages), 1) self.assertIn("waits until 16:30", self.db.messages[0]) with self.assertRaises(lls.NotFound): self.db.delete_task("zzz") self.assertEqual(self.db.messages, []) # a failed call leaves no notes class NotesAndPersistenceTests(Base): def test_notes_have_no_length_limit(self): self.db.add_task("t") text = ("x" * 99) + "\n" text *= 3000 # ~300 kB, 3000 lines text = text.rstrip("\n") self.db.set_task_notes("t", text) task = self.db.task("t") self.assertEqual(task.notes, text) self.assertEqual(task.notes_lines, 3000) self.db.set_task_notes("t", "") self.assertEqual(self.db.task("t").notes_lines, 0) def test_save_reopen_and_dirty_flag(self): self.assertFalse(self.db.dirty) self.db.add_account("A") self.db.add_task("t", "A", "testing") self.db.set_task_notes("t", "one\ntwo") self.db.set_task_status("t", "wait 11pm") self.assertTrue(self.db.dirty) self.db.save() self.assertFalse(self.db.dirty) with lls.Database(self.path) as other: other.set_clock(NOON) task = other.task("t") self.assertEqual((task.account, task.notes, task.status_label), ("A", "one\ntwo", "Wait 23:00")) def test_unsaved_changes_are_not_written(self): self.db.add_account("A") self.db.save() self.db.add_account("B") self.db.close() with lls.Database(self.path) as other: self.assertEqual([a.name for a in other.accounts()], ["A"]) def test_backup_and_restore(self): self.db.add_account("A") copy = self.db.backup(os.path.join(self.tmp.name, "copy.db")) self.assertTrue(os.path.isfile(copy)) auto = self.db.backup() # timestamped name next to the database self.assertTrue(auto.startswith(self.path) and os.path.isfile(auto)) self.db.add_account("B") self.db.restore(copy) self.assertEqual([a.name for a in self.db.accounts()], ["A"]) self.assertTrue(self.db.dirty) with self.assertRaises(lls.LLSError): self.db.restore(os.path.join(self.tmp.name, "missing.db")) def test_opening_a_bad_path_or_file(self): bad = os.path.join(self.tmp.name, "not-a-db.db") Path(bad).write_text("this is not sqlite") with self.assertRaises(lls.DatabaseError): lls.Database(bad) def test_closed_database_raises(self): self.db.close() with self.assertRaises(lls.LLSError): self.db.add_account("A") class UtilityTests(unittest.TestCase): def test_parse_time(self): self.assertEqual(lls.parse_time("4:30pm", now=NOON), NOON.replace(hour=16, minute=30).astimezone()) self.assertEqual(lls.parse_time("16:30", now=NOON).hour, 16) self.assertEqual(lls.parse_time("12am", now=NOON).day, 16) # already passed -> tomorrow for bad in ("", "6", "13pm", "25:00", "soon"): with self.assertRaises(lls.InvalidArgument): lls.parse_time(bad, now=NOON) if __name__ == "__main__": unittest.main()