task new: einen zweiten Schreibweg in den Tracker (ein Posten, keine Seite, #132)
CI / verify (push) Successful in 55s
Release / release (push) Successful in 38s

Files changed:
- CHANGES.md
- VERSION
- docs/knowledge-and-commitment.md
- instructions/wiki-ingest/SKILL.md
- tools/CONTRACT.md
- tools/chemenu/cli.py
- tools/chemenu/commands/task_cmd.py
- tools/chemenu/tasks/protocol.py
- tools/chemenu/tasks/superproductivity.py
- tools/chemenu/tests/test_instructions_cmd.py
- tools/chemenu/tests/test_superproductivity.py
- tools/chemenu/tests/test_task_cmd.py
This commit is contained in:
torben committed 2026-09-22 11:55:06 +02:00
1 parent e07d1ca42a
commit cfbe3ea83e
12 files changed
+845 -45

No files matched your search

+2
View File
@@ -32,6 +32,7 @@ try:
review_cmd,
run_budget,
search as search_module,
task_cmd,
touch as touch_module,
types_cmd,
upload_cmd,
@@ -152,6 +153,7 @@ app.add_typer(dist_cmd.app, name="dist")
app.add_typer(version_cmd.app, name="version")
app.add_typer(migrate_cmd.app, name="migrate")
app.add_typer(upstream_cmd.app, name="upstream")
app.add_typer(task_cmd.app, name="task")
app.command("new")(new_page.new_page_command)
app.command("touch")(touch_module.touch_command)
app.command("rename")(page_ops.rename_command)
+123
View File
@@ -0,0 +1,123 @@
"""`wikitool task new` - create a tracker item, no kb/ page (Gitea #132, #119
D1/D2/D4/D5/D6/D9).
The second write path into the task tracker, alongside `new project`'s own
(`chemenu.commands.new_page._ensure_tracker_project`) - and the last one that
pairing needed, per `docs/knowledge-and-commitment.md`. Unlike `new project`
this never touches `kb/`: an ingest that finds both knowledge and a
commitment in one source runs this command for the commitment and the normal
page-creation commands (`new source`, ...) for the knowledge, as two
independent steps a skill sequences - never as one transaction, because
nothing here shares state with the page-creation path the way `new project`'s
own tracker-then-page order does within a single command.
This module owns only the CLI shape - parsing, the `--project`/`--inbox`
exclusivity (#132 D4), and the `--follow-up-at` date. The one write itself is
`chemenu.tasks.protocol.TaskWriter.create_item`, dispatched through
`chemenu.tasks.build_writer` exactly like `new project` does.
"""
from __future__ import annotations
import datetime
from typing import Optional
import typer
from chemenu import config, tasks
from chemenu.commands._util import fail, success
from chemenu.errors import ValidationError
from chemenu.tasks import config as tasks_config
app = typer.Typer(
help="Create an item in the task tracker (Gitea #132) - never a kb/ page, "
"see `new project` for that pairing."
)
def _parse_follow_up_at(text: str) -> datetime.date:
try:
return datetime.date.fromisoformat(text)
except ValueError:
fail(f"--follow-up-at {text!r} must be YYYY-MM-DD.")
@app.command("new")
def task_new_command(
title: str = typer.Option(..., "--title", help="The item's title. Stored verbatim, never parsed."),
project: Optional[str] = typer.Option(
None,
"--project",
help="An existing tracker project's name (matched case-insensitively). This command "
"never searches or guesses one (Gitea #132 D6) and never creates one - use "
"`wikitool new project` first if it does not exist yet. Exactly one of --project/--inbox "
"is required.",
),
inbox: bool = typer.Option(
False,
"--inbox",
help="File into the tracker's own inbox instead of a project (Gitea #132 D4 'Weg 3') - "
"the deliberately chosen exit when no project fits, never a stand-in for an omitted "
"--project. An item filed here is invisible to `wikitool review`, since every check "
"there is reached through a project name and the inbox has none.",
),
waiting: bool = typer.Option(
False, "--waiting", help="Tag the item WAITING (#119 D9/D30) - the review's check 2 reads this."
),
follow_up_at: Optional[str] = typer.Option(
None,
"--follow-up-at",
help="YYYY-MM-DD. Only meaningful together with --waiting - it is never a due date "
"(#119 D9) and is refused without --waiting.",
),
notes: Optional[str] = typer.Option(
None,
"--notes",
help="A freetext backref, e.g. to the kb/ source page this item came from (Gitea #132 "
"D5). Stored verbatim, never parsed.",
),
):
"""Create one open item in the configured task tracker - no kb/ page.
Tracker-only by design (#132 D1): a source that carries both knowledge
and a commitment gets this command for the commitment and the normal
page-creation commands for the knowledge, run as two separate steps by
the calling skill - see `docs/knowledge-and-commitment.md`.
"""
if bool(project) == inbox:
fail(
"Exactly one of --project <name> or --inbox is required (Gitea #132 D4) - a missing "
"--project is a mistake, not a request for the tracker's inbox."
)
if follow_up_at is not None and not waiting:
fail(
"--follow-up-at only makes sense together with --waiting (#119 D9) - follow_up_at is "
"never a due date on its own."
)
follow_up_date = _parse_follow_up_at(follow_up_at) if follow_up_at is not None else None
cfg = tasks_config.read_config(config.ROOT)
if cfg is None:
fail(
f"No {config.TASKS_CONFIG_FILENAME} - no task tracker is configured, so there is "
"nowhere to create this item. Configure one first."
)
reader = tasks.build_reader(cfg)
try:
writer = tasks.build_writer(cfg, reader)
except ValidationError as exc:
fail(str(exc))
try:
writer.create_item(
title,
project_name=(None if inbox else project),
waiting=waiting,
follow_up_at=follow_up_date,
notes=notes,
)
except ValidationError as exc:
fail(str(exc))
where = "the tracker's inbox" if inbox else f"project '{project}'"
success(f"Created '{title}' in {where}.")
+42
View File
@@ -164,6 +164,48 @@ class TaskWriter(Protocol):
"""
...
def create_item(
self,
title: str,
*,
project_name: Optional[str],
waiting: bool = False,
follow_up_at: Optional[date] = None,
notes: Optional[str] = None,
) -> None:
"""Create one open item - a tracker `Posten`, never a kb/ page
(Gitea #132 D1). `title` is stored verbatim, exactly like
`WaitingItem.title` - never parsed.
`project_name=None` is the caller's own explicit choice of the
tracker's inbox (#132 D4 "Weg 3"), never a stand-in for "no project
was given" - the CLI's own `--inbox` flag is the only thing allowed
to produce it; an omitted `--project` is refused before this is ever
called. A `project_name` that is given must already exist
(case-normalized, #119 D8) - this never creates a project itself and
never searches or guesses one (#132 D6): `chemenu.errors.ValidationError`
if no such project exists.
`waiting`/`follow_up_at` set #119's own WAITING/`follow_up_at` pair
(D9/D30) - the same two machine-readable parts `WaitingItem` reads
back. Raises `ValidationError` if the provider can represent items at
all (it offers `TaskWriter`) but has no way to mark one WAITING right
now - e.g. Super Productivity's `waiting` tag does not exist yet and
tags cannot be created via its API (#132's own verified constraint):
an item is never created *without* the status it was asked for.
`notes` carries D5's freetext backref to a kb/ page - stored
verbatim, never parsed, exactly the posture `WaitingItem.title`
already has for the person named in it.
Unlike `create_project`, this never raises
`chemenu.errors.HumanInterventionRequired`: every provider offering
`TaskWriter` at all has been verified to have a real item-creation
call (#132 - the gap `create_project` hits, no project-creation
endpoint, does not exist on the item side).
"""
...
def find_project(reader: TaskReader, name: str) -> Optional[ProjectSummary]:
"""The project matching `name` case-normalized (#119 D8), or `None`.
+115 -17
View File
@@ -108,6 +108,19 @@ from chemenu.tasks.protocol import (
# - this instance's own convention, not something Super Productivity defines.
WAITING_TAG_TITLE = "waiting"
# Super Productivity's own inbox project id, verified against
# `project.const.ts`/`project.selectors.ts` on `master` (Gitea #132, 2026-09-20):
# a real project entity the store adds to itself if missing
# (`_addInboxProjectIfNecessary`), but `selectUnarchivedProjects` filters it out
# unconditionally by this exact id - so it never appears in `GET /projects`
# (nor in the snapshot path's own `project` entity state, which mirrors that
# filtering, module docstring). `create_item`'s `--inbox` route is the only
# place this module ever writes it; because of the same filter, an item filed
# there is invisible to every `chemenu.review` check that walks
# `TaskReader.projects()` - "Inbox" never appears as a project name to join
# against, not merely one this instance chooses to exclude.
INBOX_PROJECT_ID = "INBOX_PROJECT"
DEFAULT_API_BASE_URL = "http://127.0.0.1:3876"
ACCESS_API = "api"
@@ -391,29 +404,42 @@ def _expect_list(value: Any, what: str) -> list[dict]:
class _ApiClient:
"""The one HTTP transport `SuperProductivityApiReader` uses - a thin,
loudly-failing wrapper, not a general REST client."""
"""The one HTTP transport `SuperProductivityApiReader`/`SuperProductivityWriter`
use - a thin, loudly-failing wrapper, not a general REST client. `get` and
`post` (Gitea #132) share one request/error path, so a shape drift or a
new failure mode only needs handling once."""
def __init__(self, cfg: SuperProductivityConfig):
self._cfg = cfg
def get(self, path: str, *, timeout: float = 10.0) -> Any:
return self._request("GET", path, timeout=timeout)
def post(self, path: str, body: dict, *, timeout: float = 10.0) -> Any:
return self._request("POST", path, body=body, timeout=timeout)
def _request(
self, method: str, path: str, *, body: Optional[dict] = None, timeout: float = 10.0
) -> Any:
url = self._cfg.api_base_url.rstrip("/") + path
request = urllib.request.Request(
url, headers={"Authorization": f"Bearer {self._cfg.api_token}"}
)
headers = {"Authorization": f"Bearer {self._cfg.api_token}"}
data = None
if body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
request = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310
body = response.read()
response_body = response.read()
except urllib.error.HTTPError as exc:
if exc.code == 503:
raise ValidationError(
"superproductivity: API answered 503 APP_NOT_READY for "
f"{path} - the app's backend is up but its renderer is not ready yet. "
"Wait a moment and retry."
f"superproductivity: API answered 503 APP_NOT_READY for "
f"{method} {path} - the app's backend is up but its renderer is not ready "
"yet. Wait a moment and retry."
) from exc
raise ValidationError(
f"superproductivity: API returned HTTP {exc.code} for {path}."
f"superproductivity: API returned HTTP {exc.code} for {method} {path}."
) from exc
except (urllib.error.URLError, OSError) as exc:
raise ValidationError(
@@ -421,10 +447,10 @@ class _ApiClient:
"Is Super Productivity running?"
) from exc
try:
return json.loads(body)
return json.loads(response_body)
except json.JSONDecodeError as exc:
raise ValidationError(
f"superproductivity: API returned unparseable JSON for {path}."
f"superproductivity: API returned unparseable JSON for {method} {path}."
) from exc
@@ -507,15 +533,17 @@ class SuperProductivityApiReader:
class SuperProductivityWriter:
"""`TaskWriter` over the local REST API - except there is no API call
this can actually make, see the module docstring. Only offered by
`chemenu.tasks.build_writer` when `access: "api"` (Gitea #133) - on
`access: "snapshot"` the tracker is read-only from here, and that refusal
happens before this class is ever constructed."""
"""`TaskWriter` over the local REST API. `create_project` never actually
creates anything - see the module docstring; `create_item` (Gitea #132)
does, since `POST /tasks` exists where `POST /projects` does not. Only
offered by `chemenu.tasks.build_writer` when `access: "api"` (Gitea #133)
- on `access: "snapshot"` the tracker is read-only from here, and that
refusal happens before this class is ever constructed."""
def __init__(self, cfg: SuperProductivityConfig, reader):
self._cfg = cfg
self._reader = reader
self._client = _ApiClient(cfg)
def create_project(self, name: str) -> None:
"""Never creates anything. Preflights the name against the read path
@@ -542,3 +570,73 @@ class SuperProductivityWriter:
" 3. Tell the agent you have done this, so it can re-check and continue.",
verify=_verify,
)
def create_item(
self,
title: str,
*,
project_name: Optional[str],
waiting: bool = False,
follow_up_at: Optional[date] = None,
notes: Optional[str] = None,
) -> None:
"""`POST /tasks` (Gitea #132) - the endpoint `create_project` cannot
reach an equivalent of. Resolves every precondition (the target
project's own id, the `waiting` tag's own id) before making the one
write, so a missing precondition never leaves behind a half-written
item - no task without the WAITING status it was asked for."""
if project_name is None:
project_id = INBOX_PROJECT_ID
else:
if find_project(self._reader, project_name) is None:
raise ValidationError(
f"No project named '{project_name}' (case-insensitively) exists in Super "
"Productivity - this command does not create one (Gitea #132 D6). Run "
"`wikitool new project` first, or pass --inbox."
)
project_id = self._project_id(project_name)
body: dict[str, Any] = {"title": title, "projectId": project_id}
if notes:
body["notes"] = notes
if waiting:
body["tagIds"] = [self._waiting_tag_id()]
if follow_up_at is not None:
body["dueDay"] = follow_up_at.isoformat()
self._client.post("/tasks", body)
def _project_id(self, project_name: str) -> str:
"""Super Productivity's own id for `project_name`, read fresh from the
API. `ProjectSummary` (the protocol-level read shape every provider
shares) deliberately carries no id - not every provider has one - so
a writer that needs one reads it itself here rather than the generic
read path growing an SP-specific field for this one caller."""
target = normalize_project_name(project_name)
for record in _expect_list(self._client.get("/projects"), "/projects"):
if normalize_project_name(str(record.get("title", ""))) == target:
project_id = record.get("id")
if isinstance(project_id, str) and project_id:
return project_id
raise ValidationError(
f"superproductivity: project '{project_name}' matched the read path moments ago but "
"its API record now has no usable id - the response shape does not match what this "
"adapter expects."
)
def _waiting_tag_id(self) -> str:
"""The `waiting` tag's own id, or a loud refusal (Gitea #132's own
acceptance criterion): tags cannot be created via this API (only
`GET /tags` exists, module docstring), so a WAITING item is never
created without its status - the precondition is checked before
`POST /tasks` is ever called, not patched up after."""
for record in _expect_list(self._client.get("/tags"), "/tags"):
if str(record.get("title", "")).strip().casefold() == WAITING_TAG_TITLE:
tag_id = record.get("id")
if isinstance(tag_id, str) and tag_id:
return tag_id
raise ValidationError(
f"superproductivity: no tag named '{WAITING_TAG_TITLE}' exists - tags cannot be "
"created via the API (only GET /tags, Gitea #132). Create it in Super Productivity "
"first, then retry."
)
+22 -17
View File
@@ -79,23 +79,28 @@ def test_the_real_repo_publishes_every_skill():
} <= names
def test_gtd_weekly_review_skill_names_no_provider():
"""#119 D25/#127 AC1: the skill is the one document meant to read
identically in every instance, whichever task-tracker provider it runs
against - so its body must name none of them, and none of a provider's
file or API shape either. Reads the real repo's file, not a fixture,
because the claim is about what ships, not about the discovery logic."""
text = (config.INSTRUCTIONS_DIR / "gtd-weekly-review" / "SKILL.md").read_text(encoding="utf-8").lower()
forbidden = [
"super productivity",
"azure devops",
"superproductivity",
"db.json",
"rest api",
".wikitool-tasks.json",
]
hits = [term for term in forbidden if term in text]
assert not hits, f"gtd-weekly-review/SKILL.md names a provider or its shape: {hits}"
_NO_PROVIDER_FORBIDDEN_TERMS = [
"super productivity",
"azure devops",
"superproductivity",
"db.json",
"rest api",
".wikitool-tasks.json",
]
@pytest.mark.parametrize("skill_name", ["gtd-weekly-review", "wiki-ingest"])
def test_task_tracker_skills_name_no_provider(skill_name):
"""#119 D25/#127 AC1, extended by #132: any skill that can reach the task
tracker - the weekly review, and now the ingest skill's own commitment
step (`task new`) - is meant to read identically in every instance,
whichever provider it runs against, so its body must name none of them,
and none of a provider's file or API shape either. Reads the real repo's
files, not a fixture, because the claim is about what ships, not about
the discovery logic."""
text = (config.INSTRUCTIONS_DIR / skill_name / "SKILL.md").read_text(encoding="utf-8").lower()
hits = [term for term in _NO_PROVIDER_FORBIDDEN_TERMS if term in text]
assert not hits, f"{skill_name}/SKILL.md names a provider or its shape: {hits}"
def test_instructions_dev_flat_file_is_discovered(layer):
@@ -563,6 +563,157 @@ def test_api_reader_source_is_live_and_needs_no_network():
assert source.kind == sp.ACCESS_API
# --- write path: create_item (Gitea #132) --------------------------------------
def _make_write_handler(state: dict, *, token: str = "test-token"):
"""A stub API supporting `GET /projects`, `GET /tags` and `POST /tasks`
only - the three routes `create_item` ever touches. `state["posted"]`
collects every request body `POST /tasks` received, so a test can assert
on the exact fields sent (or that nothing was sent at all)."""
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self): # noqa: N802
if self.path == "/health":
self._reply(200, {"ok": True})
return
if self.headers.get("Authorization") != f"Bearer {token}":
self._reply(401, {"error": "unauthorized"})
return
if self.path == "/projects":
self._reply(200, state.get("projects", []))
return
if self.path == "/tags":
self._reply(200, state.get("tags", []))
return
if self.path == "/tasks":
# `find_project`'s preflight goes through the full
# `SuperProductivityApiReader`, which always reads all three
# routes (module docstring) - `create_item` itself never
# reads this one.
self._reply(200, state.get("tasks", []))
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)
self._reply(201, {"id": "new-task", **payload})
return
self._reply(404, {"error": "not found"})
def _reply(self, code: int, payload) -> None:
body = json.dumps(payload).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 _write_api_server(state: dict, *, token: str = "test-token"):
handler_cls = _make_write_handler(state, token=token)
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 _writer_for(server: http.server.HTTPServer, *, token: str = "test-token") -> sp.SuperProductivityWriter:
cfg_ = _api_cfg(server, token=token)
reader = sp.SuperProductivityApiReader(cfg_)
return sp.SuperProductivityWriter(cfg_, reader)
def test_create_item_posts_project_id_resolved_from_the_read_path():
state = {"projects": [
{"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
"taskIds": [], "backlogTaskIds": []},
]}
with _write_api_server(state) as server:
writer = _writer_for(server)
writer.create_item("Rueckruf beim Kunden", project_name="ship CHEMENU 7.0")
assert len(state["posted"]) == 1
assert state["posted"][0] == {"title": "Rueckruf beim Kunden", "projectId": "p1"}
def test_create_item_refuses_an_unknown_project_and_posts_nothing():
state = {"projects": []}
with _write_api_server(state) as server:
writer = _writer_for(server)
with pytest.raises(ValidationError, match="No project named"):
writer.create_item("x", project_name="No Such Project")
assert "posted" not in state
def test_create_item_inbox_route_uses_the_fixed_inbox_project_id():
state = {"projects": []}
with _write_api_server(state) as server:
writer = _writer_for(server)
writer.create_item("Beleg ablegen", project_name=None)
assert state["posted"][0]["projectId"] == sp.INBOX_PROJECT_ID
def test_create_item_sets_the_waiting_tag_and_due_day():
state = {
"projects": [{"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
"taskIds": [], "backlogTaskIds": []}],
"tags": [{"id": "tag-wait", "title": "Waiting"}],
}
with _write_api_server(state) as server:
writer = _writer_for(server)
writer.create_item(
"Nachfassen beim Elektriker", project_name="Ship Chemenu 7.0",
waiting=True, follow_up_at=date(2026, 4, 1),
)
posted = state["posted"][0]
assert posted["tagIds"] == ["tag-wait"]
assert posted["dueDay"] == "2026-04-01"
def test_create_item_carries_the_freetext_backref_in_notes():
state = {"projects": [{"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
"taskIds": [], "backlogTaskIds": []}]}
with _write_api_server(state) as server:
writer = _writer_for(server)
writer.create_item(
"Rueckruf", project_name="Ship Chemenu 7.0", notes="Source - Kundenmail 2026-09-20",
)
assert state["posted"][0]["notes"] == "Source - Kundenmail 2026-09-20"
def test_create_item_waiting_without_the_tag_refuses_and_posts_nothing():
"""Gitea #132's own acceptance criterion: a missing `waiting` tag must
fail loud, never create an item without the status it was asked for."""
state = {
"projects": [{"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
"taskIds": [], "backlogTaskIds": []}],
"tags": [],
}
with _write_api_server(state) as server:
writer = _writer_for(server)
with pytest.raises(ValidationError, match="waiting"):
writer.create_item("x", project_name="Ship Chemenu 7.0", waiting=True)
assert "posted" not in state
# --- equivalence: both access paths agree on the same fixture (Gitea #133) --------
def test_snapshot_and_api_readers_agree_on_the_same_fixture(tmp_path):
+280
View File
@@ -0,0 +1,280 @@
"""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, {"ok": 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 _reply(self, code: int, payload: Any) -> None:
body = json.dumps(payload).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