354 lines
17 KiB
Python
354 lines
17 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_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()
|