"""Step definitions for concurrency lock scenarios.""" from __future__ import annotations from datetime import UTC, datetime, timedelta from typing import Any from behave import given, then, when from behave.runner import Context from sqlalchemy import create_engine, event from sqlalchemy.orm import Session, sessionmaker from cleveragents.application.services.lock_service import LockService from cleveragents.core.exceptions import ( LockConflictError, LockExpiredError, ValidationError, ) from cleveragents.infrastructure.database.models import Base, LockModel def _build_lock_service(context: Context) -> LockService: """Build a LockService with an in-memory SQLite backend.""" engine = create_engine( "sqlite:///:memory:", echo=False, future=True, connect_args={"check_same_thread": False}, ) @event.listens_for(engine, "connect") def _fk(dbapi_conn: Any, _rec: Any) -> None: cursor = dbapi_conn.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close() Base.metadata.create_all(engine) factory: sessionmaker[Session] = sessionmaker( bind=engine, expire_on_commit=False, autoflush=False, autocommit=False, class_=Session, ) context.lock_engine = engine context.lock_session_factory = factory return LockService(session_factory=factory) # --- Background --- @given("I have a lock service backed by an in-memory database") def step_create_lock_service(context: Context) -> None: """Create a LockService backed by an in-memory SQLite database.""" context.lock_service = _build_lock_service(context) context.error = None context.result = None # --- Acquisition --- @when('I acquire a lock on plan "{plan_id}" as owner "{owner}"') def step_acquire_plan_lock(context: Context, plan_id: str, owner: str) -> None: """Acquire a plan-level lock.""" context.result = context.lock_service.acquire( owner_id=owner, resource_type="plan", resource_id=plan_id, ) @when('I acquire a lock on project "{project_id}" as owner "{owner}"') def step_acquire_project_lock(context: Context, project_id: str, owner: str) -> None: """Acquire a project-level lock.""" context.result = context.lock_service.acquire( owner_id=owner, resource_type="project", resource_id=project_id, ) @then("the lock should be acquired successfully") def step_lock_acquired(context: Context) -> None: """Assert the lock was acquired.""" assert context.result is True, f"Expected True, got {context.result}" # --- Re-entrant / fixtures --- @given('owner "{owner}" holds a lock on plan "{plan_id}"') def step_owner_holds_plan_lock(context: Context, owner: str, plan_id: str) -> None: """Ensure the specified owner holds a plan lock.""" context.lock_service.acquire( owner_id=owner, resource_type="plan", resource_id=plan_id, ) @given('owner "{owner}" holds locks on plans "{p1}" and "{p2}" and project "{proj}"') def step_owner_holds_multiple( context: Context, owner: str, p1: str, p2: str, proj: str ) -> None: """Ensure the owner holds multiple locks.""" svc: LockService = context.lock_service svc.acquire(owner_id=owner, resource_type="plan", resource_id=p1) svc.acquire(owner_id=owner, resource_type="plan", resource_id=p2) svc.acquire(owner_id=owner, resource_type="project", resource_id=proj) # --- Conflict --- @when('I try to acquire a lock on plan "{plan_id}" as owner "{owner}"') def step_try_acquire_plan_lock(context: Context, plan_id: str, owner: str) -> None: """Try to acquire a plan lock, capturing any error.""" try: context.result = context.lock_service.acquire( owner_id=owner, resource_type="plan", resource_id=plan_id, ) except LockConflictError as exc: context.error = exc @then('a lock conflict error should be raised for owner "{expected_owner}"') def step_lock_conflict_error(context: Context, expected_owner: str) -> None: """Assert a LockConflictError was raised.""" assert isinstance(context.error, LockConflictError), ( f"Expected LockConflictError, got {type(context.error)}" ) assert context.error.owner_id == expected_owner # --- Release --- @when('I release the lock on plan "{plan_id}" as owner "{owner}"') def step_release_plan_lock(context: Context, plan_id: str, owner: str) -> None: """Release a plan lock.""" context.result = context.lock_service.release( owner_id=owner, resource_type="plan", resource_id=plan_id, ) @then("the lock should be released successfully") def step_lock_released(context: Context) -> None: """Assert the lock was released.""" assert context.result is True, f"Expected True, got {context.result}" @then("the release result should be false") def step_release_false(context: Context) -> None: """Assert the release returned False.""" assert context.result is False, f"Expected False, got {context.result}" # --- Renewal --- @when('I renew the lock on plan "{plan_id}" as owner "{owner}"') def step_renew_lock(context: Context, plan_id: str, owner: str) -> None: """Renew a plan lock.""" context.result = context.lock_service.renew( owner_id=owner, resource_type="plan", resource_id=plan_id, ) @then("the lock renewal should succeed") def step_renewal_success(context: Context) -> None: """Assert the renewal succeeded.""" assert context.result is True, f"Expected True, got {context.result}" # --- Expired lock fixtures --- @given('owner "{owner}" held an expired lock on plan "{plan_id}"') def step_create_expired_lock(context: Context, owner: str, plan_id: str) -> None: """Insert an already-expired lock directly into the database.""" now = datetime.now(tz=UTC) expired_at = (now - timedelta(seconds=60)).isoformat() acquired_at = (now - timedelta(seconds=120)).isoformat() session: Session = context.lock_session_factory() lock = LockModel( owner_id=owner, resource_type="plan", resource_id=plan_id, acquired_at=acquired_at, expires_at=expired_at, ) session.add(lock) session.commit() session.close() @when('I try to renew the expired lock on plan "{plan_id}" as owner "{owner}"') def step_try_renew_expired(context: Context, plan_id: str, owner: str) -> None: """Attempt to renew an expired lock.""" try: context.result = context.lock_service.renew( owner_id=owner, resource_type="plan", resource_id=plan_id, ) except LockExpiredError as exc: context.error = exc @then("a lock expired error should be raised") def step_lock_expired_error(context: Context) -> None: """Assert a LockExpiredError was raised.""" assert isinstance(context.error, LockExpiredError), ( f"Expected LockExpiredError, got {type(context.error)}" ) # --- Cleanup --- @given("there are {count:d} expired locks in the database") def step_insert_expired_locks(context: Context, count: int) -> None: """Insert multiple expired locks into the database.""" now = datetime.now(tz=UTC) expired_at = (now - timedelta(seconds=60)).isoformat() acquired_at = (now - timedelta(seconds=120)).isoformat() session: Session = context.lock_session_factory() for i in range(count): session.add( LockModel( owner_id=f"expired-owner-{i}", resource_type="plan", resource_id=f"expired-plan-{i}", acquired_at=acquired_at, expires_at=expired_at, ) ) session.commit() session.close() @when("I run cleanup expired locks") def step_cleanup_expired(context: Context) -> None: """Run cleanup of expired locks.""" context.result = context.lock_service.cleanup_expired() @then("the expired lock count should be {count:d}") def step_expired_count(context: Context, count: int) -> None: """Assert the cleanup purged the expected count.""" assert context.result == count, f"Expected {count}, got {context.result}" # --- Graceful shutdown --- @when('I release all locks for owner "{owner}"') def step_release_all(context: Context, owner: str) -> None: """Release all locks for an owner.""" context.result = context.lock_service.release_all_for_owner(owner_id=owner) @then("the released count should be {count:d}") def step_released_count(context: Context, count: int) -> None: """Assert the release-all count.""" assert context.result == count, f"Expected {count}, got {context.result}" # --- Diagnostics --- @when("I count stale locks") def step_count_stale(context: Context) -> None: """Count stale locks.""" context.result = context.lock_service.count_stale_locks() @then("the stale lock count should be {count:d}") def step_stale_count(context: Context, count: int) -> None: """Assert the stale lock count.""" assert context.result == count, f"Expected {count}, got {context.result}" # --- is_locked --- @when('I check if plan "{plan_id}" is locked') def step_check_is_locked(context: Context, plan_id: str) -> None: """Check if a plan resource is locked.""" context.result = context.lock_service.is_locked( resource_type="plan", resource_id=plan_id, ) @then("the resource should be locked") def step_resource_locked(context: Context) -> None: """Assert the resource is locked.""" assert context.result is True, f"Expected True, got {context.result}" @then("the resource should not be locked") def step_resource_not_locked(context: Context) -> None: """Assert the resource is not locked.""" assert context.result is False, f"Expected False, got {context.result}" # --- Validation errors --- @when("I try to acquire a lock with empty owner_id") def step_try_empty_owner(context: Context) -> None: """Try to acquire with empty owner_id.""" try: context.lock_service.acquire( owner_id="", resource_type="plan", resource_id="plan-001", ) except ValidationError as exc: context.error = exc @when('I try to acquire a lock with invalid resource type "{rtype}"') def step_try_invalid_resource_type(context: Context, rtype: str) -> None: """Try to acquire with an invalid resource type.""" try: context.lock_service.acquire( owner_id="session-a", resource_type=rtype, resource_id="res-001", ) except ValidationError as exc: context.error = exc @when("I try to acquire a lock with empty resource_id") def step_try_empty_resource_id(context: Context) -> None: """Try to acquire with empty resource_id.""" try: context.lock_service.acquire( owner_id="session-a", resource_type="plan", resource_id="", ) except ValidationError as exc: context.error = exc @when("I try to acquire a lock with TTL {ttl:d}") def step_try_bad_ttl(context: Context, ttl: int) -> None: """Try to acquire with out-of-range TTL.""" try: context.lock_service.acquire( owner_id="session-a", resource_type="plan", resource_id="plan-001", ttl_seconds=ttl, ) except ValidationError as exc: context.error = exc @when("I try to create a lock service with None session factory") def step_try_none_factory(context: Context) -> None: """Try to create a LockService with None.""" try: LockService(session_factory=None) # type: ignore[arg-type] except ValidationError as exc: context.error = exc @when("I try to acquire a lock with empty resource type") def step_try_empty_resource_type(context: Context) -> None: """Try to acquire with empty resource_type.""" try: context.lock_service.acquire( owner_id="session-a", resource_type="", resource_id="plan-001", ) except ValidationError as exc: context.error = exc @when("I try to release a lock with empty owner_id") def step_try_release_empty_owner(context: Context) -> None: """Try to release with empty owner_id.""" try: context.lock_service.release( owner_id="", resource_type="plan", resource_id="plan-001", ) except ValidationError as exc: context.error = exc @when("I try to renew a lock with empty owner_id") def step_try_renew_empty_owner(context: Context) -> None: """Try to renew with empty owner_id.""" try: context.lock_service.renew( owner_id="", resource_type="plan", resource_id="plan-001", ) except ValidationError as exc: context.error = exc @when("I try to release all locks with empty owner_id") def step_try_release_all_empty_owner(context: Context) -> None: """Try to release_all_for_owner with empty owner_id.""" try: context.lock_service.release_all_for_owner(owner_id="") except ValidationError as exc: context.error = exc @when("I try to renew a lock that does not exist") def step_try_renew_nonexistent(context: Context) -> None: """Try to renew a lock that doesn't exist.""" context.result = context.lock_service.renew( owner_id="no-one", resource_type="plan", resource_id="no-such-lock", ) @then("the renew result should be false") def step_renew_false(context: Context) -> None: """Assert renew returned False.""" assert context.result is False, f"Expected False, got {context.result}" @then("a lock validation error should be raised") def step_lock_validation_error(context: Context) -> None: """Assert a ValidationError was raised for lock operations.""" assert isinstance(context.error, ValidationError), ( f"Expected ValidationError, got {type(context.error)}" )