mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-30 03:31:36 +00:00
feat: git servers own credentials, one-line project adds with auto-clone, and scheduled agents
Git servers (forge_hosts) become full credential owners: - an encrypted forge token (Fernet, HANDLER_SECRET_KEY) stored per server and never returned by the API (has_token flag only); used automatically by every project on that host and addressable as db:host:<hostname> — the reserved db: credential scheme is now live - a per-server ed25519 SSH deploy key: generated server-side, public half shown in the dashboard to paste into the forge, private half encrypted at rest and materialized 0600 only in the control container (GIT_SSH_COMMAND / core.sshCommand) Project registration gets a git-server mode: pick a registered server, type owner/name, and the API derives the remote (ssh when the server has a deploy key, https otherwise), computes root_dir under PROJECTS_ROOT, and enqueues a new 'sync' command the worker executes (clone, or ff-only pull). Spawn always pulls first, so runs start from the remote's latest state; POST /projects/:p/sync and 'handler sync' re-pull on demand. Schedules: recurring agent spawns (prefix, prompt, interval, role). The worker fires due schedules as ordinary queued spawn commands with timestamped agent names, so runs are fresh stateless agents and appear in the Activity audit trail; missed intervals collapse into one catch-up run. Dashboard: Git Servers pane shows the SSH public key (copy button) and takes a write-only token; Repositories gains the server-first add form and a Pull now button; new Schedules pane. Rebuilt static export. Also restores the missing frontend/lib (api client + format helpers) the components import. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XY1tEhQZXHZ5wci7dLc7rM
This commit is contained in:
Claude
@@ -46,8 +46,8 @@ def test_credential_config_uses_registry_base_url(env):
|
||||
assert "$FORGE_TOKEN" in value
|
||||
|
||||
|
||||
def test_db_scheme_is_reserved_not_yet_resolvable():
|
||||
def test_db_scheme_requires_host_form():
|
||||
import pytest
|
||||
|
||||
with pytest.raises(credentials.CredentialError, match="reserved"):
|
||||
with pytest.raises(credentials.CredentialError, match="db:host:<hostname>"):
|
||||
credentials.resolve("db:42")
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
"""Git servers own their credentials: the encrypted token store, the per-server SSH
|
||||
deploy key (public half visible, private half encrypted), and the resolution paths
|
||||
that hand them to forge/git — including the now-live ``db:host:<hostname>`` scheme."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import stat
|
||||
|
||||
import pytest
|
||||
|
||||
from handler.control import credentials
|
||||
from handler.db import repository as repo
|
||||
from handler.db.engine import get_engine
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def secret_key(env, monkeypatch):
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
from handler import config
|
||||
|
||||
key = Fernet.generate_key().decode()
|
||||
monkeypatch.setenv("HANDLER_SECRET_KEY", key)
|
||||
config.get_settings.cache_clear()
|
||||
return key
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ secret store
|
||||
|
||||
|
||||
def test_secretstore_roundtrip(secret_key):
|
||||
from handler import secretstore
|
||||
|
||||
assert secretstore.enabled()
|
||||
assert secretstore.decrypt(secretstore.encrypt("s3cr3t")) == "s3cr3t"
|
||||
|
||||
|
||||
def test_secretstore_refuses_without_key(env):
|
||||
from handler import secretstore
|
||||
|
||||
assert not secretstore.enabled()
|
||||
with pytest.raises(secretstore.SecretStoreError, match="HANDLER_SECRET_KEY"):
|
||||
secretstore.encrypt("s3cr3t")
|
||||
|
||||
|
||||
def test_secretstore_wrong_key_is_a_clear_error(secret_key, monkeypatch):
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
from handler import config, secretstore
|
||||
|
||||
ciphertext = secretstore.encrypt("s3cr3t")
|
||||
monkeypatch.setenv("HANDLER_SECRET_KEY", Fernet.generate_key().decode())
|
||||
config.get_settings.cache_clear()
|
||||
with pytest.raises(secretstore.SecretStoreError, match="HANDLER_SECRET_KEY changed"):
|
||||
secretstore.decrypt(ciphertext)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ ssh keys
|
||||
|
||||
|
||||
def test_generate_keypair_is_openssh_ed25519():
|
||||
from handler import sshkeys
|
||||
|
||||
private, public = sshkeys.generate_keypair("handler@github.com")
|
||||
assert private.startswith("-----BEGIN OPENSSH PRIVATE KEY-----")
|
||||
assert public.startswith("ssh-ed25519 ")
|
||||
assert public.endswith(" handler@github.com")
|
||||
|
||||
|
||||
def test_materialize_private_key_is_0600_under_projects_root(env):
|
||||
from handler import sshkeys
|
||||
|
||||
path = sshkeys.materialize_private_key("github.com", "KEYDATA")
|
||||
assert path.startswith(str(env["tmp"] / "projects"))
|
||||
assert os.path.basename(path) == "github.com"
|
||||
mode = stat.S_IMODE(os.stat(path).st_mode)
|
||||
assert mode == 0o600
|
||||
with open(path) as fh:
|
||||
assert fh.read() == "KEYDATA\n"
|
||||
|
||||
|
||||
def test_materialize_sanitizes_hostname(env):
|
||||
from handler import sshkeys
|
||||
|
||||
path = sshkeys.materialize_private_key("../evil", "K")
|
||||
assert os.path.dirname(path).endswith(".ssh")
|
||||
assert "/../" not in path[len(str(env["tmp"])):]
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ hosts API
|
||||
|
||||
|
||||
def test_create_host_with_token_and_ssh_key(client, auth, secret_key):
|
||||
r = client.post(
|
||||
"/hosts",
|
||||
json={
|
||||
"hostname": "github.com",
|
||||
"forge_type": "github",
|
||||
"token": "ghp_secret",
|
||||
"generate_ssh_key": True,
|
||||
},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.status_code == 201, r.text
|
||||
body = r.json()
|
||||
assert body["has_token"] is True
|
||||
assert body["ssh_public_key"].startswith("ssh-ed25519 ")
|
||||
# Secrets never leave the server, not even as keys in the payload.
|
||||
assert "token" not in body
|
||||
assert "token_enc" not in body
|
||||
assert "ssh_private_key_enc" not in body
|
||||
|
||||
listed = client.get("/hosts", headers=auth).json()
|
||||
assert listed[0]["has_token"] is True
|
||||
assert "token_enc" not in listed[0]
|
||||
|
||||
# The row itself holds ciphertext, not the token.
|
||||
with get_engine().begin() as conn:
|
||||
row = repo.get_host(conn, "github.com")
|
||||
assert row["token_enc"] != "ghp_secret"
|
||||
from handler import secretstore
|
||||
|
||||
assert secretstore.decrypt(row["token_enc"]) == "ghp_secret"
|
||||
|
||||
|
||||
def test_create_host_token_without_secret_key_is_400(client, auth):
|
||||
r = client.post(
|
||||
"/hosts",
|
||||
json={"hostname": "github.com", "forge_type": "github", "token": "x"},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.status_code == 400
|
||||
assert "HANDLER_SECRET_KEY" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_patch_host_rotate_and_clear(client, auth, secret_key):
|
||||
r = client.post(
|
||||
"/hosts",
|
||||
json={
|
||||
"hostname": "gitea.corp",
|
||||
"forge_type": "gitea",
|
||||
"token": "old",
|
||||
"generate_ssh_key": True,
|
||||
},
|
||||
headers=auth,
|
||||
)
|
||||
first_key = r.json()["ssh_public_key"]
|
||||
|
||||
r = client.patch(
|
||||
"/hosts/gitea.corp", json={"regenerate_ssh_key": True}, headers=auth
|
||||
)
|
||||
assert r.json()["ssh_public_key"] != first_key
|
||||
|
||||
r = client.patch("/hosts/gitea.corp", json={"clear_token": True}, headers=auth)
|
||||
assert r.json()["has_token"] is False
|
||||
|
||||
r = client.patch("/hosts/gitea.corp", json={"clear_ssh_key": True}, headers=auth)
|
||||
assert r.json()["ssh_public_key"] is None
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ resolution
|
||||
|
||||
|
||||
def test_db_host_scheme_resolves_stored_token(secret_key):
|
||||
from handler import secretstore
|
||||
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_host(
|
||||
conn, "github.com", "github", token_enc=secretstore.encrypt("tok123")
|
||||
)
|
||||
assert credentials.resolve("db:host:github.com") == "tok123"
|
||||
|
||||
|
||||
def test_db_host_scheme_missing_host_or_token(secret_key):
|
||||
with pytest.raises(credentials.CredentialError, match="not registered"):
|
||||
credentials.resolve("db:host:nowhere.example")
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_host(conn, "bare.example", "gitea")
|
||||
with pytest.raises(credentials.CredentialError, match="no stored token"):
|
||||
credentials.resolve("db:host:bare.example")
|
||||
|
||||
|
||||
def test_resolve_for_project_falls_back_to_server_token(secret_key):
|
||||
from handler import secretstore
|
||||
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_host(
|
||||
conn, "github.com", "github", token_enc=secretstore.encrypt("srv-tok")
|
||||
)
|
||||
project = {
|
||||
"id": "p",
|
||||
"git_remote": "git@github.com:me/repo.git",
|
||||
"credential_ref": None,
|
||||
}
|
||||
assert credentials.resolve_for_project(project, conn) == "srv-tok"
|
||||
# An explicit credential_ref still wins.
|
||||
os.environ["X_TOKEN"] = "own-tok"
|
||||
try:
|
||||
project["credential_ref"] = "env:X_TOKEN"
|
||||
assert credentials.resolve_for_project(project, conn) == "own-tok"
|
||||
finally:
|
||||
del os.environ["X_TOKEN"]
|
||||
|
||||
|
||||
def test_resolve_for_project_none_without_anything(env):
|
||||
with get_engine().begin() as conn:
|
||||
project = {"id": "p", "git_remote": None, "credential_ref": None}
|
||||
assert credentials.resolve_for_project(project, conn) is None
|
||||
@@ -0,0 +1,206 @@
|
||||
"""Project registration in git-server mode: pick a registered server, type owner/name,
|
||||
and Handler derives the remote + root_dir and enqueues the clone. The worker's ``sync``
|
||||
command (and the reposync module under it) is exercised against the gitops mock seam."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from handler.control import gitops, reposync, worker
|
||||
from handler.db import repository as repo
|
||||
from handler.db.engine import get_engine
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def secret_key(env, monkeypatch):
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
from handler import config
|
||||
|
||||
key = Fernet.generate_key().decode()
|
||||
monkeypatch.setenv("HANDLER_SECRET_KEY", key)
|
||||
config.get_settings.cache_clear()
|
||||
return key
|
||||
|
||||
|
||||
def _add_server(client, auth, hostname="github.com", **extra):
|
||||
body = {"hostname": hostname, "forge_type": "github", **extra}
|
||||
r = client.post("/hosts", json=body, headers=auth)
|
||||
assert r.status_code == 201, r.text
|
||||
return r.json()
|
||||
|
||||
|
||||
def test_create_project_from_server_with_ssh_key(client, auth, env, secret_key):
|
||||
_add_server(client, auth, generate_ssh_key=True)
|
||||
r = client.post(
|
||||
"/projects", json={"git_server": "github.com", "repo": "me/CoolProj"}, headers=auth
|
||||
)
|
||||
assert r.status_code == 201, r.text
|
||||
body = r.json()
|
||||
# id derived from the repo name; root under PROJECTS_ROOT; ssh remote (a key exists).
|
||||
assert body["id"] == "coolproj"
|
||||
assert body["root_dir"] == os.path.join(str(env["tmp"] / "projects"), "coolproj")
|
||||
assert body["git_remote"] == "git@github.com:me/CoolProj.git"
|
||||
# The clone is enqueued for the worker.
|
||||
assert body["sync_command_id"] is not None
|
||||
cmd = client.get(f"/commands/{body['sync_command_id']}", headers=auth).json()
|
||||
assert cmd["type"] == "sync"
|
||||
assert cmd["project_id"] == "coolproj"
|
||||
|
||||
|
||||
def test_create_project_from_server_https_without_key(client, auth, env):
|
||||
_add_server(client, auth, hostname="git.corp", forge_type="gitea",
|
||||
base_url="https://git.corp:8443")
|
||||
r = client.post(
|
||||
"/projects", json={"git_server": "git.corp", "repo": "me/repo", "id": "corp-repo"},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.status_code == 201, r.text
|
||||
assert r.json()["git_remote"] == "https://git.corp:8443/me/repo.git"
|
||||
assert r.json()["id"] == "corp-repo"
|
||||
|
||||
|
||||
def test_create_project_unknown_server_404(client, auth, env):
|
||||
r = client.post(
|
||||
"/projects", json={"git_server": "nowhere.example", "repo": "a/b"}, headers=auth
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_create_project_bad_repo_422(client, auth, env):
|
||||
r = client.post(
|
||||
"/projects", json={"git_server": "github.com", "repo": "not-owner-name"}, headers=auth
|
||||
)
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_manual_mode_still_requires_id_and_root(client, auth, env):
|
||||
r = client.post("/projects", json={"root_dir": "/tmp/x"}, headers=auth)
|
||||
assert r.status_code == 422
|
||||
r = client.post("/projects", json={"id": "x"}, headers=auth)
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_sync_endpoint_enqueues(client, auth, env, tmp_path):
|
||||
root = tmp_path / "proj"
|
||||
root.mkdir()
|
||||
client.post(
|
||||
"/projects",
|
||||
json={"id": "p1", "root_dir": str(root), "git_remote": "https://github.com/a/b.git"},
|
||||
headers=auth,
|
||||
)
|
||||
r = client.post("/projects/p1/sync", headers=auth)
|
||||
assert r.status_code == 202
|
||||
assert r.json()["type"] == "sync"
|
||||
|
||||
|
||||
def test_sync_endpoint_400_without_remote(client, auth, env, tmp_path):
|
||||
root = tmp_path / "proj2"
|
||||
root.mkdir()
|
||||
client.post("/projects", json={"id": "p2", "root_dir": str(root)}, headers=auth)
|
||||
r = client.post("/projects/p2/sync", headers=auth)
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ worker sync command
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_sync_gitops(monkeypatch):
|
||||
"""Fake the clone/pull side of the gitops seam."""
|
||||
state = {"clone": [], "pull": [], "config": [], "repos": set(), "ok": True, "out": ""}
|
||||
|
||||
def is_repo(path):
|
||||
return path in state["repos"]
|
||||
|
||||
def clone(remote, dest, env=None, config=None):
|
||||
state["clone"].append({"remote": remote, "dest": dest, "env": env or {},
|
||||
"config": config or []})
|
||||
if state["ok"]:
|
||||
state["repos"].add(dest)
|
||||
return state["ok"], state["out"]
|
||||
|
||||
def pull_ff(cwd, env=None):
|
||||
state["pull"].append({"cwd": cwd, "env": env or {}})
|
||||
return state["ok"], state["out"]
|
||||
|
||||
def config_local(cwd, key, value):
|
||||
state["config"].append({"cwd": cwd, "key": key, "value": value})
|
||||
return True, ""
|
||||
|
||||
monkeypatch.setattr(gitops, "is_repo", is_repo)
|
||||
monkeypatch.setattr(gitops, "clone", clone)
|
||||
monkeypatch.setattr(gitops, "pull_ff", pull_ff)
|
||||
monkeypatch.setattr(gitops, "config_local", config_local)
|
||||
return state
|
||||
|
||||
|
||||
def _register(conn, project_id="p", remote="https://github.com/me/repo.git", root="/tmp/r"):
|
||||
return repo.create_project(conn, project_id, root_dir=root, git_remote=remote)
|
||||
|
||||
|
||||
def test_cmd_sync_clones_then_pulls(env, fake_sync_gitops):
|
||||
with get_engine().begin() as conn:
|
||||
_register(conn)
|
||||
command = repo.enqueue_command(conn, "sync", project_id="p")
|
||||
|
||||
result = worker.execute_command(command)
|
||||
assert result["action"] == "cloned"
|
||||
assert fake_sync_gitops["clone"][0]["remote"] == "https://github.com/me/repo.git"
|
||||
|
||||
result = worker.execute_command(command)
|
||||
assert result["action"] == "pulled"
|
||||
assert fake_sync_gitops["pull"][0]["cwd"] == "/tmp/r"
|
||||
|
||||
|
||||
def test_cmd_sync_failure_is_command_error(env, fake_sync_gitops):
|
||||
fake_sync_gitops["ok"] = False
|
||||
fake_sync_gitops["out"] = "fatal: repository not found"
|
||||
with get_engine().begin() as conn:
|
||||
_register(conn)
|
||||
command = repo.enqueue_command(conn, "sync", project_id="p")
|
||||
with pytest.raises(worker.CommandError, match="repository not found"):
|
||||
worker.execute_command(command)
|
||||
|
||||
|
||||
def test_sync_uses_server_token_and_installs_helper(env, secret_key, fake_sync_gitops):
|
||||
from handler import secretstore
|
||||
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_host(conn, "github.com", "github",
|
||||
token_enc=secretstore.encrypt("srv-tok"))
|
||||
project = _register(conn)
|
||||
|
||||
result = reposync.sync_project(project)
|
||||
assert result["action"] == "cloned"
|
||||
call = fake_sync_gitops["clone"][0]
|
||||
# Token flows through the env (never argv/disk), helper is scoped to the host and
|
||||
# persisted into the fresh clone for the agents that follow.
|
||||
assert call["env"]["FORGE_TOKEN"] == "srv-tok"
|
||||
assert call["env"]["GITHUB_TOKEN"] == "srv-tok"
|
||||
assert any(k.startswith("credential.https://github.com") for k, _ in call["config"])
|
||||
assert any(c["key"].startswith("credential.") for c in fake_sync_gitops["config"])
|
||||
|
||||
|
||||
def test_sync_ssh_remote_uses_deploy_key(env, secret_key, fake_sync_gitops):
|
||||
from handler import secretstore, sshkeys
|
||||
|
||||
private, public = sshkeys.generate_keypair("handler@github.com")
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_host(conn, "github.com", "github", ssh_public_key=public,
|
||||
ssh_private_key_enc=secretstore.encrypt(private))
|
||||
project = _register(conn, remote="git@github.com:me/repo.git")
|
||||
|
||||
result = reposync.sync_project(project)
|
||||
assert result["action"] == "cloned"
|
||||
env_used = fake_sync_gitops["clone"][0]["env"]
|
||||
assert "GIT_SSH_COMMAND" in env_used
|
||||
assert "IdentitiesOnly=yes" in env_used["GIT_SSH_COMMAND"]
|
||||
# The pinned key is persisted for the agents (core.sshCommand).
|
||||
assert any(c["key"] == "core.sshCommand" for c in fake_sync_gitops["config"])
|
||||
# And the key file was materialized 0600.
|
||||
key_path = env_used["GIT_SSH_COMMAND"].split(" -i ", 1)[1].split(" ")[0]
|
||||
with open(key_path) as fh:
|
||||
assert "OPENSSH PRIVATE KEY" in fh.read()
|
||||
@@ -0,0 +1,146 @@
|
||||
"""Recurring agent spawns: the schedules API and the worker sweep that turns due
|
||||
schedules into queued ``spawn`` commands with timestamped agent names."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from handler.control import worker
|
||||
from handler.db import repository as repo
|
||||
from handler.db.engine import get_engine
|
||||
|
||||
|
||||
def _project(client, auth, tmp_path, project_id="p"):
|
||||
root = tmp_path / project_id
|
||||
root.mkdir(exist_ok=True)
|
||||
r = client.post(
|
||||
"/projects", json={"id": project_id, "root_dir": str(root)}, headers=auth
|
||||
)
|
||||
assert r.status_code == 201, r.text
|
||||
|
||||
|
||||
CONTINUE_TASK = "Read @notes.md, continue from there; before finishing, overwrite that file."
|
||||
|
||||
|
||||
def test_schedule_crud(client, auth, env, tmp_path):
|
||||
_project(client, auth, tmp_path)
|
||||
|
||||
r = client.post(
|
||||
"/projects/p/schedules",
|
||||
json={"name_prefix": "nightly", "task": CONTINUE_TASK, "interval_seconds": 3600},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.status_code == 201, r.text
|
||||
sched = r.json()
|
||||
assert sched["enabled"] is True
|
||||
assert sched["task"] == CONTINUE_TASK
|
||||
# First run fires on the worker's next pass.
|
||||
next_run = datetime.fromisoformat(sched["next_run_at"])
|
||||
assert next_run <= datetime.now(UTC) + timedelta(seconds=5)
|
||||
|
||||
assert client.get("/schedules", headers=auth).json()[0]["id"] == sched["id"]
|
||||
assert client.get("/projects/p/schedules", headers=auth).json()[0]["id"] == sched["id"]
|
||||
|
||||
r = client.patch(
|
||||
f"/schedules/{sched['id']}",
|
||||
json={"interval_seconds": 60, "enabled": False},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.json()["interval_seconds"] == 60
|
||||
assert r.json()["enabled"] is False
|
||||
|
||||
r = client.delete(f"/schedules/{sched['id']}", headers=auth)
|
||||
assert r.status_code == 200
|
||||
assert client.get("/schedules", headers=auth).json() == []
|
||||
|
||||
|
||||
def test_schedule_unknown_project_404(client, auth, env):
|
||||
r = client.post(
|
||||
"/projects/ghost/schedules",
|
||||
json={"name_prefix": "x", "task": "t", "interval_seconds": 60},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_schedule_validation_422(client, auth, env, tmp_path):
|
||||
_project(client, auth, tmp_path)
|
||||
r = client.post(
|
||||
"/projects/p/schedules",
|
||||
json={"name_prefix": "", "task": "t", "interval_seconds": 60},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.status_code == 422
|
||||
r = client.post(
|
||||
"/projects/p/schedules",
|
||||
json={"name_prefix": "x", "task": "t", "interval_seconds": 1},
|
||||
headers=auth,
|
||||
)
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_fire_due_schedules_enqueues_spawn(env, tmp_path):
|
||||
now = datetime.now(UTC)
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_project(conn, "p", root_dir=str(tmp_path))
|
||||
sched = repo.create_schedule(
|
||||
conn,
|
||||
project_id="p",
|
||||
name_prefix="nightly",
|
||||
task=CONTINUE_TASK,
|
||||
interval_seconds=3600,
|
||||
next_run_at=now - timedelta(seconds=1),
|
||||
role="junior",
|
||||
)
|
||||
|
||||
assert worker.fire_due_schedules(now) == 1
|
||||
|
||||
with get_engine().begin() as conn:
|
||||
commands = repo.list_commands(conn)
|
||||
advanced = repo.get_schedule(conn, sched["id"])
|
||||
|
||||
assert len(commands) == 1
|
||||
cmd = commands[0]
|
||||
assert cmd["type"] == "spawn"
|
||||
assert cmd["project_id"] == "p"
|
||||
assert cmd["agent_name"].startswith("nightly-")
|
||||
assert cmd["payload"]["task"] == CONTINUE_TASK
|
||||
assert cmd["payload"]["role"] == "junior"
|
||||
assert cmd["requested_by"] == f"schedule:{sched['id']}"
|
||||
|
||||
# The schedule advanced: it will not re-fire until the next interval.
|
||||
assert advanced["last_run_at"] is not None
|
||||
assert advanced["next_run_at"] > now
|
||||
assert advanced["last_command_id"] == cmd["id"]
|
||||
assert worker.fire_due_schedules(now) == 0
|
||||
|
||||
|
||||
def test_fire_skips_disabled_and_future(env, tmp_path):
|
||||
now = datetime.now(UTC)
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_project(conn, "p", root_dir=str(tmp_path))
|
||||
repo.create_schedule(
|
||||
conn, project_id="p", name_prefix="off", task="t",
|
||||
interval_seconds=60, next_run_at=now - timedelta(seconds=1), enabled=False,
|
||||
)
|
||||
repo.create_schedule(
|
||||
conn, project_id="p", name_prefix="later", task="t",
|
||||
interval_seconds=60, next_run_at=now + timedelta(hours=1),
|
||||
)
|
||||
assert worker.fire_due_schedules(now) == 0
|
||||
|
||||
|
||||
def test_missed_intervals_collapse_into_one_run(env, tmp_path):
|
||||
"""A worker that was down for hours fires once, not once per missed interval."""
|
||||
now = datetime.now(UTC)
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_project(conn, "p", root_dir=str(tmp_path))
|
||||
sched = repo.create_schedule(
|
||||
conn, project_id="p", name_prefix="hourly", task="t",
|
||||
interval_seconds=3600, next_run_at=now - timedelta(hours=10),
|
||||
)
|
||||
assert worker.fire_due_schedules(now) == 1
|
||||
with get_engine().begin() as conn:
|
||||
advanced = repo.get_schedule(conn, sched["id"])
|
||||
assert advanced["next_run_at"] == now + timedelta(hours=1)
|
||||
assert worker.fire_due_schedules(now) == 0
|
||||
Reference in New Issue
Block a user