automaton/tests/test_status.py

"""Tests for status.py enforcement script."""

import os
import sys
import tempfile
import pytest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))


@pytest.fixture
def tmp_project(tmp_path):
    """Create a temporary project with .automaton/tasks structure."""
    auto_dir = tmp_path / ".automaton"
    tasks_dir = auto_dir / "tasks"
    tasks_dir.mkdir(parents=True)
    return tmp_path


@pytest.fixture
def status_script():
    """Return path to status.py."""
    return Path.home() / ".automaton" / "scripts" / "status.py"


def _run_status(args, project=None):
    """Run status.py with given args and return (stdout, exit_code)."""
    import subprocess
    cmd = [sys.executable, str(Path.home() / ".automaton" / "scripts" / "status.py")]
    if project:
        cmd.extend(["--project", str(project)])
    cmd.extend(args)
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout.strip(), result.returncode


def _create_task(project, task_name, phase=None):
    """Create a task and optionally set its phase."""
    out, code = _run_status(["--create-task", task_name], project)
    assert code == 0, f"Failed to create task: {out}"
    if phase:
        task_dir = project / ".automaton" / "tasks" / task_name
        (task_dir / ".state").write_text(f"{phase}\n")
    return project / ".automaton" / "tasks" / task_name


class TestCreateTask:
    def test_create_valid_task(self, tmp_project):
        out, code = _run_status(["--create-task", "my-test-task"], tmp_project)
        assert code == 0
        assert "Created task 'my-test-task'" in out
        task_dir = tmp_project / ".automaton" / "tasks" / "my-test-task"
        assert task_dir.exists()
        assert (task_dir / ".state").read_text().strip() == "new"
        assert (task_dir / ".state.approvals").exists()

    def test_create_task_rejects_spaces(self, tmp_project):
        out, code = _run_status(["--create-task", "my test task"], tmp_project)
        assert code == 2
        assert "kebab-case" in out

    def test_create_task_rejects_uppercase(self, tmp_project):
        out, code = _run_status(["--create-task", "My-Task"], tmp_project)
        assert code == 2

    def test_create_task_rejects_duplicate(self, tmp_project):
        _run_status(["--create-task", "my-task"], tmp_project)
        out, code = _run_status(["--create-task", "my-task"], tmp_project)
        assert code == 2
        assert "already exists" in out


class TestStateTransitions:
    def test_transition_new_to_research(self, tmp_project):
        task_dir = _create_task(tmp_project, "trans-test", "new")
        out, code = _run_status(["--task", "trans-test", "--transition", "research"], tmp_project)
        assert code == 0
        assert "Transitioned" in out
        assert (task_dir / ".state").read_text().strip() == "research"

    def test_illegal_transition_refused(self, tmp_project):
        task_dir = _create_task(tmp_project, "illegal-test", "new")
        out, code = _run_status(["--task", "illegal-test", "--transition", "implement"], tmp_project)
        assert code == 1
        assert "Cannot transition" in out

    def test_approval_gate_blocks_transition(self, tmp_project):
        task_dir = _create_task(tmp_project, "approval-test", "research:awaiting_approval")
        out, code = _run_status(["--task", "approval-test", "--transition", "design"], tmp_project)
        assert code == 1
        assert "research:approved" in out


class TestApprovalGates:
    def test_approve_transitions_to_approved(self, tmp_project):
        task_dir = _create_task(tmp_project, "approve-test", "research:awaiting_approval")
        out, code = _run_status(["--approve", "--task", "approve-test"], tmp_project)
        assert code == 0
        assert "Approved" in out
        assert (task_dir / ".state").read_text().strip() == "research:approved"

    def test_approve_records_in_log(self, tmp_project):
        task_dir = _create_task(tmp_project, "log-test", "research:awaiting_approval")
        _run_status(["--approve", "--task", "log-test"], tmp_project)
        approvals = (task_dir / ".state.approvals").read_text()
        assert "research:approved" in approvals
        assert "user" in approvals

    def test_approve_refused_for_wrong_phase(self, tmp_project):
        _create_task(tmp_project, "wrong-phase", "implement")
        out, code = _run_status(["--approve", "--task", "wrong-phase"], tmp_project)
        assert "does not require approval" in out

    def test_approve_refused_without_awaiting(self, tmp_project):
        _create_task(tmp_project, "no-await", "research")
        out, code = _run_status(["--approve", "--task", "no-await"], tmp_project)
        assert code == 1
        assert "not awaiting approval" in out


