hermes-agent/tests/hermes_cli/test_web_server_cron_profiles.py

"""Regression tests for dashboard cron job profile routing."""

import pytest
from fastapi import HTTPException


@pytest.fixture()
def isolated_profiles(tmp_path, monkeypatch):
    """Give profile discovery an isolated default home with one named profile."""
    from hermes_cli import profiles

    default_home = tmp_path / ".hermes"
    profiles_root = default_home / "profiles"
    worker_home = profiles_root / "worker_alpha"

    for home in (default_home, worker_home):
        (home / "cron").mkdir(parents=True, exist_ok=True)
        (home / "config.yaml").write_text("model: test-model\n", encoding="utf-8")

    monkeypatch.setattr(profiles, "_get_default_hermes_home", lambda: default_home)
    monkeypatch.setattr(profiles, "_get_profiles_root", lambda: profiles_root)
    return {"default": default_home, "worker_alpha": worker_home}


def test_call_cron_for_profile_routes_storage_and_restores_globals(isolated_profiles):
    from cron import jobs as cron_jobs
    from hermes_cli import web_server

    old_cron_dir = cron_jobs.CRON_DIR
    old_jobs_file = cron_jobs.JOBS_FILE
    old_output_dir = cron_jobs.OUTPUT_DIR

    job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="run scheduled task",
        schedule="every 1h",
        name="worker-alpha-scan",
    )

    assert job["profile"] == "worker_alpha"
    assert job["profile_name"] == "worker_alpha"
    assert job["hermes_home"] == str(isolated_profiles["worker_alpha"])
    assert job["is_default_profile"] is False
    assert (isolated_profiles["worker_alpha"] / "cron" / "jobs.json").exists()
    assert not (isolated_profiles["default"] / "cron" / "jobs.json").exists()

    assert cron_jobs.CRON_DIR == old_cron_dir
    assert cron_jobs.JOBS_FILE == old_jobs_file
    assert cron_jobs.OUTPUT_DIR == old_output_dir


@pytest.mark.asyncio
async def test_list_cron_jobs_all_includes_default_and_named_profiles(isolated_profiles):
    from hermes_cli import web_server

    default_job = web_server._call_cron_for_profile(
        "default",
        "create_job",
        prompt="default heartbeat",
        schedule="every 2h",
        name="default-heartbeat",
    )
    worker_job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="worker heartbeat",
        schedule="every 3h",
        name="worker-alpha-heartbeat",
    )

    jobs = await web_server.list_cron_jobs(profile="all")
    by_id = {job["id"]: job for job in jobs}

    assert set(by_id) >= {default_job["id"], worker_job["id"]}
    assert by_id[default_job["id"]]["profile"] == "default"
    assert by_id[default_job["id"]]["is_default_profile"] is True
    assert by_id[default_job["id"]]["hermes_home"] == str(isolated_profiles["default"])
    assert by_id[worker_job["id"]]["profile"] == "worker_alpha"
    assert by_id[worker_job["id"]]["is_default_profile"] is False
    assert by_id[worker_job["id"]]["hermes_home"] == str(isolated_profiles["worker_alpha"])


@pytest.mark.asyncio
async def test_list_cron_jobs_specific_profile_filters_results(isolated_profiles):
    from hermes_cli import web_server

    web_server._call_cron_for_profile(
        "default",
        "create_job",
        prompt="default only",
        schedule="every 2h",
        name="default-only",
    )
    worker_job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="worker only",
        schedule="every 3h",
        name="worker-only",
    )

    jobs = await web_server.list_cron_jobs(profile="worker_alpha")

    assert [job["id"] for job in jobs] == [worker_job["id"]]
    assert jobs[0]["profile"] == "worker_alpha"


