Add four harder benchmark tasks

All seven newer local models passed the original four tasks 20 of 20,
so the benchmark no longer told them apart. The new tasks keep plans
explicit, as the planner is told to write them, and make the work
itself harder:

- trace-bug: a wrong multi-currency total whose cause is a shared
  rate helper; discounts and tax go through it too, so a fix at the
  call site fails the check.
- refactor-rename: rename load_config across a ten-module package,
  leaving load_config_defaults and reload_config alone.
- spec-parser: implement parse_duration from an eight-rule SPEC.md
  given three examples, and write tests for every rule.
- long-milestone: eight dependent steps adding due dates to a task
  tracker, from the model through storage versions to the CLI.

Self-tests prove that a call-site fix, a blind text replace and a
partial rename all fail their checks.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Opus 5.5 committed 2026-09-27 22:43:39 +00:00
1 parent f2c48d8b16
commit 25fd57351d
67 files changed
+2115 -2

No files matched your search

+12 -1
View File
@@ -393,7 +393,7 @@ declaration.
## Benchmarking models
The capability matrix asks whether a model *can* hold the tool contract. The
benchmark asks whether it *finishes real work*: four canned milestones, each an
benchmark asks whether it *finishes real work*: eight canned milestones, each an
approved plan over a small seed project, judged by a hidden check the model
never sees. Success means the check passes, not that the model said it was
done.
@@ -404,6 +404,17 @@ done.
| `fix-bug` | find and fix the bug a failing test exposes, without touching the test |
| `multi-file` | thread a new option through several files |
| `long-output` | find one failure in test output far longer than `max_output_chars` |
| `trace-bug` | follow a wrong total to its cause in a shared helper, and fix it there rather than at the symptom |
| `refactor-rename` | rename a function across a ten-module package, sparing two similar names a blind replace would break |
| `spec-parser` | implement a parser from an eight-rule written spec when the plan gives only three examples, with tests |
| `long-milestone` | carry out eight dependent steps (model, storage, sorting, CLI) over 25+ turns |
The first four were the original set. By September 2026 all seven newer local
models passed all of them, so the last four were added to separate models that
can hold the tool contract from models that can do a milestone's real work.
Their plans stay explicit, as the planner is told to write them; what makes
them harder is the work itself: tracing a cause, searching a codebase,
reading a spec closely, and holding a plan over many turns.
It runs the executor exactly as the agent does (the same client wiring,
context budget and `tool_choice` handling, and your config's tuning knobs),
@@ -0,0 +1,186 @@
import json
import os
import subprocess
import sys
from datetime import date
from pathlib import Path
import pytest
from tracker import storage
from tracker.core import add_task, complete, overdue, sort_tasks
from tracker.formatting import format_task
from tracker.models import Task
TODAY = date(2026, 10, 5)
# step 1
def test_task_has_an_optional_due_date() -> None:
assert Task(1, "a").due is None
assert Task(1, "a", False, date(2026, 1, 2)).due == date(2026, 1, 2)
# step 2
def test_save_writes_version_2_with_iso_dates(tmp_path: Path) -> None:
path = tmp_path / "t.json"
storage.save(path, [Task(1, "a", due=date(2026, 10, 1)), Task(2, "b")])
data = json.loads(path.read_text())
assert data["version"] == 2
assert [t["due"] for t in data["tasks"]] == ["2026-10-01", None]
def test_round_trip_keeps_dates(tmp_path: Path) -> None:
path = tmp_path / "t.json"
tasks = [Task(1, "a", due=date(2026, 10, 1)), Task(2, "b", done=True)]
storage.save(path, tasks)
assert storage.load(path) == tasks
def test_version_1_files_still_load(tmp_path: Path) -> None:
path = tmp_path / "t.json"
path.write_text(
json.dumps({"version": 1, "tasks": [{"id": 4, "title": "old", "done": True}]})
)
assert storage.load(path) == [Task(4, "old", done=True, due=None)]
@pytest.mark.parametrize("version", [0, 3, 99])
def test_other_versions_are_refused(tmp_path: Path, version: int) -> None:
path = tmp_path / "t.json"
path.write_text(json.dumps({"version": version, "tasks": []}))
with pytest.raises(ValueError, match=f"unsupported tasks file version: {version}"):
storage.load(path)
# step 3
def test_add_task_takes_a_due_date() -> None:
tasks: list[Task] = []
assert add_task(tasks, "a", due=date(2026, 1, 1)).due == date(2026, 1, 1)
assert add_task(tasks, "b").due is None
# step 4
def test_sort_order() -> None:
tasks = [
Task(1, "undated"),
Task(2, "late", due=date(2026, 12, 1)),
Task(3, "early", due=date(2026, 9, 1)),
Task(4, "done dated", done=True, due=date(2026, 1, 1)),
Task(5, "done undated", done=True),
Task(6, "early twin", due=date(2026, 9, 1)),
Task(7, "undated twin"),
]
assert [t.id for t in sort_tasks(tasks)] == [3, 6, 2, 1, 7, 4, 5]
# step 5
def test_overdue() -> None:
tasks = [
Task(1, "due today", due=TODAY),
Task(2, "yesterday", due=date(2026, 10, 4)),
Task(3, "done late", done=True, due=date(2026, 1, 1)),
Task(4, "undated"),
Task(5, "long ago", due=date(2026, 1, 1)),
]
assert [t.id for t in overdue(tasks, TODAY)] == [5, 2]
# step 6
@pytest.mark.parametrize(
("task", "line"),
[
(Task(3, "Buy milk"), "[ ] 3 Buy milk"),
(
Task(4, "Call Bob", done=True, due=date(2026, 10, 1)),
"[x] 4 Call Bob (due 2026-10-01)",
),
(Task(5, "Pay rent", due=date(2026, 9, 1)), "[!] 5 Pay rent (due 2026-09-01)"),
(Task(6, "Soon", due=date(2026, 10, 9)), "[ ] 6 Soon (due 2026-10-09)"),
(Task(7, "Today", due=TODAY), "[ ] 7 Today (due 2026-10-05)"),
],
)
def test_format(task: Task, line: str) -> None:
assert format_task(task, TODAY) == line
# step 7, end to end through the real command line
def run(tmp_path: Path, *args: str) -> subprocess.CompletedProcess[str]:
env = {
**os.environ,
"TRACKER_FILE": str(tmp_path / "t.json"),
"TRACKER_TODAY": "2026-10-05",
}
return subprocess.run(
[sys.executable, "-m", "tracker", *args],
capture_output=True,
text=True,
env=env,
cwd=Path.cwd(),
check=False,
)
def test_command_line_flow(tmp_path: Path) -> None:
assert run(tmp_path, "add", "Buy milk").returncode == 0
assert run(tmp_path, "add", "Pay rent", "--due", "2026-10-01").returncode == 0
assert run(tmp_path, "add", "Dentist", "--due", "2026-11-20").returncode == 0
assert run(tmp_path, "done", "1").returncode == 0
listing = run(tmp_path, "list")
assert listing.stdout.splitlines() == [
"[!] 2 Pay rent (due 2026-10-01)",
"[ ] 3 Dentist (due 2026-11-20)",
"[x] 1 Buy milk",
], listing.stderr
assert run(tmp_path, "list", "--overdue").stdout.splitlines() == [
"[!] 2 Pay rent (due 2026-10-01)"
]
def test_a_bad_due_date_is_a_usage_error(tmp_path: Path) -> None:
result = run(tmp_path, "add", "x", "--due", "2026-13-40")
assert result.returncode == 2
assert "invalid --due date: 2026-13-40" in result.stderr
def test_today_defaults_to_the_real_date(tmp_path: Path) -> None:
env = {**os.environ, "TRACKER_FILE": str(tmp_path / "t.json")}
env.pop("TRACKER_TODAY", None)
subprocess.run(
[sys.executable, "-m", "tracker", "add", "Ancient", "--due", "2000-01-01"],
env=env,
cwd=Path.cwd(),
check=True,
capture_output=True,
)
listing = subprocess.run(
[sys.executable, "-m", "tracker", "list", "--overdue"],
env=env,
cwd=Path.cwd(),
capture_output=True,
text=True,
check=False,
)
assert listing.stdout.splitlines() == ["[!] 1 Ancient (due 2000-01-01)"]
def test_existing_behaviour_is_kept() -> None:
tasks: list[Task] = []
add_task(tasks, "a")
add_task(tasks, "b")
complete(tasks, 1)
assert [t.id for t in sort_tasks(tasks)] == [2, 1]
# step 8
def test_the_projects_own_tests_pass_and_cover_due_dates() -> None:
result = subprocess.run(
[sys.executable, "-m", "pytest", "-q", "tests"],
capture_output=True,
text=True,
check=False,
cwd=Path.cwd(),
)
assert result.returncode == 0, result.stdout[-1500:]
written = "\n".join(p.read_text() for p in Path("tests").glob("test_*.py"))
for name in ("due", "overdue", "version"):
assert name in written, f"no test mentions {name!r}"
@@ -0,0 +1,26 @@
import pytest
from tracker.core import add_task, complete, sort_tasks
from tracker.models import Task
def test_ids_count_up() -> None:
tasks: list[Task] = []
assert [add_task(tasks, t).id for t in ("a", "b", "c")] == [1, 2, 3]
def test_a_title_is_required() -> None:
with pytest.raises(ValueError):
add_task([], " ")
def test_complete_and_sort() -> None:
tasks: list[Task] = []
for title in ("a", "b", "c"):
add_task(tasks, title)
complete(tasks, 1)
assert [t.id for t in sort_tasks(tasks)] == [2, 3, 1]
def test_completing_an_unknown_task() -> None:
with pytest.raises(KeyError):
complete([], 7)
@@ -0,0 +1,25 @@
from pathlib import Path
from tracker import storage
from tracker.cli import main
from tracker.models import Task
def test_round_trip(tmp_path: Path) -> None:
path = tmp_path / "t.json"
storage.save(path, [Task(1, "a"), Task(2, "b", done=True)])
assert storage.load(path) == [Task(1, "a"), Task(2, "b", done=True)]
def test_no_file_is_no_tasks(tmp_path: Path) -> None:
assert storage.load(tmp_path / "missing.json") == []
def test_cli_add_done_list(tmp_path: Path, capsys) -> None:
path = str(tmp_path / "t.json")
main(["--file", path, "add", "Buy milk"])
main(["--file", path, "add", "Call Bob"])
main(["--file", path, "done", "1"])
capsys.readouterr()
main(["--file", path, "list"])
assert capsys.readouterr().out.splitlines() == ["[ ] 2 Call Bob", "[x] 1 Buy milk"]
@@ -0,0 +1 @@
"""A small command-line task tracker."""
@@ -0,0 +1,3 @@
from tracker.cli import main
raise SystemExit(main())
@@ -0,0 +1,41 @@
"""python -m tracker [--file PATH] {add,list,done} ...
The file defaults to $TRACKER_FILE, or tasks.json in the current directory.
"""
import argparse
import os
from pathlib import Path
from tracker import storage
from tracker.core import add_task, complete, sort_tasks
from tracker.formatting import format_task
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="tracker")
parser.add_argument("--file", default=os.environ.get("TRACKER_FILE", "tasks.json"))
commands = parser.add_subparsers(dest="command", required=True)
add = commands.add_parser("add", help="add a task")
add.add_argument("title")
commands.add_parser("list", help="list tasks")
done = commands.add_parser("done", help="mark a task done")
done.add_argument("id", type=int)
args = parser.parse_args(argv)
path = Path(args.file)
tasks = storage.load(path)
if args.command == "add":
task = add_task(tasks, args.title)
storage.save(path, tasks)
print(f"added {task.id}")
elif args.command == "done":
try:
complete(tasks, args.id)
except KeyError as exc:
parser.error(str(exc.args[0]))
storage.save(path, tasks)
else:
for task in sort_tasks(tasks):
print(format_task(task))
return 0
@@ -0,0 +1,25 @@
"""Operations on a list of tasks."""
from tracker.models import Task
def add_task(tasks: list[Task], title: str) -> Task:
title = title.strip()
if not title:
raise ValueError("a task needs a title")
task = Task(id=max((t.id for t in tasks), default=0) + 1, title=title)
tasks.append(task)
return task
def complete(tasks: list[Task], task_id: int) -> Task:
for task in tasks:
if task.id == task_id:
task.done = True
return task
raise KeyError(f"no task {task_id}")
def sort_tasks(tasks: list[Task]) -> list[Task]:
"""Tasks not done first, then by id."""
return sorted(tasks, key=lambda t: (t.done, t.id))
@@ -0,0 +1,8 @@
"""One line per task."""
from tracker.models import Task
def format_task(task: Task) -> str:
mark = "x" if task.done else " "
return f"[{mark}] {task.id} {task.title}"
@@ -0,0 +1,10 @@
"""The task record."""
from dataclasses import dataclass
@dataclass
class Task:
id: int
title: str
done: bool = False
@@ -0,0 +1,24 @@
"""Tasks are kept in one JSON file."""
import json
from pathlib import Path
from tracker.models import Task
VERSION = 1
def load(path: Path) -> list[Task]:
"""Every task in the file; an empty list when there is no file yet."""
if not path.exists():
return []
data = json.loads(path.read_text())
return [Task(id=t["id"], title=t["title"], done=t["done"]) for t in data["tasks"]]
def save(path: Path, tasks: list[Task]) -> None:
data = {
"version": VERSION,
"tasks": [{"id": t.id, "title": t.title, "done": t.done} for t in tasks],
}
path.write_text(json.dumps(data, indent=2))
@@ -0,0 +1,76 @@
import json
from datetime import date
from pathlib import Path
import pytest
from tracker import storage
from tracker.cli import main
from tracker.core import add_task, overdue, sort_tasks
from tracker.formatting import format_task
from tracker.models import Task
def test_round_trip(tmp_path: Path) -> None:
path = tmp_path / "t.json"
tasks = [Task(1, "a"), Task(2, "b", done=True, due=date(2026, 10, 1))]
storage.save(path, tasks)
assert storage.load(path) == tasks
assert json.loads(path.read_text())["version"] == 2
def test_no_file_is_no_tasks(tmp_path: Path) -> None:
assert storage.load(tmp_path / "missing.json") == []
def test_version_1_files_load(tmp_path: Path) -> None:
path = tmp_path / "t.json"
path.write_text(
json.dumps({"version": 1, "tasks": [{"id": 1, "title": "a", "done": False}]})
)
assert storage.load(path) == [Task(1, "a")]
def test_unknown_versions_are_refused(tmp_path: Path) -> None:
path = tmp_path / "t.json"
path.write_text(json.dumps({"version": 9, "tasks": []}))
with pytest.raises(ValueError, match="unsupported tasks file version: 9"):
storage.load(path)
def test_due_dates_sort_and_overdue() -> None:
tasks: list[Task] = []
add_task(tasks, "undated")
add_task(tasks, "later", due=date(2026, 12, 1))
add_task(tasks, "sooner", due=date(2026, 9, 1))
assert [t.title for t in sort_tasks(tasks)] == ["sooner", "later", "undated"]
assert [t.title for t in overdue(tasks, date(2026, 10, 1))] == ["sooner"]
def test_format_marks_overdue() -> None:
today = date(2026, 10, 5)
assert format_task(Task(5, "Pay rent", due=date(2026, 9, 1)), today) == (
"[!] 5 Pay rent (due 2026-09-01)"
)
assert format_task(Task(3, "Buy milk"), today) == "[ ] 3 Buy milk"
def test_cli_add_done_list(tmp_path: Path, capsys, monkeypatch) -> None:
monkeypatch.setenv("TRACKER_TODAY", "2026-10-05")
path = str(tmp_path / "t.json")
main(["--file", path, "add", "Buy milk"])
main(["--file", path, "add", "Pay rent", "--due", "2026-10-01"])
main(["--file", path, "done", "1"])
capsys.readouterr()
main(["--file", path, "list"])
assert capsys.readouterr().out.splitlines() == [
"[!] 2 Pay rent (due 2026-10-01)",
"[x] 1 Buy milk",
]
main(["--file", path, "list", "--overdue"])
assert capsys.readouterr().out.splitlines() == ["[!] 2 Pay rent (due 2026-10-01)"]
def test_cli_rejects_a_bad_date(tmp_path: Path) -> None:
with pytest.raises(SystemExit) as caught:
main(["--file", str(tmp_path / "t.json"), "add", "x", "--due", "soon"])
assert caught.value.code == 2
@@ -0,0 +1,57 @@
"""python -m tracker [--file PATH] {add,list,done} ...
The file defaults to $TRACKER_FILE, or tasks.json in the current directory.
"""
import argparse
import os
from datetime import date
from pathlib import Path
from tracker import storage
from tracker.core import add_task, complete, overdue, sort_tasks
from tracker.formatting import format_task
def today() -> date:
value = os.environ.get("TRACKER_TODAY")
return date.fromisoformat(value) if value else date.today() # noqa: DTZ011 -- the plan asks for the local date
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="tracker")
parser.add_argument("--file", default=os.environ.get("TRACKER_FILE", "tasks.json"))
commands = parser.add_subparsers(dest="command", required=True)
add = commands.add_parser("add", help="add a task")
add.add_argument("title")
add.add_argument("--due", help="due date, YYYY-MM-DD")
listing = commands.add_parser("list", help="list tasks")
listing.add_argument("--overdue", action="store_true", help="only overdue tasks")
done = commands.add_parser("done", help="mark a task done")
done.add_argument("id", type=int)
args = parser.parse_args(argv)
path = Path(args.file)
tasks = storage.load(path)
now = today()
if args.command == "add":
due = None
if args.due is not None:
try:
due = date.fromisoformat(args.due)
except ValueError:
parser.error(f"invalid --due date: {args.due}")
task = add_task(tasks, args.title, due=due)
storage.save(path, tasks)
print(f"added {task.id}")
elif args.command == "done":
try:
complete(tasks, args.id)
except KeyError as exc:
parser.error(str(exc.args[0]))
storage.save(path, tasks)
else:
shown = overdue(tasks, now) if args.overdue else sort_tasks(tasks)
for task in shown:
print(format_task(task, now))
return 0
@@ -0,0 +1,34 @@
"""Operations on a list of tasks."""
from datetime import date
from tracker.models import Task
def add_task(tasks: list[Task], title: str, *, due: date | None = None) -> Task:
title = title.strip()
if not title:
raise ValueError("a task needs a title")
task = Task(id=max((t.id for t in tasks), default=0) + 1, title=title, due=due)
tasks.append(task)
return task
def complete(tasks: list[Task], task_id: int) -> Task:
for task in tasks:
if task.id == task_id:
task.done = True
return task
raise KeyError(f"no task {task_id}")
def sort_tasks(tasks: list[Task]) -> list[Task]:
"""Not done first; then dated tasks, earliest first; then undated; then id."""
return sorted(
tasks,
key=lambda t: (t.done, t.due is None, t.due or date.max, t.id),
)
def overdue(tasks: list[Task], today: date) -> list[Task]:
return [t for t in sort_tasks(tasks) if not t.done and t.due and t.due < today]
@@ -0,0 +1,16 @@
"""One line per task."""
from datetime import date
from tracker.models import Task
def format_task(task: Task, today: date) -> str:
if task.done:
mark = "x"
elif task.due and task.due < today:
mark = "!"
else:
mark = " "
due = f" (due {task.due.isoformat()})" if task.due else ""
return f"[{mark}] {task.id} {task.title}{due}"
@@ -0,0 +1,12 @@
"""The task record."""
from dataclasses import dataclass
from datetime import date
@dataclass
class Task:
id: int
title: str
done: bool = False
due: date | None = None
@@ -0,0 +1,44 @@
"""Tasks are kept in one JSON file."""
import json
from datetime import date
from pathlib import Path
from tracker.models import Task
VERSION = 2
def load(path: Path) -> list[Task]:
"""Every task in the file; an empty list when there is no file yet."""
if not path.exists():
return []
data = json.loads(path.read_text())
version = data.get("version")
if version not in (1, 2):
raise ValueError(f"unsupported tasks file version: {version}")
return [
Task(
id=t["id"],
title=t["title"],
done=t["done"],
due=date.fromisoformat(t["due"]) if t.get("due") else None,
)
for t in data["tasks"]
]
def save(path: Path, tasks: list[Task]) -> None:
data = {
"version": VERSION,
"tasks": [
{
"id": t.id,
"title": t.title,
"done": t.done,
"due": t.due.isoformat() if t.due else None,
}
for t in tasks
],
}
path.write_text(json.dumps(data, indent=2))
@@ -0,0 +1,40 @@
title: Carry out a long milestone whose steps build on each other
check: python -m pytest -q .bench_check
plan: |
Milestone 1: add due dates to the task tracker.
Run `python -m pytest -q` after each step; every test must still pass
before you start the next one. Use `datetime.date` for dates throughout.
1. `tracker/models.py`: add a field `due: date | None = None` to `Task`,
after `done`.
2. `tracker/storage.py`: `save` writes `"version": 2`, and each task's
`"due"` as an ISO string ("2026-10-01") or null. `load` reads version 1
files (their tasks have no "due"; use None) and version 2 files. Any
other version raises `ValueError("unsupported tasks file version: N")`,
with N the version found.
3. `tracker/core.py`: `add_task` takes a keyword argument
`due: date | None = None` and stores it on the new task.
4. `tracker/core.py`: change `sort_tasks` so that tasks not done come
before tasks done; within each of those two groups, tasks with a due
date come first, earliest due date first, then tasks without one; ties
are broken by id.
5. `tracker/core.py`: add `overdue(tasks: list[Task], today: date) -> list[Task]`:
tasks not done whose due date is before `today`, in `sort_tasks` order.
6. `tracker/formatting.py`: `format_task` takes a second argument
`today: date`. A task with a due date gets " (due 2026-10-01)" after its
title. An overdue task (as in step 5) shows `[!]` instead of `[ ]`.
Examples: "[ ] 3 Buy milk", "[x] 4 Call Bob (due 2026-10-01)",
"[!] 5 Pay rent (due 2026-09-01)".
7. `tracker/cli.py`: pass today's date to `format_task`. Today is the
`TRACKER_TODAY` environment variable (YYYY-MM-DD) when it is set, else
`date.today()`. `add` gains an option `--due YYYY-MM-DD`; an invalid date
must call `parser.error("invalid --due date: VALUE")`, which exits with
status 2. `list` gains a flag `--overdue` that lists only overdue tasks.
8. Add tests under `tests/` for each of steps 1-7, run `python -m pytest -q`
until everything passes, then check the whole flow by running:
`TRACKER_FILE=/tmp/tracker-check.json TRACKER_TODAY=2026-10-05 python -m tracker add "Pay rent" --due 2026-10-01`
then the same with `list --overdue`, which must print
`[!] 1 Pay rent (due 2026-10-01)`. Delete /tmp/tracker-check.json
before and after.
9. Call finish.
@@ -0,0 +1,122 @@
import re
import subprocess
import sys
from pathlib import Path
import pytest
ROOT = Path.cwd()
OLD = re.compile(r"\bload_config\b")
def project_files() -> list[Path]:
return [
p
for p in [*ROOT.rglob("*.py"), *ROOT.rglob("*.md")]
if ".bench_check" not in p.parts and "__pycache__" not in p.parts
]
def test_the_old_name_is_gone_everywhere() -> None:
left = [
f"{p.relative_to(ROOT)}:{n}: {line.strip()}"
for p in project_files()
for n, line in enumerate(p.read_text().splitlines(), 1)
if OLD.search(line)
]
assert not left, "load_config still appears:\n" + "\n".join(left)
def test_the_new_name_is_exported() -> None:
import shopkeeper
from shopkeeper.config import read_settings
assert shopkeeper.read_settings is read_settings
assert "read_settings" in shopkeeper.__all__
assert not hasattr(shopkeeper, "load_config")
def test_the_similar_names_were_left_alone() -> None:
from shopkeeper import config
assert callable(config.load_config_defaults)
assert callable(config.reload_config)
assert not hasattr(config, "read_settings_defaults")
assert not hasattr(config, "reread_settings")
readme = (ROOT / "README.md").read_text()
assert "load_config_defaults()" in readme
assert "reload_config(settings)" in readme
def test_the_new_name_appears_in_every_file_that_used_the_old_one() -> None:
for name in [
"catalog",
"pricing",
"inventory",
"notifications",
"orders",
"reports",
"storage",
"cli",
]:
source = (ROOT / "shopkeeper" / f"{name}.py").read_text()
assert "read_settings" in source, f"shopkeeper/{name}.py"
assert "read_settings" in (ROOT / "README.md").read_text()
def test_behaviour_is_unchanged(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
from shopkeeper.config import read_settings, reload_config
(tmp_path / "catalog.csv").write_text("sku,name,unit_price\nSKU-1,bolt,0.25\n")
ini = tmp_path / "shop.ini"
ini.write_text(
f"[shop]\ncurrency = EUR\ntax_percent = 20\ncatalog_file = {tmp_path / 'catalog.csv'}\n"
)
monkeypatch.delenv("SHOPKEEPER_CONFIG", raising=False)
settings = read_settings(ini)
assert (settings.currency, settings.tax_percent, settings.low_stock_threshold) == (
"EUR",
20.0,
5,
)
ini.write_text(ini.read_text().replace("EUR", "GBP"))
assert reload_config(settings).currency == "GBP"
assert read_settings().currency == "USD"
def test_the_projects_own_tests_pass() -> None:
run = subprocess.run(
[sys.executable, "-m", "pytest", "-q", "tests"],
cwd=ROOT,
capture_output=True,
text=True,
check=False,
)
assert run.returncode == 0, run.stdout[-2000:]
def test_the_command_line_still_works(tmp_path: Path) -> None:
(tmp_path / "catalog.csv").write_text("sku,name,unit_price\nSKU-2,hammer,12.50\n")
ini = tmp_path / "shop.ini"
ini.write_text(
f"[shop]\ncurrency = EUR\ntax_percent = 20\ncatalog_file = {tmp_path / 'catalog.csv'}\n"
)
run = subprocess.run(
[
sys.executable,
"-m",
"shopkeeper",
"--config",
str(ini),
"price",
"SKU-2",
"2",
],
cwd=ROOT,
capture_output=True,
text=True,
check=False,
)
assert run.stdout.strip() == "30.00 EUR", run.stderr[-1000:]
@@ -0,0 +1,22 @@
# shopkeeper
A small shop back office: catalogue, pricing, stock, orders and reports.
## Settings
Settings come from an INI file. Load them once and pass them around:
```python
from shopkeeper import load_config
settings = load_config("shop.ini")
```
`load_config()` with no path uses `SHOPKEEPER_CONFIG`, or the built-in
defaults from `load_config_defaults()` when that is unset. After editing the
file, `reload_config(settings)` reads it again.
## Command line
python -m shopkeeper --config shop.ini price SKU-1 3
python -m shopkeeper --config shop.ini stock
@@ -0,0 +1,5 @@
"""A small shop back office."""
from shopkeeper.config import Settings, load_config, load_config_defaults, reload_config
__all__ = ["Settings", "load_config", "load_config_defaults", "reload_config"]
@@ -0,0 +1,3 @@
from shopkeeper.cli import main
raise SystemExit(main())
@@ -0,0 +1,25 @@
"""The product catalogue: SKU, name and unit price, from a CSV file."""
import csv
from dataclasses import dataclass
from pathlib import Path
from shopkeeper.config import Settings, load_config
@dataclass(frozen=True)
class Product:
sku: str
name: str
unit_price: float
def read_catalog(settings: Settings | None = None) -> dict[str, Product]:
"""Every product, keyed by SKU. Uses load_config() when no settings are given."""
settings = settings or load_config()
path = Path(settings.catalog_file)
with path.open(newline="") as handle:
return {
row["sku"]: Product(row["sku"], row["name"], float(row["unit_price"]))
for row in csv.DictReader(handle)
}
@@ -0,0 +1,37 @@
"""Command line: python -m shopkeeper --config FILE COMMAND ..."""
import argparse
from pathlib import Path
from shopkeeper import storage
from shopkeeper.catalog import read_catalog
from shopkeeper.config import load_config
from shopkeeper.pricing import format_price, with_tax
from shopkeeper.reports import stock_report
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="shopkeeper")
parser.add_argument("--config", help="settings file (see load_config)")
parser.add_argument("--stock-file", default="stock.json")
commands = parser.add_subparsers(dest="command", required=True)
price = commands.add_parser(
"price", help="price of a quantity of one SKU, with tax"
)
price.add_argument("sku")
price.add_argument("quantity", type=int)
commands.add_parser("stock", help="stock report")
args = parser.parse_args(argv)
settings = load_config(args.config)
catalog = read_catalog(settings)
if args.command == "price":
amount = with_tax(catalog[args.sku].unit_price * args.quantity, settings)
print(format_price(amount, settings))
else:
print(
stock_report(
catalog, storage.load(Path(args.stock_file), settings), settings
)
)
return 0
@@ -0,0 +1,53 @@
"""Settings, read from an INI file."""
import configparser
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any
ENV_VAR = "SHOPKEEPER_CONFIG"
@dataclass(frozen=True)
class Settings:
currency: str
tax_percent: float
low_stock_threshold: int
catalog_file: str
notify_email: str
source: str | None = None # the file these came from, if any
def load_config_defaults() -> dict[str, Any]:
"""The built-in settings, used for anything the file leaves out."""
return {
"currency": "USD",
"tax_percent": 0.0,
"low_stock_threshold": 5,
"catalog_file": "catalog.csv",
"notify_email": "",
}
def load_config(path: str | Path | None = None) -> Settings:
"""Settings from `path`, or from $SHOPKEEPER_CONFIG, or the defaults.
Values missing from the file fall back to load_config_defaults().
"""
values = load_config_defaults()
chosen = path or os.environ.get(ENV_VAR)
if chosen:
parser = configparser.ConfigParser()
if not parser.read(chosen):
raise FileNotFoundError(f"no settings file at {chosen}")
section = parser["shop"] if parser.has_section("shop") else {}
for key, default in values.items():
if key in section:
values[key] = type(default)(section[key])
return Settings(**values, source=str(chosen) if chosen else None)
def reload_config(settings: Settings) -> Settings:
"""Read the settings again from the file they came from."""
return load_config(settings.source)
@@ -0,0 +1,27 @@
"""Stock levels."""
from shopkeeper.config import Settings, load_config
class Inventory:
def __init__(self, settings: Settings | None = None) -> None:
self.settings = settings or load_config()
self._stock: dict[str, int] = {}
def add(self, sku: str, quantity: int) -> None:
if quantity <= 0:
raise ValueError("quantity must be positive")
self._stock[sku] = self._stock.get(sku, 0) + quantity
def take(self, sku: str, quantity: int) -> None:
if self._stock.get(sku, 0) < quantity:
raise ValueError(f"not enough {sku} in stock")
self._stock[sku] -= quantity
def level(self, sku: str) -> int:
return self._stock.get(sku, 0)
def low_stock(self) -> list[str]:
"""SKUs below the threshold from load_config(), alphabetically."""
limit = self.settings.low_stock_threshold
return sorted(sku for sku, count in self._stock.items() if count < limit)
@@ -0,0 +1,20 @@
"""Messages to the shop owner. Collected in memory; sending is elsewhere."""
import logging
from shopkeeper.config import Settings, load_config
log = logging.getLogger(__name__)
class Notifier:
def __init__(self, settings: Settings | None = None) -> None:
self.settings = settings or load_config()
self.outbox: list[tuple[str, str]] = []
def notify(self, message: str) -> bool:
if not self.settings.notify_email:
log.info("no notify_email in load_config settings; dropped: %s", message)
return False
self.outbox.append((self.settings.notify_email, message))
return True
@@ -0,0 +1,35 @@
"""Orders: take stock, price the lines, warn when stock runs low."""
from dataclasses import dataclass, field
from shopkeeper.catalog import Product
from shopkeeper.config import Settings, load_config
from shopkeeper.inventory import Inventory
from shopkeeper.notifications import Notifier
from shopkeeper.pricing import with_tax
@dataclass
class Order:
lines: list[tuple[Product, int]] = field(default_factory=list)
def add(self, product: Product, quantity: int) -> None:
self.lines.append((product, quantity))
def net(self) -> float:
return round(sum(p.unit_price * q for p, q in self.lines), 2)
def place(
order: Order,
inventory: Inventory,
notifier: Notifier,
settings: Settings | None = None,
) -> float:
"""Take the order's stock and return its price with tax."""
settings = settings or load_config()
for product, quantity in order.lines:
inventory.take(product.sku, quantity)
for sku in inventory.low_stock():
notifier.notify(f"low stock: {sku}")
return with_tax(order.net(), settings)
@@ -0,0 +1,14 @@
"""Prices with tax, in the shop's currency."""
from shopkeeper.config import Settings, load_config
def with_tax(amount: float, settings: Settings | None = None) -> float:
settings = settings or load_config()
return round(amount * (1 + settings.tax_percent / 100), 2)
def format_price(amount: float, settings: Settings | None = None) -> str:
# The currency comes from load_config, never from the catalogue.
settings = settings or load_config()
return f"{amount:.2f} {settings.currency}"
@@ -0,0 +1,21 @@
"""Plain-text reports."""
from shopkeeper.catalog import Product
from shopkeeper.config import Settings, load_config
from shopkeeper.inventory import Inventory
from shopkeeper.pricing import format_price
def stock_report(
catalog: dict[str, Product], inventory: Inventory, settings: Settings | None = None
) -> str:
settings = settings or load_config()
lines = []
for sku in sorted(catalog):
product = catalog[sku]
value = product.unit_price * inventory.level(sku)
lines.append(
f"{sku:<8} {product.name:<16} {inventory.level(sku):>5} "
f"{format_price(value, settings):>14}"
)
return "\n".join(lines)
@@ -0,0 +1,20 @@
"""Saving and loading stock levels as JSON."""
import json
from pathlib import Path
from shopkeeper.config import Settings, load_config
from shopkeeper.inventory import Inventory
def save(inventory: Inventory, path: Path) -> None:
levels = {sku: inventory.level(sku) for sku in inventory._stock}
path.write_text(json.dumps(levels, indent=2, sort_keys=True))
def load(path: Path, settings: Settings | None = None) -> Inventory:
inventory = Inventory(settings or load_config())
for sku, count in json.loads(path.read_text()).items():
if count:
inventory.add(sku, count)
return inventory
@@ -0,0 +1,19 @@
from pathlib import Path
import pytest
@pytest.fixture
def shop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""A shop directory with a settings file and a catalogue."""
(tmp_path / "catalog.csv").write_text(
"sku,name,unit_price\nSKU-1,bolt,0.25\nSKU-2,hammer,12.50\n"
)
ini = tmp_path / "shop.ini"
ini.write_text(
"[shop]\ncurrency = EUR\ntax_percent = 20\nlow_stock_threshold = 3\n"
f"catalog_file = {tmp_path / 'catalog.csv'}\nnotify_email = owner@example.com\n"
)
monkeypatch.delenv("SHOPKEEPER_CONFIG", raising=False)
monkeypatch.chdir(tmp_path)
return ini
@@ -0,0 +1,33 @@
import pytest
from shopkeeper import load_config, load_config_defaults, reload_config
def test_defaults_without_a_file() -> None:
settings = load_config()
assert settings.currency == load_config_defaults()["currency"]
assert settings.source is None
def test_values_from_the_file(shop) -> None:
settings = load_config(shop)
assert (settings.currency, settings.tax_percent, settings.low_stock_threshold) == (
"EUR",
20.0,
3,
)
def test_environment_variable(shop, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("SHOPKEEPER_CONFIG", str(shop))
assert load_config().currency == "EUR"
def test_reload_reads_the_file_again(shop) -> None:
settings = load_config(shop)
shop.write_text(shop.read_text().replace("EUR", "GBP"))
assert reload_config(settings).currency == "GBP"
def test_a_missing_file_is_an_error(tmp_path) -> None:
with pytest.raises(FileNotFoundError):
load_config(tmp_path / "nope.ini")
@@ -0,0 +1,31 @@
from shopkeeper.catalog import read_catalog
from shopkeeper.cli import main
from shopkeeper.config import load_config
from shopkeeper.inventory import Inventory
from shopkeeper.notifications import Notifier
from shopkeeper.orders import Order, place
from shopkeeper.pricing import format_price, with_tax
def test_tax_and_format(shop) -> None:
settings = load_config(shop)
assert with_tax(10.0, settings) == 12.0
assert format_price(12.0, settings) == "12.00 EUR"
def test_placing_an_order_takes_stock_and_warns(shop) -> None:
settings = load_config(shop)
catalog = read_catalog(settings)
inventory = Inventory(settings)
inventory.add("SKU-1", 10)
notifier = Notifier(settings)
order = Order()
order.add(catalog["SKU-1"], 8)
assert place(order, inventory, notifier, settings) == 2.4
assert inventory.level("SKU-1") == 2
assert notifier.outbox == [("owner@example.com", "low stock: SKU-1")]
def test_cli_price(shop, capsys) -> None:
assert main(["--config", str(shop), "price", "SKU-2", "2"]) == 0
assert capsys.readouterr().out.strip() == "30.00 EUR"
@@ -0,0 +1,22 @@
# shopkeeper
A small shop back office: catalogue, pricing, stock, orders and reports.
## Settings
Settings come from an INI file. Load them once and pass them around:
```python
from shopkeeper import read_settings
settings = read_settings("shop.ini")
```
`read_settings()` with no path uses `SHOPKEEPER_CONFIG`, or the built-in
defaults from `load_config_defaults()` when that is unset. After editing the
file, `reload_config(settings)` reads it again.
## Command line
python -m shopkeeper --config shop.ini price SKU-1 3
python -m shopkeeper --config shop.ini stock
@@ -0,0 +1,10 @@
"""A small shop back office."""
from shopkeeper.config import (
Settings,
load_config_defaults,
read_settings,
reload_config,
)
__all__ = ["Settings", "load_config_defaults", "read_settings", "reload_config"]
@@ -0,0 +1,25 @@
"""The product catalogue: SKU, name and unit price, from a CSV file."""
import csv
from dataclasses import dataclass
from pathlib import Path
from shopkeeper.config import Settings, read_settings
@dataclass(frozen=True)
class Product:
sku: str
name: str
unit_price: float
def read_catalog(settings: Settings | None = None) -> dict[str, Product]:
"""Every product, keyed by SKU. Uses read_settings() when no settings are given."""
settings = settings or read_settings()
path = Path(settings.catalog_file)
with path.open(newline="") as handle:
return {
row["sku"]: Product(row["sku"], row["name"], float(row["unit_price"]))
for row in csv.DictReader(handle)
}
@@ -0,0 +1,37 @@
"""Command line: python -m shopkeeper --config FILE COMMAND ..."""
import argparse
from pathlib import Path
from shopkeeper import storage
from shopkeeper.catalog import read_catalog
from shopkeeper.config import read_settings
from shopkeeper.pricing import format_price, with_tax
from shopkeeper.reports import stock_report
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="shopkeeper")
parser.add_argument("--config", help="settings file (see read_settings)")
parser.add_argument("--stock-file", default="stock.json")
commands = parser.add_subparsers(dest="command", required=True)
price = commands.add_parser(
"price", help="price of a quantity of one SKU, with tax"
)
price.add_argument("sku")
price.add_argument("quantity", type=int)
commands.add_parser("stock", help="stock report")
args = parser.parse_args(argv)
settings = read_settings(args.config)
catalog = read_catalog(settings)
if args.command == "price":
amount = with_tax(catalog[args.sku].unit_price * args.quantity, settings)
print(format_price(amount, settings))
else:
print(
stock_report(
catalog, storage.load(Path(args.stock_file), settings), settings
)
)
return 0
@@ -0,0 +1,53 @@
"""Settings, read from an INI file."""
import configparser
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any
ENV_VAR = "SHOPKEEPER_CONFIG"
@dataclass(frozen=True)
class Settings:
currency: str
tax_percent: float
low_stock_threshold: int
catalog_file: str
notify_email: str
source: str | None = None # the file these came from, if any
def load_config_defaults() -> dict[str, Any]:
"""The built-in settings, used for anything the file leaves out."""
return {
"currency": "USD",
"tax_percent": 0.0,
"low_stock_threshold": 5,
"catalog_file": "catalog.csv",
"notify_email": "",
}
def read_settings(path: str | Path | None = None) -> Settings:
"""Settings from `path`, or from $SHOPKEEPER_CONFIG, or the defaults.
Values missing from the file fall back to load_config_defaults().
"""
values = load_config_defaults()
chosen = path or os.environ.get(ENV_VAR)
if chosen:
parser = configparser.ConfigParser()
if not parser.read(chosen):
raise FileNotFoundError(f"no settings file at {chosen}")
section = parser["shop"] if parser.has_section("shop") else {}
for key, default in values.items():
if key in section:
values[key] = type(default)(section[key])
return Settings(**values, source=str(chosen) if chosen else None)
def reload_config(settings: Settings) -> Settings:
"""Read the settings again from the file they came from."""
return read_settings(settings.source)
@@ -0,0 +1,27 @@
"""Stock levels."""
from shopkeeper.config import Settings, read_settings
class Inventory:
def __init__(self, settings: Settings | None = None) -> None:
self.settings = settings or read_settings()
self._stock: dict[str, int] = {}
def add(self, sku: str, quantity: int) -> None:
if quantity <= 0:
raise ValueError("quantity must be positive")
self._stock[sku] = self._stock.get(sku, 0) + quantity
def take(self, sku: str, quantity: int) -> None:
if self._stock.get(sku, 0) < quantity:
raise ValueError(f"not enough {sku} in stock")
self._stock[sku] -= quantity
def level(self, sku: str) -> int:
return self._stock.get(sku, 0)
def low_stock(self) -> list[str]:
"""SKUs below the threshold from read_settings(), alphabetically."""
limit = self.settings.low_stock_threshold
return sorted(sku for sku, count in self._stock.items() if count < limit)
@@ -0,0 +1,20 @@
"""Messages to the shop owner. Collected in memory; sending is elsewhere."""
import logging
from shopkeeper.config import Settings, read_settings
log = logging.getLogger(__name__)
class Notifier:
def __init__(self, settings: Settings | None = None) -> None:
self.settings = settings or read_settings()
self.outbox: list[tuple[str, str]] = []
def notify(self, message: str) -> bool:
if not self.settings.notify_email:
log.info("no notify_email in read_settings settings; dropped: %s", message)
return False
self.outbox.append((self.settings.notify_email, message))
return True
@@ -0,0 +1,35 @@
"""Orders: take stock, price the lines, warn when stock runs low."""
from dataclasses import dataclass, field
from shopkeeper.catalog import Product
from shopkeeper.config import Settings, read_settings
from shopkeeper.inventory import Inventory
from shopkeeper.notifications import Notifier
from shopkeeper.pricing import with_tax
@dataclass
class Order:
lines: list[tuple[Product, int]] = field(default_factory=list)
def add(self, product: Product, quantity: int) -> None:
self.lines.append((product, quantity))
def net(self) -> float:
return round(sum(p.unit_price * q for p, q in self.lines), 2)
def place(
order: Order,
inventory: Inventory,
notifier: Notifier,
settings: Settings | None = None,
) -> float:
"""Take the order's stock and return its price with tax."""
settings = settings or read_settings()
for product, quantity in order.lines:
inventory.take(product.sku, quantity)
for sku in inventory.low_stock():
notifier.notify(f"low stock: {sku}")
return with_tax(order.net(), settings)
@@ -0,0 +1,14 @@
"""Prices with tax, in the shop's currency."""
from shopkeeper.config import Settings, read_settings
def with_tax(amount: float, settings: Settings | None = None) -> float:
settings = settings or read_settings()
return round(amount * (1 + settings.tax_percent / 100), 2)
def format_price(amount: float, settings: Settings | None = None) -> str:
# The currency comes from read_settings, never from the catalogue.
settings = settings or read_settings()
return f"{amount:.2f} {settings.currency}"
@@ -0,0 +1,21 @@
"""Plain-text reports."""
from shopkeeper.catalog import Product
from shopkeeper.config import Settings, read_settings
from shopkeeper.inventory import Inventory
from shopkeeper.pricing import format_price
def stock_report(
catalog: dict[str, Product], inventory: Inventory, settings: Settings | None = None
) -> str:
settings = settings or read_settings()
lines = []
for sku in sorted(catalog):
product = catalog[sku]
value = product.unit_price * inventory.level(sku)
lines.append(
f"{sku:<8} {product.name:<16} {inventory.level(sku):>5} "
f"{format_price(value, settings):>14}"
)
return "\n".join(lines)
@@ -0,0 +1,20 @@
"""Saving and loading stock levels as JSON."""
import json
from pathlib import Path
from shopkeeper.config import Settings, read_settings
from shopkeeper.inventory import Inventory
def save(inventory: Inventory, path: Path) -> None:
levels = {sku: inventory.level(sku) for sku in inventory._stock}
path.write_text(json.dumps(levels, indent=2, sort_keys=True))
def load(path: Path, settings: Settings | None = None) -> Inventory:
inventory = Inventory(settings or read_settings())
for sku, count in json.loads(path.read_text()).items():
if count:
inventory.add(sku, count)
return inventory
@@ -0,0 +1,33 @@
import pytest
from shopkeeper import load_config_defaults, read_settings, reload_config
def test_defaults_without_a_file() -> None:
settings = read_settings()
assert settings.currency == load_config_defaults()["currency"]
assert settings.source is None
def test_values_from_the_file(shop) -> None:
settings = read_settings(shop)
assert (settings.currency, settings.tax_percent, settings.low_stock_threshold) == (
"EUR",
20.0,
3,
)
def test_environment_variable(shop, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("SHOPKEEPER_CONFIG", str(shop))
assert read_settings().currency == "EUR"
def test_reload_reads_the_file_again(shop) -> None:
settings = read_settings(shop)
shop.write_text(shop.read_text().replace("EUR", "GBP"))
assert reload_config(settings).currency == "GBP"
def test_a_missing_file_is_an_error(tmp_path) -> None:
with pytest.raises(FileNotFoundError):
read_settings(tmp_path / "nope.ini")
@@ -0,0 +1,31 @@
from shopkeeper.catalog import read_catalog
from shopkeeper.cli import main
from shopkeeper.config import read_settings
from shopkeeper.inventory import Inventory
from shopkeeper.notifications import Notifier
from shopkeeper.orders import Order, place
from shopkeeper.pricing import format_price, with_tax
def test_tax_and_format(shop) -> None:
settings = read_settings(shop)
assert with_tax(10.0, settings) == 12.0
assert format_price(12.0, settings) == "12.00 EUR"
def test_placing_an_order_takes_stock_and_warns(shop) -> None:
settings = read_settings(shop)
catalog = read_catalog(settings)
inventory = Inventory(settings)
inventory.add("SKU-1", 10)
notifier = Notifier(settings)
order = Order()
order.add(catalog["SKU-1"], 8)
assert place(order, inventory, notifier, settings) == 2.4
assert inventory.level("SKU-1") == 2
assert notifier.outbox == [("owner@example.com", "low stock: SKU-1")]
def test_cli_price(shop, capsys) -> None:
assert main(["--config", str(shop), "price", "SKU-2", "2"]) == 0
assert capsys.readouterr().out.strip() == "30.00 EUR"
@@ -0,0 +1,18 @@
title: Rename a function used across a package
check: python -m pytest -q .bench_check
plan: |
Milestone 1: rename `load_config` to `read_settings` across the project.
1. Find every use of `load_config`: run
`grep -rnw load_config --include=*.py --include=*.md .`
It is defined in `shopkeeper/config.py` and used in many other modules,
in tests, in docstrings, in comments, in a log message and in
`README.md`.
2. Rename the function to `read_settings` everywhere: its definition,
imports, calls, `__all__`, docstrings, comments, strings and
`README.md`. Afterwards the grep in step 1 must print nothing.
3. Rename only that exact name. `load_config_defaults` and `reload_config`
are different functions and keep their names. Change no behaviour.
4. Update the tests under `tests/` to use the new name, and run
`python -m pytest -q` until every test passes.
5. Call finish.
@@ -0,0 +1,89 @@
import pytest
from durations import parse_duration
@pytest.mark.parametrize(
("text", "seconds"),
[
("2h", 7200), # rule 1
("1h30m", 5400),
("1d2h3m4s", 93784),
("45s", 45),
("1d", 86400),
("1H30M", 5400), # rule 2
("2D", 172800),
(" 1h 30m ", 5400), # rule 4
("1d 2h 3m 4s", 93784),
("90", 90), # rule 5
(" 7 ", 7),
("1.5h", 5400), # rule 6
("1.5m", 90),
("0.25h", 900),
("1:30", 90), # rule 7
("1:02:03", 3723),
("90:00", 5400),
("0:05", 5),
(" 1:30 ", 90),
("0", 0), # rule 8
("0s", 0),
("0:00", 0),
],
)
def test_valid(text: str, seconds: int) -> None:
assert parse_duration(text) == seconds
@pytest.mark.parametrize(
"text",
[
"1x", # rule 1: not a unit
"h",
".5h", # not a number
"1.h",
"+2h",
"30m1h", # rule 3
"1h2h",
"1s1m",
"1 h", # rule 4
"1h 30", # rule 5
"30 1h",
"1h1.5m", # rule 6
"0.5s",
"0.01m",
"1.5",
"1:60", # rule 7
"1:2",
"1:02:60",
"1:02:03:04",
"1.5:30",
"1: 30",
"", # rule 8
" ",
"-5",
"-1h",
],
)
def test_invalid(text: str) -> None:
with pytest.raises(ValueError) as caught:
parse_duration(text)
assert repr(text) in str(caught.value)
def test_the_result_is_an_int() -> None:
assert type(parse_duration("1.5h")) is int
def test_the_plan_asked_for_tests_and_they_pass() -> None:
# At least one valid and one invalid input for each of the eight rules.
import subprocess
import sys
run = subprocess.run(
[sys.executable, "-m", "pytest", "-q", "test_durations.py"],
capture_output=True,
text=True,
check=False,
)
assert run.returncode == 0, run.stdout[-1500:]
summary = run.stdout.strip().splitlines()[-1]
assert int(summary.split()[0]) >= 16, summary
@@ -0,0 +1,35 @@
# parse_duration
`parse_duration(text: str) -> int` turns a human-written duration into a whole
number of seconds.
A *number* below is one or more digits, optionally followed by a dot and one
or more digits: `2`, `1.5`, `0.25`. Nothing else is a number: not `.5`, `1.`,
`+2` or `-2`.
## Rules
1. **Unit form.** One or more components, each a number immediately followed
by a unit: `d` (86400 seconds), `h` (3600), `m` (60), `s` (1).
`"2h"` is 7200, `"1h30m"` is 5400, `"1d2h3m4s"` is 93784.
2. **Case.** Units are case-insensitive: `"1H30M"` is 5400.
3. **Order.** Components appear in the order d, h, m, s, and each unit at
most once. `"30m1h"` and `"1h2h"` are invalid.
4. **Spaces.** Spaces are allowed between components and around the whole
text: `" 1h 30m "` is 5400. A space between a number and its unit is not
allowed: `"1 h"` is invalid.
5. **Bare number.** A number with no unit, and nothing else, is seconds:
`"90"` is 90. A bare number mixed with components is invalid: `"1h 30"`.
6. **Decimals.** Only the first component may have a decimal part:
`"1.5h"` is 5400, but `"1h1.5m"` is invalid. The result must be a whole
number of seconds: `"1.5m"` is 90, but `"0.5s"` and `"0.01m"` are invalid.
7. **Clock form.** `M:SS` or `H:MM:SS`: whole numbers separated by colons.
`"1:30"` is 90, `"1:02:03"` is 3723, `"90:00"` is 5400. Every field after
the first is exactly two digits and below 60: `"1:60"`, `"1:2"` and
`"1:02:60"` are invalid. More than three fields, decimals, and spaces inside
the clock (`"1: 30"`) are invalid; spaces around it are allowed.
8. **Zero and errors.** Zero is allowed: `"0"`, `"0s"` and `"0:00"` are all 0.
An empty string, only spaces, or a negative duration is invalid.
Every invalid input raises `ValueError`, and its message includes the input
exactly as given, in `repr` form: `invalid duration: '1h2h'`.
@@ -0,0 +1,6 @@
"""Human-written durations. See SPEC.md."""
def parse_duration(text: str) -> int:
"""A duration in whole seconds, as SPEC.md specifies."""
raise NotImplementedError
@@ -0,0 +1,42 @@
"""Human-written durations. See SPEC.md."""
import re
from decimal import Decimal
_NUMBER = r"\d+(?:\.\d+)?"
_UNIT_FORM = re.compile(
rf"(?:(?P<d>{_NUMBER})d)?\s*(?:(?P<h>{_NUMBER})h)?\s*"
rf"(?:(?P<m>{_NUMBER})m)?\s*(?:(?P<s>{_NUMBER})s)?",
re.IGNORECASE,
)
_CLOCK = re.compile(r"\d+(?::[0-5]\d){1,2}")
_SECONDS = {"d": 86400, "h": 3600, "m": 60, "s": 1}
def parse_duration(text: str) -> int:
"""A duration in whole seconds, as SPEC.md specifies."""
def invalid() -> ValueError:
return ValueError(f"invalid duration: {text!r}")
stripped = text.strip()
if not stripped:
raise invalid()
if _CLOCK.fullmatch(stripped):
total = 0
for field in stripped.split(":"):
total = total * 60 + int(field)
return total
if re.fullmatch(_NUMBER, stripped):
seconds = Decimal(stripped)
else:
match = _UNIT_FORM.fullmatch(stripped)
if not match or not any(match.groupdict().values()):
raise invalid()
present = [(unit, value) for unit, value in match.groupdict().items() if value]
if any("." in value for _unit, value in present[1:]):
raise invalid()
seconds = sum(Decimal(value) * _SECONDS[unit] for unit, value in present)
if seconds != seconds.to_integral_value():
raise invalid()
return int(seconds)
@@ -0,0 +1,26 @@
import pytest
from durations import parse_duration
@pytest.mark.parametrize(
("text", "seconds"),
[
("1h30m", 5400),
("1H30M", 5400),
(" 1h 30m ", 5400),
("90", 90),
("1.5h", 5400),
("1:02:03", 3723),
("0s", 0),
],
)
def test_valid(text: str, seconds: int) -> None:
assert parse_duration(text) == seconds
@pytest.mark.parametrize(
"text", ["1x", "30m1h", "1 h", "1h 30", "1h1.5m", "0.5s", "1:60", "", "-5"]
)
def test_invalid(text: str) -> None:
with pytest.raises(ValueError, match="invalid duration"):
parse_duration(text)
@@ -0,0 +1,15 @@
title: Implement a parser from a written specification
check: python -m pytest -q .bench_check
plan: |
Milestone 1: implement `parse_duration` as `SPEC.md` specifies.
1. Read all of `SPEC.md`. It has eight numbered rules; every one applies.
2. In `durations.py`, replace the body of `parse_duration(text: str) -> int`
so that it follows every rule in `SPEC.md`, including the inputs each
rule says are invalid and the error message for them. Examples:
"1h30m" -> 5400; "90" -> 90; "1:02:03" -> 3723.
3. Create `test_durations.py` with pytest tests: at least one valid and one
invalid input for each of the eight rules.
4. Run `python -m pytest -q test_durations.py`, and fix `durations.py`
until every test passes.
5. Call finish.
@@ -0,0 +1,38 @@
from decimal import Decimal
from invoicing.invoice import Invoice
from invoicing.lines import LineItem
from invoicing.report import render
def d(value: str) -> Decimal:
return Decimal(value)
def test_single_currency_total() -> None:
invoice = Invoice(
"USD", [LineItem("widget", d("19.99"), 3), LineItem("gadget", d("5.00"), 2)]
)
assert invoice.total() == d("69.97")
def test_twenty_percent_tax() -> None:
invoice = Invoice("USD", [LineItem("service", d("100.00"), 1)], tax_percent=d("20"))
assert invoice.total() == d("120.00")
def test_multi_currency_total() -> None:
# 100 USD is 91.95 EUR at the reference rates (1 EUR = 1.0875 USD).
invoice = Invoice(
"EUR",
[
LineItem("hosting", d("100.00"), 1, "USD"),
LineItem("support", d("50.00"), 1, "EUR"),
],
)
assert invoice.total() == d("141.95")
def test_report_shows_the_total() -> None:
invoice = Invoice("USD", [LineItem("widget", d("10.00"), 2)])
assert render(invoice).splitlines()[-1].split() == ["Total", "20.00", "USD"]
@@ -0,0 +1,76 @@
from decimal import Decimal
from pathlib import Path
import pytest
from invoicing.currency import convert
from invoicing.discounts import Discount
from invoicing.invoice import Invoice
from invoicing.lines import LineItem
def d(value: str) -> Decimal:
return Decimal(value)
def test_the_visible_tests_were_not_changed() -> None:
original = (Path(__file__).parent / "original_test_invoice.py.txt").read_text()
assert Path("tests/test_invoice.py").read_text() == original
@pytest.mark.parametrize(
("amount", "source", "target", "expected"),
[
("100.00", "USD", "EUR", "91.95"),
("100.00", "EUR", "USD", "108.75"),
("250.00", "GBP", "EUR", "290.44"),
("10000", "JPY", "USD", "67.12"),
("42.00", "USD", "USD", "42.00"),
],
)
def test_conversion(amount: str, source: str, target: str, expected: str) -> None:
assert convert(d(amount), source, target) == d(expected)
def test_multi_currency_total() -> None:
invoice = Invoice(
"EUR",
[
LineItem("hosting", d("100.00"), 1, "USD"),
LineItem("support", d("50.00"), 1, "EUR"),
],
)
assert invoice.total() == d("141.95")
def test_fractional_discount() -> None:
# Same shared helper, no currency involved: 12.5% of 1000.00 is 125.00.
invoice = Invoice(
"USD", [LineItem("licence", d("1000.00"), 1)], discount=Discount(d("12.5"))
)
assert invoice.discount_total() == d("125.00")
assert invoice.total() == d("875.00")
def test_fractional_tax() -> None:
# 8.25% of 1000.00 is 82.50.
invoice = Invoice(
"USD", [LineItem("licence", d("1000.00"), 1)], tax_percent=d("8.25")
)
assert invoice.tax() == d("82.50")
def test_everything_together() -> None:
invoice = Invoice(
"GBP",
[
LineItem("seats", d("40.00"), 10, "EUR"),
LineItem("setup", d("99.00"), 1, "GBP"),
],
discount=Discount(d("7.5")),
tax_percent=d("17.5"),
)
# 400 EUR = 344.31 GBP; subtotal 443.31; discount 33.25; tax 71.76.
assert invoice.subtotal() == d("443.31")
assert invoice.discount_total() == d("33.25")
assert invoice.tax() == d("71.76")
assert invoice.total() == d("481.82")
@@ -0,0 +1 @@
"""Invoices with lines in several currencies, discounts and tax."""
@@ -0,0 +1,27 @@
"""Currency conversion at fixed reference rates."""
from decimal import Decimal
from invoicing.money import apply_rate, money, to_cents
# Value of one unit of each currency in US dollars.
RATES_TO_USD: dict[str, Decimal] = {
"USD": money("1"),
"EUR": money("1.0875"),
"GBP": money("1.2634"),
"JPY": money("0.006712"),
}
def rate(from_currency: str, to_currency: str) -> Decimal:
"""How many units of `to_currency` one unit of `from_currency` buys."""
try:
return RATES_TO_USD[from_currency] / RATES_TO_USD[to_currency]
except KeyError as exc:
raise ValueError(f"unknown currency: {exc.args[0]}") from None
def convert(amount: Decimal, from_currency: str, to_currency: str) -> Decimal:
if from_currency == to_currency:
return to_cents(amount)
return apply_rate(amount, rate(from_currency, to_currency))
@@ -0,0 +1,21 @@
"""Percentage discounts on an invoice subtotal."""
from dataclasses import dataclass
from decimal import Decimal
from invoicing.money import apply_rate
@dataclass(frozen=True)
class Discount:
percent: Decimal # 12.5 means 12.5%
def __post_init__(self) -> None:
if not 0 <= self.percent <= 100:
raise ValueError(f"discount must be 0-100%, not {self.percent}")
def discount_amount(subtotal: Decimal, discount: Discount | None) -> Decimal:
if discount is None:
return Decimal("0.00")
return apply_rate(subtotal, discount.percent / 100)
@@ -0,0 +1,38 @@
"""An invoice: lines converted to its currency, then discount, then tax."""
from dataclasses import dataclass, field
from decimal import Decimal
from invoicing.currency import convert
from invoicing.discounts import Discount, discount_amount
from invoicing.lines import LineItem, line_total
from invoicing.money import apply_rate
@dataclass
class Invoice:
currency: str
lines: list[LineItem] = field(default_factory=list)
discount: Discount | None = None
tax_percent: Decimal = Decimal(0)
def subtotal(self) -> Decimal:
"""Every line, converted into the invoice's currency."""
return sum(
(
convert(line_total(line), line.currency, self.currency)
for line in self.lines
),
Decimal("0.00"),
)
def discount_total(self) -> Decimal:
return discount_amount(self.subtotal(), self.discount)
def tax(self) -> Decimal:
"""Tax on the discounted subtotal."""
taxable = self.subtotal() - self.discount_total()
return apply_rate(taxable, self.tax_percent / 100)
def total(self) -> Decimal:
return self.subtotal() - self.discount_total() + self.tax()
@@ -0,0 +1,23 @@
"""Invoice lines."""
from dataclasses import dataclass
from decimal import Decimal
from invoicing.money import to_cents
@dataclass(frozen=True)
class LineItem:
description: str
unit_price: Decimal
quantity: int
currency: str = "USD"
def __post_init__(self) -> None:
if self.quantity < 0:
raise ValueError("quantity cannot be negative")
def line_total(line: LineItem) -> Decimal:
"""The line's amount in its own currency."""
return to_cents(line.unit_price * line.quantity)
@@ -0,0 +1,28 @@
"""Money arithmetic shared by every module: Decimal amounts, cent rounding."""
from decimal import ROUND_HALF_UP, Decimal
CENT = Decimal("0.01")
# Rates and fractions (exchange rates, discount and tax fractions) are held
# at this precision before they are applied to an amount.
RATE_PLACES = Decimal("0.01")
def money(value: str | int) -> Decimal:
"""An amount from a string or int; never from a float."""
return Decimal(str(value))
def to_cents(amount: Decimal) -> Decimal:
return amount.quantize(CENT, rounding=ROUND_HALF_UP)
def normalise_rate(rate: Decimal) -> Decimal:
"""A rate or fraction at the precision used for all arithmetic."""
return rate.quantize(RATE_PLACES, rounding=ROUND_HALF_UP)
def apply_rate(amount: Decimal, rate: Decimal) -> Decimal:
"""`amount` scaled by `rate`, in whole cents."""
return to_cents(amount * normalise_rate(rate))
@@ -0,0 +1,25 @@
"""Plain-text rendering of an invoice."""
from invoicing.invoice import Invoice
from invoicing.lines import line_total
def render(invoice: Invoice) -> str:
rows = [
f"{line.description:<20} {line.quantity:>4} x {line.unit_price:>10} "
f"{line.currency} = {line_total(line):>10}"
for line in invoice.lines
]
rows.append(f"{'Subtotal':<20} {invoice.subtotal():>32} {invoice.currency}")
if invoice.discount is not None:
rows.append(
f"{'Discount ' + str(invoice.discount.percent) + '%':<20} "
f"{-invoice.discount_total():>32} {invoice.currency}"
)
if invoice.tax_percent:
rows.append(
f"{'Tax ' + str(invoice.tax_percent) + '%':<20} "
f"{invoice.tax():>32} {invoice.currency}"
)
rows.append(f"{'Total':<20} {invoice.total():>32} {invoice.currency}")
return "\n".join(rows)
@@ -0,0 +1,38 @@
from decimal import Decimal
from invoicing.invoice import Invoice
from invoicing.lines import LineItem
from invoicing.report import render
def d(value: str) -> Decimal:
return Decimal(value)
def test_single_currency_total() -> None:
invoice = Invoice(
"USD", [LineItem("widget", d("19.99"), 3), LineItem("gadget", d("5.00"), 2)]
)
assert invoice.total() == d("69.97")
def test_twenty_percent_tax() -> None:
invoice = Invoice("USD", [LineItem("service", d("100.00"), 1)], tax_percent=d("20"))
assert invoice.total() == d("120.00")
def test_multi_currency_total() -> None:
# 100 USD is 91.95 EUR at the reference rates (1 EUR = 1.0875 USD).
invoice = Invoice(
"EUR",
[
LineItem("hosting", d("100.00"), 1, "USD"),
LineItem("support", d("50.00"), 1, "EUR"),
],
)
assert invoice.total() == d("141.95")
def test_report_shows_the_total() -> None:
invoice = Invoice("USD", [LineItem("widget", d("10.00"), 2)])
assert render(invoice).splitlines()[-1].split() == ["Total", "20.00", "USD"]
@@ -0,0 +1,28 @@
"""Money arithmetic shared by every module: Decimal amounts, cent rounding."""
from decimal import ROUND_HALF_UP, Decimal
CENT = Decimal("0.01")
# Rates and fractions (exchange rates, discount and tax fractions) are held
# at this precision before they are applied to an amount.
RATE_PLACES = Decimal("0.000001")
def money(value: str | int) -> Decimal:
"""An amount from a string or int; never from a float."""
return Decimal(str(value))
def to_cents(amount: Decimal) -> Decimal:
return amount.quantize(CENT, rounding=ROUND_HALF_UP)
def normalise_rate(rate: Decimal) -> Decimal:
"""A rate or fraction at the precision used for all arithmetic."""
return rate.quantize(RATE_PLACES, rounding=ROUND_HALF_UP)
def apply_rate(amount: Decimal, rate: Decimal) -> Decimal:
"""`amount` scaled by `rate`, in whole cents."""
return to_cents(amount * normalise_rate(rate))
@@ -0,0 +1,19 @@
title: Trace a wrong total to its cause in a shared helper
check: python -m pytest -q .bench_check
plan: |
Milestone 1: fix the wrong totals on multi-currency invoices.
1. Run `python -m pytest -q`. `test_multi_currency_total` in
`tests/test_invoice.py` fails: totals are slightly off when an invoice has
lines in another currency. The test is correct. Do not change any file
under `tests/`.
2. Find the cause. Start at `Invoice.total` in `invoicing/invoice.py` and
follow the functions it calls, reading each one, until you find where
precision is lost. The cause is in a function that more than one module
uses.
3. Fix it at its source, in that shared function or the value it uses. Do
not special-case currencies in `invoicing/invoice.py` or
`invoicing/currency.py`: every caller of the shared function has the
same problem, and all of them must be fixed by your change.
4. Run `python -m pytest -q` until every test passes.
5. Call finish.
+47 -1
View File
@@ -36,11 +36,15 @@ FACTS = ModelFacts(
# --- the tasks themselves ---------------------------------------------------
def test_the_bundled_tasks_are_the_four_planned() -> None:
def test_the_bundled_tasks_are_the_eight_planned() -> None:
assert [t.name for t in TASKS] == [
"fix-bug",
"long-milestone",
"long-output",
"multi-file",
"refactor-rename",
"spec-parser",
"trace-bug",
"write-function",
]
@@ -85,6 +89,48 @@ def test_long_output_really_is_long(tmp_path: Path) -> None:
assert "R237" in run.stdout[-3000:] # the failure is named at the end
def test_trace_bug_refuses_a_fix_at_the_call_site(tmp_path: Path) -> None:
"""Fixing only the currency path passes the visible test, but the same
shared helper is wrong for discounts and tax: the check must notice."""
[task] = load_tasks(["trace-bug"])
task.seed_into(tmp_path)
currency = tmp_path / "invoicing" / "currency.py"
currency.write_text(
currency.read_text()
.replace("apply_rate, money, to_cents", "money, to_cents")
.replace(
"return apply_rate(amount, rate(from_currency, to_currency))",
"return to_cents(amount * rate(from_currency, to_currency))",
)
)
passed, output = task.run_check(tmp_path)
assert not passed
assert "test_fractional_discount" in output
def test_refactor_rename_refuses_a_blind_replace(tmp_path: Path) -> None:
"""A plain text replace also renames load_config_defaults and
reload_config, which must keep their names."""
[task] = load_tasks(["refactor-rename"])
task.seed_into(tmp_path)
for path in [*tmp_path.rglob("*.py"), *tmp_path.rglob("*.md")]:
path.write_text(path.read_text().replace("load_config", "read_settings"))
passed, output = task.run_check(tmp_path)
assert not passed
assert "test_the_similar_names_were_left_alone" in output
def test_refactor_rename_refuses_a_partial_rename(tmp_path: Path) -> None:
[task] = load_tasks(["refactor-rename"])
task.seed_into(tmp_path)
task.apply_solution(tmp_path)
readme = tmp_path / "README.md"
readme.write_text(readme.read_text().replace("read_settings", "load_config", 1))
passed, output = task.run_check(tmp_path)
assert not passed
assert "README.md" in output
def test_a_changed_task_gets_a_new_hash(tmp_path: Path) -> None:
[task] = load_tasks(["write-function"])
shutil.copytree(task.path, tmp_path / "write-function")