diff --git a/.gitignore b/.gitignore index e44eafd..626ebb2 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,7 @@ build/ .mypy_cache/ .ruff_cache/ uv.lock +.idea/ .DS_Store # === agent-memory: AI infrastructure (personal / per-machine — do not commit) === diff --git a/README.md b/README.md index 5fb74ce..1a89343 100644 --- a/README.md +++ b/README.md @@ -26,10 +26,10 @@ graph task calls a Python function exactly as if it were local. ```python # app.py -from mercury_composable import AppException, platform, preload +from mercury_composable import AppException, Body, platform, preload @preload(route="hello.python", instances=10) -def handle_event(headers: dict, body): +def handle_event(headers: dict[str, str], body: Body): if not isinstance(body, dict) or "text" not in body: raise AppException(400, "missing 'text'") return {"text": str(body["text"]).upper(), "language": "python"} @@ -66,7 +66,7 @@ now executes the Python function, with trace context carried end to end. ## The function contract A handler receives the same two-part input as an engine `TypedLambdaFunction` — -`(headers: dict, body)` — and returns the reply body (or an `EventEnvelope` for full +`(headers: dict[str, str], body: Body)` — `Body` is any MsgPack value — and returns the reply body (or an `EventEnvelope` for full control of status and reply headers). `async def` and plain `def` are both supported; synchronous handlers run in a thread-pool executor so the event loop never blocks. @@ -75,6 +75,8 @@ synchronous handlers run in a thread-pool executor so the event loop never block exception handler or the graph's `error.*` contract. - `get_trace()` exposes `trace_id` / `trace_path` / `cid`; `annotate_trace(k, v)` sends an annotation back on the reply envelope. +- Outside a hosted function (batch jobs, tests), `trace_context(trace_id, trace_path)` + establishes the context your `PostOffice` calls inherit — the node `runWithTrace` twin. - Functions must be stateless; anything you must keep belongs to the caller's flow model or state machine. @@ -125,6 +127,19 @@ orchestration** — those live in the engines. It provides functions plus the mi foundation utilities, keeping Python fast to prototype with while the composable core guarantees the architecture. +## Development + +```bash +uv venv .venv && uv pip install -e '.[dev]' # environment (uv-managed python) +.venv/bin/pytest -q # tests +uvx ruff check . # lint (config in pyproject.toml) +uvx basedpyright # type check (config in pyproject.toml) +``` + +PyCharm: use interpreter type **uv** pointing at the project `.venv`, and set +*Settings → Tools → Python Integrated Tools → Package requirements file* to +`pyproject.toml` so the requirements inspection reads `[project.dependencies]`. + ## License Apache 2.0 — see [LICENSE.txt](LICENSE.txt). diff --git a/examples/demo_app.py b/examples/demo_app.py index 80cb91d..d0f3c45 100644 --- a/examples/demo_app.py +++ b/examples/demo_app.py @@ -10,14 +10,26 @@ target: 'http://127.0.0.1:8086/api/event' """ -from mercury_composable import AppException, annotate_trace, get_logger, platform, preload +from mercury_composable import ( + AppException, + Body, + annotate_trace, + get_logger, + platform, + preload, +) log = get_logger(__name__) @preload(route="hello.python", instances=10) -def handle_event(headers: dict, body): - """Uppercase transform - the polyglot hello world.""" +def handle_event(_headers: dict[str, str], body: Body): + """Uppercase transform - the polyglot hello world. + + The (headers, body) two-part signature is the function contract (the + TypedLambdaFunction mirror) - a handler that does not need headers keeps + the parameter, underscore-prefixed per Python convention. + """ if not isinstance(body, dict) or "text" not in body: raise AppException(400, "missing 'text'") annotate_trace("language", "python") @@ -26,7 +38,7 @@ def handle_event(headers: dict, body): @preload(route="hello.declarative", instances=10) -async def declarative_echo(headers: dict, body): +async def declarative_echo(headers: dict[str, str], body: Body): """Echo for the composable-example declarative Event-over-HTTP demo.""" return {"body": body, "headers": headers, "language": "python"} diff --git a/pyproject.toml b/pyproject.toml index 398c26b..17ff60f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ Homepage = "https://github.com/Accenture/mercury-python" Documentation = "https://accenture.github.io/mercury-composable" [project.optional-dependencies] -dev = ["pytest>=8", "pytest-asyncio>=0.23"] +dev = ["pytest>=8", "pytest-asyncio>=0.23", "ruff>=0.9"] [project.scripts] mercury-serve = "mercury_composable.cli:main" @@ -35,6 +35,27 @@ mercury-serve = "mercury_composable.cli:main" [tool.hatch.build.targets.wheel] packages = ["src/mercury_composable"] +[tool.ruff] +line-length = 100 +target-version = "py310" +# the agent-skills layer is tool-managed by agent-memory (overwritten on upgrade); +# style fixes for it belong upstream, not here +extend-exclude = ["agent-skills"] + +[tool.ruff.lint] +extend-select = ["I"] + +[tool.basedpyright] +# standard mode plus the contract-relevant strictness: every function parameter +# is annotated, unit tests included (the wire contract is typed - Body, +# dict[str, str]); the reportUnknown* warning family of the editor's +# "recommended" mode is deliberately not chased. agent-skills is tool-managed +# by agent-memory and excluded. +typeCheckingMode = "standard" +reportMissingParameterType = "error" +include = ["src", "examples", "tests"] +exclude = ["agent-skills", ".venv"] + [tool.pytest.ini_options] asyncio_mode = "auto" testpaths = ["tests"] diff --git a/src/mercury_composable/__init__.py b/src/mercury_composable/__init__.py index b0bdd2f..050ef12 100644 --- a/src/mercury_composable/__init__.py +++ b/src/mercury_composable/__init__.py @@ -11,18 +11,36 @@ from .client import PostOffice from .config import AppConfig, app_config, load_config -from .envelope import EventEnvelope, iso_utc +from .envelope import Body, EventEnvelope, iso_utc from .exceptions import AppException, CompactFormatError from .log import get_logger -from .registry import FunctionRegistry, default_registry, preload +from .registry import FunctionRegistry, Handler, default_registry, preload from .server import EventApiServer, Platform, platform -from .trace import TraceInfo, annotate_trace, get_trace +from .trace import TraceInfo, annotate_trace, get_trace, trace_context __version__ = "0.1.0" __all__ = [ - "AppConfig", "AppException", "CompactFormatError", "EventApiServer", - "EventEnvelope", "FunctionRegistry", "Platform", "PostOffice", "TraceInfo", - "annotate_trace", "app_config", "default_registry", "get_logger", - "get_trace", "iso_utc", "load_config", "platform", "preload", "__version__", + "AppConfig", + "AppException", + "Body", + "CompactFormatError", + "EventApiServer", + "EventEnvelope", + "FunctionRegistry", + "Handler", + "Platform", + "PostOffice", + "TraceInfo", + "__version__", + "annotate_trace", + "app_config", + "default_registry", + "get_logger", + "get_trace", + "iso_utc", + "load_config", + "platform", + "preload", + "trace_context", ] diff --git a/src/mercury_composable/cli.py b/src/mercury_composable/cli.py index 76dcb98..06b796b 100644 --- a/src/mercury_composable/cli.py +++ b/src/mercury_composable/cli.py @@ -25,10 +25,15 @@ def main() -> int: help="Configuration file (default: resources/application.yml|properties)") # -Dkey=value runtime overrides are consumed by AppConfig from sys.argv args, _unknown = parser.parse_known_args() + # argparse Namespace attributes are untyped - pin the types at the boundary + app_arg: str = args.app + port_arg: int | None = args.port + host_arg: str = args.host + config_arg: str | None = args.config from .config import DEFAULT_CANDIDATES, load_config - app_path = os.path.abspath(args.app) - config_path = args.config + app_path = os.path.abspath(app_arg) + config_path: str | None = config_arg if config_path is None and not any(os.path.isfile(c) for c in DEFAULT_CANDIDATES): # fall back to a resources folder next to the application file app_dir = os.path.dirname(app_path) @@ -56,7 +61,7 @@ def main() -> int: print("No functions registered - use @preload(route=..., instances=...)", file=sys.stderr) return 1 - platform.run(port=args.port, host=args.host) + platform.run(port=port_arg, host=host_arg) return 0 diff --git a/src/mercury_composable/client.py b/src/mercury_composable/client.py index 779417b..93af7d3 100644 --- a/src/mercury_composable/client.py +++ b/src/mercury_composable/client.py @@ -15,7 +15,7 @@ from __future__ import annotations import re -from typing import Any, Dict, Optional +from typing import Any import aiohttp @@ -27,32 +27,52 @@ _W3C_SPAN_ID = re.compile(r"^[0-9a-f]{16}$") +def _build_event(route: str, body: Any, headers: dict[str, str] | None, + from_route: str | None, cid: str | None) -> EventEnvelope: + """Build the outbound envelope, inheriting the current trace context.""" + event = EventEnvelope(to=route, body=body, headers=headers or {}) + if from_route: + event.set_from(from_route) + info = get_trace() + if info and info.trace_id: + event.set_trace(info.trace_id, info.trace_path or route) + effective_cid = cid or (info.cid if info else None) + if effective_cid: + event.set_correlation_id(effective_cid) + return event + + class PostOffice: """Event-over-HTTP client for calling functions on peer applications.""" - def __init__(self, endpoint: Optional[str] = None, - security_headers: Optional[Dict[str, str]] = None): + def __init__(self, endpoint: str | None = None, + security_headers: dict[str, str] | None = None): self.endpoint = endpoint self.security_headers = dict(security_headers or {}) - self._session: Optional[aiohttp.ClientSession] = None + self._session: aiohttp.ClientSession | None = None - async def _get_session(self) -> aiohttp.ClientSession: - if self._session is None or self._session.closed: - self._session = aiohttp.ClientSession() - return self._session + def _get_session(self) -> aiohttp.ClientSession: + # called from the running event loop only (inside request/send) + session = self._session + if session is None or session.closed: + session = aiohttp.ClientSession() + self._session = session + return session async def close(self) -> None: if self._session is not None and not self._session.closed: await self._session.close() - async def __aenter__(self) -> "PostOffice": + # PYI034 wants '-> Self', which needs python >= 3.11; switch when the + # floor moves past 3.10 + async def __aenter__(self) -> PostOffice: # noqa: PYI034 return self - async def __aexit__(self, *_exc) -> None: + async def __aexit__(self, *_exc: object) -> None: await self.close() def _http_headers(self, timeout_ms: int, is_async: bool, - event: EventEnvelope) -> Dict[str, str]: + event: EventEnvelope) -> dict[str, str]: headers = { "content-type": "application/octet-stream", "accept": "*/*", @@ -71,28 +91,15 @@ def _http_headers(self, timeout_ms: int, is_async: bool, headers["traceparent"] = f"00-{event.trace_id}-{event.span_id}-01" return headers - def _build_event(self, route: str, body: Any, headers: Optional[Dict[str, str]], - from_route: Optional[str], cid: Optional[str]) -> EventEnvelope: - event = EventEnvelope(to=route, body=body, headers=headers or {}) - if from_route: - event.set_from(from_route) - info = get_trace() - if info and info.trace_id: - event.set_trace(info.trace_id, info.trace_path or route) - effective_cid = cid or (info.cid if info else None) - if effective_cid: - event.set_correlation_id(effective_cid) - return event - - async def _call(self, route: str, body: Any, headers: Optional[Dict[str, str]], - timeout_ms: int, endpoint: Optional[str], is_async: bool, - from_route: Optional[str], cid: Optional[str]) -> EventEnvelope: + async def _call(self, route: str, body: Any, headers: dict[str, str] | None, + timeout_ms: int, endpoint: str | None, is_async: bool, + from_route: str | None, cid: str | None) -> EventEnvelope: url = endpoint or self.endpoint if not url: raise ValueError("Missing event endpoint - " "e.g. PostOffice(endpoint='http://peer:8085/api/event')") - event = self._build_event(route, body, headers, from_route, cid) - session = await self._get_session() + event = _build_event(route, body, headers, from_route, cid) + session = self._get_session() # +100 ms cushion so the HTTP client does not time out before the target client_timeout = aiohttp.ClientTimeout(total=(max(100, timeout_ms) + 100) / 1000) async with session.post(url, data=event.to_bytes(), @@ -106,21 +113,21 @@ async def _call(self, route: str, body: Any, headers: Optional[Dict[str, str]], f"Invalid event-over-http response - {e}") from e async def request(self, route: str, body: Any = None, *, - headers: Optional[Dict[str, str]] = None, + headers: dict[str, str] | None = None, timeout_ms: int = 30000, - endpoint: Optional[str] = None, - from_route: Optional[str] = None, - cid: Optional[str] = None) -> EventEnvelope: + endpoint: str | None = None, + from_route: str | None = None, + cid: str | None = None) -> EventEnvelope: """RPC call: returns the target function's reply envelope.""" return await self._call(route, body, headers, timeout_ms, endpoint, False, from_route, cid) async def send(self, route: str, body: Any = None, *, - headers: Optional[Dict[str, str]] = None, + headers: dict[str, str] | None = None, timeout_ms: int = 30000, - endpoint: Optional[str] = None, - from_route: Optional[str] = None, - cid: Optional[str] = None) -> EventEnvelope: + endpoint: str | None = None, + from_route: str | None = None, + cid: str | None = None) -> EventEnvelope: """Drop-n-forget: returns the peer's 202 delivery acknowledgement envelope.""" return await self._call(route, body, headers, timeout_ms, endpoint, True, from_route, cid) diff --git a/src/mercury_composable/config.py b/src/mercury_composable/config.py index 4bba6c3..926d0a4 100644 --- a/src/mercury_composable/config.py +++ b/src/mercury_composable/config.py @@ -33,11 +33,11 @@ import re import sys import threading -from typing import Any, List, Optional +from typing import Any import yaml -_REF = re.compile(r"\$\{([^}]+)\}") +_REF = re.compile(r"\$\{([^}]+)}") DEFAULT_CANDIDATES = [ "resources/application.yml", @@ -46,7 +46,7 @@ ] -def _flatten(prefix: str, node: Any, out: dict) -> None: +def _flatten(prefix: str, node: Any, out: dict[str, Any]) -> None: if isinstance(node, dict): for k, v in node.items(): key = f"{prefix}.{k}" if prefix else str(k) @@ -55,8 +55,8 @@ def _flatten(prefix: str, node: Any, out: dict) -> None: out[prefix] = node -def _parse_properties(text: str) -> dict: - result: dict = {} +def _parse_properties(text: str) -> dict[str, Any]: + result: dict[str, Any] = {} for line in text.splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: @@ -66,9 +66,9 @@ def _parse_properties(text: str) -> dict: return result -def parse_d_args(argv: List[str]) -> dict: +def parse_d_args(argv: list[str]) -> dict[str, Any]: """Extract -Dkey=value runtime overrides (Java/Rust engine syntax).""" - overrides: dict = {} + overrides: dict[str, Any] = {} for arg in argv: if arg.startswith("-D") and "=" in arg: key, _, value = arg[2:].partition("=") @@ -80,9 +80,9 @@ def parse_d_args(argv: List[str]) -> dict: class AppConfig: """Flat, dot-addressed application configuration.""" - def __init__(self, path: Optional[str] = None, argv: Optional[List[str]] = None): - self._store: dict = {} - self._overrides: dict = parse_d_args(sys.argv[1:] if argv is None else argv) + def __init__(self, path: str | None = None, argv: list[str] | None = None): + self._store: dict[str, Any] = {} + self._overrides: dict[str, Any] = parse_d_args(sys.argv[1:] if argv is None else argv) self._source = "none" candidates = [path] if path else DEFAULT_CANDIDATES for candidate in candidates: @@ -99,7 +99,7 @@ def _load(self, path: str) -> None: text = f.read() if path.endswith((".yml", ".yaml")): data = yaml.safe_load(text) or {} - flat: dict = {} + flat: dict[str, Any] = {} _flatten("", data, flat) self._store = flat else: @@ -125,7 +125,7 @@ def get(self, key: str, default: Any = None) -> Any: return value return default - def get_property(self, key: str, default: Optional[str] = None) -> Optional[str]: + def get_property(self, key: str, default: str | None = None) -> str | None: value = self.get(key, default) return None if value is None else str(value) @@ -138,9 +138,9 @@ def _substitute(self, value: str, default: Any = None) -> Any: resolved = self._resolve_ref(match.group(1)) return resolved if resolved is not None else default - def repl(m: "re.Match[str]") -> str: - resolved = self._resolve_ref(m.group(1)) - return "" if resolved is None else str(resolved) + def repl(m: re.Match[str]) -> str: + replacement = self._resolve_ref(m.group(1)) + return "" if replacement is None else str(replacement) return _REF.sub(repl, value) @@ -158,21 +158,24 @@ def _resolve_ref(self, ref: str) -> Any: _lock = threading.Lock() -_instance: Optional[AppConfig] = None +_instance: AppConfig | None = None def app_config() -> AppConfig: """The shared AppConfig singleton (created on first use).""" global _instance with _lock: - if _instance is None: - _instance = AppConfig() - return _instance + instance = _instance + if instance is None: + instance = AppConfig() + _instance = instance + return instance -def load_config(path: Optional[str] = None) -> AppConfig: +def load_config(path: str | None = None) -> AppConfig: """Replace the shared AppConfig (used by the CLI before startup).""" global _instance with _lock: - _instance = AppConfig(path) - return _instance + instance = AppConfig(path) + _instance = instance + return instance diff --git a/src/mercury_composable/envelope.py b/src/mercury_composable/envelope.py index a51d7ed..90da93a 100644 --- a/src/mercury_composable/envelope.py +++ b/src/mercury_composable/envelope.py @@ -16,12 +16,16 @@ import uuid from datetime import datetime, timezone -from typing import Any, Dict, Optional +from typing import Any, TypeAlias, cast import msgpack from .exceptions import CompactFormatError +# any MsgPack value - the payload universe of the standard wire format: +# a map, array, string, integer, float, boolean, binary or nil +Body: TypeAlias = "None | bool | int | float | str | bytes | list[Body] | dict[str, Body]" + # wire field names (standard format) _ID = "id" _TO = "to" @@ -43,7 +47,7 @@ _EXCEPTION = "exception" -def iso_utc(dt: Optional[datetime] = None) -> str: +def iso_utc(dt: datetime | None = None) -> str: """ISO-8601 UTC with millisecond precision, e.g. 2026-07-21T12:00:00.000Z""" value = dt or datetime.now(timezone.utc) if value.tzinfo is None: @@ -61,59 +65,59 @@ def _pack_default(obj: Any) -> Any: class EventEnvelope: """In-memory event with the same field vocabulary as the Java/Rust engines.""" - def __init__(self, to: Optional[str] = None, body: Any = None, - headers: Optional[Dict[str, str]] = None): + def __init__(self, to: str | None = None, body: Any = None, + headers: dict[str, str] | None = None): self.id: str = str(uuid.uuid4()).replace("-", "") self.to = to - self.sender: Optional[str] = None # wire field "from" - self.reply_to: Optional[str] = None - self.cid: Optional[str] = None - self.trace_id: Optional[str] = None - self.trace_path: Optional[str] = None - self.span_id: Optional[str] = None - self.status: Optional[int] = None # None encodes as absent (default 200) - self.headers: Dict[str, str] = dict(headers or {}) + self.sender: str | None = None # wire field "from" + self.reply_to: str | None = None + self.cid: str | None = None + self.trace_id: str | None = None + self.trace_path: str | None = None + self.span_id: str | None = None + self.status: int | None = None # None encodes as absent (default 200) + self.headers: dict[str, str] = dict(headers or {}) self.body: Any = body - self.exec_time: Optional[float] = None - self.round_trip: Optional[float] = None - self.tags: Dict[str, str] = {} - self.annotations: Dict[str, Any] = {} - self.stack: Optional[str] = None - self.obj_type: Optional[str] = None - self.exception: Optional[bytes] = None # language-native, opaque here + self.exec_time: float | None = None + self.round_trip: float | None = None + self.tags: dict[str, str] = {} + self.annotations: dict[str, Any] = {} + self.stack: str | None = None + self.obj_type: str | None = None + self.exception: bytes | None = None # language-native, opaque here # --- fluent helpers mirroring the engine API vocabulary --- - def set_to(self, route: str) -> "EventEnvelope": + def set_to(self, route: str) -> EventEnvelope: self.to = route return self - def set_from(self, route: str) -> "EventEnvelope": + def set_from(self, route: str) -> EventEnvelope: self.sender = route return self - def set_header(self, key: str, value: Any) -> "EventEnvelope": + def set_header(self, key: str, value: Any) -> EventEnvelope: self.headers[str(key)] = str(value) return self - def set_body(self, body: Any) -> "EventEnvelope": + def set_body(self, body: Any) -> EventEnvelope: self.body = body return self - def set_status(self, status: int) -> "EventEnvelope": + def set_status(self, status: int) -> EventEnvelope: self.status = int(status) return self - def set_correlation_id(self, cid: str) -> "EventEnvelope": + def set_correlation_id(self, cid: str) -> EventEnvelope: self.cid = cid return self - def set_trace(self, trace_id: str, trace_path: str) -> "EventEnvelope": + def set_trace(self, trace_id: str, trace_path: str) -> EventEnvelope: self.trace_id = trace_id self.trace_path = trace_path return self - def set_reply_to(self, route: Optional[str]) -> "EventEnvelope": + def set_reply_to(self, route: str | None) -> EventEnvelope: self.reply_to = route return self @@ -125,8 +129,8 @@ def has_error(self) -> bool: # --- wire codec (standard format) --- - def to_map(self) -> Dict[str, Any]: - result: Dict[str, Any] = {_ID: self.id, _HEADERS: dict(self.headers)} + def to_map(self) -> dict[str, Any]: + result: dict[str, Any] = {_ID: self.id, _HEADERS: dict(self.headers)} optional = [ (_TO, self.to), (_FROM, self.sender), (_REPLY_TO, self.reply_to), (_CID, self.cid), (_TRACE_ID, self.trace_id), (_TRACE_PATH, self.trace_path), @@ -144,12 +148,15 @@ def to_map(self) -> Dict[str, Any]: return result def to_bytes(self) -> bytes: - return msgpack.packb(self.to_map(), use_bin_type=True, default=_pack_default) + # packb returns None only in the legacy stream mode - never here + return cast(bytes, msgpack.packb(self.to_map(), use_bin_type=True, default=_pack_default)) @classmethod - def from_map(cls, data: Dict[str, Any]) -> "EventEnvelope": + def from_map(cls, data: dict[str, Any]) -> EventEnvelope: event = cls() - event.id = str(data.get(_ID)) if data.get(_ID) is not None else event.id + raw_id: Any = data.get(_ID) + if raw_id is not None: + event.id = str(raw_id) event.to = data.get(_TO) event.sender = data.get(_FROM) event.reply_to = data.get(_REPLY_TO) @@ -157,27 +164,34 @@ def from_map(cls, data: Dict[str, Any]) -> "EventEnvelope": event.trace_id = data.get(_TRACE_ID) event.trace_path = data.get(_TRACE_PATH) event.span_id = data.get(_SPAN_ID) - status = data.get(_STATUS) - event.status = int(status) if status is not None else None - headers = data.get(_HEADERS) - event.headers = {str(k): str(v) for k, v in headers.items()} if isinstance(headers, dict) else {} + raw_status: Any = data.get(_STATUS) + if raw_status is not None: + event.status = int(raw_status) + raw_headers = data.get(_HEADERS) + if isinstance(raw_headers, dict): + event.headers = {str(k): str(v) for k, v in raw_headers.items()} event.body = data.get(_BODY) - exec_time = data.get(_EXEC_TIME) - event.exec_time = float(exec_time) if exec_time is not None else None - round_trip = data.get(_ROUND_TRIP) - event.round_trip = float(round_trip) if round_trip is not None else None - tags = data.get(_TAGS) - event.tags = {str(k): str(v) for k, v in tags.items()} if isinstance(tags, dict) else {} - annotations = data.get(_ANNOTATIONS) - event.annotations = dict(annotations) if isinstance(annotations, dict) else {} + raw_exec_time: Any = data.get(_EXEC_TIME) + if raw_exec_time is not None: + event.exec_time = float(raw_exec_time) + raw_round_trip: Any = data.get(_ROUND_TRIP) + if raw_round_trip is not None: + event.round_trip = float(raw_round_trip) + raw_tags = data.get(_TAGS) + if isinstance(raw_tags, dict): + event.tags = {str(k): str(v) for k, v in raw_tags.items()} + raw_annotations = data.get(_ANNOTATIONS) + if isinstance(raw_annotations, dict): + event.annotations = dict(raw_annotations) event.stack = data.get(_STACK) event.obj_type = data.get(_OBJ_TYPE) - exception = data.get(_EXCEPTION) - event.exception = bytes(exception) if isinstance(exception, (bytes, bytearray)) else None + raw_exception = data.get(_EXCEPTION) + if isinstance(raw_exception, (bytes, bytearray)): + event.exception = bytes(raw_exception) return event @classmethod - def from_bytes(cls, data: bytes) -> "EventEnvelope": + def from_bytes(cls, data: bytes) -> EventEnvelope: try: decoded = msgpack.unpackb(data, raw=False) except Exception as e: @@ -192,5 +206,5 @@ def from_bytes(cls, data: bytes) -> "EventEnvelope": return cls.from_map(decoded) def __repr__(self) -> str: - return (f"EventEnvelope(id={self.id!r}, to={self.to!r}, " - f"status={self.get_status()}, headers={self.headers!r})") + return (f"EventEnvelope(id='{self.id}', to='{self.to or ''}', " + f"status={self.get_status()}, headers={self.headers})") diff --git a/src/mercury_composable/log.py b/src/mercury_composable/log.py index 26d602d..55ac4a4 100644 --- a/src/mercury_composable/log.py +++ b/src/mercury_composable/log.py @@ -25,7 +25,6 @@ import os import sys import time -from typing import Optional from .config import app_config @@ -37,7 +36,10 @@ def format(self, record: logging.LogRecord) -> str: ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(record.created)) ms = int(record.msecs) level = f"{record.levelname:<5}" - return f"{ts}.{ms:03d} {level} {record.name}:{record.lineno} - {record.getMessage()}" + line = f"{ts}.{ms:03d} {level} {record.name}:{record.lineno} - {record.getMessage()}" + if record.exc_info: + line += "\n" + self.formatException(record.exc_info) + return line class EngineJsonFormatter(logging.Formatter): @@ -54,6 +56,8 @@ def format(self, record: logging.LogRecord) -> str: info = get_trace() if info and info.trace_id: entry["trace_id"] = info.trace_id + if record.exc_info: + entry["exception"] = self.formatException(record.exc_info) return json.dumps(entry, ensure_ascii=False) @@ -73,7 +77,7 @@ def _setup() -> None: _configured = True -def get_logger(name: Optional[str] = None) -> logging.Logger: +def get_logger(name: str | None = None) -> logging.Logger: """A logger writing engine-consistent log lines.""" _setup() return logging.getLogger(name or "mercury") diff --git a/src/mercury_composable/registry.py b/src/mercury_composable/registry.py index d62beb2..00e1576 100644 --- a/src/mercury_composable/registry.py +++ b/src/mercury_composable/registry.py @@ -3,7 +3,7 @@ Mirrors the engines' PreLoad vocabulary: a function is registered under a route name with an instance count (its concurrency limit) and a private flag. -Handlers take ``(headers: dict, body)`` — the same two-part input as a +Handlers take ``(headers: dict[str, str], body)`` — the same two-part input as a TypedLambdaFunction — and return the reply body (or an EventEnvelope for full control of status and reply headers). Both ``async def`` and plain ``def`` handlers are supported; synchronous handlers run in the default executor so @@ -14,8 +14,15 @@ import inspect import re +from collections.abc import Callable from dataclasses import dataclass -from typing import Any, Callable, Dict, Optional +from typing import Any + +from .envelope import Body + +# the function contract: (headers, body) in, reply body (or EventEnvelope) out - +# mirrors the node package's exported Handler type +Handler = Callable[[dict[str, str], Body], Any] _ROUTE_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]*$") @@ -32,7 +39,7 @@ def validate_route(route: str) -> str: @dataclass class ServiceDef: route: str - handler: Callable[[Dict[str, str], Any], Any] + handler: Handler instances: int = 10 private: bool = False is_async: bool = False @@ -40,9 +47,9 @@ class ServiceDef: class FunctionRegistry: def __init__(self) -> None: - self._services: Dict[str, ServiceDef] = {} + self._services: dict[str, ServiceDef] = {} - def register(self, route: str, handler: Callable, *, + def register(self, route: str, handler: Handler, *, instances: int = 10, private: bool = False) -> ServiceDef: route = validate_route(route) service = ServiceDef( @@ -55,13 +62,13 @@ def register(self, route: str, handler: Callable, *, self._services[route] = service return service - def get(self, route: str) -> Optional[ServiceDef]: + def get(self, route: str) -> ServiceDef | None: return self._services.get(route) def exists(self, route: str) -> bool: return route in self._services - def routes(self) -> Dict[str, ServiceDef]: + def routes(self) -> dict[str, ServiceDef]: return dict(self._services) @@ -75,10 +82,10 @@ def preload(route: str, instances: int = 10, private: bool = False): Usage:: @preload(route="hello.python", instances=10) - def handle_event(headers: dict, body): + def handle_event(headers: dict[str, str], body): return {"text": body["text"].upper()} """ - def wrapper(fn: Callable) -> Callable: + def wrapper(fn: Handler) -> Handler: default_registry.register(route, fn, instances=instances, private=private) return fn return wrapper diff --git a/src/mercury_composable/server.py b/src/mercury_composable/server.py index 20f0c85..cecd290 100644 --- a/src/mercury_composable/server.py +++ b/src/mercury_composable/server.py @@ -24,13 +24,12 @@ import contextvars import time import traceback -from typing import Optional from aiohttp import web from .config import app_config from .envelope import EventEnvelope, iso_utc -from .exceptions import AppException, CompactFormatError +from .exceptions import AppException from .log import get_logger from .registry import FunctionRegistry, ServiceDef, default_registry from .trace import TraceInfo, _reset_trace, _set_trace @@ -50,7 +49,7 @@ def _transport_error(status: int, message: str) -> web.Response: return web.Response(status=status, body=reply.to_bytes(), content_type=OCTET_STREAM) -def _handler_headers(event: EventEnvelope) -> dict: +def _handler_headers(event: EventEnvelope) -> dict[str, str]: headers = {k: v for k, v in event.headers.items() if k.lower() != X_EVENT_API and not k.lower().startswith("my_")} my_cid = event.tags.get(MY_CID_TAG) @@ -60,9 +59,9 @@ def _handler_headers(event: EventEnvelope) -> dict: class EventApiServer: - def __init__(self, registry: Optional[FunctionRegistry] = None): + def __init__(self, registry: FunctionRegistry | None = None): self.registry = registry or default_registry - self._semaphores: dict = {} + self._semaphores: dict[str, asyncio.Semaphore] = {} def _semaphore(self, service: ServiceDef) -> asyncio.Semaphore: semaphore = self._semaphores.get(service.route) @@ -72,7 +71,7 @@ def _semaphore(self, service: ServiceDef) -> asyncio.Semaphore: return semaphore async def _invoke(self, service: ServiceDef, event: EventEnvelope, - headers: dict) -> EventEnvelope: + headers: dict[str, str]) -> EventEnvelope: """Run the handler under its trace context and shape the outcome as a reply.""" info = TraceInfo(trace_id=event.trace_id, trace_path=event.trace_path, cid=event.cid) token = _set_trace(info) @@ -83,15 +82,17 @@ async def _invoke(self, service: ServiceDef, event: EventEnvelope, result = await service.handler(headers, event.body) else: # copy_context() carries the trace contextvar into the executor thread - call = contextvars.copy_context().run + context = contextvars.copy_context() result = await asyncio.get_running_loop().run_in_executor( - None, call, service.handler, headers, event.body) + None, lambda: context.run(service.handler, headers, event.body)) reply = result if isinstance(result, EventEnvelope) else EventEnvelope(body=result) except AppException as e: reply = EventEnvelope().set_status(e.status).set_body(e.message) except asyncio.CancelledError: raise - except Exception as e: + except Exception as e: # noqa: BLE001 - the host converts ANY handler failure + # into the portable error contract (envelope status 500 + message + stack), + # mirroring the engines; letting it propagate would drop the reply reply = EventEnvelope().set_status(500).set_body(str(e)) reply.stack = traceback.format_exc(limit=20) finally: @@ -112,7 +113,8 @@ async def handle_event(self, request: web.Request) -> web.Response: is_async = request.headers.get(X_ASYNC, "") == "true" try: event = EventEnvelope.from_bytes(raw) - except (CompactFormatError, ValueError) as e: + # CompactFormatError is a ValueError - one catch covers the codec errors + except ValueError as e: return _transport_error(400, str(e)) if not event.to: return _transport_error(400, "Missing routing path") @@ -139,18 +141,25 @@ async def handle_event(self, request: web.Request) -> web.Response: event.to, reply.get_status(), reply.exec_time, event.trace_id) return web.Response(status=200, body=reply.to_bytes(), content_type=OCTET_STREAM) - def _log_async_outcome(self, route: str): - def callback(task: asyncio.Task) -> None: + @staticmethod + def _log_async_outcome(route: str): + def callback(task: asyncio.Task[EventEnvelope]) -> None: + # noinspection PyBroadException try: reply = task.result() if reply.has_error(): log.warning("Async event %s ended with status %d - %s", route, reply.get_status(), reply.body) - except Exception as e: - log.error("Async event %s failed - %s", route, e) + except Exception: + # deliberate log-only sink: a drop-n-forget event has no requester + # to answer, so any failure is logged with its traceback, never raised + log.exception("Async event %s failed", route) return callback - async def handle_health(self, _request: web.Request) -> web.Response: + # aiohttp handlers must be coroutines - async is the framework contract + # even though this one has nothing to await + @staticmethod + async def handle_health(_request: web.Request) -> web.Response: return web.Response(text="OK") def create_app(self) -> web.Application: @@ -163,10 +172,10 @@ def create_app(self) -> web.Application: class Platform: """Runs the Event API host for the default (or a given) registry.""" - def __init__(self, registry: Optional[FunctionRegistry] = None): + def __init__(self, registry: FunctionRegistry | None = None): self.registry = registry or default_registry - def run(self, port: Optional[int] = None, host: str = "127.0.0.1") -> None: + def run(self, port: int | None = None, host: str = "127.0.0.1") -> None: config = app_config() app_name = config.get_property("application.name", "application") actual_port = int(port if port is not None else config.get("rest.server.port", 8085)) diff --git a/src/mercury_composable/trace.py b/src/mercury_composable/trace.py index 71d2342..379ee25 100644 --- a/src/mercury_composable/trace.py +++ b/src/mercury_composable/trace.py @@ -11,28 +11,47 @@ from __future__ import annotations import contextvars +from collections.abc import Iterator +from contextlib import contextmanager from dataclasses import dataclass, field -from typing import Any, Dict, Optional +from typing import Any @dataclass class TraceInfo: - trace_id: Optional[str] = None - trace_path: Optional[str] = None - cid: Optional[str] = None - annotations: Dict[str, Any] = field(default_factory=dict) + trace_id: str | None = None + trace_path: str | None = None + cid: str | None = None + annotations: dict[str, Any] = field(default_factory=dict) -_current: contextvars.ContextVar[Optional[TraceInfo]] = contextvars.ContextVar( +_current: contextvars.ContextVar[TraceInfo | None] = contextvars.ContextVar( "mercury_trace", default=None ) -def get_trace() -> Optional[TraceInfo]: +def get_trace() -> TraceInfo | None: """The trace context of the event being handled, if any.""" return _current.get() +@contextmanager +def trace_context(trace_id: str, trace_path: str, + cid: str | None = None) -> Iterator[TraceInfo]: + """Establish a trace context around a block - the node runWithTrace twin. + + Useful for callers outside a hosted function (batch jobs, tests) whose + PostOffice calls should carry a trace: the client inherits the context + into the outbound envelope. + """ + info = TraceInfo(trace_id=trace_id, trace_path=trace_path, cid=cid) + token = _set_trace(info) + try: + yield info + finally: + _reset_trace(token) + + def annotate_trace(key: str, value: Any) -> None: """Attach an annotation to the current trace (returned on the reply envelope).""" info = _current.get() @@ -40,9 +59,9 @@ def annotate_trace(key: str, value: Any) -> None: info.annotations[str(key)] = value -def _set_trace(info: Optional[TraceInfo]) -> contextvars.Token: +def _set_trace(info: TraceInfo | None) -> contextvars.Token[TraceInfo | None]: return _current.set(info) -def _reset_trace(token: contextvars.Token) -> None: +def _reset_trace(token: contextvars.Token[TraceInfo | None]) -> None: _current.reset(token) diff --git a/tests/test_client.py b/tests/test_client.py index c38f00a..0ad7ed2 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,22 +1,25 @@ """PostOffice client tests against the in-process host (full wrapper loop).""" +from collections.abc import AsyncIterator + import pytest_asyncio from aiohttp import web -from mercury_composable import EventEnvelope, FunctionRegistry, PostOffice +from mercury_composable import Body, FunctionRegistry, PostOffice from mercury_composable.server import EventApiServer -from mercury_composable.trace import TraceInfo, _reset_trace, _set_trace +from mercury_composable.trace import trace_context def build_registry() -> FunctionRegistry: registry = FunctionRegistry() - async def echo(headers, body): + async def echo(headers: dict[str, str], body: Body): return {"headers": headers, "body": body} - async def whoami(headers, body): + async def whoami(_headers: dict[str, str], _body: Body): from mercury_composable import get_trace info = get_trace() + assert info is not None return {"trace_id": info.trace_id, "trace_path": info.trace_path, "cid": info.cid} registry.register("client.echo", echo) @@ -25,18 +28,18 @@ async def whoami(headers, body): @pytest_asyncio.fixture -async def endpoint(): +async def endpoint() -> AsyncIterator[str]: server = EventApiServer(build_registry()) runner = web.AppRunner(server.create_app()) await runner.setup() site = web.TCPSite(runner, "127.0.0.1", 0) await site.start() - port = site._server.sockets[0].getsockname()[1] + port = runner.addresses[0][1] yield f"http://127.0.0.1:{port}/api/event" await runner.cleanup() -async def test_rpc_round_trip(endpoint): +async def test_rpc_round_trip(endpoint: str): async with PostOffice(endpoint=endpoint) as po: reply = await po.request("client.echo", body={"hello": "world"}, headers={"h1": "v1"}, timeout_ms=5000) @@ -45,24 +48,21 @@ async def test_rpc_round_trip(endpoint): assert reply.body["headers"]["h1"] == "v1" -async def test_trace_context_propagates_through_client(endpoint): - token = _set_trace(TraceInfo(trace_id="trace-777", trace_path="TEST /client", cid="cid-42")) - try: +async def test_trace_context_propagates_through_client(endpoint: str): + with trace_context("trace-777", "TEST /client", cid="cid-42"): async with PostOffice(endpoint=endpoint) as po: reply = await po.request("client.whoami", body={}, timeout_ms=5000) - finally: - _reset_trace(token) assert reply.body == {"trace_id": "trace-777", "trace_path": "TEST /client", "cid": "cid-42"} -async def test_error_reply_is_returned_not_raised(endpoint): +async def test_error_reply_is_returned_not_raised(endpoint: str): async with PostOffice(endpoint=endpoint) as po: reply = await po.request("no.such.route", body={}, timeout_ms=5000) assert reply.get_status() == 404 assert reply.body == "Route no.such.route not found" -async def test_drop_n_forget_ack(endpoint): +async def test_drop_n_forget_ack(endpoint: str): async with PostOffice(endpoint=endpoint) as po: ack = await po.send("client.echo", body={"fire": "forget"}, timeout_ms=5000) assert ack.get_status() == 202 diff --git a/tests/test_config.py b/tests/test_config.py index df271fc..6ded039 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,17 +1,20 @@ """AppConfig tests: resources/ convention, -D overrides, ${ENV:default} substitution.""" import os +from pathlib import Path + +import pytest from mercury_composable import AppConfig -def write(path, text): +def write(path: str, text: str) -> None: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write(text) -def test_resources_location_convention(tmp_path, monkeypatch): +def test_resources_location_convention(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): write(str(tmp_path / "resources" / "application.yml"), "application:\n name: 'demo-app'\nrest:\n server:\n port: 8086\n") monkeypatch.chdir(tmp_path) @@ -21,7 +24,7 @@ def test_resources_location_convention(tmp_path, monkeypatch): assert config.get("rest.server.port") == 8086 -def test_properties_format(tmp_path): +def test_properties_format(tmp_path: Path): path = str(tmp_path / "application.properties") write(path, "# comment\nrest.server.port=8087\napplication.name=props-app\n") config = AppConfig(path=path, argv=[]) @@ -29,7 +32,7 @@ def test_properties_format(tmp_path): assert config.get("application.name") == "props-app" -def test_d_argument_overrides_win(tmp_path): +def test_d_argument_overrides_win(tmp_path: Path): path = str(tmp_path / "application.yml") write(path, "rest:\n server:\n port: 8086\n") config = AppConfig(path=path, argv=["-Drest.server.port=9999", "-Dnew.key=live"]) @@ -37,7 +40,7 @@ def test_d_argument_overrides_win(tmp_path): assert config.get("new.key") == "live" -def test_set_is_runtime_override(tmp_path): +def test_set_is_runtime_override(tmp_path: Path): path = str(tmp_path / "application.yml") write(path, "some:\n key: 'original'\n") config = AppConfig(path=path, argv=[]) @@ -46,7 +49,7 @@ def test_set_is_runtime_override(tmp_path): assert config.get("some.key") == "changed" -def test_env_substitution_with_default(tmp_path, monkeypatch): +def test_env_substitution_with_default(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): path = str(tmp_path / "application.yml") write(path, "peer:\n url: 'http://127.0.0.1:${PEER_PORT:8085}/api/event'\n" "missing: '${NOT_SET_ANYWHERE}'\n") diff --git a/tests/test_envelope.py b/tests/test_envelope.py index 3cda42e..cdcf963 100644 --- a/tests/test_envelope.py +++ b/tests/test_envelope.py @@ -2,15 +2,22 @@ import base64 import json -import os from datetime import datetime, timezone +from pathlib import Path import msgpack import pytest + +def packb(obj: object) -> bytes: + """msgpack.packb is typed bytes | None (None only in stream mode).""" + data = msgpack.packb(obj, use_bin_type=True) + assert data is not None + return data + from mercury_composable import CompactFormatError, EventEnvelope, iso_utc -VECTORS = os.path.join(os.path.dirname(__file__), "vectors", "vectors.json") +VECTORS = Path(__file__).parent / "vectors" / "vectors.json" def test_round_trip_all_fields(): @@ -51,20 +58,20 @@ def test_unset_fields_are_omitted_and_headers_id_always_present(): def test_absent_and_nil_are_equivalent(): - explicit_nil = msgpack.packb({"id": "x1", "headers": {}, "body": None}, use_bin_type=True) + explicit_nil = packb({"id": "x1", "headers": {}, "body": None}) decoded = EventEnvelope.from_bytes(explicit_nil) assert decoded.body is None assert decoded.get_status() == 200 # default when unset def test_unknown_keys_are_ignored(): - payload = msgpack.packb({"id": "x2", "headers": {}, "future_field": 42}, use_bin_type=True) + payload = packb({"id": "x2", "headers": {}, "future_field": 42}) decoded = EventEnvelope.from_bytes(payload) assert decoded.id == "x2" def test_compact_format_detected_and_rejected(): - compact = msgpack.packb({"0": "e1", "T": "hello.world"}, use_bin_type=True) + compact = packb({"0": "e1", "T": "hello.world"}) with pytest.raises(CompactFormatError): EventEnvelope.from_bytes(compact) @@ -89,11 +96,12 @@ def _decoded_wire_fields(envelope: EventEnvelope) -> dict: def test_golden_vectors_conformance(): - with open(VECTORS, "r", encoding="utf-8") as f: + with VECTORS.open("r", encoding="utf-8") as f: catalog = json.load(f) standard = [v for v in catalog["vectors"] if v["format"] == "standard"] compact = [v for v in catalog["vectors"] if v["format"] == "compact"] - assert standard and compact + assert standard + assert compact for vector in standard: raw = base64.b64decode(vector["base64"]) decoded = EventEnvelope.from_bytes(raw) @@ -105,5 +113,6 @@ def test_golden_vectors_conformance(): for key, expected in vector["expect"].items(): assert again.get(key) == expected, f"{vector['name']} re-encoded: field '{key}'" for vector in compact: + raw = base64.b64decode(vector["base64"]) with pytest.raises(CompactFormatError): - EventEnvelope.from_bytes(base64.b64decode(vector["base64"])) + EventEnvelope.from_bytes(raw) diff --git a/tests/test_log.py b/tests/test_log.py new file mode 100644 index 0000000..abb3e55 --- /dev/null +++ b/tests/test_log.py @@ -0,0 +1,35 @@ +"""Formatter tests: engine pattern and exception rendering.""" + +import logging +import sys + +from mercury_composable.log import EngineJsonFormatter, EngineTextFormatter + + +def _record_with_exception() -> logging.LogRecord: + try: + raise RuntimeError("boom-x") + except RuntimeError: + return logging.LogRecord( + name="unit.test", level=logging.ERROR, pathname=__file__, + lineno=42, msg="Async event %s failed", args=("demo.route",), + exc_info=sys.exc_info()) + + +def test_text_formatter_engine_pattern_and_traceback(): + line = EngineTextFormatter().format(_record_with_exception()) + first = line.splitlines()[0] + # %d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger:%line - %msg + assert " ERROR unit.test:42 - Async event demo.route failed" in first + assert "Traceback" in line + assert "RuntimeError: boom-x" in line + + +def test_json_formatter_carries_exception(): + import json + + entry = json.loads(EngineJsonFormatter().format(_record_with_exception())) + assert entry["level"] == "ERROR" + assert entry["logger"] == "unit.test:42" + assert entry["message"] == "Async event demo.route failed" + assert "RuntimeError: boom-x" in entry["exception"] diff --git a/tests/test_server.py b/tests/test_server.py index e7b6b5b..23bd389 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,15 +1,21 @@ """Event API host tests: engine-mirrored semantics over real HTTP.""" import asyncio +from collections.abc import AsyncIterator import aiohttp import msgpack -import pytest import pytest_asyncio from aiohttp import web -from mercury_composable import (AppException, EventEnvelope, FunctionRegistry, - annotate_trace, get_trace) +from mercury_composable import ( + AppException, + Body, + EventEnvelope, + FunctionRegistry, + annotate_trace, + get_trace, +) from mercury_composable.server import EventApiServer OCTET = "application/octet-stream" @@ -18,30 +24,31 @@ def build_registry() -> FunctionRegistry: registry = FunctionRegistry() - async def echo(headers, body): + async def echo(headers: dict[str, str], body: Body): return {"headers": headers, "body": body} - def upper(headers, body): # synchronous handler runs in the executor + def upper(_headers: dict[str, str], body: Body): # sync handler runs in the executor info = get_trace() - return {"text": str(body.get("text", "")).upper(), + text = body.get("text", "") if isinstance(body, dict) else "" + return {"text": str(text).upper(), "trace_id": info.trace_id if info else None, "cid": info.cid if info else None} - async def annotated(headers, body): + async def annotated(_headers: dict[str, str], _body: Body): annotate_trace("checked", "yes") return {"ok": True} - async def app_error(headers, body): + async def app_error(_headers: dict[str, str], _body: Body): raise AppException(400, "missing 'text'") - async def boom(headers, body): + async def boom(_headers: dict[str, str], _body: Body): raise RuntimeError("kaboom") - async def slow(headers, body): + async def slow(_headers: dict[str, str], _body: Body): await asyncio.sleep(5) return {"late": True} - async def secret(headers, body): + async def secret(_headers: dict[str, str], _body: Body): return {"secret": True} registry.register("unit.echo", echo) @@ -55,36 +62,39 @@ async def secret(headers, body): @pytest_asyncio.fixture -async def server_url(aiohttp_server=None): +async def server_url() -> AsyncIterator[str]: server = EventApiServer(build_registry()) runner = web.AppRunner(server.create_app()) await runner.setup() site = web.TCPSite(runner, "127.0.0.1", 0) await site.start() - port = site._server.sockets[0].getsockname()[1] + port = runner.addresses[0][1] yield f"http://127.0.0.1:{port}" await runner.cleanup() -async def post_event(url: str, event: EventEnvelope, *, ttl="10000", extra=None): +async def post_event(url: str, event: EventEnvelope, *, ttl: str = "10000", + extra: dict[str, str] | None = None) -> tuple[int, EventEnvelope]: headers = {"content-type": OCTET, "x-ttl": ttl} headers.update(extra or {}) - async with aiohttp.ClientSession() as session: - async with session.post(f"{url}/api/event", data=event.to_bytes(), - headers=headers) as response: - return response.status, EventEnvelope.from_bytes(await response.read()) + async with ( + aiohttp.ClientSession() as session, + session.post(f"{url}/api/event", data=event.to_bytes(), headers=headers) as response, + ): + return response.status, EventEnvelope.from_bytes(await response.read()) -async def test_rpc_success_with_exec_time(server_url): +async def test_rpc_success_with_exec_time(server_url: str): status, reply = await post_event(server_url, EventEnvelope(to="unit.echo", body={"a": 1})) assert status == 200 assert reply.get_status() == 200 assert reply.body["body"] == {"a": 1} assert reply.sender == "unit.echo" - assert reply.exec_time is not None and reply.exec_time >= 0 + assert reply.exec_time is not None + assert reply.exec_time >= 0 -async def test_sync_handler_sees_trace_context(server_url): +async def test_sync_handler_sees_trace_context(server_url: str): event = EventEnvelope(to="unit.upper", body={"text": "hello"}) event.set_trace("trace-100", "TEST /upper").set_correlation_id("cid-9") status, reply = await post_event(server_url, event) @@ -92,7 +102,7 @@ async def test_sync_handler_sees_trace_context(server_url): assert reply.body == {"text": "HELLO", "trace_id": "trace-100", "cid": "cid-9"} -async def test_reserved_header_hygiene_and_my_cid_injection(server_url): +async def test_reserved_header_hygiene_and_my_cid_injection(server_url: str): event = EventEnvelope(to="unit.echo", body={}) event.set_header("x-event-api", "callback").set_header("my_secret", "x") event.set_header("content-type", "application/json") @@ -105,12 +115,12 @@ async def test_reserved_header_hygiene_and_my_cid_injection(server_url): assert delivered["content-type"] == "application/json" -async def test_annotations_ride_the_reply(server_url): +async def test_annotations_ride_the_reply(server_url: str): _, reply = await post_event(server_url, EventEnvelope(to="unit.annotated", body={})) assert reply.annotations == {"checked": "yes"} -async def test_app_exception_is_portable_error_on_http_200(server_url): +async def test_app_exception_is_portable_error_on_http_200(server_url: str): status, reply = await post_event(server_url, EventEnvelope(to="unit.app.error", body={})) assert status == 200 # handler-level errors ride HTTP 200, engine-style assert reply.get_status() == 400 @@ -118,45 +128,48 @@ async def test_app_exception_is_portable_error_on_http_200(server_url): assert reply.stack is None -async def test_unexpected_exception_maps_to_500_with_stack(server_url): +async def test_unexpected_exception_maps_to_500_with_stack(server_url: str): status, reply = await post_event(server_url, EventEnvelope(to="unit.boom", body={})) assert status == 200 assert reply.get_status() == 500 assert reply.body == "kaboom" + assert reply.stack is not None assert "RuntimeError" in reply.stack -async def test_unknown_route_404(server_url): +async def test_unknown_route_404(server_url: str): status, reply = await post_event(server_url, EventEnvelope(to="no.where", body={})) assert status == 404 assert reply.get_status() == 404 assert reply.body == "Route no.where not found" -async def test_private_route_403(server_url): +async def test_private_route_403(server_url: str): status, reply = await post_event(server_url, EventEnvelope(to="unit.secret", body={})) assert status == 403 assert reply.body == "unit.secret is private" -async def test_missing_routing_path_400(server_url): +async def test_missing_routing_path_400(server_url: str): status, reply = await post_event(server_url, EventEnvelope(body={"x": 1})) assert status == 400 assert reply.body == "Missing routing path" -async def test_compact_request_rejected_400(server_url): +async def test_compact_request_rejected_400(server_url: str): compact = msgpack.packb({"0": "e1", "T": "unit.echo"}, use_bin_type=True) - async with aiohttp.ClientSession() as session: - async with session.post(f"{server_url}/api/event", data=compact, - headers={"content-type": OCTET, "x-ttl": "5000"}) as response: - assert response.status == 400 - reply = EventEnvelope.from_bytes(await response.read()) + async with ( + aiohttp.ClientSession() as session, + session.post(f"{server_url}/api/event", data=compact, + headers={"content-type": OCTET, "x-ttl": "5000"}) as response, + ): + assert response.status == 400 + reply = EventEnvelope.from_bytes(await response.read()) assert reply.get_status() == 400 assert "standard" in str(reply.body) -async def test_timeout_408_mirrors_engine_message(server_url): +async def test_timeout_408_mirrors_engine_message(server_url: str): status, reply = await post_event(server_url, EventEnvelope(to="unit.slow", body={}), ttl="1000") assert status == 408 @@ -164,7 +177,7 @@ async def test_timeout_408_mirrors_engine_message(server_url): assert reply.body == "Timeout for 1000 ms" -async def test_async_drop_n_forget_202_ack(server_url): +async def test_async_drop_n_forget_202_ack(server_url: str): status, reply = await post_event(server_url, EventEnvelope(to="unit.echo", body={}), extra={"x-async": "true"}) assert status == 202 @@ -174,8 +187,10 @@ async def test_async_drop_n_forget_202_ack(server_url): assert "time" in reply.body -async def test_health_endpoint(server_url): - async with aiohttp.ClientSession() as session: - async with session.get(f"{server_url}/health") as response: - assert response.status == 200 - assert await response.text() == "OK" +async def test_health_endpoint(server_url: str): + async with ( + aiohttp.ClientSession() as session, + session.get(f"{server_url}/health") as response, + ): + assert response.status == 200 + assert await response.text() == "OK"