"""Tests for tasks_common pure logic — no I/O beyond tmp_path, no network.""" import json from pathlib import Path import pytest import tasks_common from tasks_common import ( FILENAME_RE, build_task_content, build_task_filename, extract_section, format_age, format_result, format_time, goal_summary, load_preset_names, parse_filename, parse_frontmatter, parse_kv, parse_timestamp, render_list, resolve_preset, ) # --------------------------------------------------------------------------- # parse_frontmatter # --------------------------------------------------------------------------- def test_parse_frontmatter_basic(): content = "---\ncreated: 2026-01-01T10:00:00+00:00\nslug: my-task\n---\n\n# Goal\n\nDo something.\n" fm, body = parse_frontmatter(content) assert fm["created"] == "2026-01-01T10:00:00+00:00" assert fm["slug"] == "my-task" assert "# Goal" in body def test_parse_frontmatter_quoted_chat_id(): content = '---\nchat_id: "12345"\nchannel: telegram\n---\nbody\n' fm, body = parse_frontmatter(content) assert fm["chat_id"] == "12345" assert fm["channel"] == "telegram" assert body == "body\n" def test_parse_frontmatter_no_match(): content = "No frontmatter here." fm, body = parse_frontmatter(content) assert fm == {} assert body == content def test_parse_frontmatter_roundtrip(): original = "---\nfoo: bar\nbaz: qux\n---\nbody text\n" fm, body = parse_frontmatter(original) assert fm == {"foo": "bar", "baz": "qux"} assert body == "body text\n" # --------------------------------------------------------------------------- # parse_kv # --------------------------------------------------------------------------- def test_parse_kv_basic(): text = "completed: 2026-01-02T11:00:00+00:00\nduration_seconds: 42\nstatus: done\n" kv = parse_kv(text) assert kv["completed"] == "2026-01-02T11:00:00+00:00" assert kv["duration_seconds"] == "42" assert kv["status"] == "done" def test_parse_kv_empty(): assert parse_kv("") == {} def test_parse_kv_no_colon_lines_ignored(): kv = parse_kv("no colon here\nkey: value\n") assert kv == {"key": "value"} # --------------------------------------------------------------------------- # FILENAME_RE / parse_filename # --------------------------------------------------------------------------- @pytest.mark.parametrize("name,expected", [ ("2026-01-15T143022-my-task.md", ("2026-01-15T143022", "my-task")), ("2026-12-31T235959-qdrant-vs-weaviate.md", ("2026-12-31T235959", "qdrant-vs-weaviate")), ("2026-06-07_15_00_00_123456-qdrant-vs-weaviate.md", ("2026-06-07_15_00_00_123456", "qdrant-vs-weaviate")), ("2026-06-07_09_05_59_000001-deploy-api.md", ("2026-06-07_09_05_59_000001", "deploy-api")), ]) def test_filename_re_matches(name, expected): assert parse_filename(name) == expected @pytest.mark.parametrize("name", [ "not-a-task.md", "2026-01-15-missing-time.md", "2026-01-15T1430-short.md", "2026-06-07_15_00_123456-too-few-groups.md", ]) def test_filename_re_no_match(name): assert parse_filename(name) is None def test_filename_re_direct_old_format(): assert FILENAME_RE.match("2026-06-02T120000-test-slug.md") is not None def test_filename_re_direct_new_format(): assert FILENAME_RE.match("2026-06-07_15_00_00_123456-test-slug.md") is not None # --------------------------------------------------------------------------- # format_time # --------------------------------------------------------------------------- def test_format_time_old_format(): assert format_time("2026-06-02T143022") == "14:30" def test_format_time_new_format(): assert format_time("2026-06-07_15_00_00_123456") == "15:00" def test_format_time_invalid(): assert format_time("not-a-time") == "not-a-time" # --------------------------------------------------------------------------- # parse_timestamp # --------------------------------------------------------------------------- def test_parse_timestamp_old_format(): dt = parse_timestamp("2026-06-02T143022") assert dt.hour == 14 assert dt.minute == 30 assert dt.second == 22 def test_parse_timestamp_new_format(): dt = parse_timestamp("2026-06-07_15_00_00_123456") assert dt.hour == 15 assert dt.minute == 0 assert dt.microsecond == 123456 def test_parse_timestamp_invalid(): import pytest as _pytest with _pytest.raises(ValueError): parse_timestamp("not-a-timestamp") # --------------------------------------------------------------------------- # format_age # --------------------------------------------------------------------------- def test_format_age_invalid(): assert format_age("bad-value") == "?" def test_format_age_future_clamps_to_zero(): # A timestamp far in the future still returns a non-negative age string. result = format_age("2099-01-01T000000") assert result.endswith("ago") or result == "0s ago" # --------------------------------------------------------------------------- # render_list / goal_summary # --------------------------------------------------------------------------- def _make_task_path(tmp_path: Path, name: str, goal: str = "Test goal.") -> Path: path = tmp_path / name path.write_text( f'---\nslug: test\nchat_id: "1"\nchannel: telegram\n---\n\n# Goal\n\n{goal}\n' ) return path def test_render_list_basic(tmp_path): paths = [ _make_task_path(tmp_path, "2026-06-01T120000-alpha.md", "Alpha goal."), _make_task_path(tmp_path, "2026-06-02T130000-beta.md", "Beta goal."), ] out = render_list(paths, len(paths)) assert "- `alpha`" in out assert "- `beta`" in out assert "|" not in out # no markdown table grammar def test_render_list_shows_goal(tmp_path): paths = [_make_task_path(tmp_path, "2026-06-01T120000-mytask.md", "Research Qdrant.")] out = render_list(paths, 1) assert "- `mytask`" in out assert "\n Research Qdrant." in out # goal on its own indented line def test_render_list_truncation_note(tmp_path): paths = [_make_task_path(tmp_path, "2026-06-01T120000-alpha.md")] out = render_list(paths, 5) assert "(+ 4 older)" in out def test_render_list_no_truncation_note_when_exact(tmp_path): paths = [_make_task_path(tmp_path, "2026-06-01T120000-alpha.md")] out = render_list(paths, 1) assert "older" not in out def test_render_list_unknown_filename(tmp_path): path = tmp_path / "weird-name.md" path.write_text("no frontmatter") out = render_list([path], 1) assert "- `weird-name.md`" in out assert "—" not in out # no table placeholders def test_render_list_new_format_filename(tmp_path): paths = [_make_task_path(tmp_path, "2026-06-07_15_00_00_123456-new-slug.md", "New task.")] out = render_list(paths, 1) assert "- `new-slug`" in out assert "New task." in out assert "15:00" in out def test_render_list_omits_goal_line_when_empty(tmp_path): path = tmp_path / "2026-06-01T120000-nogoal.md" path.write_text('---\nslug: nogoal\n---\n\nNo goal section here.\n') out = render_list([path], 1) assert out.startswith("- `nogoal` · 12:00 · ") assert "\n " not in out # no indented goal line def test_goal_summary_truncates(tmp_path): long_goal = "A" * 100 path = _make_task_path(tmp_path, "2026-06-01T120000-long.md", long_goal) summary = goal_summary(path) assert len(summary) <= 80 assert summary.endswith("…") def test_goal_summary_missing_file(): from pathlib import Path as _Path assert goal_summary(_Path("/nonexistent/file.md")) == "" # --------------------------------------------------------------------------- # extract_section # --------------------------------------------------------------------------- def test_extract_section_found(): text = "# Goal\n\nDo something useful.\n\n# Constraints\n\n- bullet\n" assert extract_section(text, "Goal") == "Do something useful." def test_extract_section_not_found(): assert extract_section("# Goal\n\ntext\n", "Result") is None def test_extract_section_stops_at_next_heading(): text = "# Goal\n\ngoal text\n\n# Result\n\nresult text\n" assert extract_section(text, "Goal") == "goal text" assert extract_section(text, "Result") == "result text" # --------------------------------------------------------------------------- # format_result (uses tmp_path) # --------------------------------------------------------------------------- def _make_task_file(tmp_path: Path, slug: str, goal: str, result_text: str) -> Path: filename = f"2026-06-01T120000-{slug}.md" path = tmp_path / "done" / filename path.parent.mkdir(parents=True, exist_ok=True) content = ( f"---\ncreated: 2026-06-01T12:00:00+02:00\nchannel: telegram\n" f'chat_id: "99"\nslug: {slug}\n---\n\n' f"# Goal\n\n{goal}\n\n# Result\n\n{result_text}\n\n" f"---\ncompleted: 2026-06-01T12:05:00+02:00\nduration_seconds: 300\nstatus: done\n" ) path.write_text(content) return path def test_format_result_contains_slug(tmp_path): path = _make_task_file(tmp_path, "my-slug", "Research X.", "Found Y.") output = format_result(path) assert "my-slug" in output def test_format_result_contains_goal(tmp_path): path = _make_task_file(tmp_path, "task-one", "Research X.", "Found Y.") output = format_result(path) assert "Research X." in output def test_format_result_contains_result(tmp_path): path = _make_task_file(tmp_path, "task-two", "Research X.", "Found Y.") output = format_result(path) assert "Found Y." in output def test_format_result_contains_meta(tmp_path): path = _make_task_file(tmp_path, "task-three", "Do it.", "Done.") output = format_result(path) assert "300" in output # duration_seconds assert "done" in output # --------------------------------------------------------------------------- # build_task_filename # --------------------------------------------------------------------------- def test_build_task_filename_old_format(): name = build_task_filename("2026-06-02T153045", "my-slug") assert name == "2026-06-02T153045-my-slug.md" assert FILENAME_RE.match(name) is not None def test_build_task_filename_new_format(): name = build_task_filename("2026-06-07_15_00_00_123456", "my-slug") assert name == "2026-06-07_15_00_00_123456-my-slug.md" assert FILENAME_RE.match(name) is not None # --------------------------------------------------------------------------- # build_task_content # --------------------------------------------------------------------------- FIXED_TS = "2026-06-02T153045" FIXED_ISO = "2026-06-02T15:30:45+02:00" def test_build_task_content_frontmatter_fields(): content = build_task_content( created_iso=FIXED_ISO, channel="telegram", chat_id="42", slug="test-task", goal="Do the thing.", constraints=[], ) fm, body = parse_frontmatter(content) assert fm["created"] == FIXED_ISO assert fm["channel"] == "telegram" assert fm["chat_id"] == "42" assert fm["slug"] == "test-task" def test_build_task_content_goal_section(): content = build_task_content( created_iso=FIXED_ISO, channel="telegram", chat_id="42", slug="test-task", goal="Do the thing.", constraints=[], ) _, body = parse_frontmatter(content) assert extract_section(body, "Goal") == "Do the thing." def test_build_task_content_constraints_section_has_no_interaction(): content = build_task_content( created_iso=FIXED_ISO, channel="telegram", chat_id="42", slug="test-task", goal="Do the thing.", constraints=[], ) _, body = parse_frontmatter(content) constraints = extract_section(body, "Constraints") assert constraints is not None assert "No user interaction" in constraints def test_build_task_content_extra_constraints(): content = build_task_content( created_iso=FIXED_ISO, channel="telegram", chat_id="42", slug="test-task", goal="Do the thing.", constraints=["Max 5 minutes.", "Output must be JSON."], ) _, body = parse_frontmatter(content) constraints = extract_section(body, "Constraints") assert "Max 5 minutes." in constraints assert "Output must be JSON." in constraints def test_build_task_content_parseable_by_daemon(): """The content produced must be parseable by parse_frontmatter as tasks-daemon does.""" content = build_task_content( created_iso=FIXED_ISO, channel="websocket", chat_id="777", slug="daemon-check", goal="Verify parsing.", constraints=[], ) fm, body = parse_frontmatter(content) assert fm.get("chat_id") == "777" assert fm.get("channel") == "websocket" assert "# Goal" in body assert "# Constraints" in body def test_build_task_content_omits_model_by_default(): content = build_task_content( created_iso=FIXED_ISO, channel="telegram", chat_id="42", slug="test-task", goal="Do the thing.", constraints=[], ) fm, _ = parse_frontmatter(content) assert "model" not in fm def test_build_task_content_includes_model_when_set(): content = build_task_content( created_iso=FIXED_ISO, channel="telegram", chat_id="42", slug="test-task", goal="Do the thing.", constraints=[], model="kimi-k2.6-openrouter", ) fm, _ = parse_frontmatter(content) assert fm["model"] == "kimi-k2.6-openrouter" # --------------------------------------------------------------------------- # resolve_preset # --------------------------------------------------------------------------- PRESETS = ["glm-5.1-ollama", "kimi-k2.6-openrouter", "qwen-3.5-ollama", "qwen-3.6-plus"] def test_resolve_preset_exact_case_insensitive(): assert resolve_preset("Kimi-K2.6-OpenRouter", PRESETS) == "kimi-k2.6-openrouter" def test_resolve_preset_unique_substring(): assert resolve_preset("kimi", PRESETS) == "kimi-k2.6-openrouter" assert resolve_preset("glm", PRESETS) == "glm-5.1-ollama" def test_resolve_preset_unknown_raises_with_available(): with pytest.raises(KeyError) as exc: resolve_preset("gpt5", PRESETS) assert "not found" in exc.value.args[0] assert "kimi-k2.6-openrouter" in exc.value.args[0] def test_resolve_preset_ambiguous_raises_with_candidates(): with pytest.raises(KeyError) as exc: resolve_preset("qwen", PRESETS) assert "ambiguous" in exc.value.args[0] assert "qwen-3.5-ollama" in exc.value.args[0] assert "qwen-3.6-plus" in exc.value.args[0] # --------------------------------------------------------------------------- # load_preset_names (uses tmp_path + monkeypatched CONFIG) # --------------------------------------------------------------------------- def test_load_preset_names_reads_camelcase(tmp_path, monkeypatch): config = tmp_path / "config.json" config.write_text(json.dumps({"modelPresets": {"b-preset": {}, "a-preset": {}}})) monkeypatch.setattr(tasks_common, "CONFIG", config) assert load_preset_names() == ["a-preset", "b-preset"] def test_load_preset_names_reads_snake_case(tmp_path, monkeypatch): config = tmp_path / "config.json" config.write_text(json.dumps({"model_presets": {"kimi": {}, "glm": {}}})) monkeypatch.setattr(tasks_common, "CONFIG", config) assert load_preset_names() == ["glm", "kimi"] def test_load_preset_names_empty_when_absent(tmp_path, monkeypatch): config = tmp_path / "config.json" config.write_text(json.dumps({"channels": {}})) monkeypatch.setattr(tasks_common, "CONFIG", config) assert load_preset_names() == []