# Aidderall MCP Server - Hierarchical task management for AI assistants
# Copyright (C) 2024 Briam R. <briamr@gmail.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import tempfile
from pathlib import Path
import pytest
from src.models import MainTask, SubTask, TaskStatus
from src.persistence import StateStore
from src.task_manager import TaskManager
@pytest.fixture
def tmp_db(tmp_path):
return tmp_path / "test_state.db"
@pytest.fixture
def store(tmp_db):
return StateStore(db_path=tmp_db)
@pytest.fixture
def manager(store):
return TaskManager(store=store)
class TestStateStore:
def test_empty_store(self, store):
assert not store.has_state()
global_tasks, completed, manual_id = store.load()
assert global_tasks == []
assert completed == []
assert manual_id is None
def test_save_and_load_single_task(self, store):
task = MainTask(title="Test", body="Body")
task.status = TaskStatus.CURRENT
store.save([task], [], None)
assert store.has_state()
global_tasks, completed, manual_id = store.load()
assert len(global_tasks) == 1
assert global_tasks[0].id == task.id
assert global_tasks[0].title == "Test"
assert global_tasks[0].body == "Body"
assert global_tasks[0].status == TaskStatus.CURRENT
def test_save_and_load_hierarchy(self, store):
main = MainTask(title="Main", body="Main body")
main.status = TaskStatus.PENDING
sub1 = SubTask(title="Sub 1", body="Sub body 1")
sub1.status = TaskStatus.PENDING
sub2 = SubTask(title="Sub 2", body="Sub body 2")
sub2.status = TaskStatus.CURRENT
main.sub_tasks = [sub1, sub2]
store.save([main], [], None)
global_tasks, _, _ = store.load()
assert len(global_tasks) == 1
loaded_main = global_tasks[0]
assert len(loaded_main.sub_tasks) == 2
assert loaded_main.sub_tasks[0].title == "Sub 1"
assert loaded_main.sub_tasks[1].title == "Sub 2"
assert loaded_main.sub_tasks[1].status == TaskStatus.CURRENT
def test_completed_tasks_reference_sharing(self, store):
main = MainTask(title="Main", body="Body")
main.status = TaskStatus.COMPLETED
store.save([main], [main], None)
global_tasks, completed, _ = store.load()
assert len(completed) == 1
# Same object reference should be shared
assert completed[0] is global_tasks[0]
def test_completed_tasks_orphaned(self, store):
"""Completed task whose original was removed from hierarchy."""
orphan = SubTask(title="Orphan", body="Was removed")
orphan.status = TaskStatus.COMPLETED
store.save([], [orphan], None)
global_tasks, completed, _ = store.load()
assert len(global_tasks) == 0
assert len(completed) == 1
assert completed[0].title == "Orphan"
def test_manual_focus_roundtrip(self, store):
main = MainTask(title="Main", body="Body")
main.status = TaskStatus.PENDING
sub = SubTask(title="Sub", body="Sub body")
sub.status = TaskStatus.CURRENT
main.sub_tasks = [sub]
store.save([main], [], sub.id)
_, _, manual_id = store.load()
assert manual_id == sub.id
def test_overwrite_on_save(self, store):
task1 = MainTask(title="First", body="Body 1")
store.save([task1], [], None)
task2 = MainTask(title="Second", body="Body 2")
store.save([task2], [], None)
global_tasks, _, _ = store.load()
assert len(global_tasks) == 1
assert global_tasks[0].title == "Second"
class TestTaskManagerPersistence:
def test_state_survives_new_instance(self, tmp_db):
store1 = StateStore(db_path=tmp_db)
mgr1 = TaskManager(store=store1)
task = mgr1.create_new_task("Persistent Task", "Should survive")
sub = mgr1.extend_current_task("Sub Task", "Also survives")
task_id = task.id
sub_id = sub.id
store1.close()
store2 = StateStore(db_path=tmp_db)
mgr2 = TaskManager(store=store2)
assert len(mgr2.global_tasks) == 1
assert mgr2.global_tasks[0].id == task_id
assert mgr2.global_tasks[0].title == "Persistent Task"
assert len(mgr2.global_tasks[0].sub_tasks) == 1
assert mgr2.global_tasks[0].sub_tasks[0].id == sub_id
store2.close()
def test_complete_persists(self, tmp_db):
store1 = StateStore(db_path=tmp_db)
mgr1 = TaskManager(store=store1)
mgr1.create_new_task("Task", "Body")
mgr1.complete_current_task()
store1.close()
store2 = StateStore(db_path=tmp_db)
mgr2 = TaskManager(store=store2)
assert mgr2.is_zen_state
assert len(mgr2.completed_tasks) == 1
assert mgr2.completed_tasks[0].title == "Task"
store2.close()
def test_switch_focus_persists(self, tmp_db):
store1 = StateStore(db_path=tmp_db)
mgr1 = TaskManager(store=store1)
main1 = mgr1.create_new_task("Task A", "Body A")
main2 = mgr1.create_new_task("Task B", "Body B")
mgr1.switch_focus(main1.id)
store1.close()
store2 = StateStore(db_path=tmp_db)
mgr2 = TaskManager(store=store2)
assert mgr2.current_task is not None
assert mgr2.current_task.id == main1.id
store2.close()
def test_remove_persists(self, tmp_db):
store1 = StateStore(db_path=tmp_db)
mgr1 = TaskManager(store=store1)
task = mgr1.create_new_task("Doomed", "Will be removed")
mgr1.create_new_task("Survivor", "Stays")
mgr1.remove_task(task.id)
store1.close()
store2 = StateStore(db_path=tmp_db)
mgr2 = TaskManager(store=store2)
assert len(mgr2.global_tasks) == 1
assert mgr2.global_tasks[0].title == "Survivor"
store2.close()
def test_update_body_persists(self, tmp_db):
store1 = StateStore(db_path=tmp_db)
mgr1 = TaskManager(store=store1)
mgr1.create_new_task("Task", "Original body")
mgr1.update_current_task("Updated body")
store1.close()
store2 = StateStore(db_path=tmp_db)
mgr2 = TaskManager(store=store2)
assert mgr2.current_task is not None
assert mgr2.current_task.body == "Updated body"
store2.close()
def test_no_store_works_as_before(self):
"""TaskManager without a store should work identically to before."""
mgr = TaskManager()
task = mgr.create_new_task("Ephemeral", "In-memory only")
assert mgr.current_task == task
mgr.complete_current_task()
assert mgr.is_zen_state
def test_complex_scenario_roundtrip(self, tmp_db):
store1 = StateStore(db_path=tmp_db)
mgr1 = TaskManager(store=store1)
main1 = mgr1.create_new_task("Design Feature", "Design specs")
sub1_1 = mgr1.extend_current_task("Research", "Research notes")
sub1_2 = mgr1.extend_current_task("Interview", "Interview results")
mgr1.complete_current_task() # complete sub1_2
main2 = mgr1.create_new_task("Urgent Bug", "Bug report")
store1.close()
store2 = StateStore(db_path=tmp_db)
mgr2 = TaskManager(store=store2)
assert len(mgr2.global_tasks) == 2
assert mgr2.global_tasks[0].title == "Design Feature"
assert mgr2.global_tasks[1].title == "Urgent Bug"
assert len(mgr2.global_tasks[0].sub_tasks) == 2
assert mgr2.global_tasks[0].sub_tasks[1].status == TaskStatus.COMPLETED
assert len(mgr2.completed_tasks) == 1
assert mgr2.completed_tasks[0].title == "Interview"
# Completed task should be same object as the one in hierarchy
assert mgr2.completed_tasks[0] is mgr2.global_tasks[0].sub_tasks[1]
store2.close()
