test(migrations): cover import script translation, skip list, and prompt rewrite
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.
This commit is contained in:
322
tests/test_import_openclaw_jobs.py
Normal file
322
tests/test_import_openclaw_jobs.py
Normal file
@@ -0,0 +1,322 @@
|
||||
"""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"
|
||||
Reference in New Issue
Block a user