torben
b0c64772cc
Files changed: - .gitea/scripts/start-radicale.sh - .gitea/sp-live/Dockerfile - .gitea/sp-live/resolve-version.sh - .gitea/workflows/ci.yml - .gitea/workflows/sp-live-image.yml - .gitea/workflows/tracker-live.yml - .gitignore - CHANGES.md - DEVELOPMENT.md - INSTALL.md - VERSION - instructions/dev/doc-pull-through.md - instructions/dev/stack-dev/SKILL.md - instructions/dev/testing-conventions.md - instructions/dev/tracker-testing.md - kb/gtd/INDEX.md - kb/gtd/technik/Chemenu 8.0.0 freigeben.md - kb/gtd/technik/Windows nativ unterstützen.md - kb/index.md - tools/CONTRACT.md - tools/chemenu/commands/docs_verify.py - tools/chemenu/commands/doctor.py - tools/chemenu/commands/review_cmd.py - tools/chemenu/commands/task_cmd.py - tools/chemenu/config.py - tools/chemenu/tasks/config.py - tools/chemenu/tests/conftest.py - tools/chemenu/tests/fixtures/sp/MANIFEST.json - tools/chemenu/tests/fixtures/sp/api/health.json - tools/chemenu/tests/fixtures/sp/api/projects.json - tools/chemenu/tests/fixtures/sp/api/tags.json - tools/chemenu/tests/fixtures/sp/api/tasks.json - tools/chemenu/tests/fixtures/sp/seed-backup.json - tools/chemenu/tests/record_sp_fixtures.py - tools/chemenu/tests/sp_headless.py - tools/chemenu/tests/test_doctor.py - tools/chemenu/tests/test_review.py - tools/chemenu/tests/test_sp_recorded.py - tools/chemenu/tests/test_task_cmd.py - tools/chemenu/tests/test_tasks_config.py - tools/chemenu/tests/test_tracker_live.py - tools/chemenu/tests/tracker_live.py - tools/pytest.ini
425 lines
17 KiB
Python
425 lines
17 KiB
Python
"""Tests for `wikitool task new` (Gitea #132) - the CLI adapter over
|
|
`chemenu.tasks.protocol.TaskWriter.create_item`. Mirrors the fixture shape
|
|
`test_new_page.py` uses for `new project`'s own tracker calls: a stub Super
|
|
Productivity local REST API on a loopback socket, `.wikitool-tasks.json`
|
|
pointed at it, and `config.ROOT` repointed at `tmp_path` via the shared
|
|
`kb_dir` fixture. No real Super Productivity instance is ever started
|
|
(`instructions/dev/testing-conventions.md`).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import http.server
|
|
import json
|
|
import threading
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
runner = CliRunner()
|
|
|
|
_TASKS_THRESHOLDS = {
|
|
"stalled_waiting_days": 14, "unpaged_project_weeks": 3, "someday_stale_months": 5,
|
|
}
|
|
_API_TOKEN = "test-token"
|
|
|
|
|
|
def _ms(year: int, month: int, day: int) -> int:
|
|
return int(datetime(year, month, day, tzinfo=timezone.utc).timestamp() * 1000)
|
|
|
|
|
|
def _write_tasks_config(root: Path, base_url: str) -> None:
|
|
(root / ".wikitool-tasks.json").write_text(
|
|
json.dumps({
|
|
"schema": 1, "provider": "superproductivity", "thresholds": _TASKS_THRESHOLDS,
|
|
"superproductivity": {"access": "api", "api_base_url": base_url, "api_token": _API_TOKEN},
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _make_handler(state: dict, *, token: str = _API_TOKEN):
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self): # noqa: N802
|
|
if self.path == "/health":
|
|
self._reply(200, {"server": "up", "rendererReady": True})
|
|
return
|
|
if self.headers.get("Authorization") != f"Bearer {token}":
|
|
self._reply(401, {"error": "unauthorized"})
|
|
return
|
|
if self.path in ("/projects", "/tasks", "/tags"):
|
|
self._reply(200, state.get(self.path.lstrip("/"), []))
|
|
return
|
|
self._reply(404, {"error": "not found"})
|
|
|
|
def do_POST(self): # noqa: N802
|
|
if self.headers.get("Authorization") != f"Bearer {token}":
|
|
self._reply(401, {"error": "unauthorized"})
|
|
return
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
payload = json.loads(self.rfile.read(length)) if length else {}
|
|
if self.path == "/tasks":
|
|
state.setdefault("posted", []).append(payload)
|
|
task_id = f"posted-{len(state['posted'])}"
|
|
record = {"id": task_id, "isDone": False, **payload}
|
|
state.setdefault("tasks", []).append(record)
|
|
for project in state.get("projects", []):
|
|
if project.get("id") == payload.get("projectId"):
|
|
project.setdefault("taskIds", []).append(task_id)
|
|
self._reply(201, record)
|
|
return
|
|
self._reply(404, {"error": "not found"})
|
|
|
|
def do_PATCH(self): # noqa: N802
|
|
if self.headers.get("Authorization") != f"Bearer {token}":
|
|
self._reply(401, {"error": "unauthorized"})
|
|
return
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
payload = json.loads(self.rfile.read(length)) if length else {}
|
|
if self.path.startswith("/tasks/"):
|
|
task_id = self.path[len("/tasks/"):]
|
|
record = next((t for t in state.get("tasks", []) if t.get("id") == task_id), None)
|
|
if record is None:
|
|
self._reply(404, {"code": "TASK_NOT_FOUND", "message": "Task not found"})
|
|
return
|
|
record.update(payload)
|
|
state.setdefault("patched", []).append((task_id, payload))
|
|
self._reply(200, record)
|
|
return
|
|
self._reply(404, {"error": "not found"})
|
|
|
|
def _reply(self, code: int, payload: Any) -> None:
|
|
# The app's real envelope (v19.1.0, Gitea #162).
|
|
if code < 400:
|
|
envelope = {"ok": True, "data": payload}
|
|
else:
|
|
envelope = {"ok": False, "error": {
|
|
"code": payload.get("code", "ERROR"),
|
|
"message": payload.get("message", str(payload.get("error", ""))),
|
|
}}
|
|
body = json.dumps(envelope).encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args): # silence stderr noise during the test run
|
|
pass
|
|
|
|
return Handler
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _api_server(state: dict):
|
|
handler_cls = _make_handler(state)
|
|
server = http.server.HTTPServer(("127.0.0.1", 0), handler_cls)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield server
|
|
finally:
|
|
server.shutdown()
|
|
thread.join(timeout=2)
|
|
|
|
|
|
def _base_url(server: http.server.HTTPServer) -> str:
|
|
return f"http://127.0.0.1:{server.server_address[1]}"
|
|
|
|
|
|
def _invoke(monkeypatch, kb_dir, args):
|
|
import chemenu.config as config
|
|
from chemenu.cli import app
|
|
|
|
monkeypatch.setattr(config, "KB_DIR", kb_dir)
|
|
return runner.invoke(app, args)
|
|
|
|
|
|
def _project_record(title: str = "Ship Chemenu 7.0") -> dict:
|
|
return {"id": "p1", "title": title, "created": _ms(2026, 1, 1), "taskIds": [], "backlogTaskIds": []}
|
|
|
|
|
|
# --- argument validation, no tracker call needed -----------------------------
|
|
|
|
|
|
def test_neither_project_nor_inbox_is_refused(monkeypatch, kb_dir):
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "new", "--title", "x"])
|
|
assert result.exit_code == 1
|
|
assert "Exactly one of --project" in result.output
|
|
|
|
|
|
def test_both_project_and_inbox_is_refused(monkeypatch, kb_dir):
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "x", "--project", "Ship Chemenu 7.0", "--inbox",
|
|
])
|
|
assert result.exit_code == 1
|
|
assert "Exactly one of --project" in result.output
|
|
|
|
|
|
def test_follow_up_at_without_waiting_is_refused(monkeypatch, kb_dir):
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "x", "--project", "Ship Chemenu 7.0",
|
|
"--follow-up-at", "2026-04-01",
|
|
])
|
|
assert result.exit_code == 1
|
|
assert "--waiting" in result.output
|
|
|
|
|
|
def test_malformed_follow_up_at_is_refused(monkeypatch, kb_dir):
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "x", "--project", "Ship Chemenu 7.0",
|
|
"--waiting", "--follow-up-at", "not-a-date",
|
|
])
|
|
assert result.exit_code == 1
|
|
assert "YYYY-MM-DD" in result.output
|
|
|
|
|
|
def test_no_tracker_configured_is_refused(monkeypatch, kb_dir):
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "x", "--project", "Ship Chemenu 7.0",
|
|
])
|
|
assert result.exit_code == 1
|
|
assert "no task tracker is configured" in result.output
|
|
|
|
|
|
# --- against a stub Super Productivity API -----------------------------------
|
|
|
|
|
|
def test_creates_an_item_in_the_named_project(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"projects": [_project_record()]}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "Rueckruf beim Kunden", "--project", "ship CHEMENU 7.0",
|
|
])
|
|
assert result.exit_code == 0, result.output
|
|
assert state["posted"] == [{"title": "Rueckruf beim Kunden", "projectId": "p1"}]
|
|
|
|
|
|
def test_unknown_project_is_refused_and_creates_nothing(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"projects": []}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "x", "--project", "No Such Project",
|
|
])
|
|
assert result.exit_code == 1
|
|
assert "No project named" in result.output
|
|
assert "posted" not in state
|
|
|
|
|
|
def test_inbox_route_files_into_the_fixed_inbox_project(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"projects": []}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "new", "--title", "Beleg ablegen", "--inbox"])
|
|
assert result.exit_code == 0, result.output
|
|
assert state["posted"][0]["projectId"] == "INBOX_PROJECT"
|
|
|
|
|
|
def test_waiting_with_follow_up_at_and_notes(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"projects": [_project_record()], "tags": [{"id": "tag-wait", "title": "waiting"}]}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "Nachfassen", "--project", "Ship Chemenu 7.0",
|
|
"--waiting", "--follow-up-at", "2026-04-01",
|
|
"--notes", "Source - Kundenmail 2026-09-20",
|
|
])
|
|
assert result.exit_code == 0, result.output
|
|
posted = state["posted"][0]
|
|
assert posted["tagIds"] == ["tag-wait"]
|
|
assert posted["dueDay"] == "2026-04-01"
|
|
assert posted["notes"] == "Source - Kundenmail 2026-09-20"
|
|
|
|
|
|
def test_missing_waiting_tag_is_refused_and_creates_nothing(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"projects": [_project_record()], "tags": []}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "x", "--project", "Ship Chemenu 7.0", "--waiting",
|
|
])
|
|
assert result.exit_code == 1
|
|
assert "waiting" in result.output
|
|
assert "posted" not in state
|
|
|
|
|
|
# --- round-trip through the read path (Gitea #132's own acceptance criterion) ---
|
|
|
|
|
|
def test_created_item_appears_in_review_as_an_open_and_overdue_waiting_item(monkeypatch, kb_dir):
|
|
"""Not just "was it posted" - read it back the way `wikitool review`
|
|
actually would, over the same live API, and confirm both check 2's
|
|
finding and the plain open-item count see it."""
|
|
from chemenu.review import CHECK_WAITING_OVERDUE, run_review
|
|
|
|
root = kb_dir.parent
|
|
state = {"projects": [_project_record()], "tags": [{"id": "tag-wait", "title": "waiting"}]}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "Nachfassen beim Elektriker",
|
|
"--project", "Ship Chemenu 7.0", "--waiting", "--follow-up-at", "2026-03-01",
|
|
])
|
|
assert result.exit_code == 0, result.output
|
|
|
|
report = run_review(root, today=date(2026, 4, 1))
|
|
|
|
waiting_findings = [f for f in report.findings if f.check == CHECK_WAITING_OVERDUE]
|
|
assert len(waiting_findings) == 1
|
|
assert "Nachfassen beim Elektriker" in waiting_findings[0].message
|
|
|
|
|
|
def test_snapshot_access_has_no_write_path(monkeypatch, kb_dir, tmp_path):
|
|
"""Gitea #133: `access: "snapshot"` never offers a `TaskWriter` at all -
|
|
`task new` must refuse the same way `new project` does, exit 1, naming
|
|
the `access: "api"` instance to use instead."""
|
|
root = kb_dir.parent
|
|
backups_dir = root / "backups"
|
|
backups_dir.mkdir()
|
|
(backups_dir / "2026-01-01_000000.json").write_text(
|
|
json.dumps({
|
|
"project": {"ids": [], "entities": {}},
|
|
"task": {"ids": [], "entities": {}},
|
|
"tag": {"ids": [], "entities": {}},
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
(root / ".wikitool-tasks.json").write_text(
|
|
json.dumps({
|
|
"schema": 1, "provider": "superproductivity", "thresholds": _TASKS_THRESHOLDS,
|
|
"superproductivity": {"access": "snapshot", "backups_dir": str(backups_dir)},
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
result = _invoke(monkeypatch, kb_dir, [
|
|
"task", "new", "--title", "x", "--project", "Ship Chemenu 7.0",
|
|
])
|
|
assert result.exit_code == 1
|
|
assert "access: 'api'" in result.output
|
|
|
|
|
|
# --- `task list` (Gitea #138) -------------------------------------------------
|
|
|
|
|
|
def test_list_shows_id_title_and_waiting_marker(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {
|
|
"projects": [{**_project_record(), "taskIds": ["t1", "t2"]}],
|
|
"tasks": [
|
|
{"id": "t1", "title": "Nachfassen beim Elektriker", "isDone": False,
|
|
"tagIds": ["tag-wait"]},
|
|
{"id": "t2", "title": "Kickoff-Meeting vorbereiten", "isDone": False, "tagIds": []},
|
|
],
|
|
"tags": [{"id": "tag-wait", "title": "waiting"}],
|
|
}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "list", "--project", "ship CHEMENU 7.0"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "t1\tNachfassen beim Elektriker [WAITING]" in result.output
|
|
assert "t2\tKickoff-Meeting vorbereiten" in result.output
|
|
assert "t2\tKickoff-Meeting vorbereiten [WAITING]" not in result.output
|
|
|
|
|
|
def test_list_empty_project_says_so(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"projects": [_project_record()], "tasks": []}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "list", "--project", "Ship Chemenu 7.0"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "No open items" in result.output
|
|
|
|
|
|
# --- `task close` (Gitea #138) ------------------------------------------------
|
|
|
|
|
|
def test_close_marks_the_item_done(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"tasks": [{"id": "t1", "title": "x", "isDone": False}]}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "close", "--id", "t1"])
|
|
assert result.exit_code == 0, result.output
|
|
assert state["patched"] == [("t1", {"isDone": True})]
|
|
|
|
|
|
def test_close_unknown_id_is_refused_and_writes_nothing(monkeypatch, kb_dir):
|
|
root = kb_dir.parent
|
|
state = {"tasks": []}
|
|
with _api_server(state) as server:
|
|
_write_tasks_config(root, _base_url(server))
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "close", "--id", "no-such-id"])
|
|
assert result.exit_code == 1
|
|
assert "patched" not in state
|
|
|
|
|
|
def test_close_on_snapshot_access_is_refused_the_same_way_as_task_new(monkeypatch, kb_dir):
|
|
"""Gitea #138 E7/#133: closing is a write like creating - `access:
|
|
'snapshot'` never offers a `TaskWriter`, so `task close` refuses with
|
|
the exact message `task new` already gives on the same configuration."""
|
|
root = kb_dir.parent
|
|
backups_dir = root / "backups"
|
|
backups_dir.mkdir()
|
|
(backups_dir / "2026-01-01_000000.json").write_text(
|
|
json.dumps({
|
|
"project": {"ids": [], "entities": {}},
|
|
"task": {"ids": [], "entities": {}},
|
|
"tag": {"ids": [], "entities": {}},
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
(root / ".wikitool-tasks.json").write_text(
|
|
json.dumps({
|
|
"schema": 1, "provider": "superproductivity", "thresholds": _TASKS_THRESHOLDS,
|
|
"superproductivity": {"access": "snapshot", "backups_dir": str(backups_dir)},
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "close", "--id", "t1"])
|
|
assert result.exit_code == 1
|
|
assert "access: 'api'" in result.output
|
|
|
|
|
|
# --- WIKITOOL_TASKS_CONFIG (Gitea #156) ----------------------------------------------
|
|
|
|
|
|
def test_an_override_naming_a_missing_file_fails_naming_the_path(monkeypatch, kb_dir):
|
|
monkeypatch.setenv("WIKITOOL_TASKS_CONFIG", str(kb_dir.parent / "no-such-profile.json"))
|
|
for args in (
|
|
["task", "list", "--project", "x"],
|
|
["task", "close", "--id", "t1"],
|
|
["task", "new", "--title", "x", "--inbox"],
|
|
):
|
|
result = _invoke(monkeypatch, kb_dir, args)
|
|
assert result.exit_code == 1, args
|
|
flat = "".join(result.output.split())
|
|
assert "no-such-profile.json" in flat, args
|
|
assert "notasktrackerisconfigured" not in flat, args
|
|
|
|
|
|
def test_an_override_points_a_command_at_another_tracker_without_touching_the_root_file(
|
|
monkeypatch, kb_dir, tmp_path,
|
|
):
|
|
root = kb_dir.parent
|
|
state = {"projects": [{**_project_record(), "taskIds": ["t1"]}],
|
|
"tasks": [{"id": "t1", "title": "Ueber das Profil", "isDone": False, "tagIds": []}]}
|
|
with _api_server(state) as server:
|
|
profile_dir = tmp_path / "profiles"
|
|
profile_dir.mkdir()
|
|
_write_tasks_config(profile_dir, _base_url(server))
|
|
profile = profile_dir / ".wikitool-tasks.json"
|
|
monkeypatch.setenv("WIKITOOL_TASKS_CONFIG", str(profile))
|
|
result = _invoke(monkeypatch, kb_dir, ["task", "list", "--project", "Ship Chemenu 7.0"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "Ueber das Profil" in result.output
|
|
assert not (root / ".wikitool-tasks.json").exists()
|