Marius Mutu
dd8f40774f
18 tests: --dry-run safety, UTC -> Bucharest hour-shift vs. already-tagged Bucharest passthrough, antfarm/night-execute/YouTube: skip list behavior, cd ~/clawd and absolute /home/moltbot/clawd/ rewrites, clawd-archive / clawdbot negative-match guard, duplicate-name preserving existing entry, --skip-disabled / --skip / --channel flags, non-cron schedule safe-skip, translate_job enabled/model field preservation.
323 lines
10 KiB
Python
323 lines
10 KiB
Python
"""Tests for tools/migrations/import_openclaw_jobs_2026-04.py
|
|
|
|
The script is a one-shot translator. We test:
|
|
- --dry-run does not mutate the target file
|
|
- UTC -> Bucharest cron conversion actually changes the hour field
|
|
- antfarm/* are excluded by the default skip list
|
|
- clawd path rewrites happen for `cd ~/clawd` and `/home/moltbot/clawd/`
|
|
- `clawd-archive` / `clawdbot` etc. are NOT matched
|
|
- Duplicate job names in target cause a skip-with-warning, not a crash
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT_PATH = (
|
|
PROJECT_ROOT / "tools" / "migrations" / "import_openclaw_jobs_2026-04.py"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def mod():
|
|
"""Load the migration script as a Python module."""
|
|
spec = importlib.util.spec_from_file_location(
|
|
"import_openclaw_jobs_2026_04", SCRIPT_PATH,
|
|
)
|
|
assert spec is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_openclaw_job(
|
|
name: str,
|
|
expr: str = "0 6 * * *",
|
|
tz: str | None = None,
|
|
enabled: bool = True,
|
|
message: str = "hello",
|
|
model: str | None = None,
|
|
kind: str = "cron",
|
|
):
|
|
sched = {"kind": kind, "expr": expr}
|
|
if tz:
|
|
sched["tz"] = tz
|
|
payload = {"kind": "agentTurn", "message": message}
|
|
if model:
|
|
payload["model"] = model
|
|
return {
|
|
"id": f"uuid-{name}",
|
|
"agentId": "echo",
|
|
"name": name,
|
|
"enabled": enabled,
|
|
"schedule": sched,
|
|
"sessionTarget": "isolated",
|
|
"payload": payload,
|
|
"state": {},
|
|
"delivery": {"mode": "none"},
|
|
}
|
|
|
|
|
|
def _write_source(tmp_path: Path, jobs: list[dict]) -> Path:
|
|
src = tmp_path / "openclaw_jobs.json"
|
|
src.write_text(json.dumps({"version": 1, "jobs": jobs}), encoding="utf-8")
|
|
return src
|
|
|
|
|
|
def _write_target(tmp_path: Path, jobs: list[dict]) -> Path:
|
|
tgt = tmp_path / "echo_jobs.json"
|
|
tgt.write_text(json.dumps(jobs), encoding="utf-8")
|
|
return tgt
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_dry_run_does_not_write(mod, tmp_path, capsys):
|
|
src = _write_source(tmp_path, [_make_openclaw_job("foo")])
|
|
tgt = _write_target(tmp_path, [])
|
|
before = tgt.read_text(encoding="utf-8")
|
|
|
|
rc = mod.run([
|
|
"--dry-run",
|
|
"--source", str(src),
|
|
"--target", str(tgt),
|
|
])
|
|
assert rc == 0
|
|
after = tgt.read_text(encoding="utf-8")
|
|
assert before == after
|
|
captured = capsys.readouterr()
|
|
assert "DRY-RUN" in captured.out
|
|
|
|
|
|
def test_utc_to_bucharest_conversion_shifts_hours(mod):
|
|
"""A UTC cron 0 6 * * * should translate to a non-'0 6' Bucharest expr.
|
|
|
|
Offset varies with DST (+2 or +3), so assert the conversion happened,
|
|
not the exact value.
|
|
"""
|
|
new, warnings = mod.convert_cron_utc_to_bucharest(
|
|
"0 6 * * *", src_tz=None,
|
|
)
|
|
assert new != "0 6 * * *", f"expected conversion, got {new!r}"
|
|
# Minute, day-of-month, month, day-of-week must be unchanged.
|
|
parts = new.split()
|
|
assert parts[0] == "0"
|
|
assert parts[2] == "*"
|
|
assert parts[3] == "*"
|
|
assert parts[4] == "*"
|
|
# Hour is now 8 (winter) or 9 (summer).
|
|
assert parts[1] in {"8", "9"}
|
|
|
|
|
|
def test_bucharest_tz_source_is_unchanged(mod):
|
|
"""If openclaw already marks tz=Europe/Bucharest, we must NOT re-shift."""
|
|
new, warnings = mod.convert_cron_utc_to_bucharest(
|
|
"0 7 * * *", src_tz="Europe/Bucharest",
|
|
)
|
|
assert new == "0 7 * * *"
|
|
assert warnings == []
|
|
|
|
|
|
def test_skip_by_default_excludes_antfarm(mod, tmp_path):
|
|
src = _write_source(tmp_path, [
|
|
_make_openclaw_job("antfarm/feature-dev/planner"),
|
|
_make_openclaw_job("antfarm/feature-dev/developer"),
|
|
_make_openclaw_job("keep-this-one"),
|
|
])
|
|
tgt = _write_target(tmp_path, [])
|
|
|
|
rc = mod.run(["--source", str(src), "--target", str(tgt)])
|
|
assert rc == 0
|
|
result = json.loads(tgt.read_text(encoding="utf-8"))
|
|
names = {j["name"] for j in result}
|
|
assert "keep-this-one" in names
|
|
assert not any(n.startswith("antfarm/") for n in names)
|
|
|
|
|
|
def test_skip_by_default_excludes_night_execute(mod, tmp_path):
|
|
src = _write_source(tmp_path, [_make_openclaw_job("night-execute")])
|
|
tgt = _write_target(tmp_path, [])
|
|
rc = mod.run(["--source", str(src), "--target", str(tgt)])
|
|
assert rc == 0
|
|
result = json.loads(tgt.read_text(encoding="utf-8"))
|
|
assert result == []
|
|
|
|
|
|
def test_youtube_prefix_is_auto_skipped(mod, tmp_path):
|
|
src = _write_source(tmp_path, [
|
|
_make_openclaw_job("YouTube: abc123"),
|
|
_make_openclaw_job("not-youtube"),
|
|
])
|
|
tgt = _write_target(tmp_path, [])
|
|
rc = mod.run(["--source", str(src), "--target", str(tgt)])
|
|
assert rc == 0
|
|
names = {j["name"] for j in json.loads(tgt.read_text(encoding="utf-8"))}
|
|
assert "not-youtube" in names
|
|
assert not any(n.startswith("YouTube:") for n in names)
|
|
|
|
|
|
def test_prompt_rewrite_clawd_to_echo_core(mod):
|
|
text = "Hey, please run this: cd ~/clawd && python3 tools/foo.py"
|
|
new, subs = mod.rewrite_prompt_paths(text)
|
|
assert "~/clawd" not in new
|
|
assert "~/echo-core" in new
|
|
assert "cd ~/clawd" not in new
|
|
assert "cd ~/echo-core" in new
|
|
assert len(subs) == 1
|
|
|
|
|
|
def test_prompt_rewrite_absolute_clawd_path(mod):
|
|
text = "read /home/moltbot/clawd/memory/foo.md now"
|
|
new, subs = mod.rewrite_prompt_paths(text)
|
|
assert "/home/moltbot/echo-core/memory/foo.md" in new
|
|
assert "/home/moltbot/clawd/memory/foo.md" not in new
|
|
assert len(subs) == 1
|
|
|
|
|
|
def test_prompt_rewrite_does_not_match_clawd_archive(mod):
|
|
"""Substrings like `clawd-archive` and `clawdbot` must stay untouched."""
|
|
text = (
|
|
"Folder: /home/moltbot/clawd-archive-old/stuff\n"
|
|
"Other folder: /home/moltbot/clawdbot/data\n"
|
|
"cd ~/clawd-archive\n"
|
|
)
|
|
new, subs = mod.rewrite_prompt_paths(text)
|
|
assert new == text, f"unexpected substitution: {subs}"
|
|
assert subs == []
|
|
|
|
|
|
def test_prompt_rewrite_cd_clawd_with_trailing_space(mod):
|
|
# Whitespace boundary on the shell form
|
|
text = "cd ~/clawd\nls"
|
|
new, _ = mod.rewrite_prompt_paths(text)
|
|
assert "cd ~/echo-core" in new
|
|
|
|
|
|
def test_duplicate_name_warns_skips(mod, tmp_path, capsys):
|
|
"""When target already has a job 'foo', importing 'foo' must preserve
|
|
the existing entry and not write a duplicate."""
|
|
existing = {
|
|
"name": "foo",
|
|
"cron": "0 0 * * *",
|
|
"channel": "echo-core",
|
|
"model": "haiku",
|
|
"prompt": "existing!",
|
|
"allowed_tools": [],
|
|
"enabled": True,
|
|
}
|
|
tgt = _write_target(tmp_path, [existing])
|
|
src = _write_source(tmp_path, [_make_openclaw_job("foo", message="new prompt")])
|
|
|
|
rc = mod.run(["--source", str(src), "--target", str(tgt)])
|
|
assert rc == 0
|
|
|
|
out = capsys.readouterr().out
|
|
assert "DUPE" in out
|
|
assert "already in target" in out
|
|
|
|
result = json.loads(tgt.read_text(encoding="utf-8"))
|
|
assert len(result) == 1
|
|
assert result[0]["prompt"] == "existing!" # untouched
|
|
assert result[0]["model"] == "haiku"
|
|
|
|
|
|
def test_skip_disabled_flag(mod, tmp_path):
|
|
src = _write_source(tmp_path, [
|
|
_make_openclaw_job("enabled-one", enabled=True),
|
|
_make_openclaw_job("disabled-one", enabled=False),
|
|
])
|
|
tgt = _write_target(tmp_path, [])
|
|
rc = mod.run(["--source", str(src), "--target", str(tgt), "--skip-disabled"])
|
|
assert rc == 0
|
|
names = {j["name"] for j in json.loads(tgt.read_text(encoding="utf-8"))}
|
|
assert names == {"enabled-one"}
|
|
|
|
|
|
def test_non_cron_schedule_is_skipped(mod, tmp_path):
|
|
# 'at' (one-shot) is not supported in echo-core; must be skipped, not crash.
|
|
src = _write_source(
|
|
tmp_path,
|
|
[_make_openclaw_job("one-shot", kind="at")],
|
|
)
|
|
# Force kind=at by overriding schedule dict
|
|
data = json.loads(src.read_text(encoding="utf-8"))
|
|
data["jobs"][0]["schedule"] = {
|
|
"kind": "at", "at": "2026-02-09T13:00:00.000Z",
|
|
}
|
|
src.write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
tgt = _write_target(tmp_path, [])
|
|
rc = mod.run(["--source", str(src), "--target", str(tgt)])
|
|
assert rc == 0
|
|
assert json.loads(tgt.read_text(encoding="utf-8")) == []
|
|
|
|
|
|
def test_extra_skip_cli_flag(mod, tmp_path):
|
|
src = _write_source(tmp_path, [
|
|
_make_openclaw_job("a"),
|
|
_make_openclaw_job("b"),
|
|
_make_openclaw_job("c"),
|
|
])
|
|
tgt = _write_target(tmp_path, [])
|
|
rc = mod.run([
|
|
"--source", str(src), "--target", str(tgt),
|
|
"--skip", "a,c",
|
|
])
|
|
assert rc == 0
|
|
names = {j["name"] for j in json.loads(tgt.read_text(encoding="utf-8"))}
|
|
assert names == {"b"}
|
|
|
|
|
|
def test_default_channel_override(mod, tmp_path):
|
|
src = _write_source(tmp_path, [_make_openclaw_job("foo")])
|
|
tgt = _write_target(tmp_path, [])
|
|
rc = mod.run([
|
|
"--source", str(src), "--target", str(tgt),
|
|
"--channel", "echo-sprijin",
|
|
])
|
|
assert rc == 0
|
|
data = json.loads(tgt.read_text(encoding="utf-8"))
|
|
assert data[0]["channel"] == "echo-sprijin"
|
|
|
|
|
|
def test_translate_job_preserves_enabled_flag(mod):
|
|
j = _make_openclaw_job("foo", enabled=True)
|
|
echo, _ = mod.translate_job(j, "echo-work")
|
|
assert echo is not None
|
|
assert echo["enabled"] is True
|
|
|
|
j = _make_openclaw_job("bar", enabled=False)
|
|
echo, _ = mod.translate_job(j, "echo-work")
|
|
assert echo is not None
|
|
assert echo["enabled"] is False
|
|
|
|
|
|
def test_translate_job_defaults_model_to_sonnet(mod):
|
|
j = _make_openclaw_job("foo", model=None)
|
|
echo, _ = mod.translate_job(j, "echo-work")
|
|
assert echo is not None
|
|
assert echo["model"] == "sonnet"
|
|
|
|
|
|
def test_translate_job_respects_explicit_model(mod):
|
|
j = _make_openclaw_job("foo", model="opus")
|
|
echo, _ = mod.translate_job(j, "echo-work")
|
|
assert echo is not None
|
|
assert echo["model"] == "opus"
|