@pytest.mark.asyncio
async def test_create_cron_job_normalizes_representative_core_fields(
    isolated_profiles, tmp_path
):
    from hermes_cli import web_server

    scripts_dir = isolated_profiles["worker_alpha"] / "scripts"
    scripts_dir.mkdir()
    (scripts_dir / "collect-status.py").write_text("print('ok')\n", encoding="utf-8")

    job = await web_server.create_cron_job(
        web_server.CronJobCreate(
            prompt="summarize upstream status",
            schedule="every 1h",
            name="full-core-mapping",
            base_url="https://example.invalid/v1/",
            script=str(scripts_dir / "collect-status.py"),
            no_agent=True,
        ),
        profile="worker_alpha",
    )

    assert job["name"] == "full-core-mapping"
    assert job["base_url"] == "https://example.invalid/v1"
    assert job["script"] == "collect-status.py"
    assert job["no_agent"] is True


@pytest.mark.asyncio
async def test_cron_mutation_without_profile_finds_named_profile_job(isolated_profiles):
    from hermes_cli import web_server

    worker_job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="managed by named profile",
        schedule="every 1h",
        name="named-profile-job",
    )

    paused = await web_server.pause_cron_job(worker_job["id"])
    assert paused["profile"] == "worker_alpha"
    assert paused["enabled"] is False

    default_jobs = await web_server.list_cron_jobs(profile="default")
    worker_jobs = await web_server.list_cron_jobs(profile="worker_alpha")

    assert default_jobs == []
    assert len(worker_jobs) == 1
    assert worker_jobs[0]["id"] == worker_job["id"]
    assert worker_jobs[0]["enabled"] is False


@pytest.mark.asyncio
async def test_update_cron_job_normalizes_dashboard_core_fields(isolated_profiles, tmp_path):
    from hermes_cli import web_server

    scripts_dir = isolated_profiles["worker_alpha"] / "scripts"
    scripts_dir.mkdir()
    (scripts_dir / "collect.py").write_text("print('ok')\n", encoding="utf-8")
    job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="managed by named profile",
        schedule="every 1h",
        name="normalizes-dashboard-fields",
    )

    updated = await web_server.update_cron_job(
        job["id"],
        web_server.CronJobUpdate(
            updates={
                "base_url": "https://example.invalid/v1/",
                "script": str(scripts_dir / "collect.py"),
                "context_from": "",
                "no_agent": True,
            }
        ),
        profile="worker_alpha",
    )

    assert updated["base_url"] == "https://example.invalid/v1"
    assert updated["script"] == "collect.py"
    assert updated["context_from"] is None
    assert updated["no_agent"] is True


@pytest.mark.asyncio
async def test_create_cron_job_rejects_script_outside_profile_scripts(
    isolated_profiles, tmp_path
):
    from hermes_cli import web_server

    outside = tmp_path / "outside.py"
    outside.write_text("print('nope')\n", encoding="utf-8")

    with pytest.raises(HTTPException) as exc:
        await web_server.create_cron_job(
            web_server.CronJobCreate(
                schedule="every 1h",
                script=str(outside),
                no_agent=True,
            ),
            profile="worker_alpha",
        )

    assert exc.value.status_code == 400
    assert "inside" in exc.value.detail


@pytest.mark.asyncio
async def test_create_cron_job_rejects_empty_agent_job(isolated_profiles):
    from hermes_cli import web_server

    with pytest.raises(HTTPException) as exc:
        await web_server.create_cron_job(
            web_server.CronJobCreate(schedule="every 1h"),
            profile="worker_alpha",
        )

    assert exc.value.status_code == 400
    assert "prompt, skill, or script" in exc.value.detail


@pytest.mark.asyncio
async def test_update_cron_job_no_agent_reuses_existing_script(isolated_profiles):
    from hermes_cli import web_server

    scripts_dir = isolated_profiles["worker_alpha"] / "scripts"
    scripts_dir.mkdir()
    (scripts_dir / "collect.py").write_text("print('ok')\n", encoding="utf-8")

    job = await web_server.create_cron_job(
        web_server.CronJobCreate(
            schedule="every 1h",
            script=str(scripts_dir / "collect.py"),
        ),
        profile="worker_alpha",
    )

    updated = await web_server.update_cron_job(
        job["id"],
        web_server.CronJobUpdate(updates={"no_agent": True}),
        profile="worker_alpha",
    )

    assert updated["no_agent"] is True
    assert updated["script"] == "collect.py"


