Last active
November 25, 2025 03:32
-
-
Save jakkaj/b73f113f27bafce0bb14a9745034aa45 to your computer and use it in GitHub Desktop.
[md-export] constitution-code-4
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from datetime import datetime | |
| # Test doubles implement the same interfaces as production code | |
| class FakeBookingRepository(BookingRepository): | |
| """In-memory repository for testing - no HTTP, no database.""" | |
| def __init__(self): | |
| self._bookings: dict[str, Booking] = {} | |
| def find_by_id(self, booking_id: BookingId) -> Booking | None: | |
| return self._bookings.get(booking_id.value) | |
| def save(self, booking: Booking) -> None: | |
| self._bookings[booking.id.value] = booking | |
| class FakePaymentAdapter(PaymentAdapter): | |
| """Predictable payment adapter for testing - no Stripe SDK.""" | |
| def __init__(self, should_succeed: bool = True): | |
| self._should_succeed = should_succeed | |
| self.charges: list[PaymentRequest] = [] | |
| def charge(self, request: PaymentRequest) -> PaymentResult: | |
| self.charges.append(request) | |
| return PaymentResult( | |
| transaction_id="fake-txn-123", | |
| success=self._should_succeed, | |
| ) | |
| def test_given_valid_booking_command_when_creating_booking_then_charges_payment_and_persists(): | |
| """ | |
| Test Doc: | |
| - Why: Verifies the core booking creation flow orchestrates payment and persistence correctly | |
| - Contract: create_booking charges payment via adapter, then saves confirmed booking via repository | |
| - Usage Notes: Inject FakeBookingRepository and FakePaymentAdapter; command requires user_id, departure, origin, destination | |
| - Quality Contribution: Critical path - booking creation is the primary user journey | |
| - Worked Example: SYD→MEL booking for user-1 on 2025-06-15 → payment charged, booking saved with same ID | |
| """ | |
| # Arrange - construct fakes and inject into real service | |
| fake_repo = FakeBookingRepository() | |
| fake_payments = FakePaymentAdapter(should_succeed=True) | |
| service = BookingApplicationServiceImpl( | |
| bookings=fake_repo, | |
| payments=fake_payments, | |
| ) | |
| command = CreateBookingCommand( | |
| user_id="user-1", | |
| departure=datetime(2025, 6, 15, 10, 0), | |
| origin="SYD", | |
| destination="MEL", | |
| ) | |
| # Act | |
| booking = service.create_booking(command) | |
| # Assert - payment was charged with correct booking | |
| assert len(fake_payments.charges) == 1 | |
| assert fake_payments.charges[0].booking_id == booking.id | |
| # Assert - booking was persisted | |
| saved = fake_repo.find_by_id(booking.id) | |
| assert saved is not None | |
| assert saved.user_id == "user-1" |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from abc import ABC, abstractmethod | |
| # Domain command + IDs used at the boundary of the application layer | |
| @dataclass | |
| class BookingId: | |
| value: str | |
| @dataclass | |
| class CreateBookingCommand: | |
| user_id: str | |
| departure: datetime | |
| origin: str | |
| destination: str | |
| # Application service interface | |
| class BookingApplicationService(ABC): | |
| @abstractmethod | |
| def create_booking(self, command: CreateBookingCommand) -> "Booking": | |
| pass | |
| @abstractmethod | |
| def get_booking(self, booking_id: BookingId) -> "Booking | None": | |
| pass | |
| class BookingApplicationServiceImpl(BookingApplicationService): | |
| def __init__( | |
| self, | |
| bookings: "BookingRepository", | |
| payments: "PaymentAdapter", | |
| ): | |
| self._bookings = bookings | |
| self._payments = payments | |
| def create_booking(self, command: CreateBookingCommand) -> "Booking": | |
| # Validation & business rules | |
| # (no HTTP, no SDK types) | |
| self._validate_command(command) | |
| provisional = Booking.provisional( | |
| user_id=command.user_id, | |
| departure=command.departure, | |
| origin=command.origin, | |
| destination=command.destination, | |
| ) | |
| # Orchestration via adapters/repositories | |
| payment_result = self._payments.charge( | |
| PaymentRequest.for_booking(provisional), | |
| ) | |
| confirmed = provisional.confirm(payment_result) | |
| self._bookings.save(confirmed) | |
| return confirmed | |
| def get_booking(self, booking_id: BookingId) -> "Booking | None": | |
| return self._bookings.find_by_id(booking_id) | |
| def _validate_command(self, command: CreateBookingCommand) -> None: | |
| # Pure validation; raise domain-level errors only | |
| pass |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # Interface lives in the application/domain module | |
| from abc import ABC, abstractmethod | |
| from typing import Optional | |
| import httpx | |
| class BookingRepository(ABC): | |
| @abstractmethod | |
| async def find_by_id(self, booking_id: BookingId) -> Optional["Booking"]: | |
| pass | |
| @abstractmethod | |
| async def save(self, booking: "Booking") -> None: | |
| pass | |
| # Implementation lives in an infrastructure module | |
| class HttpBookingRepository(BookingRepository): | |
| def __init__(self, client: httpx.AsyncClient): | |
| self._client = client | |
| async def find_by_id(self, booking_id: BookingId) -> Optional["Booking"]: | |
| response = await self._client.get( | |
| f"https://api.example.com/bookings/{booking_id.value}" | |
| ) | |
| if response.status_code == 404: | |
| return None | |
| # Map HTTP/JSON → domain | |
| return Booking.from_dict(response.json()) | |
| async def save(self, booking: "Booking") -> None: | |
| await self._client.post( | |
| "https://api.example.com/bookings", | |
| json=booking.to_dict(), | |
| headers={"Content-Type": "application/json"}, | |
| ) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # Interface visible to application services | |
| from abc import ABC, abstractmethod | |
| from dataclasses import dataclass | |
| class PaymentAdapter(ABC): | |
| @abstractmethod | |
| async def charge(self, request: "PaymentRequest") -> "PaymentResult": | |
| pass | |
| # Domain-facing request/result types | |
| @dataclass | |
| class PaymentRequest: | |
| booking_id: BookingId | |
| amount: "Money" | |
| @classmethod | |
| def for_booking(cls, booking: "Booking") -> "PaymentRequest": | |
| return cls( | |
| booking_id=booking.id, | |
| amount=booking.total_price, | |
| ) | |
| @dataclass | |
| class PaymentResult: | |
| transaction_id: str | |
| success: bool | |
| # Infrastructure implementation wrapping a vendor SDK | |
| class StripePaymentAdapter(PaymentAdapter): | |
| def __init__(self, stripe: "StripeClient"): | |
| self._stripe = stripe # vendor SDK type | |
| async def charge(self, request: PaymentRequest) -> PaymentResult: | |
| session = await self._stripe.charge( | |
| amount_cents=request.amount.cents, | |
| metadata={"booking_id": request.booking_id.value}, | |
| ) | |
| return PaymentResult( | |
| transaction_id=session.id, | |
| success=session.status == "succeeded", | |
| ) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment