codexpi/tests/test_ecosystem.py

from __future__ import annotations

import uuid

import pytest

from codexpi.ecosystem_service import PI_DOCTOR_PROMPT


def initialize_minimal_portfolio(app, projects):
    owner = app.state.db.one("SELECT id FROM users LIMIT 1")["id"]
    for slug in ("pi-doctor", "codex-commander", "autonomous-dev-studio"):
        portfolio = app.state.db.one("SELECT * FROM portfolio_projects WHERE slug=?", (slug,))
        path = projects / slug
        path.mkdir(exist_ok=True)
        workspace_id = str(uuid.uuid4())
        app.state.db.execute(
            "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
            (workspace_id, slug, str(path), owner, "2026-01-01T00:00:00+00:00"),
        )
        conversation = app.state.codex.create_conversation(workspace_id, f"{slug} V0", user_id=owner)
        app.state.db.execute("UPDATE conversations SET portfolio_project_id=? WHERE id=?", (portfolio["id"], conversation["id"]))
        app.state.db.execute("UPDATE portfolio_projects SET workspace_id=?,owner_id=?,status='INITIALIZED' WHERE id=?", (workspace_id, owner, portfolio["id"]))
    return owner


def test_pi_doctor_agent_has_exact_persistent_prompt(logged_in):
    _, app, projects, _, _ = logged_in
    owner = initialize_minimal_portfolio(app, projects)

    agents = app.state.ecosystem.seed_agents(owner)

    doctor = next(item for item in agents if item["key"] == "pi-doctor-repair-agent")
    assert doctor["name"] == "Pi Doctor — Raspberry Pi Repair Agent"
    assert doctor["system_prompt"] == PI_DOCTOR_PROMPT
    session = app.state.db.one("SELECT title,bootstrap_context FROM conversations WHERE id=?", (doctor["conversation_id"],))
    assert session == {"title": doctor["name"], "bootstrap_context": PI_DOCTOR_PROMPT}

    app.state.db.execute("UPDATE conversations SET codex_thread_id='existing-thread',bootstrap_context=NULL WHERE id=?", (doctor["conversation_id"],))
    app.state.ecosystem.seed_agents(owner)
    persisted = app.state.db.one("SELECT bootstrap_context FROM conversations WHERE id=?", (doctor["conversation_id"],))
    assert persisted["bootstrap_context"] == PI_DOCTOR_PROMPT


def test_commander_seeds_seven_roles_with_bounded_permissions(logged_in):
    _, app, projects, _, _ = logged_in
    owner = initialize_minimal_portfolio(app, projects)

    agents = app.state.ecosystem.seed_agents(owner)

    commander = [item for item in agents if item["key"].startswith("commander-")]
    assert len(commander) == 7
    assert all("root" not in item["permissions"] for item in commander)
    assert all(item["conversation_id"] for item in commander)
    assert len({item["conversation_id"] for item in commander}) == 7


def test_file_scope_lock_prevents_two_agents_on_same_files(logged_in):
    _, app, projects, _, _ = logged_in
    owner = initialize_minimal_portfolio(app, projects)
    agents = app.state.ecosystem.seed_agents(owner)
    target = app.state.db.one("SELECT id FROM portfolio_projects WHERE slug='pi-doctor'")["id"]

    app.state.ecosystem.assign_agent(agents[1]["id"], target, "A", "Première tâche", ["backend/api"])

    with pytest.raises(RuntimeError, match="déjà réservé"):
        app.state.ecosystem.assign_agent(agents[2]["id"], target, "B", "Seconde tâche", ["backend/api/routes.py"])

    with pytest.raises(RuntimeError, match="relatif"):
        app.state.ecosystem.assign_agent(agents[2]["id"], target, "C", "Périmètre invalide", ["../secrets"])


def test_repair_plan_is_immutable_and_arbitrary_operation_is_refused(logged_in):
    _, app, *_ = logged_in
    owner = app.state.db.one("SELECT id FROM users LIMIT 1")["id"]
    plan = app.state.ecosystem.create_plan(None, owner, "run_shell", {"command": "id"}, "Commande libre", "Privilèges", True, "Restaurer")

    with pytest.raises(RuntimeError, match="modifié"):
        app.state.ecosystem.approve_plan(plan["id"], "0" * 64, owner)

    approved = app.state.ecosystem.approve_plan(plan["id"], plan["plan_hash"], owner)
    assert approved["status"] == "APPROVED"
    with pytest.raises(PermissionError, match="diagnostic uniquement"):
        app.state.ecosystem.execute_diagnostic(plan["id"], plan["plan_hash"])


