diff --git a/tests/test_import_openclaw_jobs.py b/tests/test_import_openclaw_jobs.py new file mode 100644 index 0000000..ccf0150 --- /dev/null +++ b/tests/test_import_openclaw_jobs.py @@ -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"