Merge pull request 'Add four harder benchmark tasks' (#53) from harder-bench-tasks into main
CI / checks (push) Successful in 1m17s
CI / checks (push) Successful in 1m17s
This commit was merged in pull request #53.
This commit is contained in:
commit
8467b8747f
67 files changed
+2117
-2
No files matched your search
+12
-1
@@ -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))
|
||||
+76
@@ -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)
|
||||
+20
@@ -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,21 @@
|
||||
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. The rates in
|
||||
`invoicing/currency.py` have up to six decimal places (JPY is 0.006712),
|
||||
and no step may round a rate or fraction to fewer places than that.
|
||||
4. Run `python -m pytest -q` until every test passes.
|
||||
5. Call finish.
|
||||
@@ -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")
|
||||
|
||||
Reference in new issue
Block a user