fix(events): eliminate inline type ignores in event_bus_steps.py
CI / lint (pull_request) Successful in 42s
CI / build (pull_request) Successful in 39s
CI / quality (pull_request) Successful in 1m22s
CI / typecheck (pull_request) Successful in 1m30s
CI / security (pull_request) Successful in 2m0s
CI / helm (pull_request) Successful in 24s
CI / push-validation (pull_request) Successful in 26s
CI / integration_tests (pull_request) Successful in 4m44s
CI / unit_tests (pull_request) Failing after 5m26s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 3s

Replace all # type: ignore[arg-type] and # type: ignore[misc] comments
with pyright-compatible alternatives: typing.cast() for intentional
type mismatches in error-handling tests, and explicit stub method bodies
for Protocol subclasses.
This commit is contained in:
2026-05-15 11:09:43 +00:00
committed by Forgejo
parent 4dfd183f66
commit 03cb038b53
+29 -10
View File
@@ -6,7 +6,8 @@ LoggingEventBus, service event emission, and DI container registration.
from __future__ import annotations
from typing import Any
from collections.abc import Callable
from typing import Any, cast
from unittest.mock import MagicMock
from behave import given, then, when # type: ignore[import-untyped]
@@ -256,7 +257,7 @@ def step_bus_has_stream(ctx: Context) -> None:
def step_emit_non_domain_event(ctx: Context) -> None:
raised = False
try:
ctx.bus.emit("not-an-event") # type: ignore[arg-type]
ctx.bus.emit(cast(DomainEvent, "not-an-event"))
except TypeError:
raised = True
assert raised, "Expected TypeError for non-DomainEvent"
@@ -266,7 +267,7 @@ def step_emit_non_domain_event(ctx: Context) -> None:
def step_subscribe_bad_type(ctx: Context) -> None:
raised = False
try:
ctx.bus.subscribe("plan.created", lambda e: None) # type: ignore[arg-type]
ctx.bus.subscribe(cast(EventType, "plan.created"), lambda e: None)
except TypeError:
raised = True
assert raised, "Expected TypeError for non-EventType event_type"
@@ -276,7 +277,9 @@ def step_subscribe_bad_type(ctx: Context) -> None:
def step_subscribe_non_callable(ctx: Context) -> None:
raised = False
try:
ctx.bus.subscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type]
ctx.bus.subscribe(
EventType.PLAN_CREATED, cast(Callable[[DomainEvent], None], "not-callable")
)
except TypeError:
raised = True
assert raised, "Expected TypeError for non-callable handler"
@@ -304,7 +307,7 @@ def step_unsubscribe_nonexistent_returns_false(ctx: Context) -> None:
def step_unsubscribe_bad_type(ctx: Context) -> None:
raised = False
try:
ctx.bus.unsubscribe("plan.created", lambda e: None) # type: ignore[arg-type]
ctx.bus.unsubscribe(cast(EventType, "plan.created"), lambda e: None)
except TypeError:
raised = True
assert raised, "Expected TypeError for non-EventType event_type in unsubscribe"
@@ -314,7 +317,9 @@ def step_unsubscribe_bad_type(ctx: Context) -> None:
def step_unsubscribe_non_callable(ctx: Context) -> None:
raised = False
try:
ctx.bus.unsubscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type]
ctx.bus.unsubscribe(
EventType.PLAN_CREATED, cast(Callable[[DomainEvent], None], "not-callable")
)
except TypeError:
raised = True
assert raised, "Expected TypeError for non-callable handler in unsubscribe"
@@ -324,8 +329,15 @@ def step_unsubscribe_non_callable(ctx: Context) -> None:
def step_protocol_stubs_callable(ctx: Context) -> None:
"""Exercise Protocol method stubs directly to ensure 100% coverage."""
class _BareImpl(EventBus): # type: ignore[misc]
pass
class _BareImpl(EventBus):
def emit(self, event: DomainEvent) -> None: ...
def subscribe(
self, event_type: EventType, handler: Callable[[DomainEvent], None]
) -> None: ...
def unsubscribe(
self, event_type: EventType, handler: Callable[[DomainEvent], None]
) -> bool:
return True
obj = _BareImpl()
event = _make_event("plan.created")
@@ -337,8 +349,15 @@ def step_protocol_stubs_callable(ctx: Context) -> None:
def step_protocol_unsubscribe_callable(ctx: Context) -> None:
"""Exercise Protocol unsubscribe stub directly to ensure coverage."""
class _BareImpl(EventBus): # type: ignore[misc]
pass
class _BareImpl(EventBus):
def emit(self, event: DomainEvent) -> None: ...
def subscribe(
self, event_type: EventType, handler: Callable[[DomainEvent], None]
) -> None: ...
def unsubscribe(
self, event_type: EventType, handler: Callable[[DomainEvent], None]
) -> bool:
return True
obj = _BareImpl()
obj.unsubscribe(EventType.PLAN_CREATED, lambda e: None)