class TestValidateFolder:
    def test_valid_folder_passes(self, tmp_project):
        task_dir = _create_task(tmp_project, "valid-test", "research")
        # Write SPEC.md (expected artifact for research)
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--validate-folder", "--task", "valid-test"], tmp_project)
        assert "PASS" in out

    def test_folder_without_state_flagged(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        task_dir = tasks_dir / "manual-task"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--validate-folder", "--task", "manual-task"], tmp_project)
        assert "no .state file" in out or "manually" in out

    def test_out_of_order_artifacts_flagged(self, tmp_project):
        task_dir = _create_task(tmp_project, "skip-test", "research")
        (task_dir / "SPEC.md").write_text("# Spec")
        (task_dir / "IMPLEMENTATION.md").write_text("# Impl")
        out, code = _run_status(["--validate-folder", "--task", "skip-test"], tmp_project)
        assert "FAIL" in out
        assert "IMPLEMENTATION.md" in out


class TestArtifactValidation:
    def test_missing_required_artifact_blocks_transition(self, tmp_project):
        _create_task(tmp_project, "no-spec", "research")
        out, code = _run_status(["--task", "no-spec", "--transition", "design"], tmp_project)
        assert code == 1
        assert "SPEC.md" in out

    def test_forbidden_artifact_blocks_transition(self, tmp_project):
        task_dir = _create_task(tmp_project, "skip-artifact", "research")
        (task_dir / "SPEC.md").write_text("# Spec")
        (task_dir / "IMPLEMENTATION.md").write_text("# Impl")
        out, code = _run_status(["--task", "skip-artifact", "--transition", "design"], tmp_project)
        assert code == 1
        assert "out-of-order" in out.lower() or "forbidden" in out.lower()


class TestShowTask:
    def test_show_task_displays_phase(self, tmp_project):
        _create_task(tmp_project, "show-test", "implement")
        out, code = _run_status(["--task", "show-test"], tmp_project)
        assert code == 0
        assert "Phase: implement" in out

    def test_show_task_displays_allowed_forbidden(self, tmp_project):
        _create_task(tmp_project, "show-impl", "implement")
        out, code = _run_status(["--task", "show-impl"], tmp_project)
        assert "Allowed actions" in out
        assert "Edit code" in out
        assert "Forbidden actions" in out


class TestList:
    def test_list_shows_tasks(self, tmp_project):
        _create_task(tmp_project, "list-a", "research")
        _create_task(tmp_project, "list-b", "implement")
        out, code = _run_status(["--list"], tmp_project)
        assert code == 0
        assert "list-a" in out
        assert "list-b" in out


class TestToolHooks:
    def test_can_edit_denied_in_research(self, tmp_project):
        _create_task(tmp_project, "edit-no", "research")
        out, code = _run_status(["--can-edit", "--task", "edit-no"], tmp_project)
        assert code == 1
        assert "DENIED" in out

    def test_can_edit_allowed_in_implement(self, tmp_project):
        _create_task(tmp_project, "edit-yes", "implement")
        out, code = _run_status(["--can-edit", "--task", "edit-yes"], tmp_project)
        assert code == 0
        assert "ALLOWED" in out

    def test_scope_check_in_scope(self, tmp_project):
        test_file = tmp_project / "src" / "main.py"
        test_file.parent.mkdir(parents=True, exist_ok=True)
        test_file.write_text("# test")
        out, code = _run_status(["--scope-check", "--task", "scope-test", "--file", str(test_file)], tmp_project)
        assert "IN_SCOPE" in out

    def test_scope_check_out_of_scope(self, tmp_project):
        _create_task(tmp_project, "scope-test", "research")
        out, code = _run_status(["--scope-check", "--task", "scope-test", "--file", "/tmp/some_random_file.py"], tmp_project)
        assert code == 1
        assert "OUT_OF_SCOPE" in out

    def test_scope_check_framework_out_of_scope_for_project(self, tmp_project):
        _create_task(tmp_project, "scope-proj", "research")
        framework_file = Path.home() / ".automaton" / "scripts" / "status.py"
        if framework_file.exists():
            out, code = _run_status(["--scope-check", "--task", "scope-proj", "--file", str(framework_file)], tmp_project)
            assert code == 1
            assert "OUT_OF_SCOPE" in out


class TestProjectScoping:
    def test_no_project_errors_without_flag(self):
        import subprocess
        cmd = [sys.executable, str(Path.home() / ".automaton" / "scripts" / "status.py"), "--list"]
        result = subprocess.run(cmd, capture_output=True, text=True, cwd="/tmp")
        assert result.returncode != 0

    def test_project_flag_targets_correct_tasks(self, tmp_project):
        _create_task(tmp_project, "scoped-task", "research")
        out, code = _run_status(["--list"], tmp_project)
        assert code == 0
        assert "scoped-task" in out


class TestUntrackedTasks:
    def test_transition_refuses_untracked_task(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        task_dir = tasks_dir / "untracked-task"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--transition", "design", "--task", "untracked-task"], tmp_project)
        assert code == 1
        assert "no .state file" in out

    def test_can_edit_refuses_untracked_task(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        task_dir = tasks_dir / "untracked-edit"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--can-edit", "--task", "untracked-edit"], tmp_project)
        assert code == 1
        assert "no .state file" in out

    def test_show_task_refuses_untracked_task(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        task_dir = tasks_dir / "untracked-show"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--task", "untracked-show"], tmp_project)
        assert code == 1
        assert "no .state file" in out

    def test_list_shows_untracked_task(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        task_dir = tasks_dir / "untracked-list"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--list"], tmp_project)
        assert code == 0
        assert "UNTRACKED" in out

    def test_upgrade_bootstraps_state_file(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        task_dir = tasks_dir / "upgrade-test"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--upgrade", "--task", "upgrade-test"], tmp_project)
        assert code == 0
        assert "Bootstrapped" in out
        state_file = task_dir / ".state"
        assert state_file.exists()

    def test_upgrade_all_tasks(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        t1 = tasks_dir / "task-a"
        t1.mkdir()
        (t1 / "SPEC.md").write_text("# Spec")
        t2 = tasks_dir / "task-b"
        t2.mkdir()
        out, code = _run_status(["--upgrade"], tmp_project)
        assert code == 0
        assert (t1 / ".state").exists()
        assert (t2 / ".state").exists()


class TestCanEditProject:
    def test_can_edit_no_task_allowed_when_implement(self, tmp_project):
        _create_task(tmp_project, "edit-impl", "implement")
        out, code = _run_status(["--can-edit"], tmp_project)
        assert code == 0
        assert "ALLOWED" in out
        assert "edit-impl" in out

    def test_can_edit_no_task_allowed_when_doc_review(self, tmp_project):
        _create_task(tmp_project, "edit-doc", "doc_review")
        out, code = _run_status(["--can-edit"], tmp_project)
        assert code == 0
        assert "ALLOWED" in out
        assert "edit-doc" in out

    def test_can_edit_no_task_denied_when_no_edit_phase(self, tmp_project):
        _create_task(tmp_project, "edit-res", "research")
        out, code = _run_status(["--can-edit"], tmp_project)
        assert code == 1
        assert "DENIED" in out

    def test_can_edit_no_task_denied_when_no_tasks(self, tmp_project):
        out, code = _run_status(["--can-edit"], tmp_project)
        assert code == 1
        assert "DENIED" in out

    def test_can_edit_with_file_scope(self, tmp_project):
        _create_task(tmp_project, "edit-scope", "implement")
        test_file = tmp_project / "src" / "main.py"
        test_file.parent.mkdir(parents=True, exist_ok=True)
        test_file.write_text("# test")
        out, code = _run_status(["--can-edit", "--file", str(test_file)], tmp_project)
        assert code == 0
        assert "ALLOWED" in out

    def test_can_edit_with_file_out_of_scope(self, tmp_project):
        _create_task(tmp_project, "edit-scope-out", "implement")
        out, code = _run_status(["--can-edit", "--file", "/tmp/some_random_file.py"], tmp_project)
        assert code == 1
        assert "outside project" in out

    def test_can_edit_json_output(self, tmp_project):
        _create_task(tmp_project, "edit-json", "implement")
        out, code = _run_status(["--can-edit", "--json"], tmp_project)
        assert code == 0
        assert '"allowed": true' in out.lower() or '"allowed": True' in out

    def test_can_edit_json_denied(self, tmp_project):
        _create_task(tmp_project, "edit-json-denied", "research")
        out, code = _run_status(["--can-edit", "--json"], tmp_project)
        assert code == 1
        assert '"allowed": false' in out.lower() or '"allowed": False' in out


class TestAudit:
    def test_audit_reports_manual_task(self, tmp_project):
        tasks_dir = tmp_project / ".automaton" / "tasks"
        task_dir = tasks_dir / "manual-only"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--audit"], tmp_project)
        assert "manually created" in out

    def test_audit_clean_task(self, tmp_project):
        task_dir = _create_task(tmp_project, "clean-audit", "research")
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--audit"], tmp_project)
        assert "PASS" in out


class TestInferStateFromArtifacts:
    def test_spec_only_infers_research(self, tmp_project):
        task_dir = tmp_project / ".automaton" / "tasks" / "spec-only"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--upgrade", "--task", "spec-only"], tmp_project)
        assert code == 0
        state = (task_dir / ".state").read_text().strip()
        assert state == "research"

    def test_implementation_only_infers_bug_find(self, tmp_project):
        task_dir = tmp_project / ".automaton" / "tasks" / "impl-only"
        task_dir.mkdir()
        (task_dir / "IMPLEMENTATION.md").write_text("# Impl")
        out, code = _run_status(["--upgrade", "--task", "impl-only"], tmp_project)
        assert code == 0
        state = (task_dir / ".state").read_text().strip()
        assert state == "bug_find"

    def test_bug_report_only_infers_adversarial_bug_find(self, tmp_project):
        task_dir = tmp_project / ".automaton" / "tasks" / "bug-only"
        task_dir.mkdir()
        (task_dir / "BUG_REPORT.md").write_text("# Bugs")
        out, code = _run_status(["--upgrade", "--task", "bug-only"], tmp_project)
        assert code == 0
        state = (task_dir / ".state").read_text().strip()
        assert state == "adversarial_bug_find"

    def test_spec_and_decomposition_infers_decomposition(self, tmp_project):
        task_dir = tmp_project / ".automaton" / "tasks" / "decomp-test"
        task_dir.mkdir()
        (task_dir / "SPEC.md").write_text("# Spec")
        (task_dir / "DECOMPOSITION.md").write_text("# Decomp")
        out, code = _run_status(["--upgrade", "--task", "decomp-test"], tmp_project)
        assert code == 0
        state = (task_dir / ".state").read_text().strip()
        assert state == "decomposition"


class TestValidateFolderCorruptedState:
    def test_corrupted_empty_state_file_errors(self, tmp_project):
        task_dir = tmp_project / ".automaton" / "tasks" / "corrupt-test"
        task_dir.mkdir()
        (task_dir / ".state").write_text("")
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--validate-folder", "--task", "corrupt-test"], tmp_project)
        assert code == 1
        assert "corrupted" in out.lower()

    def test_whitespace_state_file_errors(self, tmp_project):
        task_dir = tmp_project / ".automaton" / "tasks" / "ws-test"
        task_dir.mkdir()
        (task_dir / ".state").write_text("   \n")
        (task_dir / "SPEC.md").write_text("# Spec")
        out, code = _run_status(["--validate-folder", "--task", "ws-test"], tmp_project)
        assert code == 1
        assert "corrupted" in out.lower()