@pytest.mark.asyncio
async def test_dashboard_cron_rejects_missing_context_from(isolated_profiles):
    from hermes_cli import web_server

    with pytest.raises(HTTPException) as create_exc:
        await web_server.create_cron_job(
            web_server.CronJobCreate(
                prompt="process missing upstream",
                schedule="every 1h",
                context_from=["missing-job-id"],
            ),
            profile="worker_alpha",
        )

    assert create_exc.value.status_code == 400
    assert "missing-job-id" in create_exc.value.detail

    job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="managed by named profile",
        schedule="every 1h",
        name="context-update-target",
    )

    with pytest.raises(HTTPException) as update_exc:
        await web_server.update_cron_job(
            job["id"],
            web_server.CronJobUpdate(
                updates={
                    "context_from": ["missing-job-id"],
                }
            ),
            profile="worker_alpha",
        )

    assert update_exc.value.status_code == 400
    assert "missing-job-id" in update_exc.value.detail


@pytest.mark.asyncio
async def test_dashboard_cron_context_from_is_profile_scoped(isolated_profiles):
    from hermes_cli import web_server

    default_job = web_server._call_cron_for_profile(
        "default",
        "create_job",
        prompt="default upstream",
        schedule="every 1h",
        name="default-upstream",
    )
    worker_upstream = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="worker upstream",
        schedule="every 1h",
        name="worker-upstream",
    )

    with pytest.raises(HTTPException):
        await web_server.create_cron_job(
            web_server.CronJobCreate(
                prompt="worker downstream",
                schedule="every 1h",
                context_from=[default_job["id"]],
            ),
            profile="worker_alpha",
        )

    job = await web_server.create_cron_job(
        web_server.CronJobCreate(
            prompt="worker downstream",
            schedule="every 1h",
            context_from=[worker_upstream["id"]],
        ),
        profile="worker_alpha",
    )

    assert job["context_from"] == [worker_upstream["id"]]


@pytest.mark.asyncio
async def test_update_cron_job_refreshes_snapshots_when_unpinning(
    isolated_profiles,
    monkeypatch,
):
    from hermes_cli import runtime_provider, web_server

    monkeypatch.setattr(
        runtime_provider,
        "resolve_runtime_provider",
        lambda **kwargs: {"provider": "worker-provider"},
    )

    job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="managed by named profile",
        schedule="every 1h",
        name="pinned-job",
        provider="fixed-provider",
        model="fixed-model",
    )

    assert job["provider_snapshot"] is None
    assert job["model_snapshot"] is None

    updated = await web_server.update_cron_job(
        job["id"],
        web_server.CronJobUpdate(
            updates={
                "provider": None,
                "model": None,
            }
        ),
        profile="worker_alpha",
    )

    assert updated["provider"] is None
    assert updated["model"] is None
    assert updated["provider_snapshot"] == "worker-provider"
    assert updated["model_snapshot"] == "test-model"


@pytest.mark.asyncio
async def test_dashboard_cron_noop_inference_fields_keep_existing_snapshots(
    isolated_profiles,
    monkeypatch,
):
    from hermes_cli import runtime_provider, web_server

    current_provider = {"name": "initial-provider"}
    monkeypatch.setattr(
        runtime_provider,
        "resolve_runtime_provider",
        lambda **kwargs: {"provider": current_provider["name"]},
    )

    job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="managed by named profile",
        schedule="every 1h",
        name="dashboard-edit-job",
    )

    assert job["provider_snapshot"] == "initial-provider"
    assert job["model_snapshot"] == "test-model"

    current_provider["name"] = "changed-provider"
    (isolated_profiles["worker_alpha"] / "config.yaml").write_text(
        "model: changed-model\n",
        encoding="utf-8",
    )

    updated = await web_server.update_cron_job(
        job["id"],
        web_server.CronJobUpdate(
            updates={
                "name": "dashboard-edit-job-renamed",
                "provider": None,
                "model": None,
                "base_url": None,
                "no_agent": False,
            }
        ),
        profile="worker_alpha",
    )

    assert updated["name"] == "dashboard-edit-job-renamed"
    assert updated["provider_snapshot"] == "initial-provider"
    assert updated["model_snapshot"] == "test-model"


