Local Development Guide

This guide walks through setting up the service for local development, running it, and executing the test suite.

Prerequisites

1. Clone the repository

git clone https://github.com/ITlusions/ITL.Platform.SubscriptionVending.git
cd ITL.Platform.SubscriptionVending

2. Install dependencies

The project uses Hatchling as its build backend. Install the package together with the development extras:

pip install -e ".[dev]"

This installs:

• All runtime dependencies (fastapi, uvicorn, pydantic, Azure SDK packages, etc.)
• Development tools: pytest, pytest-asyncio, httpx

3. Configure the environment

Copy the example configuration and fill in the required values:

cp .env.example .env

At minimum you must set VENDING_AZURE_TENANT_ID. For a pure local/mock run you can set it to any non-empty string.

To enable the mock webhook endpoint (useful for testing without a real Azure subscription):

VENDING_MOCK_MODE=true

See configuration.md for a full description of all variables, and secrets.md for guidance on which variables are secrets and how to handle them securely.

4. Run the service

uvicorn subscription_vending.main:app --reload --host 0.0.0.0 --port 8000

The service is available at http://localhost:8000.

5. Project layout

src/subscription_vending/
├── main.py                  application factory, router registration
│
├── core/
│   ├── config.py            Settings (Pydantic) + get_settings() @lru_cache singleton
│   ├── context.py           StepContext, ProvisioningResult — pure dataclasses, no I/O
│   ├── job.py               ProvisioningJob — retry-queue payload
│   ├── enums.py             RetryStrategy enum
│   ├── base.py              BaseStep ABC — plugin contract
│   ├── events.py            lifecycle event bus (STARTED / SUCCEEDED / FAILED / COMPLETED)
│   ├── registry.py          step + gate registries, register_step(), register_gate(), _toposort()
│   ├── protocols.py         Azure port contracts (typing.Protocol, @runtime_checkable)
│   └── exceptions.py        typed exception hierarchy (AppError → ProvisioningError, etc.)
│
├── workflow/
│   ├── engine.py            WorkflowEngine class
│   └── steps.py             built-in provisioning steps 1–6
│
├── cli/                     Click CLI (`vending` command)
│   ├── __init__.py
│   ├── main.py              provision / preflight / status / enqueue / config subcommands
│   └── monitor.py           jobs / events subcommands
│
├── schemas/
│   └── event_grid.py        HTTP schemas: EventGridEvent, WebhookResponse, HealthResponse
│
├── handlers/                FastAPI routers (driving adapters), each as a sub-package
│   ├── event_grid/          POST /webhook/
│   ├── worker/              POST /worker/process-job
│   ├── preflight/           POST /webhook/preflight
│   ├── replay/              POST /webhook/replay
│   ├── mock/                POST /webhook/test (VENDING_MOCK_MODE)
│   └── jobs/                GET+DELETE /jobs/* — queue monitoring & management
│
├── infrastructure/          I/O adapters
│   ├── azure/               Azure SDK calls (management groups, RBAC, policy, tags, notifications)
│   └── queue/               retry dispatcher + Azure Storage Queue client
│
└── extensions/              auto-discovered plugins (self-register at import time)
    ├── __init__.py          autodiscover() + re-exports for extension authors
    ├── webhook_notify.py    optional: POST result to HTTPS webhook
    ├── api_notify.py        optional: POST result to REST API (Bearer token)
    ├── servicenow_check.py  optional gate: validate ServiceNow ticket before provisioning
    └── servicenow_feedback.py  optional step: PATCH ServiceNow ticket with outcome

Key conventions

• Import StepContext / ProvisioningResult from core.context; import register_step / register_gate from core.registry. Both are also re-exported from workflow and extensions for convenience.
• Use get_settings() everywhere instead of Settings(). The singleton is cached after the first call. The authoritative location is core.config.
• New HTTP schemas go in schemas/ — never in core/.
• New Azure SDK calls go in infrastructure/azure/ — never in workflow/ or handlers/.
• New retry/queue logic goes in infrastructure/queue/ — never inline in handlers.
• Raise typed exceptions from core/exceptions.py rather than appending plain strings to ctx.result.errors (the plain-string pattern is still supported for backward compatibility).
• Extensions are active when their controlling env var is set. To write a new extension, create a .py file in extensions/ — it will be auto-discovered at startup. Import everything you need from subscription_vending.extensions (re-exported there).

6. Running with Docker Compose

docker-compose up --build

Docker Compose starts the service with VENDING_MOCK_MODE=true. The service is available at http://localhost:8000.

To rebuild the image after code changes:

docker-compose up --build --force-recreate

7. Triggering the mock workflow

With mock mode enabled, you can trigger a full provisioning run without connecting to Azure:

curl -X POST http://localhost:8000/webhook/test \
  -H "Content-Type: application/json" \
  -d '{
    "subscription_id": "00000000-0000-0000-0000-000000000001",
    "subscription_name": "local-test-sub",
    "management_group_id": "ITL-Development"
  }'

Add "dry_run": true to skip all Azure mutations and outbound HTTP calls — only log output is produced:

curl -X POST http://localhost:8000/webhook/test \
  -H "Content-Type: application/json" \
  -d '{
    "subscription_id": "00000000-0000-0000-0000-000000000001",
    "subscription_name": "local-test-sub",
    "management_group_id": "ITL-Development",
    "dry_run": true
  }'

Note: “Mock mode” only enables the /webhook/test endpoint — it does not stub out Azure SDK calls. The full provisioning workflow is executed, including calls to the Azure Management Groups, RBAC, and Policy APIs. These calls will fail in a local environment without valid Azure credentials, but each step’s error is caught, logged, and appended to the result without crashing the service. This makes the endpoint useful for verifying request routing and partial workflow logic.

8. Running tests

pytest

Or with verbose output:

pytest -v

To run a specific test file:

pytest tests/test_workflow.py -v

The test suite uses pytest-asyncio in auto mode, so all async test functions are automatically treated as async tests.

When patching settings in tests, call get_settings.cache_clear() before applying monkeypatch env vars:

from subscription_vending.config import get_settings

def test_something(monkeypatch):
    get_settings.cache_clear()
    monkeypatch.setenv("VENDING_AZURE_TENANT_ID", "test-tenant")
    settings = get_settings()
    ...

Test structure

9. CLI reference

The vending CLI provides local and remote-mode access to the service. Install the CLI extras:

pip install -e ".[cli]"

All commands that call a live API accept --remote <URL> (or VENDING_API_URL env var) and -v / --verbose to print request timing to stderr.

Provisioning

# Provision a subscription locally (runs WorkflowEngine in-process)
vending provision --sub-id <id> --sub-name <name> [--mg-id <mg>] [--dry-run]

# Provision via a running API
vending provision --sub-id <id> --sub-name <name> --remote http://vending:8000 [--secret <secret>]

# Dry-run preflight check
vending preflight --sub-id <id> --sub-name <name> [--remote http://vending:8000]

# Show current worker / queue status
vending status

Configuration

# Show active configuration (local: reads Settings; remote: GET /config)
vending config show [--remote http://vending:8000] [-o json|table]

# Validate configuration and connectivity
vending config validate [--remote http://vending:8000]

Queue management

# Enqueue a job directly (bypasses Event Grid)
vending enqueue --sub-id <id> --sub-name <name> [--mg-id <mg>] [--remote http://vending:8000]

# Peek provisioning queue
vending jobs list [--remote http://vending:8000] [--count 10]

# Peek dead-letter queue
vending jobs dlq  [--remote http://vending:8000] [--count 10]

# Approximate message counts for both queues
vending jobs stats [--remote http://vending:8000]

# Continuously poll the provisioning queue
vending jobs watch [--remote http://vending:8000] [--interval 5]

# Find a specific job by ID (searches both queues)
vending jobs get <job_id> [--remote http://vending:8000]

# Clear all messages from the dead-letter queue (asks for confirmation)
vending jobs purge [--remote http://vending:8000] [--yes]

Event Grid connectivity

# Check Event Grid topic endpoint reachability
vending events test [--remote http://vending:8000]

Local queue commands (--account / --conn-str) connect directly to Azure Storage Queue. Set VENDING_STORAGE_ACCOUNT_NAME or VENDING_STORAGE_CONNECTION_STRING to avoid passing them on every invocation.

8. Project structure recap

src/subscription_vending/   # Application source
  azure/                    # Azure SDK wrappers (management groups, RBAC, policy, etc.)
  core/                     # Shared internals (BaseStep ABC, lifecycle event bus)
  extensions/               # Auto-discovered extension modules
  handlers/                 # FastAPI route handlers
tests/                      # Pytest test suite
infra/                      # Bicep IaC templates
k8s/                        # Kubernetes manifests
docs/                       # Documentation
.env.example                # Annotated environment-variable template
pyproject.toml              # Build metadata and dependencies
Dockerfile                  # Container image definition
docker-compose.yml          # Local development compose file

Tips

• Hot reload: --reload in the uvicorn command restarts the server on every file change.
• Log level: Set the LOG_LEVEL environment variable (or pass --log-level debug to uvicorn) for more verbose output.
• Interactive API docs: The Swagger UI at /docs lets you call all endpoints directly from the browser without curl.