396 lines
19 KiB
Python
396 lines
19 KiB
Python
"""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_account_wait("A", "6pm")
|
|
self.db.set_account_wait("B", "2pm")
|
|
self.db.set_account_wait("C", "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"])
|
|
|
|
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_wait_belongs_to_the_account(self):
|
|
self.db.set_account_wait("A", "4:30pm")
|
|
self.assertEqual(self.statuses(), {"a-testing": "Wait 16:30", "a-prompt": "Wait 16:30",
|
|
"a-done": "Done", "b-prompt": "Prompt"})
|
|
self.assertEqual(self.db.task("a-testing").stored_status, lls.Status.TESTING) # kept underneath
|
|
self.assertEqual(self.db.task("a-testing").until, NOON.replace(hour=16, minute=30).astimezone())
|
|
account = self.db.account("A")
|
|
self.assertEqual((account.level, account.wait_state), (lls.Level.BUSY, lls.WaitState.WAITING))
|
|
|
|
self.at(17) # the wait is over -> Limit reset until cleared
|
|
self.assertEqual(self.db.task("a-prompt").status, lls.Status.LIMIT_RESET)
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.LIMIT_RESET)
|
|
|
|
self.assertTrue(self.db.clear_account_wait("A"))
|
|
self.assertFalse(self.db.clear_account_wait("A"))
|
|
self.assertEqual(self.db.task("a-testing").status, lls.Status.TESTING)
|
|
self.assertEqual(self.db.task("a-prompt").status, lls.Status.PROMPT)
|
|
|
|
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))
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.WAITING)
|
|
|
|
def test_set_wait_with_datetime_and_seconds(self):
|
|
self.db.set_account_wait("A", NOON.replace(hour=15))
|
|
self.assertEqual(self.db.task("a-prompt").status_label, "Wait 15:00")
|
|
self.db.set_account_wait("A", int(NOON.replace(hour=20).timestamp()))
|
|
self.assertEqual(self.db.task("a-prompt").status_label, "Wait 20:00")
|
|
|
|
def test_task_wait_sets_the_account_wait(self):
|
|
self.db.set_task_status("b-prompt", "wait 3pm")
|
|
self.assertEqual(self.db.account("B").wait_state, lls.WaitState.WAITING)
|
|
self.assertEqual(self.db.task("b-prompt").status_label, "Wait 15:00")
|
|
|
|
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.at(17) # over: a time is required again
|
|
with self.assertRaises(lls.InvalidArgument):
|
|
self.db.set_task_status("b-prompt", "wait")
|
|
|
|
def test_explicit_testing_or_prompt_clears_a_finished_wait(self):
|
|
self.db.set_account_wait("A", "1pm")
|
|
self.at(14)
|
|
self.assertEqual(self.db.task("a-prompt").status, lls.Status.LIMIT_RESET)
|
|
self.db.add_task("implicit", "A") # no status given: nothing cleared
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.LIMIT_RESET)
|
|
self.db.set_task_status("a-testing", "uploading") # not an "in use" status
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.LIMIT_RESET)
|
|
self.db.add_task("resumed", "A", "testing") # explicit: the account is in use again
|
|
self.assertTrue(any("back in use" in m for m in self.db.messages))
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.NONE) # (queries keep the messages)
|
|
self.assertEqual(self.db.task("resumed").status, lls.Status.TESTING)
|
|
self.assertEqual(self.db.task("a-prompt").status, lls.Status.PROMPT)
|
|
self.assertTrue(any("back in use" in m for m in self.db.messages))
|
|
|
|
def test_explicit_testing_or_prompt_shows_that_status_while_waiting(self):
|
|
self.db.set_account_wait("A", "6pm")
|
|
self.db.set_task_status("a-testing", "testing") # explicit
|
|
task = self.db.task("a-testing")
|
|
self.assertEqual((task.status, task.status_label, task.overrides_wait), (lls.Status.TESTING, "Testing", True))
|
|
self.assertTrue(any("ignores it" in m for m in self.db.messages))
|
|
|
|
self.db.set_task_status("a-prompt", "prompt")
|
|
self.assertEqual(self.db.task("a-prompt").status_label, "Prompt")
|
|
self.db.add_task("n1", "A", "testing")
|
|
self.assertEqual(self.db.task("n1").status, lls.Status.TESTING)
|
|
self.db.add_task("n2", "A") # no status given: follows the wait
|
|
self.assertEqual(self.db.task("n2").status, lls.Status.WAIT)
|
|
self.assertFalse(self.db.task("n2").overrides_wait)
|
|
|
|
# the account itself is still waiting, nothing was cleared
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.WAITING)
|
|
|
|
def test_other_tasks_keep_waiting_and_wait_makes_a_task_follow_again(self):
|
|
self.db.set_account_wait("A", "6pm")
|
|
self.db.add_task("a-other", "A")
|
|
self.db.set_task_status("a-testing", "testing")
|
|
self.assertEqual(self.db.task("a-other").status, lls.Status.WAIT)
|
|
self.assertEqual(self.db.task("a-prompt").status, lls.Status.WAIT)
|
|
|
|
self.db.set_task_status("a-testing", "wait") # no time: follow the running wait again
|
|
task = self.db.task("a-testing")
|
|
self.assertEqual((task.status, task.overrides_wait), (lls.Status.WAIT, False))
|
|
self.assertIn("now follows account 'A'", self.db.messages[0])
|
|
|
|
def test_a_new_wait_applies_to_every_task_again(self):
|
|
self.db.set_account_wait("A", "6pm")
|
|
self.db.set_task_status("a-testing", "testing")
|
|
self.db.set_account_wait("A", "7pm") # the account hit its limit again
|
|
task = self.db.task("a-testing")
|
|
self.assertEqual((task.status_label, task.overrides_wait), ("Wait 19:00", False))
|
|
|
|
def test_unwait_clears_the_overrides_too(self):
|
|
self.db.set_account_wait("A", "6pm")
|
|
self.db.set_task_status("a-testing", "testing")
|
|
self.assertTrue(self.db.clear_account_wait("A"))
|
|
self.assertFalse(self.db.task("a-testing").overrides_wait)
|
|
self.db.set_account_wait("A", "8pm")
|
|
self.assertEqual(self.db.task("a-testing").status_label, "Wait 20:00")
|
|
|
|
def test_an_override_survives_the_wait_ending(self):
|
|
self.db.set_account_wait("A", "1pm")
|
|
self.at(12, 30)
|
|
self.db.set_task_status("a-testing", "testing") # pinned while the wait runs
|
|
self.at(14) # the wait is over
|
|
self.assertEqual(self.db.task("a-testing").status, lls.Status.TESTING)
|
|
self.assertEqual(self.db.task("a-prompt").status, lls.Status.LIMIT_RESET)
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.LIMIT_RESET)
|
|
# resuming another task after the reset clears the finished wait for everyone
|
|
self.db.set_task_status("a-prompt", "prompt")
|
|
self.assertEqual(self.db.account("A").wait_state, lls.WaitState.NONE)
|
|
self.assertEqual(self.db.task("a-testing").status, lls.Status.TESTING)
|
|
self.assertEqual(self.db.task("a-prompt").status, lls.Status.PROMPT)
|
|
|
|
def test_the_override_ends_when_the_status_changes_or_the_task_moves(self):
|
|
self.db.set_account_wait("A", "6pm")
|
|
self.db.set_task_status("a-testing", "testing")
|
|
self.db.set_task_status("a-testing", "uploading")
|
|
self.assertFalse(self.db.task("a-testing").overrides_wait)
|
|
self.db.set_task_status("a-testing", "testing")
|
|
self.assertTrue(self.db.task("a-testing").overrides_wait)
|
|
self.db.set_task_account("a-testing", "B") # B is not waiting
|
|
task = self.db.task("a-testing")
|
|
self.assertEqual((task.account, task.status, task.overrides_wait), ("B", lls.Status.TESTING, False))
|
|
|
|
def test_a_wait_no_task_shows_when_every_task_ignores_it(self):
|
|
self.db.set_account_wait("A", "6pm")
|
|
self.db.set_task_status("a-testing", "testing")
|
|
self.db.set_task_status("a-prompt", "prompt") # a-done is Done
|
|
account = self.db.account("A")
|
|
self.assertEqual((account.level, account.wait_state, account.wait_listed),
|
|
(lls.Level.BUSY, lls.WaitState.WAITING, True))
|
|
|
|
def test_overrides_are_saved(self):
|
|
self.db.set_account_wait("A", "6pm")
|
|
self.db.set_task_status("a-testing", "testing")
|
|
self.db.save()
|
|
with lls.Database(self.path) as other:
|
|
other.set_clock(NOON)
|
|
self.assertTrue(other.task("a-testing").overrides_wait)
|
|
self.assertEqual(other.task("a-prompt").status, lls.Status.WAIT)
|
|
|
|
def test_account_wait_nothing_shows(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.db.add_task("only-done", "Empty", "done")
|
|
self.assertTrue(self.db.account("Empty").wait_listed)
|
|
self.assertEqual(self.db.account("Empty").tasks, ())
|
|
|
|
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_account_wait("A", "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()
|