@pytest.mark.asyncio
async def test_update_cron_job_clears_snapshots_for_no_agent(
    isolated_profiles,
    monkeypatch,
):
    from hermes_cli import runtime_provider, web_server

    monkeypatch.setattr(
        runtime_provider,
        "resolve_runtime_provider",
        lambda **kwargs: {"provider": "worker-provider"},
    )
    scripts_dir = isolated_profiles["worker_alpha"] / "scripts"
    scripts_dir.mkdir()
    (scripts_dir / "collect.py").write_text("print('ok')\n", encoding="utf-8")

    job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="managed by named profile",
        schedule="every 1h",
        name="agent-to-script-job",
    )

    assert job["provider_snapshot"] == "worker-provider"
    assert job["model_snapshot"] == "test-model"

    updated = await web_server.update_cron_job(
        job["id"],
        web_server.CronJobUpdate(
            updates={
                "script": str(scripts_dir / "collect.py"),
                "no_agent": True,
            }
        ),
        profile="worker_alpha",
    )

    assert updated["provider_snapshot"] is None
    assert updated["model_snapshot"] is None


@pytest.mark.asyncio
async def test_update_cron_job_rejects_id_mutation(isolated_profiles):
    """Dashboard surfaces a 400 (not a 500 or silent rename) when an
    id-mutation attempt is rejected by cron/jobs.update_job."""
    from hermes_cli import web_server

    worker_job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="managed by named profile",
        schedule="every 1h",
        name="immutable-id-job",
    )

    with pytest.raises(HTTPException) as exc:
        await web_server.update_cron_job(
            worker_job["id"],
            web_server.CronJobUpdate(updates={"id": "../escape"}),
            profile="worker_alpha",
        )

    assert exc.value.status_code == 400
    assert "id" in exc.value.detail
    worker_jobs = await web_server.list_cron_jobs(profile="worker_alpha")
    assert [job["id"] for job in worker_jobs] == [worker_job["id"]]


@pytest.mark.asyncio
async def test_cron_delete_with_profile_deletes_only_target_profile(isolated_profiles):
    from hermes_cli import web_server

    default_job = web_server._call_cron_for_profile(
        "default",
        "create_job",
        prompt="same-ish default",
        schedule="every 1h",
        name="shared-name",
    )
    worker_job = web_server._call_cron_for_profile(
        "worker_alpha",
        "create_job",
        prompt="same-ish worker",
        schedule="every 1h",
        name="shared-name-worker",
    )

    deleted = await web_server.delete_cron_job(worker_job["id"], profile="worker_alpha")
    assert deleted == {"ok": True}

    remaining_default = await web_server.list_cron_jobs(profile="default")
    remaining_worker = await web_server.list_cron_jobs(profile="worker_alpha")
    assert [job["id"] for job in remaining_default] == [default_job["id"]]
    assert remaining_worker == []


@pytest.mark.asyncio
async def test_cron_profile_validation_errors(isolated_profiles):
    from hermes_cli import web_server

    with pytest.raises(HTTPException) as bad_name:
        await web_server.list_cron_jobs(profile="../bad")
    assert bad_name.value.status_code == 400

    with pytest.raises(HTTPException) as missing:
        await web_server.list_cron_jobs(profile="missing_profile")
    assert missing.value.status_code == 404