diff --git a/.gitignore b/.gitignore index 5c09459..4230208 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,6 @@ uv.lock # oh-my-claudecode runtime state .omc/ + +# local working files (plans, research briefs) +.cache/ diff --git a/docs/mcp.md b/docs/mcp.md index 9519077..3149980 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -29,16 +29,24 @@ python-design-patterns-mcp --http --host 127.0.0.1 --port 8734 | Tool | What it does | |---|---| | `list_patterns(group?, verdict?)` | Catalog listing, filterable | -| `get_pattern(pattern_id, variant?)` | Full prose + example source (`naive`/`pythonic`/`real_world`/`all`) | +| `get_pattern(pattern_id, variant?)` | Full prose (+ legacy variant source files: `naive.py` / `pythonic.py` / `real_world.py`, or `all`) | | `search_patterns(query, limit?)` | BM25 full-text search over names, aliases, problems, symptoms, prose | -| `run_example(pattern_id, variant)` | Executes the vendored example in a sandboxed subprocess; returns real stdout | +| `get_pattern_docs(pattern_id, doc)` | A migrated pattern's teaching doc: `fundamentals`, `implementation`, or `examples` | +| `list_examples(pattern_id)` | A migrated pattern's runnable mini-projects | +| `run_example(pattern_id, variant?/example?)` | Executes a vendored example (legacy `variant` or migrated `example`) in a sandboxed subprocess; returns real stdout | +| `read_source(pattern_id)` | A migrated pattern's own implementation (`pattern/` package) | | `recommend_pattern(problem_statement, limit?)` | Ranked candidates with caveats; `prefer-alternative` verdicts tell you what to write instead | +Migrated (module-shape) patterns follow three access levels: scan docs +(`get_pattern_docs`) → run a use case (`list_examples` + `run_example`) → +read the source (`read_source`). + ## Resources - `catalog://index` — the whole catalog as JSON - `pattern:///` — one pattern's prose -- `pattern:////` — one example's source +- `pattern:////` — one legacy example's source +- `pattern:////docs/` — one migrated pattern's teaching doc ## Prompts @@ -47,6 +55,7 @@ python-design-patterns-mcp --http --host 127.0.0.1 --port 8734 ## Sandbox contract `run_example` executes only files resolved from the catalog index — the -`(id, variant)` pair is a dictionary lookup, never joined into a path. The +`(id, variant)` / `(id, example)` pair is a dictionary lookup, never joined +into a path. The subprocess runs `python -I` in a temp cwd with a scrubbed environment, a 10s timeout, and 64KB output caps. There is no arbitrary-code-execution tool. diff --git a/patterns/behavioral/chain_of_responsibility/README.md b/patterns/behavioral/chain_of_responsibility/README.md index aad0e2a..20b9cfc 100644 --- a/patterns/behavioral/chain_of_responsibility/README.md +++ b/patterns/behavioral/chain_of_responsibility/README.md @@ -14,30 +14,17 @@ stdlib_sightings: [logging propagation, urllib.request opener chain] # Chain of Responsibility -## Problem - -A support ticket should be handled by the first tier able to deal with it; -an HTTP request passes middleware until something produces a response. The -sender must not know which handler will answer. - -## Naive solution - -`naive.py` threads successor pointers through handler objects, GoF-style: -each handler either handles or forwards to `self.successor`. - -## Pythonic solution - -A chain is a *list of callables* tried in order — the first non-`None` answer -wins. Registration is appending; reordering is list surgery; the -fell-off-the-end case is explicit. That's the whole pattern. - -## In the wild - -`logging` propagation is a chain: a record climbs the logger hierarchy, -offered to each logger's handlers on the way up. `urllib.request` passes -requests through its chain of openers/handlers until one claims the scheme. - -## Verdict - -**Prefer an alternative:** a list and a loop. Objects with successor -pointers, only if handlers already are stateful objects. +Pass a request along an ordered line of handlers until one takes it — without +the sender knowing which. **Verdict: prefer an alternative** — in Python the +chain is callables in a list, not objects with successor pointers. + +| Where | What | +|---|---| +| [`pattern/`](pattern/) | The importable code: `Chain`, `Handler`, `UnhandledRequestError` | +| [`docs/`](docs/) | [Fundamentals](docs/fundamentals.md) · [Implementation guide](docs/implementation.md) · [External examples](docs/examples.md) | +| [`examples/ticket_escalation/`](examples/ticket_escalation/) | Mini-project: support-ticket routing built on `pattern/` | +| [`tests/`](tests/) | Behavioral tests for the pattern and the mini-project | + +```bash +uv run python -m patterns.behavioral.chain_of_responsibility.examples.ticket_escalation +``` diff --git a/patterns/behavioral/chain_of_responsibility/__init__.py b/patterns/behavioral/chain_of_responsibility/__init__.py index 1885894..eebf149 100644 --- a/patterns/behavioral/chain_of_responsibility/__init__.py +++ b/patterns/behavioral/chain_of_responsibility/__init__.py @@ -1 +1,12 @@ -"""Chain of Responsibility: first handler that can, does. Verdict: a list and a loop.""" +"""Chain of Responsibility — public API. + +>>> from patterns.behavioral.chain_of_responsibility import Chain +""" + +from patterns.behavioral.chain_of_responsibility.pattern import ( + Chain, + Handler, + UnhandledRequestError, +) + +__all__ = ["Chain", "Handler", "UnhandledRequestError"] diff --git a/patterns/behavioral/chain_of_responsibility/docs/examples.md b/patterns/behavioral/chain_of_responsibility/docs/examples.md new file mode 100644 index 0000000..50b4606 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/docs/examples.md @@ -0,0 +1,38 @@ +# Chain of Responsibility — where it lives outside this repo + +Cited, real implementations to study (or point an agent at) when designing or +reviewing chain-shaped code. + +## Python standard library + +- **`logging` propagation.** A record emitted on a child logger climbs the + dot-separated logger hierarchy, offered to each ancestor's handlers until + `propagate` stops it — a chain wired by naming convention. + [docs.python.org/3/library/logging.html#logging.Logger.propagate](https://docs.python.org/3/library/logging.html#logging.Logger.propagate) +- **`urllib.request.OpenerDirector`.** Openers hold an ordered list of + `BaseHandler`s; each protocol method is tried on each handler in order until + one returns a non-`None` response — decline-by-`None`, exactly this module's + contract. [docs.python.org/3/library/urllib.request.html#urllib.request.OpenerDirector](https://docs.python.org/3/library/urllib.request.html#urllib.request.OpenerDirector) + +## Major ecosystems + +- **Django middleware.** Requests descend an ordered middleware stack; any + layer may short-circuit by returning a response, otherwise it delegates + inward. Ordering is explicit configuration (`MIDDLEWARE`), and the docs + discuss it as policy. + [docs.djangoproject.com/en/stable/topics/http/middleware/](https://docs.djangoproject.com/en/stable/topics/http/middleware/) +- **pluggy `firstresult` hooks** (the engine under pytest). Hook + implementations run in registration order until the first non-`None` result + wins — Chain of Responsibility offered as a library feature flag. + [pluggy.readthedocs.io/en/stable/#first-result-only](https://pluggy.readthedocs.io/en/stable/#first-result-only) +- **WSGI middleware (PEP 3333).** Applications wrap applications; each layer + answers or passes inward. The chain here is built by function composition + rather than a list. + [peps.python.org/pep-3333/](https://peps.python.org/pep-3333/) + +## What to notice across all of them + +Every production example makes two decisions the GoF text leaves open: the +**decline convention** (`None`, `propagate=False`, "call the next app") and +the **unhandled policy** (logging's `lastResort` handler, urllib raising +`URLError`, Django's 404). When reviewing chain code, check both are explicit. diff --git a/patterns/behavioral/chain_of_responsibility/docs/fundamentals.md b/patterns/behavioral/chain_of_responsibility/docs/fundamentals.md new file mode 100644 index 0000000..951e646 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/docs/fundamentals.md @@ -0,0 +1,87 @@ +# Chain of Responsibility — fundamentals + +## Intent + +Avoid coupling the sender of a request to its receiver by giving more than one +handler a chance to act. The request travels an ordered chain until one handler +takes it; the sender never knows — and never needs to know — which one will. + +## Participants + +| Role | Classic (GoF) form | Python form | +|---|---|---| +| Handler contract | Abstract class with a successor pointer | Any callable `(request) -> answer \| None` — `None` means "decline, try the next" | +| Concrete handlers | Subclasses overriding an `_attempt` hook | Plain functions (or any callable) | +| The chain itself | Implicit in the successor links | An explicit ordered collection — `Chain` in [`pattern/chain.py`](../pattern/chain.py) | +| Client | Talks to the head of the chain | Calls `chain.handle(request)` | + +## Mechanism + +1. Handlers are placed in a deliberate order. +2. A request is offered to each handler in turn. +3. A handler either returns an answer (the chain stops) or declines by + returning `None` (the chain continues). +4. If every handler declines, the *caller's* chosen policy applies — raise + (`handle`) or fall back to a default (`handle_or`). GoF leaves this case + undefined; making it explicit is the one improvement you should always add. + +## The classic form, and what Python absorbs + +The textbook implementation threads a successor pointer through handler +*objects* — each one both does its work and forwards to the next: + +```python +class Handler(ABC): + def __init__(self, successor: Handler | None = None) -> None: + self.successor = successor # every handler carries the wiring + + def handle(self, severity: int) -> str: + answer = self._attempt(severity) + if answer is not None: + return answer + if self.successor is None: + return "unhandled" # the fall-off-the-end case, buried + return self.successor.handle(severity) + + @abstractmethod + def _attempt(self, severity: int) -> str | None: ... + + +class Helpdesk(Handler): ... + + +class Engineer(Handler): ... + + +class Management(Handler): ... + + +chain = Helpdesk(Engineer(Management())) # order hidden in nesting +``` + +Three classes, an ABC, and pointer bookkeeping — because 1994 languages had no +first-class functions. In Python the same design collapses: handlers are +functions, the chain is a list, dispatch is a loop. That collapse *is* the +pattern's Python lesson: what survives is not the class diagram but the two +ideas — **decline by convention** and **order as policy**. + +## When to use it + +- Several handlers could serve a request and the right one is known only at + runtime (escalation tiers, fallback strategies, middleware). +- You want to add, remove, or reorder handling policies without touching the + sender. + +## When not to use it + +- Exactly one receiver is ever right → a plain function call or a dict lookup. +- Every handler must see the request (notification, not handling) → that is + Observer, not a chain. +- The dispatch key is a simple value → `dict[key, handler]` beats scanning. + +## Verdict: prefer an alternative + +A list of callables and a loop is the whole pattern (this module's `Chain` is +that loop with a name and an explicit unhandled policy). Reach for +successor-pointer objects only when handlers are already stateful objects that +own their forwarding decision. diff --git a/patterns/behavioral/chain_of_responsibility/docs/implementation.md b/patterns/behavioral/chain_of_responsibility/docs/implementation.md new file mode 100644 index 0000000..2124c86 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/docs/implementation.md @@ -0,0 +1,79 @@ +# Chain of Responsibility — putting it into a system + +## The smell it fixes + +An `if/elif` ladder that keeps growing, where each arm is really a policy +owned by a different concern: + +```python +def route(ticket): + if is_faq(ticket): + ... + elif ticket.severity >= 5: + ... + elif ticket.severity <= 2: + ... + else: + ... +``` + +Every new policy edits this one function. The chain inverts that: each policy +becomes a handler that owns its own "is this mine?" test, and the router +becomes data — an ordered list you configure. + +## Steps + +1. **Define the request and answer types.** Small frozen dataclasses work + well; the types make `mypy` police the handler contract for you. +2. **Extract each ladder arm into a handler** `(request) -> answer | None`. + The arm's condition becomes the handler's decline test (`return None`). +3. **Choose the order deliberately.** Order is policy: put short-circuiting + handlers (cache hits, emergencies) before general ones — `chain.insert(0, h)` + and `chain.remove(h)` edit that policy at runtime. Write a test that + pins the order's observable behavior. +4. **Decide the unhandled policy at the call site.** `chain.handle(req)` + raises `UnhandledRequestError`; `chain.handle_or(req, default)` substitutes + a fallback. Never let "no handler" pass silently. +5. **Assemble the chain in one place** (a `build_*_chain()` factory), so the + whole routing policy is readable — and swappable in tests. + +```python +from patterns.behavioral.chain_of_responsibility import Chain + +chain: Chain[Ticket, Resolution] = Chain([auto_responder, incident_commander, helpdesk]) +chain.register(on_call) # or grow it later / use as decorator +resolution = chain.handle_or(ticket, triage(ticket)) +``` + +## Python idioms that keep it small + +- Handlers are **plain functions** until they need state; then any callable + object or `functools.partial(handler, config)` slots in unchanged. +- `chain.register` as a **decorator** turns registration into a one-liner at + definition site — the same move `singledispatch` and Flask routes use. +- Parameterize, don't subclass: `partial(severity_gate, max_severity=2)` + replaces a class hierarchy of near-identical handlers. + +## Pitfalls + +- **Silent fall-off-the-end** — the GoF form's biggest trap; the unhandled + case must be a visible decision (step 4). +- **`None` as a real answer.** The decline convention reserves `None`; if your + domain needs "the answer is nothing", wrap answers or use a sentinel. +- **Order coupling nobody wrote down.** If swapping two handlers changes + behavior, a test must fail. Test the chain's routing table, not just each + handler. +- **Handlers that mutate the request** turn a dispatch chain into a pipeline — + a different pattern with different guarantees. Keep requests immutable. +- **Overlapping predicates** make the first match arbitrary; keep each + handler's claim test exclusive enough that order expresses priority, not + accident. + +## Worked example + +[`examples/ticket_escalation/`](../examples/ticket_escalation/) applies every +step above to support-ticket routing — run it with: + +```bash +uv run python -m patterns.behavioral.chain_of_responsibility.examples.ticket_escalation +``` diff --git a/patterns/behavioral/chain_of_responsibility/examples/__init__.py b/patterns/behavioral/chain_of_responsibility/examples/__init__.py new file mode 100644 index 0000000..dc47780 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/examples/__init__.py @@ -0,0 +1 @@ +"""Mini-projects demonstrating the Chain of Responsibility in practice.""" diff --git a/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/__init__.py b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/__init__.py new file mode 100644 index 0000000..fd4aa0e --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/__init__.py @@ -0,0 +1,15 @@ +"""Support-ticket escalation built on the Chain of Responsibility. + +Run it: ``uv run python -m patterns.behavioral.chain_of_responsibility.examples.ticket_escalation`` +""" + +from patterns.behavioral.chain_of_responsibility.examples.ticket_escalation.handlers import ( + build_escalation_chain, + route, +) +from patterns.behavioral.chain_of_responsibility.examples.ticket_escalation.models import ( + Resolution, + Ticket, +) + +__all__ = ["Resolution", "Ticket", "build_escalation_chain", "route"] diff --git a/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/__main__.py b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/__main__.py new file mode 100644 index 0000000..1abe37f --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/__main__.py @@ -0,0 +1,23 @@ +"""Demo: a morning's tickets through the escalation chain.""" + +from __future__ import annotations + +from patterns.behavioral.chain_of_responsibility.examples.ticket_escalation.handlers import route +from patterns.behavioral.chain_of_responsibility.examples.ticket_escalation.models import Ticket + + +def main() -> None: + inbox = [ + Ticket("T-1", "Can't log in", 1, frozenset({"password-reset"})), + Ticket("T-2", "Wrong charge on invoice", 2, frozenset({"billing"})), + Ticket("T-3", "Export breaks on large files", 4, frozenset({"bug"})), + Ticket("T-4", "API returning 500s for everyone", 3, frozenset({"outage"})), + Ticket("T-5", "Feature idea: dark mode", 0, frozenset()), + ] + for ticket in inbox: + resolution = route(ticket) + print(f"{ticket.id} [{ticket.subject}] -> {resolution.team}: {resolution.action}") + + +if __name__ == "__main__": + main() diff --git a/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/handlers.py b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/handlers.py new file mode 100644 index 0000000..7db0ec5 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/handlers.py @@ -0,0 +1,65 @@ +"""Escalation policies as handlers, and the chain that orders them. + +Each handler claims a ticket by returning a ``Resolution`` or declines with +``None``. The chain's order *is* the escalation policy: knowledge-base +auto-replies first, outages jump every queue, then the human tiers. +""" + +from __future__ import annotations + +from patterns.behavioral.chain_of_responsibility.examples.ticket_escalation.models import ( + Resolution, + Ticket, +) +from patterns.behavioral.chain_of_responsibility.pattern import Chain + +KNOWLEDGE_BASE = { + "password-reset": "KB-101: resetting your password", + "invoice-copy": "KB-204: downloading past invoices", +} + + +def auto_responder(ticket: Ticket) -> Resolution | None: + """Answer known FAQ topics instantly, without a human.""" + for tag in ticket.tags: + if tag in KNOWLEDGE_BASE: + return Resolution(ticket.id, "bot", f"sent {KNOWLEDGE_BASE[tag]}") + return None + + +def incident_commander(ticket: Ticket) -> Resolution | None: + """Outages and severity-5 tickets bypass every queue.""" + if ticket.severity >= 5 or "outage" in ticket.tags: + return Resolution(ticket.id, "incident", "declared incident, paged commander") + return None + + +def helpdesk(ticket: Ticket) -> Resolution | None: + """First human tier: routine tickets.""" + if 1 <= ticket.severity <= 2: + return Resolution(ticket.id, "helpdesk", "assigned to helpdesk queue") + return None + + +def engineering_on_call(ticket: Ticket) -> Resolution | None: + """Second human tier: defects and anything the helpdesk can't take.""" + if 3 <= ticket.severity <= 4: + return Resolution(ticket.id, "on-call", "paged engineering on-call") + return None + + +def build_escalation_chain() -> Chain[Ticket, Resolution]: + return Chain( + [ + auto_responder, + incident_commander, # before the human tiers: outages jump the queue + helpdesk, + engineering_on_call, + ] + ) + + +def route(ticket: Ticket, chain: Chain[Ticket, Resolution] | None = None) -> Resolution: + """Route one ticket; anything no policy claims goes to human triage.""" + escalation = chain if chain is not None else build_escalation_chain() + return escalation.handle_or(ticket, Resolution(ticket.id, "triage", "queued for human triage")) diff --git a/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/models.py b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/models.py new file mode 100644 index 0000000..ed2b4ce --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/examples/ticket_escalation/models.py @@ -0,0 +1,24 @@ +"""Domain types for the ticket-escalation mini-project.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class Ticket: + """A customer support ticket. Severity: 1 (question) .. 5 (outage).""" + + id: str + subject: str + severity: int + tags: frozenset[str] = field(default_factory=frozenset) + + +@dataclass(frozen=True) +class Resolution: + """Where a ticket ended up and why.""" + + ticket_id: str + team: str + action: str diff --git a/patterns/behavioral/chain_of_responsibility/naive.py b/patterns/behavioral/chain_of_responsibility/naive.py deleted file mode 100644 index e062941..0000000 --- a/patterns/behavioral/chain_of_responsibility/naive.py +++ /dev/null @@ -1,50 +0,0 @@ -"""The Gang of Four chain: successor pointers through handler objects.""" - -from __future__ import annotations - -from abc import ABC, abstractmethod - - -class Handler(ABC): - def __init__(self, successor: Handler | None = None) -> None: - self.successor = successor - - def handle(self, severity: int) -> str: - answer = self._attempt(severity) - if answer is not None: - return answer - if self.successor is None: - return "unhandled" - return self.successor.handle(severity) - - @abstractmethod - def _attempt(self, severity: int) -> str | None: ... - - -class Helpdesk(Handler): - def _attempt(self, severity: int) -> str | None: - return "helpdesk resolves it" if severity <= 1 else None - - -class Engineer(Handler): - def _attempt(self, severity: int) -> str | None: - return "engineer resolves it" if severity <= 3 else None - - -class Management(Handler): - def _attempt(self, severity: int) -> str | None: - return "management escalation" if severity <= 5 else None - - -def build_chain() -> Handler: - return Helpdesk(Engineer(Management())) - - -def main() -> None: - chain = build_chain() - for severity in (1, 3, 5, 9): - print(f"severity {severity}: {chain.handle(severity)}") - - -if __name__ == "__main__": - main() diff --git a/patterns/behavioral/chain_of_responsibility/pattern/__init__.py b/patterns/behavioral/chain_of_responsibility/pattern/__init__.py new file mode 100644 index 0000000..33c6b88 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/pattern/__init__.py @@ -0,0 +1,9 @@ +"""The Chain of Responsibility pattern, importable as library code.""" + +from patterns.behavioral.chain_of_responsibility.pattern.chain import ( + Chain, + Handler, + UnhandledRequestError, +) + +__all__ = ["Chain", "Handler", "UnhandledRequestError"] diff --git a/patterns/behavioral/chain_of_responsibility/pattern/chain.py b/patterns/behavioral/chain_of_responsibility/pattern/chain.py new file mode 100644 index 0000000..a1ffdfd --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/pattern/chain.py @@ -0,0 +1,68 @@ +"""Chain of Responsibility as an importable, typed building block. + +A handler is any callable that returns an answer or ``None`` to decline. +``Chain`` tries its handlers in order; the first non-``None`` answer wins. +What an unhandled request means is the caller's decision: ``handle`` raises, +``handle_or`` falls back to a default. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable, Iterator +from typing import Generic, TypeVar + +Req = TypeVar("Req") +Res = TypeVar("Res") + +Handler = Callable[[Req], "Res | None"] + + +class UnhandledRequestError(LookupError): + """No handler in the chain accepted the request.""" + + +class Chain(Generic[Req, Res]): + """An ordered chain of handlers; the first non-``None`` answer wins.""" + + def __init__(self, handlers: Iterable[Handler[Req, Res]] = ()) -> None: + self._handlers: list[Handler[Req, Res]] = list(handlers) + + def register(self, handler: Handler[Req, Res]) -> Handler[Req, Res]: + """Append a handler to the end of the chain; usable as a decorator.""" + self._handlers.append(handler) + return handler + + def insert(self, index: int, handler: Handler[Req, Res]) -> None: + """Insert a handler at ``index`` — order is policy, so it is editable.""" + self._handlers.insert(index, handler) + + def remove(self, handler: Handler[Req, Res]) -> None: + """Remove a handler; ``ValueError`` if it is not in the chain.""" + self._handlers.remove(handler) + + def handle(self, request: Req) -> Res: + """Return the first handler's answer; raise if every handler declines.""" + for handler in self._handlers: + answer = handler(request) + if answer is not None: + return answer + raise UnhandledRequestError(f"no handler accepted {request!r}") + + def handle_or(self, request: Req, default: Res) -> Res: + """Like ``handle``, but fall back to ``default`` instead of raising. + + Only this chain's own exhaustion falls back: an + ``UnhandledRequestError`` raised *inside* a handler (say, a nested + chain's ``handle``) propagates — it is a routing bug, not a decline. + """ + for handler in self._handlers: + answer = handler(request) + if answer is not None: + return answer + return default + + def __iter__(self) -> Iterator[Handler[Req, Res]]: + return iter(self._handlers) + + def __len__(self) -> int: + return len(self._handlers) diff --git a/patterns/behavioral/chain_of_responsibility/pythonic.py b/patterns/behavioral/chain_of_responsibility/pythonic.py deleted file mode 100644 index ac57252..0000000 --- a/patterns/behavioral/chain_of_responsibility/pythonic.py +++ /dev/null @@ -1,43 +0,0 @@ -"""The chain as a list of callables and one loop. - -Each handler returns an answer or None; the first answer wins, and the -unhandled case is explicit at the end of the loop. -""" - -from __future__ import annotations - -from collections.abc import Callable, Sequence - -Handler = Callable[[int], str | None] - - -def helpdesk(severity: int) -> str | None: - return "helpdesk resolves it" if severity <= 1 else None - - -def engineer(severity: int) -> str | None: - return "engineer resolves it" if severity <= 3 else None - - -def management(severity: int) -> str | None: - return "management escalation" if severity <= 5 else None - - -CHAIN: list[Handler] = [helpdesk, engineer, management] - - -def handle(severity: int, chain: Sequence[Handler] | None = None) -> str: - for handler in chain if chain is not None else CHAIN: - answer = handler(severity) - if answer is not None: - return answer - return "unhandled" # falling off the end is a decision, made visible - - -def main() -> None: - for severity in (1, 3, 5, 9): - print(f"severity {severity}: {handle(severity)}") - - -if __name__ == "__main__": - main() diff --git a/patterns/behavioral/chain_of_responsibility/real_world.py b/patterns/behavioral/chain_of_responsibility/real_world.py deleted file mode 100644 index 025fd39..0000000 --- a/patterns/behavioral/chain_of_responsibility/real_world.py +++ /dev/null @@ -1,35 +0,0 @@ -"""``logging`` propagation: a record climbs the logger hierarchy. - -A child logger with no handlers still gets its records delivered -- they -propagate up the chain until some ancestor's handler takes them. -""" - -from __future__ import annotations - -import logging - - -def chain_delivery(sink: list[str]) -> None: - """Log on the child; watch the parent's handler receive it.""" - - class ListHandler(logging.Handler): - def emit(self, record: logging.LogRecord) -> None: - sink.append(f"{record.name}: {record.getMessage()}") - - parent = logging.getLogger("cor_demo") - parent.handlers.clear() - parent.setLevel(logging.INFO) - parent.addHandler(ListHandler()) - - child = logging.getLogger("cor_demo.web.requests") # no handlers of its own - child.info("timeout on /api") - - -def main() -> None: - sink: list[str] = [] - chain_delivery(sink) - print(sink) - - -if __name__ == "__main__": - main() diff --git a/patterns/behavioral/chain_of_responsibility/tests/test_chain.py b/patterns/behavioral/chain_of_responsibility/tests/test_chain.py new file mode 100644 index 0000000..2dee9f8 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/tests/test_chain.py @@ -0,0 +1,116 @@ +"""Behavioral tests for the Chain building block.""" + +from __future__ import annotations + +import pytest + +from patterns.behavioral.chain_of_responsibility import ( + Chain, + Handler, + UnhandledRequestError, +) + + +def helpdesk(severity: int) -> str | None: + return "helpdesk" if severity <= 1 else None + + +def engineer(severity: int) -> str | None: + return "engineer" if severity <= 3 else None + + +def management(severity: int) -> str | None: + return "management" if severity <= 5 else None + + +class TestDispatch: + def test_first_capable_handler_wins(self) -> None: + chain: Chain[int, str] = Chain([helpdesk, engineer, management]) + assert chain.handle(1) == "helpdesk" + assert chain.handle(3) == "engineer" + assert chain.handle(5) == "management" + + def test_order_is_policy(self) -> None: + reordered: Chain[int, str] = Chain([management, helpdesk]) + assert reordered.handle(1) == "management" + + def test_declining_handlers_are_skipped_not_consulted_again(self) -> None: + calls: list[str] = [] + + def declines(severity: int) -> str | None: + calls.append("declines") + return None + + def answers(severity: int) -> str | None: + calls.append("answers") + return "ok" + + def never_reached(severity: int) -> str | None: # pragma: no cover + calls.append("never") + return "late" + + chain: Chain[int, str] = Chain([declines, answers, never_reached]) + assert chain.handle(1) == "ok" + assert calls == ["declines", "answers"] + + +class TestUnhandledPolicy: + def test_handle_raises_with_the_request_in_the_message(self) -> None: + chain: Chain[int, str] = Chain([helpdesk]) + with pytest.raises(UnhandledRequestError, match="9"): + chain.handle(9) + + def test_handle_or_falls_back_to_default(self) -> None: + chain: Chain[int, str] = Chain([helpdesk]) + assert chain.handle_or(9, "triage") == "triage" + + def test_handle_or_propagates_a_nested_chains_unhandled_error(self) -> None: + # A handler that delegates to a misconfigured inner chain is a routing + # bug, not a decline — the outer default must NOT paper over it. + inner: Chain[int, str] = Chain([helpdesk]) + + def delegate(severity: int) -> str | None: + return inner.handle(severity) + + outer: Chain[int, str] = Chain([delegate]) + with pytest.raises(UnhandledRequestError): + outer.handle_or(9, "default") + + def test_empty_chain_is_explicitly_unhandled(self) -> None: + empty: Chain[int, str] = Chain() + with pytest.raises(UnhandledRequestError): + empty.handle(1) + + +class TestRegistration: + def test_register_appends_and_returns_the_handler(self) -> None: + chain: Chain[int, str] = Chain([helpdesk]) + returned = chain.register(engineer) + assert returned is engineer + assert list(chain) == [helpdesk, engineer] + assert chain.handle(3) == "engineer" + + def test_insert_reorders_policy(self) -> None: + chain: Chain[int, str] = Chain([helpdesk]) + chain.insert(0, management) + assert chain.handle(1) == "management" + assert list(chain) == [management, helpdesk] + + def test_remove_deletes_and_raises_on_unknown(self) -> None: + chain: Chain[int, str] = Chain([helpdesk, engineer]) + chain.remove(helpdesk) + assert chain.handle(1) == "engineer" + with pytest.raises(ValueError): + chain.remove(helpdesk) + + def test_register_works_as_a_decorator(self) -> None: + chain: Chain[int, str] = Chain() + + @chain.register + def catch_all(severity: int) -> str | None: + return "caught" + + handler: Handler[int, str] = catch_all + assert handler(0) == "caught" + assert len(chain) == 1 + assert chain.handle(99) == "caught" diff --git a/patterns/behavioral/chain_of_responsibility/tests/test_chain_of_responsibility.py b/patterns/behavioral/chain_of_responsibility/tests/test_chain_of_responsibility.py deleted file mode 100644 index 97f7c6e..0000000 --- a/patterns/behavioral/chain_of_responsibility/tests/test_chain_of_responsibility.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Behavioral tests for all three chain-of-responsibility variants.""" - -from patterns.behavioral.chain_of_responsibility import naive, pythonic, real_world - - -class TestNaive: - def test_first_capable_handler_wins(self) -> None: - chain = naive.build_chain() - assert chain.handle(1) == "helpdesk resolves it" - assert chain.handle(3) == "engineer resolves it" - assert chain.handle(5) == "management escalation" - - def test_falling_off_the_end(self) -> None: - assert naive.build_chain().handle(9) == "unhandled" - - -class TestPythonic: - def test_list_chain_matches_naive(self) -> None: - assert pythonic.handle(1) == "helpdesk resolves it" - assert pythonic.handle(3) == "engineer resolves it" - assert pythonic.handle(9) == "unhandled" - - def test_reordering_is_list_surgery(self) -> None: - reordered: list[pythonic.Handler] = [pythonic.management, pythonic.helpdesk] - assert pythonic.handle(1, reordered) == "management escalation" - - def test_empty_chain_is_explicitly_unhandled(self) -> None: - assert pythonic.handle(1, []) == "unhandled" - - -class TestRealWorld: - def test_record_propagates_to_ancestor_handler(self) -> None: - sink: list[str] = [] - real_world.chain_delivery(sink) - assert sink == ["cor_demo.web.requests: timeout on /api"] diff --git a/patterns/behavioral/chain_of_responsibility/tests/test_ticket_escalation.py b/patterns/behavioral/chain_of_responsibility/tests/test_ticket_escalation.py new file mode 100644 index 0000000..240a897 --- /dev/null +++ b/patterns/behavioral/chain_of_responsibility/tests/test_ticket_escalation.py @@ -0,0 +1,66 @@ +"""Behavioral tests for the ticket-escalation mini-project.""" + +from __future__ import annotations + +import pytest + +from patterns.behavioral.chain_of_responsibility.examples.ticket_escalation import ( + Ticket, + build_escalation_chain, + route, +) +from patterns.behavioral.chain_of_responsibility.examples.ticket_escalation.__main__ import main + + +def ticket(severity: int, tags: frozenset[str] = frozenset(), id_: str = "T-1") -> Ticket: + return Ticket(id_, "subject", severity, tags) + + +class TestRouting: + def test_faq_topics_are_answered_by_the_bot(self) -> None: + resolution = route(ticket(1, frozenset({"password-reset"}))) + assert resolution.team == "bot" + assert "KB-101" in resolution.action + + def test_routine_tickets_go_to_helpdesk(self) -> None: + assert route(ticket(2)).team == "helpdesk" + + def test_defects_page_engineering(self) -> None: + assert route(ticket(4, frozenset({"bug"}))).team == "on-call" + + def test_outages_jump_the_queue_regardless_of_severity(self) -> None: + low_severity_outage = ticket(2, frozenset({"outage"})) + assert route(low_severity_outage).team == "incident" + + def test_severity_five_is_an_incident_without_any_tag(self) -> None: + assert route(ticket(5)).team == "incident" + + def test_unclaimed_tickets_fall_back_to_human_triage(self) -> None: + feature_idea = ticket(0) + resolution = route(feature_idea) + assert resolution.team == "triage" + assert resolution.ticket_id == feature_idea.id + + def test_faq_beats_outage_because_the_bot_is_first(self) -> None: + both = ticket(5, frozenset({"password-reset", "outage"})) + assert route(both).team == "bot" + + +class TestChainShape: + def test_the_policy_is_four_handlers_in_documented_order(self) -> None: + chain = build_escalation_chain() + assert [h.__name__ for h in chain] == [ + "auto_responder", + "incident_commander", + "helpdesk", + "engineering_on_call", + ] + + +class TestDemo: + def test_main_routes_the_sample_inbox(self, capsys: pytest.CaptureFixture[str]) -> None: + main() + out = capsys.readouterr().out + assert "T-1" in out and "bot" in out + assert "T-4" in out and "incident" in out + assert "T-5" in out and "triage" in out diff --git a/pyproject.toml b/pyproject.toml index 8303eb7..07b2c0d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "python-design-patterns" version = "1.0.0" -description = "Design patterns in Python: naive, pythonic, and real-world examples, with an MCP server for agents." +description = "Design patterns as importable Python modules — docs, runnable mini-projects, and an MCP server for agents." readme = "README.md" license = { file = "LICENSE" } authors = [{ name = "SuperElectron" }] @@ -27,7 +27,7 @@ Homepage = "https://github.com/SuperElectron/python-design-patterns" Reference = "https://python-patterns.guide/" [project.scripts] -python-design-patterns-mcp = "design_patterns_mcp.server:main" +python-design-patterns-mcp = "design_patterns.mcp.server:main" [dependency-groups] dev = [ @@ -44,7 +44,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["src/design_patterns", "src/design_patterns_mcp", "patterns"] +packages = ["src/design_patterns", "patterns"] [tool.ruff] line-length = 100 diff --git a/src/design_patterns/catalog.py b/src/design_patterns/catalog.py index f34952c..1743be5 100644 --- a/src/design_patterns/catalog.py +++ b/src/design_patterns/catalog.py @@ -16,9 +16,12 @@ Verdict = Literal["pythonic", "use-with-care", "prefer-alternative"] VariantName = Literal["naive", "pythonic", "real_world"] +Shape = Literal["module", "legacy"] +DocName = Literal["fundamentals", "implementation", "examples"] VERDICTS: tuple[str, ...] = get_args(Verdict) VARIANTS: tuple[str, ...] = get_args(VariantName) +DOC_NAMES: tuple[str, ...] = get_args(DocName) _REQUIRED_KEYS = frozenset({"id", "name", "guide_url", "problem", "symptoms", "verdict", "caveats"}) @@ -51,10 +54,47 @@ def group(self) -> str: def slug(self) -> str: return self.id.split("/", 1)[1] + @property + def shape(self) -> Shape: + """``module`` units keep code in ``pattern/``; ``legacy`` units ship flat variant files. + + Any module-shape marker (``pattern/``, ``docs/``, ``examples/``) claims + the unit for strict validation, so a half-migration fails CI loudly + instead of quietly loading as legacy. + """ + markers = ("pattern", "docs", "examples") + return "module" if any((self.path / m).is_dir() for m in markers) else "legacy" + def variants(self) -> dict[str, Path]: - """The example files this unit actually ships.""" + """The flat example files a legacy unit ships (empty for module units).""" return {v: self.path / f"{v}.py" for v in VARIANTS if (self.path / f"{v}.py").is_file()} + def docs(self) -> dict[str, Path]: + """The unit's teaching docs (fundamentals/implementation/examples), if present.""" + return { + d: self.path / "docs" / f"{d}.md" + for d in DOC_NAMES + if (self.path / "docs" / f"{d}.md").is_file() + } + + def examples(self) -> dict[str, Path]: + """Runnable mini-project packages: ``examples//`` with a ``__main__.py``.""" + examples_dir = self.path / "examples" + if not examples_dir.is_dir(): + return {} + return { + child.name: child + for child in sorted(examples_dir.iterdir()) + if child.is_dir() and (child / "__main__.py").is_file() + } + + def sources(self) -> dict[str, Path]: + """The pattern's own code: ``pattern/*.py``, keyed by filename (module units).""" + pattern_dir = self.path / "pattern" + if not pattern_dir.is_dir(): + return {} + return {p.name: p for p in sorted(pattern_dir.glob("*.py"))} + def _split_frontmatter(text: str, readme: Path) -> tuple[str, str]: if not text.startswith("---\n"): @@ -119,11 +159,40 @@ def _parse_pattern(readme: Path, root: Path) -> Pattern: prose=prose, path=unit_dir, ) - if not pattern.variants(): - raise CatalogError(f"{readme}: unit ships no naive/pythonic/real_world example") + _validate_shape(pattern, readme) return pattern +def _validate_shape(pattern: Pattern, readme: Path) -> None: + """Module-shape units get strict structural validation; legacy units keep the old rule.""" + if pattern.shape == "legacy": + if not pattern.variants(): + raise CatalogError(f"{readme}: unit ships no naive/pythonic/real_world example") + return + + unit = pattern.path + if stale := sorted(pattern.variants()): + raise CatalogError(f"{readme}: module unit still ships legacy variant files: {stale}") + missing_docs = [d for d in DOC_NAMES if not (unit / "docs" / f"{d}.md").is_file()] + if missing_docs: + raise CatalogError(f"{readme}: module unit missing docs/: {missing_docs}") + if not (unit / "pattern" / "__init__.py").is_file(): + raise CatalogError(f"{readme}: module unit's pattern/ package has no __init__.py") + examples = pattern.examples() + if not examples: + raise CatalogError( + f"{readme}: module unit ships no runnable examples//__main__.py" + ) + if not (unit / "examples" / "__init__.py").is_file(): + raise CatalogError(f"{readme}: examples/ is not a package (no __init__.py)") + for name, path in examples.items(): + if not (path / "__init__.py").is_file(): + raise CatalogError(f"{readme}: example {name!r} is not a package (no __init__.py)") + tests_dir = unit / "tests" + if not any(tests_dir.glob("test_*.py")): + raise CatalogError(f"{readme}: module unit has no tests/test_*.py") + + @dataclass(frozen=True) class Catalog: """All validated pattern units, ordered by id.""" @@ -145,7 +214,10 @@ def to_json(self) -> str: for p in self.patterns: entry = asdict(p) del entry["prose"], entry["path"] + entry["shape"] = p.shape entry["variants"] = sorted(p.variants()) + entry["docs"] = sorted(p.docs()) + entry["examples"] = sorted(p.examples()) entries.append(entry) return json.dumps(entries, indent=2) diff --git a/src/design_patterns_mcp/__init__.py b/src/design_patterns/mcp/__init__.py similarity index 100% rename from src/design_patterns_mcp/__init__.py rename to src/design_patterns/mcp/__init__.py diff --git a/src/design_patterns_mcp/sandbox.py b/src/design_patterns/mcp/sandbox.py similarity index 64% rename from src/design_patterns_mcp/sandbox.py rename to src/design_patterns/mcp/sandbox.py index 1e69529..bd0acf1 100644 --- a/src/design_patterns_mcp/sandbox.py +++ b/src/design_patterns/mcp/sandbox.py @@ -1,8 +1,8 @@ """Sandboxed execution of catalog example files -- and nothing else. The contract: only paths resolved from the catalog index are runnable. -The (id, variant) pair is looked up, never joined into a path, so there is -no traversal and no arbitrary-file execution surface. +The (id, variant) / (id, example) pair is looked up, never joined into a +path, so there is no traversal and no arbitrary-file execution surface. """ from __future__ import annotations @@ -26,22 +26,12 @@ class RunResult: timed_out: bool = False -def run_example(catalog: Catalog, pattern_id: str, variant: str) -> RunResult: - """Execute one vendored example in a subprocess and capture its output.""" - pattern = catalog.get(pattern_id) # KeyError for unknown ids -- by design - variants = pattern.variants() - if variant not in variants: - raise KeyError(f"{pattern_id} has no variant {variant!r} (has: {sorted(variants)})") - path = variants[variant] # resolved by the catalog, never by the caller - if not path.is_file(): # a real check, not an assert: survives python -O - raise FileNotFoundError(f"catalog names {path} but it does not exist") - - repo_root = pattern.path.parents[2] - module = f"patterns.{pattern.group}.{pattern.slug}.{variant}" +def _run_module(repo_root: str, module: str) -> RunResult: + """Run ``python -I -m `` with a scrubbed env, scratch cwd, and output caps.""" # -I ignores PYTHONPATH by design, so the repo root (resolved by the # catalog, never by the caller) is injected in the bootstrap itself. bootstrap = ( - f"import sys, runpy; sys.path.insert(0, {str(repo_root)!r}); " + f"import sys, runpy; sys.path.insert(0, {repo_root!r}); " f"runpy.run_module({module!r}, run_name='__main__')" ) with tempfile.TemporaryDirectory() as scratch_cwd: @@ -68,3 +58,33 @@ def run_example(catalog: Catalog, pattern_id: str, variant: str) -> RunResult: stdout=completed.stdout[:MAX_OUTPUT_BYTES], stderr=completed.stderr[:MAX_OUTPUT_BYTES], ) + + +def run_example(catalog: Catalog, pattern_id: str, variant: str) -> RunResult: + """Execute one vendored legacy example file in a subprocess and capture its output.""" + pattern = catalog.get(pattern_id) # KeyError for unknown ids -- by design + variants = pattern.variants() + if variant not in variants: + raise KeyError(f"{pattern_id} has no variant {variant!r} (has: {sorted(variants)})") + path = variants[variant] # resolved by the catalog, never by the caller + if not path.is_file(): # a real check, not an assert: survives python -O + raise FileNotFoundError(f"catalog names {path} but it does not exist") + + repo_root = pattern.path.parents[2] + module = f"patterns.{pattern.group}.{pattern.slug}.{variant}" + return _run_module(str(repo_root), module) + + +def run_example_package(catalog: Catalog, pattern_id: str, example: str) -> RunResult: + """Execute a module-shape unit's ``examples/`` mini-project package.""" + pattern = catalog.get(pattern_id) # KeyError for unknown ids -- by design + examples = pattern.examples() + if example not in examples: + raise KeyError(f"{pattern_id} has no example {example!r} (has: {sorted(examples)})") + path = examples[example] # resolved by the catalog, never by the caller + if not (path / "__main__.py").is_file(): + raise FileNotFoundError(f"catalog names {path} but it has no __main__.py") + + repo_root = pattern.path.parents[2] + module = f"patterns.{pattern.group}.{pattern.slug}.examples.{example}" + return _run_module(str(repo_root), module) diff --git a/src/design_patterns_mcp/search.py b/src/design_patterns/mcp/search.py similarity index 100% rename from src/design_patterns_mcp/search.py rename to src/design_patterns/mcp/search.py diff --git a/src/design_patterns/mcp/server.py b/src/design_patterns/mcp/server.py new file mode 100644 index 0000000..2abf98e --- /dev/null +++ b/src/design_patterns/mcp/server.py @@ -0,0 +1,326 @@ +"""The python-design-patterns MCP server. + +Tools, resources, and prompts over the pattern catalog. Run over stdio by +default (``python-design-patterns-mcp``) or streamable HTTP (``--http``). +""" + +from __future__ import annotations + +import argparse +from functools import lru_cache +from typing import Any + +from mcp.server import MCPServer +from mcp.server.mcpserver.exceptions import ResourceError, ToolError + +from design_patterns.catalog import DOC_NAMES, Catalog, Pattern, load_catalog +from design_patterns.mcp.sandbox import run_example as _run_example +from design_patterns.mcp.sandbox import run_example_package as _run_example_package +from design_patterns.mcp.search import SearchIndex + + +# Lazy initialization (see patterns/python/global_object): importing this +# module must not do disk I/O; the catalog loads on first use, once. +@lru_cache(maxsize=1) +def get_catalog() -> Catalog: + return load_catalog() + + +@lru_cache(maxsize=1) +def get_index() -> SearchIndex: + return SearchIndex(get_catalog()) + + +mcp = MCPServer( + "python-design-patterns", + instructions=( + "Design patterns in Python: 32 units covering all 23 GoF patterns, " + "Python-native patterns, and modern additions, each with an honest " + "verdict. Start with search_patterns or recommend_pattern; verdicts of " + "'prefer-alternative' tell you what to write instead. Migrated " + "(module-shape) units offer three access levels: get_pattern_docs " + "(fundamentals/implementation/examples), then list_examples + " + "run_example(example=...) for runnable mini-projects, then read_source " + "(the pattern/ package). Legacy (not yet migrated) units instead ship " + "flat variant files, served via get_pattern(variant=...) and " + "run_example(variant=...) with variant one of 'naive', 'pythonic', " + "'real_world' (their literal filenames)." + ), +) + + +def _summary(pattern: Pattern) -> dict[str, Any]: + return { + "id": pattern.id, + "name": pattern.name, + "problem": pattern.problem, + "verdict": pattern.verdict, + } + + +def _detail(pattern: Pattern, include_source: str | None) -> dict[str, Any]: + detail: dict[str, Any] = { + **_summary(pattern), + "shape": pattern.shape, + "aliases": list(pattern.aliases), + "guide_url": pattern.guide_url, + "symptoms": list(pattern.symptoms), + "caveats": list(pattern.caveats), + "stdlib_sightings": list(pattern.stdlib_sightings), + "variants": sorted(pattern.variants()), + "prose": pattern.prose, + } + if pattern.shape == "module": + detail["docs"] = sorted(pattern.docs()) + detail["examples"] = sorted(pattern.examples()) + detail["note"] = ( + "module-shape unit: variants are legacy-only. Read docs via " + "get_pattern_docs, run mini-projects via list_examples + " + "run_example(example=...), read code via read_source." + ) + return detail + if include_source: + variants = pattern.variants() + wanted = sorted(variants) if include_source == "all" else [include_source] + detail["source"] = {name: variants[name].read_text() for name in wanted if name in variants} + return detail + + +@mcp.tool() +def list_patterns(group: str | None = None, verdict: str | None = None) -> list[dict[str, Any]]: + """List catalog patterns, optionally filtered by group (creational, + structural, behavioral, python, principle, modern) or verdict + (pythonic, use-with-care, prefer-alternative).""" + patterns = get_catalog().patterns + if group is not None: + patterns = tuple(p for p in patterns if p.group == group) + if verdict is not None: + patterns = tuple(p for p in patterns if p.verdict == verdict) + return [_summary(p) for p in patterns] + + +@mcp.tool() +def get_pattern(pattern_id: str, variant: str | None = None) -> dict[str, Any]: + """Fetch one pattern's full documentation. pattern_id is '/' + (e.g. 'structural/decorator'). For legacy units, variant ('naive', + 'pythonic', 'real_world', or 'all') includes that flat file's source; + module-shape units serve code via read_source instead.""" + return _detail(_get(pattern_id), variant) + + +@mcp.tool() +def search_patterns(query: str, limit: int = 5) -> list[dict[str, Any]]: + """Full-text search across pattern names, aliases, problems, symptoms, + and prose. Returns the best matches with scores.""" + return [{**_summary(h.pattern), "score": h.score} for h in get_index().search(query, limit)] + + +def _get(pattern_id: str) -> Pattern: + try: + return get_catalog().get(pattern_id) + except KeyError: + known = ", ".join(get_catalog().ids()) + raise ValueError(f"unknown pattern {pattern_id!r}; known ids: {known}") from None + + +def _require_module_shape(pattern: Pattern) -> Pattern: + if pattern.shape != "module": + raise ToolError( + f"{pattern.id} is not yet migrated to the module shape " + "(no pattern/, docs/, examples/); use get_pattern with a variant instead" + ) + return pattern + + +@mcp.tool() +def run_example( + pattern_id: str, variant: str | None = None, example: str | None = None +) -> dict[str, Any]: + """Execute one of a pattern's vendored examples in a sandboxed subprocess + and return its real output. For migrated (module-shape) patterns pass + example=; for legacy patterns pass + variant='naive'|'pythonic'|'real_world'. Exactly one of the two.""" + if (variant is None) == (example is None): + raise ValueError("pass exactly one of 'variant' (legacy) or 'example' (module-shape)") + if example is not None: + result = _run_example_package(get_catalog(), pattern_id, example) + else: + assert variant is not None + result = _run_example(get_catalog(), pattern_id, variant) + return { + "exit_code": result.exit_code, + "stdout": result.stdout, + "stderr": result.stderr, + "timed_out": result.timed_out, + } + + +@mcp.tool() +def get_pattern_docs(pattern_id: str, doc: str) -> str: + """Read one of a migrated pattern's teaching docs: 'fundamentals' (intent, + participants, mechanism, classic-form contrast), 'implementation' (how to + introduce it into a real system), or 'examples' (cited external usages).""" + pattern = _require_module_shape(_get(pattern_id)) + docs = pattern.docs() + if doc not in docs: + raise ValueError(f"doc must be one of {sorted(DOC_NAMES)}; {pattern_id} has {sorted(docs)}") + return docs[doc].read_text(encoding="utf-8") + + +@mcp.tool() +def list_examples(pattern_id: str) -> list[dict[str, Any]]: + """List a migrated pattern's runnable mini-projects (examples/); + run one with run_example(pattern_id, example=).""" + pattern = _require_module_shape(_get(pattern_id)) + return [ + { + "name": name, + "modules": sorted(p.name for p in path.glob("*.py")), + "run": f"run_example(pattern_id={pattern.id!r}, example={name!r})", + } + for name, path in pattern.examples().items() + ] + + +@mcp.tool() +def read_source(pattern_id: str) -> dict[str, str]: + """Read a migrated pattern's own implementation: every file in its + pattern/ package, keyed by filename.""" + pattern = _require_module_shape(_get(pattern_id)) + return {name: path.read_text(encoding="utf-8") for name, path in pattern.sources().items()} + + +@mcp.tool() +def recommend_pattern(problem_statement: str, limit: int = 3) -> list[dict[str, Any]]: + """Describe a design problem in plain words; get ranked candidate patterns, + each with its caveats and verdict attached. A 'prefer-alternative' verdict + means the pythonic variant shows what to write instead.""" + recommendations = [] + for hit in get_index().search(problem_statement, limit): + p = hit.pattern + rec = { + **_summary(p), + "score": hit.score, + "caveats": list(p.caveats), + "stdlib_sightings": list(p.stdlib_sightings), + } + if p.verdict == "prefer-alternative": + where = ( + f"read get_pattern_docs({p.id!r}, 'fundamentals') and read_source({p.id!r})" + if p.shape == "module" + else "see this unit's pythonic.py" + ) + rec["note"] = ( + f"The guide's honest answer is usually not {p.name}: " + f"{where} for what to write instead." + ) + recommendations.append(rec) + return recommendations + + +@mcp.resource("catalog://index") +def catalog_index() -> str: + """The whole catalog as JSON: every pattern's metadata and variants.""" + return get_catalog().to_json() + + +@mcp.resource("pattern://{group}/{slug}") +def pattern_doc(group: str, slug: str) -> str: + """One pattern's README prose.""" + return _get(f"{group}/{slug}").prose + + +@mcp.resource("pattern://{group}/{slug}/{variant}") +def pattern_source(group: str, slug: str, variant: str) -> str: + """One legacy pattern's example source (naive | pythonic | real_world).""" + pattern = _get(f"{group}/{slug}") + variants = pattern.variants() + if variant not in variants: + hint = ( + "module-shape unit: use pattern:///docs/ or the read_source tool" + if pattern.shape == "module" + else f"has: {sorted(variants)}" + ) + raise ResourceError(f"{pattern.id} has no variant {variant!r} ({hint})") + return variants[variant].read_text() + + +@mcp.resource("pattern://{group}/{slug}/docs/{doc}") +def pattern_docs_resource(group: str, slug: str, doc: str) -> str: + """One migrated pattern's teaching doc (fundamentals | implementation | examples).""" + pattern = _get(f"{group}/{slug}") + docs = pattern.docs() + if doc not in docs: + hint = ( + f"has: {sorted(docs)}" + if pattern.shape == "module" + else "unit not yet migrated to the module shape; use get_pattern instead" + ) + raise ResourceError(f"{pattern.id} has no doc {doc!r} ({hint})") + return docs[doc].read_text(encoding="utf-8") + + +@mcp.prompt() +def refactor_toward(pattern_id: str, code: str) -> str: + """Ask for a refactor of the given code toward one catalog pattern.""" + pattern = _get(pattern_id) + caveats = "\n".join(f"- {c}" for c in pattern.caveats) + reference = ( + f"read_source({pattern.id!r}) and get_pattern_docs({pattern.id!r}, 'implementation')" + if pattern.shape == "module" + else "this unit's pythonic.py file" + ) + return ( + f"Refactor the following code toward the {pattern.name} pattern " + f"({pattern.id}), as shown by {reference}.\n" + f"Verdict for this pattern: {pattern.verdict}. Honor these caveats:\n" + f"{caveats}\n\nCode:\n```python\n{code}\n```" + ) + + +@mcp.prompt() +def explain_pattern(pattern_id: str, audience: str = "an intermediate Python developer") -> str: + """Ask for an explanation of one pattern, tuned to an audience.""" + pattern = _get(pattern_id) + contrast = ( + f"the classic-form vs Python contrast in get_pattern_docs({pattern.id!r}, 'fundamentals')" + if pattern.shape == "module" + else "the contrast between this unit's classic-form and pythonic example files" + ) + return ( + f"Explain the {pattern.name} pattern to {audience}. Problem it solves: " + f"{pattern.problem} Use {contrast}, state " + f"the verdict ({pattern.verdict}) plainly, and show where the stdlib " + f"already uses it ({', '.join(pattern.stdlib_sightings)})." + ) + + +@mcp.prompt() +def choose_pattern(problem: str) -> str: + """Ask which pattern (if any!) fits a described problem.""" + return ( + f"A developer describes this problem:\n\n{problem}\n\n" + "Using the python-design-patterns catalog (search_patterns / " + "recommend_pattern), name the best-fitting pattern or say plainly that " + "no pattern is needed. If the top candidate's verdict is " + "'prefer-alternative', recommend the alternative the unit itself " + "documents instead." + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description="python-design-patterns MCP server") + parser.add_argument( + "--http", action="store_true", help="serve streamable HTTP instead of stdio" + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8734) + args = parser.parse_args() + if args.http: + mcp.run(transport="streamable-http", host=args.host, port=args.port) + else: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/src/design_patterns_mcp/server.py b/src/design_patterns_mcp/server.py deleted file mode 100644 index 1cab21f..0000000 --- a/src/design_patterns_mcp/server.py +++ /dev/null @@ -1,216 +0,0 @@ -"""The python-design-patterns MCP server. - -Tools, resources, and prompts over the pattern catalog. Run over stdio by -default (``python-design-patterns-mcp``) or streamable HTTP (``--http``). -""" - -from __future__ import annotations - -import argparse -from functools import lru_cache -from typing import Any - -from mcp.server import MCPServer - -from design_patterns.catalog import Catalog, Pattern, load_catalog -from design_patterns_mcp.sandbox import run_example as _run_example -from design_patterns_mcp.search import SearchIndex - - -# Lazy initialization (see patterns/python/global_object): importing this -# module must not do disk I/O; the catalog loads on first use, once. -@lru_cache(maxsize=1) -def get_catalog() -> Catalog: - return load_catalog() - - -@lru_cache(maxsize=1) -def get_index() -> SearchIndex: - return SearchIndex(get_catalog()) - - -mcp = MCPServer( - "python-design-patterns", - instructions=( - "Design patterns in Python: 32 units covering all 23 GoF patterns, " - "Python-native patterns, and modern additions. Each unit has prose, a " - "naive (GoF-literal) example, a pythonic example, a real_world stdlib " - "sighting, and an honest verdict. Start with search_patterns or " - "recommend_pattern; verdicts of 'prefer-alternative' tell you what to " - "write instead." - ), -) - - -def _summary(pattern: Pattern) -> dict[str, Any]: - return { - "id": pattern.id, - "name": pattern.name, - "problem": pattern.problem, - "verdict": pattern.verdict, - } - - -def _detail(pattern: Pattern, include_source: str | None) -> dict[str, Any]: - detail: dict[str, Any] = { - **_summary(pattern), - "aliases": list(pattern.aliases), - "guide_url": pattern.guide_url, - "symptoms": list(pattern.symptoms), - "caveats": list(pattern.caveats), - "stdlib_sightings": list(pattern.stdlib_sightings), - "variants": sorted(pattern.variants()), - "prose": pattern.prose, - } - if include_source: - variants = pattern.variants() - wanted = sorted(variants) if include_source == "all" else [include_source] - detail["source"] = {name: variants[name].read_text() for name in wanted if name in variants} - return detail - - -@mcp.tool() -def list_patterns(group: str | None = None, verdict: str | None = None) -> list[dict[str, Any]]: - """List catalog patterns, optionally filtered by group (creational, - structural, behavioral, python, principle, modern) or verdict - (pythonic, use-with-care, prefer-alternative).""" - patterns = get_catalog().patterns - if group is not None: - patterns = tuple(p for p in patterns if p.group == group) - if verdict is not None: - patterns = tuple(p for p in patterns if p.verdict == verdict) - return [_summary(p) for p in patterns] - - -@mcp.tool() -def get_pattern(pattern_id: str, variant: str | None = None) -> dict[str, Any]: - """Fetch one pattern's full documentation. pattern_id is '/' - (e.g. 'structural/decorator'). variant: 'naive', 'pythonic', 'real_world', - or 'all' to include example source code.""" - try: - pattern = get_catalog().get(pattern_id) - except KeyError: - known = ", ".join(get_catalog().ids()) - raise ValueError(f"unknown pattern {pattern_id!r}; known ids: {known}") from None - return _detail(pattern, variant) - - -@mcp.tool() -def search_patterns(query: str, limit: int = 5) -> list[dict[str, Any]]: - """Full-text search across pattern names, aliases, problems, symptoms, - and prose. Returns the best matches with scores.""" - return [{**_summary(h.pattern), "score": h.score} for h in get_index().search(query, limit)] - - -@mcp.tool() -def run_example(pattern_id: str, variant: str) -> dict[str, Any]: - """Execute one of a pattern's vendored example files ('naive', 'pythonic', - 'real_world') in a sandboxed subprocess and return its real output.""" - result = _run_example(get_catalog(), pattern_id, variant) - return { - "exit_code": result.exit_code, - "stdout": result.stdout, - "stderr": result.stderr, - "timed_out": result.timed_out, - } - - -@mcp.tool() -def recommend_pattern(problem_statement: str, limit: int = 3) -> list[dict[str, Any]]: - """Describe a design problem in plain words; get ranked candidate patterns, - each with its caveats and verdict attached. A 'prefer-alternative' verdict - means the pythonic variant shows what to write instead.""" - recommendations = [] - for hit in get_index().search(problem_statement, limit): - p = hit.pattern - rec = { - **_summary(p), - "score": hit.score, - "caveats": list(p.caveats), - "stdlib_sightings": list(p.stdlib_sightings), - } - if p.verdict == "prefer-alternative": - rec["note"] = ( - f"The guide's honest answer is usually not {p.name}: " - f"see this unit's pythonic.py for what to write instead." - ) - recommendations.append(rec) - return recommendations - - -@mcp.resource("catalog://index") -def catalog_index() -> str: - """The whole catalog as JSON: every pattern's metadata and variants.""" - return get_catalog().to_json() - - -@mcp.resource("pattern://{group}/{slug}") -def pattern_doc(group: str, slug: str) -> str: - """One pattern's README prose.""" - return get_catalog().get(f"{group}/{slug}").prose - - -@mcp.resource("pattern://{group}/{slug}/{variant}") -def pattern_source(group: str, slug: str, variant: str) -> str: - """One pattern's example source (naive | pythonic | real_world).""" - pattern = get_catalog().get(f"{group}/{slug}") - variants = pattern.variants() - if variant not in variants: - raise KeyError(f"{pattern.id} has no variant {variant!r}") - return variants[variant].read_text() - - -@mcp.prompt() -def refactor_toward(pattern_id: str, code: str) -> str: - """Ask for a refactor of the given code toward one catalog pattern.""" - pattern = get_catalog().get(pattern_id) - caveats = "\n".join(f"- {c}" for c in pattern.caveats) - return ( - f"Refactor the following code toward the {pattern.name} pattern " - f"({pattern.id}), as done in this catalog's pythonic variant.\n" - f"Verdict for this pattern: {pattern.verdict}. Honor these caveats:\n" - f"{caveats}\n\nCode:\n```python\n{code}\n```" - ) - - -@mcp.prompt() -def explain_pattern(pattern_id: str, audience: str = "an intermediate Python developer") -> str: - """Ask for an explanation of one pattern, tuned to an audience.""" - pattern = get_catalog().get(pattern_id) - return ( - f"Explain the {pattern.name} pattern to {audience}. Problem it solves: " - f"{pattern.problem} Use the catalog's naive-vs-pythonic contrast, state " - f"the verdict ({pattern.verdict}) plainly, and show where the stdlib " - f"already uses it ({', '.join(pattern.stdlib_sightings)})." - ) - - -@mcp.prompt() -def choose_pattern(problem: str) -> str: - """Ask which pattern (if any!) fits a described problem.""" - return ( - f"A developer describes this problem:\n\n{problem}\n\n" - "Using the python-design-patterns catalog (search_patterns / " - "recommend_pattern), name the best-fitting pattern or say plainly that " - "no pattern is needed. If the top candidate's verdict is " - "'prefer-alternative', recommend the alternative its pythonic variant " - "shows instead." - ) - - -def main() -> None: - parser = argparse.ArgumentParser(description="python-design-patterns MCP server") - parser.add_argument( - "--http", action="store_true", help="serve streamable HTTP instead of stdio" - ) - parser.add_argument("--host", default="127.0.0.1") - parser.add_argument("--port", type=int, default=8734) - args = parser.parse_args() - if args.http: - mcp.run(transport="streamable-http", host=args.host, port=args.port) - else: - mcp.run() - - -if __name__ == "__main__": - main() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..4bcc19c --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,50 @@ +"""Shared fixtures: a synthetic module-shape unit for engine tests. + +The real catalog's module-shape pilot is being built alongside this code, so +engine tests use a self-contained synthetic unit instead of assuming any real +unit's API. +""" + +from pathlib import Path + +import pytest + +from design_patterns.catalog import Catalog, load_catalog + + +def write_module_unit(root: Path) -> Path: + """Build ``/creational/thing`` as a complete, runnable module-shape unit.""" + unit = root / "creational" / "thing" + (unit / "pattern").mkdir(parents=True) + for pkg in (root, root / "creational", unit, unit / "pattern"): + (pkg / "__init__.py").write_text("") + (unit / "README.md").write_text( + "---\n" + "id: creational/thing\nname: Thing\nguide_url: null\n" + 'problem: "Build a thing."\nsymptoms: ["thing needed"]\nverdict: pythonic\ncaveats: []\n' + "---\n\n# Thing\n" + ) + (unit / "pattern" / "thing.py").write_text("def build() -> str:\n return 'built a thing'\n") + docs = unit / "docs" + docs.mkdir() + for name in ("fundamentals", "implementation", "examples"): + (docs / f"{name}.md").write_text(f"# {name} of Thing\n") + project = unit / "examples" / "demo" + project.mkdir(parents=True) + (unit / "examples" / "__init__.py").write_text("") + (project / "__init__.py").write_text("") + (project / "__main__.py").write_text( + "from patterns.creational.thing.pattern.thing import build\n\nprint(build())\n" + ) + tests = unit / "tests" + tests.mkdir() + (tests / "test_thing.py").write_text("def test_ok() -> None:\n assert True\n") + return unit + + +@pytest.fixture +def module_catalog(tmp_path: Path) -> Catalog: + """A catalog whose ``patterns/`` root holds one synthetic module-shape unit.""" + root = tmp_path / "patterns" + write_module_unit(root) + return load_catalog(root) diff --git a/tests/test_catalog.py b/tests/test_catalog.py index cfc3aef..8d5b8f4 100644 --- a/tests/test_catalog.py +++ b/tests/test_catalog.py @@ -19,9 +19,24 @@ def test_loads_all_units(self) -> None: assert len(catalog.patterns) == 32 assert "structural/decorator" in catalog.ids() - def test_every_unit_ships_all_three_variants(self) -> None: + def test_catalog_contains_both_shapes_during_migration(self) -> None: + # The pilot migrated at least one unit; a silent regression of a + # module unit back to legacy shape must fail here, not skip a branch. + shapes = {p.shape for p in load_catalog().patterns} + assert "module" in shapes + module_ids = {p.id for p in load_catalog().patterns if p.shape == "module"} + assert "behavioral/chain_of_responsibility" in module_ids + + def test_every_unit_ships_its_shape_completely(self) -> None: for pattern in load_catalog().patterns: - assert sorted(pattern.variants()) == ["naive", "pythonic", "real_world"], pattern.id + if pattern.shape == "module": + assert sorted(pattern.docs()) == ["examples", "fundamentals", "implementation"], ( + pattern.id + ) + assert pattern.examples(), pattern.id + assert pattern.sources(), pattern.id + else: + assert sorted(pattern.variants()) == ["naive", "pythonic", "real_world"], pattern.id def test_verdicts_are_from_the_vocabulary(self) -> None: for pattern in load_catalog().patterns: @@ -99,5 +114,97 @@ def test_empty_tree_fails(self, tmp_path: Path) -> None: load_catalog(tmp_path) +def _write_module_unit(root: Path, group: str, slug: str, frontmatter: str) -> Path: + """A minimal valid module-shape unit: pattern/ + docs/ + examples/ + tests/.""" + unit = root / group / slug + (unit / "pattern").mkdir(parents=True) + (unit / "README.md").write_text(f"---\n{frontmatter}\n---\n\n# x\n") + (unit / "__init__.py").write_text("") + (unit / "pattern" / "__init__.py").write_text("") + (unit / "pattern" / "thing.py").write_text("def build() -> str:\n return 'thing'\n") + docs = unit / "docs" + docs.mkdir() + for name in ("fundamentals", "implementation", "examples"): + (docs / f"{name}.md").write_text(f"# {name}\n") + project = unit / "examples" / "demo" + project.mkdir(parents=True) + (unit / "examples" / "__init__.py").write_text("") + (project / "__init__.py").write_text("") + (project / "__main__.py").write_text("print('demo ran')\n") + tests = unit / "tests" + tests.mkdir() + (tests / "test_thing.py").write_text("def test_ok() -> None:\n assert True\n") + return unit + + +class TestModuleShapeValidation: + def test_valid_module_unit_loads_with_shape_fields(self, tmp_path: Path) -> None: + _write_module_unit(tmp_path, "creational", "thing", GOOD) + pattern = load_catalog(tmp_path).get("creational/thing") + assert pattern.shape == "module" + assert sorted(pattern.docs()) == ["examples", "fundamentals", "implementation"] + assert sorted(pattern.examples()) == ["demo"] + assert sorted(pattern.sources()) == ["__init__.py", "thing.py"] + assert pattern.variants() == {} + + def test_missing_doc_fails(self, tmp_path: Path) -> None: + unit = _write_module_unit(tmp_path, "creational", "thing", GOOD) + (unit / "docs" / "implementation.md").unlink() + with pytest.raises(CatalogError, match=r"missing docs.*implementation"): + load_catalog(tmp_path) + + def test_no_example_fails(self, tmp_path: Path) -> None: + unit = _write_module_unit(tmp_path, "creational", "thing", GOOD) + (unit / "examples" / "demo" / "__main__.py").unlink() + with pytest.raises(CatalogError, match="no runnable examples"): + load_catalog(tmp_path) + + def test_example_not_a_package_fails(self, tmp_path: Path) -> None: + unit = _write_module_unit(tmp_path, "creational", "thing", GOOD) + (unit / "examples" / "demo" / "__init__.py").unlink() + with pytest.raises(CatalogError, match="not a package"): + load_catalog(tmp_path) + + def test_empty_tests_fails(self, tmp_path: Path) -> None: + unit = _write_module_unit(tmp_path, "creational", "thing", GOOD) + (unit / "tests" / "test_thing.py").unlink() + with pytest.raises(CatalogError, match="no tests"): + load_catalog(tmp_path) + + def test_pattern_package_needs_init(self, tmp_path: Path) -> None: + unit = _write_module_unit(tmp_path, "creational", "thing", GOOD) + (unit / "pattern" / "__init__.py").unlink() + with pytest.raises(CatalogError, match=r"no __init__\.py"): + load_catalog(tmp_path) + + def test_stale_legacy_variant_files_fail(self, tmp_path: Path) -> None: + unit = _write_module_unit(tmp_path, "creational", "thing", GOOD) + (unit / "pythonic.py").write_text("def main() -> None: ...\n") + with pytest.raises(CatalogError, match="legacy variant files"): + load_catalog(tmp_path) + + def test_half_migration_is_claimed_and_fails_loudly(self, tmp_path: Path) -> None: + # docs/ alone marks the unit module-shape; strict validation then + # demands the rest instead of silently classifying it legacy. + unit = tmp_path / "creational" / "thing" + (unit / "docs").mkdir(parents=True) + (unit / "README.md").write_text(f"---\n{GOOD}\n---\n\n# x\n") + with pytest.raises(CatalogError, match="module unit missing docs"): + load_catalog(tmp_path) + + def test_examples_dir_needs_init(self, tmp_path: Path) -> None: + unit = _write_module_unit(tmp_path, "creational", "thing", GOOD) + (unit / "examples" / "__init__.py").unlink() + with pytest.raises(CatalogError, match=r"examples/ is not a package"): + load_catalog(tmp_path) + + def test_index_json_carries_shape(self, tmp_path: Path) -> None: + _write_module_unit(tmp_path, "creational", "thing", GOOD) + entries = json.loads(load_catalog(tmp_path).to_json()) + assert entries[0]["shape"] == "module" + assert entries[0]["examples"] == ["demo"] + assert entries[0]["docs"] == ["examples", "fundamentals", "implementation"] + + def test_find_patterns_root_from_repo() -> None: assert find_patterns_root().name == "patterns" diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index d0db7a5..6c00f18 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2,10 +2,13 @@ import json +import pytest from mcp import Client from mcp.types import TextResourceContents -from design_patterns_mcp.server import mcp +import design_patterns.mcp.server as server_module +from design_patterns.catalog import Catalog +from design_patterns.mcp.server import mcp class TestTools: @@ -72,6 +75,99 @@ async def test_recommend_attaches_caveats_and_alternative_note(self) -> None: assert singleton["caveats"] +class TestModuleShapeTools: + """The three new access levels, against a synthetic migrated unit.""" + + @pytest.fixture(autouse=True) + def _use_module_catalog(self, module_catalog: Catalog, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(server_module, "get_catalog", lambda: module_catalog) + + async def test_get_pattern_docs(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "get_pattern_docs", {"pattern_id": "creational/thing", "doc": "fundamentals"} + ) + assert not result.is_error + assert "fundamentals of Thing" in str(result.content[0]) + + async def test_get_pattern_docs_rejects_unknown_doc(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "get_pattern_docs", {"pattern_id": "creational/thing", "doc": "naive"} + ) + assert result.is_error + + async def test_list_examples_names_the_run_call(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("list_examples", {"pattern_id": "creational/thing"}) + assert result.structured_content is not None + examples = result.structured_content["result"] + assert [e["name"] for e in examples] == ["demo"] + assert "run_example" in examples[0]["run"] + + async def test_run_example_by_package(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "run_example", {"pattern_id": "creational/thing", "example": "demo"} + ) + assert result.structured_content is not None + run = result.structured_content + assert run["exit_code"] == 0 and "built a thing" in run["stdout"] + + async def test_run_example_requires_exactly_one_selector(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("run_example", {"pattern_id": "creational/thing"}) + assert result.is_error + + async def test_read_source_returns_pattern_package(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("read_source", {"pattern_id": "creational/thing"}) + assert result.structured_content is not None + sources = result.structured_content + assert "built a thing" in sources["thing.py"] + + async def test_docs_resource(self) -> None: + async with Client(mcp) as client: + doc = await client.read_resource("pattern://creational/thing/docs/implementation") + first = doc.contents[0] + assert isinstance(first, TextResourceContents) + assert "implementation of Thing" in first.text + + +class TestLegacyShapeErrors: + """Module-shape tools refuse un-migrated units with a clear message, not a crash.""" + + async def test_get_pattern_docs_on_legacy_unit(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "get_pattern_docs", {"pattern_id": "structural/decorator", "doc": "fundamentals"} + ) + assert result.is_error + assert "not yet migrated" in str(result.content[0]) + + async def test_list_examples_on_legacy_unit(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("list_examples", {"pattern_id": "structural/decorator"}) + assert result.is_error + + async def test_read_source_on_legacy_unit(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("read_source", {"pattern_id": "structural/decorator"}) + assert result.is_error + + async def test_docs_resource_error_text_reaches_client(self) -> None: + # ResourceError (not ValueError) is required for the hint to survive + # the SDK's template wrapper — this pins that the text gets through. + async with Client(mcp) as client: + with pytest.raises(Exception, match="not yet migrated"): + await client.read_resource("pattern://structural/decorator/docs/fundamentals") + + async def test_variant_resource_error_text_reaches_client(self) -> None: + async with Client(mcp) as client: + with pytest.raises(Exception, match="module-shape unit"): + await client.read_resource("pattern://behavioral/chain_of_responsibility/naive") + + class TestResources: async def test_catalog_index_resource(self) -> None: async with Client(mcp) as client: diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index 57b99e2..a5b20c4 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -2,8 +2,8 @@ import pytest -from design_patterns.catalog import load_catalog -from design_patterns_mcp.sandbox import run_example +from design_patterns.catalog import Catalog, load_catalog +from design_patterns.mcp.sandbox import run_example, run_example_package CATALOG = load_catalog() @@ -34,15 +34,48 @@ def test_failing_example_reports_not_raises(self) -> None: assert isinstance(result.stderr, str) +class TestPackageSandbox: + def test_runs_a_module_example_that_imports_the_pattern(self, module_catalog: Catalog) -> None: + result = run_example_package(module_catalog, "creational/thing", "demo") + assert result.exit_code == 0, result.stderr + assert "built a thing" in result.stdout + assert not result.timed_out + + def test_unknown_example_is_refused(self, module_catalog: Catalog) -> None: + with pytest.raises(KeyError, match="no example"): + run_example_package(module_catalog, "creational/thing", "nope") + + def test_traversal_shaped_example_is_refused(self, module_catalog: Catalog) -> None: + with pytest.raises(KeyError): + run_example_package(module_catalog, "creational/thing", "../../../tmp/evil") + + def test_unknown_pattern_id_is_refused(self, module_catalog: Catalog) -> None: + with pytest.raises(KeyError): + run_example_package(module_catalog, "../../etc/passwd", "demo") + + def test_legacy_unit_has_no_packages(self) -> None: + with pytest.raises(KeyError, match="no example"): + run_example_package(CATALOG, "structural/flyweight", "pythonic") + + def test_runs_the_real_pilot_unit(self) -> None: + # The migrated unit itself, through the python -I -m path CI must cover. + result = run_example_package( + CATALOG, "behavioral/chain_of_responsibility", "ticket_escalation" + ) + assert result.exit_code == 0, result.stderr + assert "triage" in result.stdout + assert not result.timed_out + + class TestSearchIndex: def test_symptom_search_hits_the_right_unit(self) -> None: - from design_patterns_mcp.search import SearchIndex + from design_patterns.mcp.search import SearchIndex index = SearchIndex(CATALOG) top = index.search("undo redo history snapshot", limit=3) assert top and top[0].pattern.id in {"behavioral/memento", "behavioral/command"} def test_no_match_returns_empty(self) -> None: - from design_patterns_mcp.search import SearchIndex + from design_patterns.mcp.search import SearchIndex assert SearchIndex(CATALOG).search("zzzqqqxxx") == []