76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
import pytest
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import json
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
from src import native_orchestrator
|
|
|
|
def test_read_plan_nonexistent():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
result = native_orchestrator.read_plan("nonexistent_track", tmpdir)
|
|
assert result == ""
|
|
|
|
def test_write_and_read_plan():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
content = "# Test Plan\n- [x] Task 1\n- [ ] Task 2"
|
|
native_orchestrator.write_plan("test_track", content, tmpdir)
|
|
result = native_orchestrator.read_plan("test_track", tmpdir)
|
|
assert result == content
|
|
|
|
def test_parse_plan_tasks():
|
|
content = """# Plan
|
|
- [x] Completed task
|
|
- [ ] Pending task
|
|
- [x] Another done"""
|
|
tasks = native_orchestrator.parse_plan_tasks(content)
|
|
assert len(tasks) == 3
|
|
assert tasks[0]["completed"] == True
|
|
assert tasks[1]["completed"] == False
|
|
assert tasks[2]["completed"] == True
|
|
|
|
def test_read_metadata_nonexistent():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
result = native_orchestrator.read_metadata("nonexistent_track", tmpdir)
|
|
assert result == {}
|
|
|
|
def test_write_and_read_metadata():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
data = {"name": "Test Track", "status": "active"}
|
|
native_orchestrator.write_metadata("test_track", data, tmpdir)
|
|
result = native_orchestrator.read_metadata("test_track", tmpdir)
|
|
assert result["name"] == "Test Track"
|
|
assert result["status"] == "active"
|
|
|
|
def test_get_track_dir():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
track_dir = native_orchestrator.get_track_dir("my_track", tmpdir)
|
|
expected = Path(tmpdir) / "conductor" / "tracks" / "my_track"
|
|
assert track_dir == expected
|
|
|
|
def test_get_archive_dir():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
archive_dir = native_orchestrator.get_archive_dir(tmpdir)
|
|
expected = Path(tmpdir) / "conductor" / "archive"
|
|
assert archive_dir == expected
|
|
|
|
def test_native_orchestrator_class():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
orch = native_orchestrator.NativeOrchestrator(tmpdir)
|
|
assert orch.base_dir == Path(tmpdir)
|
|
|
|
data = {"name": "Test"}
|
|
orch.save_track("track1", data)
|
|
loaded = orch.load_track("track1")
|
|
assert loaded["name"] == "Test"
|
|
|
|
plan_content = "# Plan\n- [x] Done"
|
|
orch.save_plan("track1", plan_content)
|
|
loaded_plan = orch.load_plan("track1")
|
|
assert loaded_plan == plan_content
|
|
|
|
tasks = orch.get_track_tasks("track1")
|
|
assert len(tasks) == 1
|