from pathlib import Path
import argparse
import zipfile
ROOT = Path(__file__).resolve().parent
PROGRAM = r'''import argparse
import csv
import re
import sys
from datetime import date
FIELDS = ("task_id", "status", "due_date", "team")
def iso_date(value):
if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
raise ValueError("date must use YYYY-MM-DD")
return date.fromisoformat(value)
def load_tasks(path):
rows = []
with open(path, newline="", encoding="utf-8") as source:
reader = csv.DictReader(source, strict=True)
missing = set(FIELDS) - set(reader.fieldnames or [])
if missing:
raise ValueError("missing required columns: " + ", ".join(sorted(missing)))
for number, raw in enumerate(reader, start=2):
if None in raw or any(raw[name] is None for name in FIELDS):
raise ValueError(f"row {number}: malformed CSV record")
if not raw["task_id"].strip():
raise ValueError(f"row {number}: task_id must not be blank")
if not raw["team"].strip():
raise ValueError(f"row {number}: team must not be blank")
if raw["status"] not in ("open", "completed"):
raise ValueError(f"row {number}: status must be open or completed")
try:
due = iso_date(raw["due_date"])
except ValueError as exc:
raise ValueError(f"row {number}: invalid due_date") from exc
row = {name: raw[name] for name in FIELDS}
row["due_date"] = due
rows.append(row)
return rows
def select_tasks(rows, as_of):
return [row for row in rows
if row["status"] == "open" and row["due_date"] < as_of]
def main():
parser = argparse.ArgumentParser(description="Report open overdue tasks")
parser.add_argument("source")
parser.add_argument("--as-of", required=True, type=iso_date)
args = parser.parse_args()
try:
rows = load_tasks(args.source)
except (OSError, ValueError, csv.Error) as exc:
parser.error(str(exc))
selected = select_tasks(rows, args.as_of)
writer = csv.DictWriter(sys.stdout, fieldnames=FIELDS, lineterminator="\n")
writer.writeheader()
writer.writerows(selected)
if __name__ == "__main__":
main()
'''
SAMPLE = '''task_id,status,due_date,team
T201,open,2026-04-14,Support
T202,open,2026-04-13,Ops
T203,completed,2026-04-12,Support
T204,open,2026-04-15,Support
'''
TESTS = r'''import csv
import io
import subprocess
import sys
import tempfile
import unittest
from datetime import date
from pathlib import Path
from task_report import FIELDS, load_tasks, select_tasks
ROOT = Path(__file__).resolve().parent
SAMPLE = ROOT / "task_sample.csv"
class BaselineTests(unittest.TestCase):
def cli(self, source, *flags):
return subprocess.run(
[sys.executable, str(ROOT / "task_report.py"), str(source),
"--as-of", "2026-04-15", *flags],
capture_output=True, text=True)
def test_selection(self):
selected = select_tasks(load_tasks(SAMPLE), date(2026, 4, 15))
self.assertEqual([row["task_id"] for row in selected], ["T201", "T202"])
def test_real_cli(self):
result = self.cli(SAMPLE)
self.assertEqual(result.returncode, 0, result.stderr)
reader = csv.DictReader(io.StringIO(result.stdout))
self.assertEqual(reader.fieldnames, list(FIELDS))
self.assertEqual(list(reader), [
dict(task_id="T201", status="open", due_date="2026-04-14", team="Support"),
dict(task_id="T202", status="open", due_date="2026-04-13", team="Ops")])
def test_source_errors_emit_no_report(self):
invalid_sources = [
"task_id,status,due_date\nT1,open,2026-04-14\n",
"task_id,status,due_date,team\n ,open,2026-04-14,Ops\n",
"task_id,status,due_date,team\nT1,open,2026-04-14, \n",
"task_id,status,due_date,team\nT1,pending,2026-04-14,Ops\n",
"task_id,status,due_date,team\nT1,open,2026-02-30,Ops\n"]
with tempfile.TemporaryDirectory() as folder:
source = Path(folder) / "invalid.csv"
for text in invalid_sources:
with self.subTest(text=text):
source.write_text(text, encoding="utf-8")
result = self.cli(source)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, "")
self.assertIn("error:", result.stderr)
if __name__ == "__main__":
unittest.main()
'''
EXTENSION = r'''
# Focused acceptance checks: expectations, not recorded execution results.
class TeamAcceptanceTests(unittest.TestCase):
cli = BaselineTests.cli
def expected(self, ids, padded=False, dates=False):
records = {
"T201": dict(task_id="T201", status="open", due_date="2026-04-14",
team=" Support " if padded else "Support"),
"T202": dict(task_id="T202", status="open", due_date="2026-04-13", team="Ops")}
result = [records[key].copy() for key in ids]
if dates:
for row in result:
row["due_date"] = date.fromisoformat(row["due_date"])
return result
def assert_report(self, result, ids, padded=False):
self.assertEqual(result.returncode, 0, result.stderr)
reader = csv.DictReader(io.StringIO(result.stdout))
self.assertEqual(reader.fieldnames, list(FIELDS))
records = list(reader)
self.assertEqual(records, self.expected(ids, padded=padded))
self.assertTrue({row["task_id"] for row in records}.isdisjoint({"T203", "T204"}))
if not ids:
self.assertEqual(result.stdout, "task_id,status,due_date,team\n")
def test_selection_teams(self):
rows = load_tasks(SAMPLE)
cutoff = date(2026, 4, 15)
self.assertEqual(select_tasks(rows, cutoff), self.expected(["T201", "T202"], dates=True))
# Only normalized values enter selection. CLI owns supplied-value trimming.
for team, ids in [(None, ["T201", "T202"]), ("Support", ["T201"]),
("Ops", ["T202"]), ("support", []), ("Unknown", [])]:
with self.subTest(team=team):
selected = select_tasks(rows, cutoff, team)
self.assertEqual(selected, self.expected(ids, dates=True))
self.assertTrue({row["task_id"] for row in selected}.isdisjoint({"T203", "T204"}))
def test_cli_teams(self):
for flags, ids in [((), ["T201", "T202"]),
(("--team", "Support"), ["T201"]),
(("--team", "Ops"), ["T202"]),
(("--team", " Support "), ["T201"]),
(("--team", "support"), []),
(("--team", "Unknown"), [])]:
with self.subTest(flags=flags):
self.assert_report(self.cli(SAMPLE, *flags), ids)
def test_record_side_trimming(self):
with tempfile.TemporaryDirectory() as folder:
source = Path(folder) / "padded.csv"
source.write_text(SAMPLE.read_text(encoding="utf-8").replace(
",Support\n", ", Support \n"), encoding="utf-8")
rows = load_tasks(source)
self.assertEqual(rows[0]["team"], " Support ")
self.assertEqual(select_tasks(rows, date(2026, 4, 15), "Support"),
self.expected(["T201"], padded=True, dates=True))
for supplied in ("Support", " Support "):
self.assert_report(self.cli(source, "--team", supplied), ["T201"], padded=True)
def test_blank_supplied_team(self):
# Blank is rejected at the CLI, not passed into selection.
for supplied in ("", " "):
with self.subTest(supplied=supplied):
result = self.cli(SAMPLE, "--team", supplied)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, "")
self.assertIn("error:", result.stderr)
self.assertIn("--team", result.stderr)
self.assertIn("blank", result.stderr.lower())
def test_full_source_validation_with_filter(self):
cases = [
("task_id,status,due_date\nT1,open,2026-04-14\n",
"Unknown", "missing required columns: team"),
("task_id,status,due_date,team\nT201,open,2026-04-14,Support\n"
"T202,pending,2026-04-13,Ops\n",
"Support", "status must be open or completed")]
with tempfile.TemporaryDirectory() as folder:
source = Path(folder) / "invalid.csv"
for text, team, message in cases:
with self.subTest(team=team):
source.write_text(text, encoding="utf-8")
result = self.cli(source, "--team", team)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, "")
self.assertIn("error:", result.stderr)
self.assertIn(message, result.stderr)
'''
GUIDE = '''# Task report workspace
Python 3.10+, standard library only. Run commands in this directory.
Use python3 or py -3 consistently if that is your interpreter command.
Setup does not execute the report or tests. All results remain unverified.
## Baseline stage
1. Inspect task_report.py, task_sample.csv, and test_task_report.py.
2. Predict the selected task IDs before running anything. Keep your prediction
separate from observed evidence (for example in prediction.txt).
3. Run:
python task_report.py task_sample.csv --as-of 2026-04-15
python -m unittest -v test_task_report.BaselineTests
4. Record each actual command, stdout, stderr, and exit status in evidence.md.
That note starts empty. Do not turn expected output into observed evidence.
Obtain the exit status immediately after each command ($? in a POSIX shell,
$LASTEXITCODE in PowerShell), or use subprocess.run to capture it.
Expected, not observed: the baseline CLI exits zero and emits the four-column
header followed by T201 and T202, in that order. Baseline tests should pass.
If execution is unavailable, label your written trace unverified.
baseline.zip is the untouched downloadable baseline: it contains only the three
original exercise files. Extract it into a separate directory for recovery;
do not overwrite unfinished work without saving it first.
## Extension-check stage (same workspace)
When ready, run:
python workspace.py extension
This appends acceptance checks to test_task_report.py, retaining all original
baseline assertions. It does not implement --team or execute tests. Run the
checks before or alongside your edits; they are expected to fail until the
feature is connected.
Edit task_report.py using the lesson: add the parser argument, normalize the
supplied value, reject blank input, extend the function signature and predicate,
and connect the caller. Those are the learner's edits; setup supplies no finished
implementation. Keep full-source validation before selection.
Run focused checks, the real CLI, and the unchanged regression checks:
python -m unittest -v test_task_report.TeamAcceptanceTests
python task_report.py task_sample.csv --as-of 2026-04-15 --team Support
python task_report.py task_sample.csv --as-of 2026-04-15
python -m unittest -v
Record actual results and exit statuses in evidence.md, including failures.
Expected, not observed: Support selects T201; Ops selects T202; padded Support
matches Support; lowercase support and Unknown produce header-only success.
Blank supplied values fail without CSV output. T203 and T204 remain excluded.
Record-side trimming affects matching, not the original team text in output.
The checks also require invalid other-team rows and missing headers to fail.
'''
BASELINE = {"task_report.py": PROGRAM, "task_sample.csv": SAMPLE,
"test_task_report.py": TESTS}
def setup():
# Exclusive creation preserves working edits and any existing evidence.
for name, content in {**BASELINE, "WORKSPACE.md": GUIDE, "evidence.md": ""}.items():
path = ROOT / name
if not path.exists():
with path.open("x", encoding="utf-8", newline="") as target:
target.write(content)
archive = ROOT / "baseline.zip"
if not archive.exists():
with zipfile.ZipFile(archive, "x", compression=zipfile.ZIP_DEFLATED) as saved:
for name, content in BASELINE.items():
saved.writestr(name, content)
print("Workspace ready. Read WORKSPACE.md. No report or tests have been run by setup.")
def add_extension_checks():
path = ROOT / "test_task_report.py"
if not path.exists():
raise SystemExit("Run python workspace.py first.")
text = path.read_text(encoding="utf-8")
if "class TeamAcceptanceTests(" in text:
print("Acceptance checks already present; no files changed.")
return
marker = 'if __name__ == "__main__":\n unittest.main()'
if text.count(marker) != 1:
raise SystemExit("Test entry point changed; append acceptance checks manually before unittest.main().")
path.write_text(text.replace(marker, EXTENSION + "\n\n" + marker), encoding="utf-8")
print("Acceptance checks added. No implementation changed; no checks executed.")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Create the task report workspace or add acceptance checks")
parser.add_argument("stage", nargs="?", choices=("baseline", "extension"), default="baseline")
args = parser.parse_args()
if args.stage == "baseline":
setup()
else:
add_extension_checks()