def test_incidents_are_deduplicated_and_counted(logged_in):
    _, app, *_ = logged_in
    incident = {"fingerprint": "mount-storage1", "severity": "warning", "service": "/media/Storage1", "symptoms": "NAS unavailable", "details": {"state": "network_or_nas_unavailable"}}

    assert app.state.ecosystem.ingest_incidents([incident, incident]) == 2

    row = app.state.db.one("SELECT * FROM pi_doctor_incidents WHERE fingerprint='mount-storage1'")
    assert row["occurrences"] == 2
    assert row["status"] == "OPEN"


def test_diagnostic_execution_records_result(logged_in):
    _, app, *_ = logged_in
    owner = app.state.db.one("SELECT id FROM users LIMIT 1")["id"]
    plan = app.state.ecosystem.create_plan(None, owner, "diagnostic_snapshot", {}, "Instantané", "Charge brève", False, "Aucune restauration")
    app.state.ecosystem.approve_plan(plan["id"], plan["plan_hash"], owner)

    execution = app.state.ecosystem.execute_diagnostic(plan["id"], plan["plan_hash"])

    assert execution["status"] == "SUCCEEDED"
    assert "cpu_percent" in execution["result"]


def test_collector_snapshot_is_authenticated_persisted_and_used_by_dashboard(logged_in):
    client, app, *_ = logged_in
    token = (app.state.settings.data_dir / "pi-doctor" / "ingest.token").read_text(encoding="utf-8").strip()
    snapshot = {
        "system": {"timestamp": "2026-10-27T12:00:00+00:00", "cpu_percent": 12.5, "memory_percent": 44,
                   "temperature_c": 52.3, "boot_time": "2026-10-27T08:00:00+00:00", "kernel_error_count": 2,
                   "previous_boot_error_count": 1, "abnormal_restart_count": 0},
        "storage": {"root_percent": 63, "mounts": [{"path": "/media/Storage1", "fstype": "cifs", "state": "mounted", "percent": 71}]},
        "docker": {"available": True, "containers": [{"name": "jellyfin", "state": "running", "status": "Up"}]},
        "network": {"nas_smb": {"configured": True, "reachable": True}},
        "services": {"jellyfin": {"available": True}, "codexpi": {"available": True}},
    }

    assert client.post("/api/pi-doctor/snapshots/import", json={"snapshot": snapshot}).status_code == 401
    imported = client.post(
        "/api/pi-doctor/snapshots/import", json={"snapshot": snapshot}, headers={"X-Pi-Doctor-Token": token},
    )
    assert imported.status_code == 201
    dashboard = client.get("/api/pi-doctor").json()
    assert dashboard["snapshot"]["source"] == "collector"
    assert dashboard["snapshot"]["temperature_c"] == 52.3
    assert dashboard["snapshot"]["mounts"][0]["status"] == "mounted"
    assert dashboard["snapshot"]["docker"]["containers"][0]["name"] == "jellyfin"

    token_path = app.state.settings.data_dir / "pi-doctor" / "ingest.token"
    token_path.chmod(0o644)
    refused = client.post(
        "/api/pi-doctor/snapshots/import", json={"snapshot": snapshot}, headers={"X-Pi-Doctor-Token": token},
    )
    assert refused.status_code == 503


def test_agent_assignment_actions_require_completion_before_approval(logged_in):
    client, app, projects, _, headers = logged_in
    owner = initialize_minimal_portfolio(app, projects)
    agent = next(item for item in app.state.ecosystem.seed_agents(owner) if item["key"] == "commander-backend")
    target = app.state.db.one("SELECT id FROM portfolio_projects WHERE slug='pi-doctor'")["id"]
    assignment = app.state.ecosystem.assign_agent(agent["id"], target, "API", "Ajouter une route", ["backend/api"])

    pending = client.post(f"/api/ecosystem/assignments/{assignment['id']}/action", headers=headers, json={"action": "approve"})
    assert pending.status_code == 409
    paused = client.post(f"/api/ecosystem/assignments/{assignment['id']}/action", headers=headers, json={"action": "pause"})
    assert paused.status_code == 200
    assert paused.json()["status"] == "PAUSED"
    resumed = client.post(f"/api/ecosystem/assignments/{assignment['id']}/action", headers=headers, json={"action": "resume"})
    assert resumed.status_code == 200
    assert resumed.json()["status"] == "QUEUED"
    rejected = client.post(f"/api/ecosystem/assignments/{assignment['id']}/action", headers=headers, json={"action": "reject"})
    assert rejected.status_code == 200
    assert rejected.json()["status"] == "REJECTED"
    listed = client.get("/api/ecosystem/agents").json()
    listed_assignment = next(item for item in listed if item["id"] == agent["id"])["assignments"][0]
    assert listed_assignment["status"] == "REJECTED"