freemo
08868e1e55
Add Behave BDD scenarios covering: - ServerLifecycle.start() with mocked uvicorn (full lifecycle) - run_server() convenience function with Settings defaults and overrides - Signal handler installation and non-main-thread edge case - TeamCollaborationService validation edge cases (empty args) - SessionRegistry validation edge cases and get_active_sessions - VersionConflictError negative version validation - Untracked resource version stamp creation path These tests cover previously uncovered validation branches and the server lifecycle start/shutdown paths to maintain >=97% coverage.
1040 lines
36 KiB
Python
1040 lines
36 KiB
Python
"""Step definitions for team collaboration feature tests.
|
|
|
|
Covers team roles, permissions, user sessions, session registry,
|
|
version stamps, conflict resolution, and the TeamCollaborationService.
|
|
|
|
Based on issue #863 — feat(server): team collaboration features.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from time import sleep
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.application.services.team_collab_service import (
|
|
TeamCollaborationService,
|
|
)
|
|
from cleveragents.domain.models.core.team_collab import (
|
|
TEAM_ROLE_PERMISSIONS,
|
|
ConflictDetectionResult,
|
|
ConflictResolutionStrategy,
|
|
SessionRegistry,
|
|
TeamCollaborationError,
|
|
TeamMember,
|
|
TeamMemberExistsError,
|
|
TeamMemberNotFoundError,
|
|
TeamPermission,
|
|
TeamRole,
|
|
UserSession,
|
|
VersionConflictError,
|
|
VersionStamp,
|
|
)
|
|
|
|
ULID_RE = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$")
|
|
|
|
|
|
# ── TeamRole enum ──────────────────────────────────────────────────
|
|
|
|
|
|
@given("the TeamRole enum")
|
|
def step_given_team_role_enum(context: Context) -> None:
|
|
context.enum_cls = TeamRole
|
|
|
|
|
|
@then('the TeamRole enum should contain "{value}"')
|
|
def step_then_team_role_contains(context: Context, value: str) -> None:
|
|
members = [m.value for m in TeamRole]
|
|
assert value in members, f"{value} not in TeamRole: {members}"
|
|
|
|
|
|
# ── TeamPermission enum ────────────────────────────────────────────
|
|
|
|
|
|
@given("the TeamPermission enum")
|
|
def step_given_team_permission_enum(context: Context) -> None:
|
|
context.enum_cls = TeamPermission
|
|
|
|
|
|
@then('the TeamPermission enum should contain "{value}"')
|
|
def step_then_team_permission_contains(context: Context, value: str) -> None:
|
|
members = [m.value for m in TeamPermission]
|
|
assert value in members, f"{value} not in TeamPermission: {members}"
|
|
|
|
|
|
# ── ConflictResolutionStrategy enum ────────────────────────────────
|
|
|
|
|
|
@given("the ConflictResolutionStrategy enum")
|
|
def step_given_conflict_strategy_enum(context: Context) -> None:
|
|
context.enum_cls = ConflictResolutionStrategy
|
|
|
|
|
|
@then('the ConflictResolutionStrategy enum should contain "{value}"')
|
|
def step_then_conflict_strategy_contains(context: Context, value: str) -> None:
|
|
members = [m.value for m in ConflictResolutionStrategy]
|
|
assert value in members, f"{value} not in ConflictResolutionStrategy: {members}"
|
|
|
|
|
|
# ── TEAM_ROLE_PERMISSIONS constant ─────────────────────────────────
|
|
|
|
|
|
@given("the TEAM_ROLE_PERMISSIONS constant")
|
|
def step_given_team_role_permissions(context: Context) -> None:
|
|
context.role_permissions = TEAM_ROLE_PERMISSIONS
|
|
|
|
|
|
@then('the "{role}" team role should have {count:d} allowed permissions')
|
|
def step_then_team_role_has_n_permissions(
|
|
context: Context, role: str, count: int
|
|
) -> None:
|
|
team_role = TeamRole(role)
|
|
perms = TEAM_ROLE_PERMISSIONS[team_role]
|
|
assert len(perms) == count, f"{role} has {len(perms)} permissions, expected {count}"
|
|
|
|
|
|
# ── TeamMember model ───────────────────────────────────────────────
|
|
|
|
|
|
@given('a team member "{user_id}" with role "{role}" and email "{email}"')
|
|
def step_given_team_member(
|
|
context: Context, user_id: str, role: str, email: str
|
|
) -> None:
|
|
context.team_member = TeamMember(
|
|
user_id=user_id,
|
|
display_name=user_id.title(),
|
|
email=email,
|
|
role=TeamRole(role),
|
|
)
|
|
|
|
|
|
@then('the team member user_id should be "{expected}"')
|
|
def step_then_member_user_id(context: Context, expected: str) -> None:
|
|
assert context.team_member.user_id == expected
|
|
|
|
|
|
@then('the team member role should be "{expected}"')
|
|
def step_then_member_role(context: Context, expected: str) -> None:
|
|
assert context.team_member.role == TeamRole(expected)
|
|
|
|
|
|
@then('the team member email should be "{expected}"')
|
|
def step_then_member_email(context: Context, expected: str) -> None:
|
|
assert context.team_member.email == expected
|
|
|
|
|
|
@when('I try to create a team member with email "{email}"')
|
|
def step_when_create_member_bad_email(context: Context, email: str) -> None:
|
|
context.validation_error = None
|
|
try:
|
|
TeamMember(
|
|
user_id="test",
|
|
display_name="Test",
|
|
email=email,
|
|
role=TeamRole.MEMBER,
|
|
)
|
|
except ValidationError as exc:
|
|
context.validation_error = exc
|
|
|
|
|
|
@then("a team member validation error should be raised")
|
|
def step_then_team_member_validation_error_raised(context: Context) -> None:
|
|
assert context.validation_error is not None, "Expected ValidationError"
|
|
|
|
|
|
@then('the team member should have "{permission}" permission')
|
|
def step_then_member_has_permission(context: Context, permission: str) -> None:
|
|
perm = TeamPermission(permission)
|
|
assert context.team_member.has_permission(perm), (
|
|
f"Expected member to have {permission}"
|
|
)
|
|
|
|
|
|
@then('the team member should not have "{permission}" permission')
|
|
def step_then_member_no_permission(context: Context, permission: str) -> None:
|
|
perm = TeamPermission(permission)
|
|
assert not context.team_member.has_permission(perm), (
|
|
f"Expected member NOT to have {permission}"
|
|
)
|
|
|
|
|
|
# ── UserSession model ──────────────────────────────────────────────
|
|
|
|
|
|
@given('a user session for user "{user_id}" on project "{project_id}"')
|
|
def step_given_user_session(context: Context, user_id: str, project_id: str) -> None:
|
|
context.user_session = UserSession(
|
|
session_id=str(SessionRegistry().generate_session_id()),
|
|
user_id=user_id,
|
|
display_name=user_id.title(),
|
|
project_id=project_id,
|
|
)
|
|
|
|
|
|
@then('the user session user_id should be "{expected}"')
|
|
def step_then_session_user_id(context: Context, expected: str) -> None:
|
|
assert context.user_session.user_id == expected
|
|
|
|
|
|
@then('the user session project_id should be "{expected}"')
|
|
def step_then_session_project_id(context: Context, expected: str) -> None:
|
|
assert context.user_session.project_id == expected
|
|
|
|
|
|
@then("the user session should be active")
|
|
def step_then_session_active(context: Context) -> None:
|
|
assert context.user_session.is_active is True
|
|
|
|
|
|
@when("the session is deactivated")
|
|
def step_when_session_deactivated(context: Context) -> None:
|
|
context.user_session.deactivate()
|
|
|
|
|
|
@then("the user session should not be active")
|
|
def step_then_session_not_active(context: Context) -> None:
|
|
assert context.user_session.is_active is False
|
|
|
|
|
|
@when("the session is touched")
|
|
def step_when_session_touched(context: Context) -> None:
|
|
context.old_last_active = context.user_session.last_active
|
|
sleep(0.01)
|
|
context.user_session.touch()
|
|
|
|
|
|
@then("the last_active timestamp should be updated")
|
|
def step_then_last_active_updated(context: Context) -> None:
|
|
assert context.user_session.last_active >= context.old_last_active
|
|
|
|
|
|
# ── VersionStamp model ─────────────────────────────────────────────
|
|
|
|
|
|
@given('a version stamp for resource "{resource_id}" by user "{user_id}"')
|
|
def step_given_version_stamp(context: Context, resource_id: str, user_id: str) -> None:
|
|
context.version_stamp = VersionStamp(
|
|
resource_id=resource_id,
|
|
updated_by=user_id,
|
|
)
|
|
|
|
|
|
@then("the version stamp version should be {expected:d}")
|
|
def step_then_stamp_version(context: Context, expected: int) -> None:
|
|
assert context.version_stamp.version == expected
|
|
|
|
|
|
@then('the version stamp updated_by should be "{expected}"')
|
|
def step_then_stamp_updated_by(context: Context, expected: str) -> None:
|
|
assert context.version_stamp.updated_by == expected
|
|
|
|
|
|
@when('the version stamp is incremented by "{user_id}"')
|
|
def step_when_stamp_incremented(context: Context, user_id: str) -> None:
|
|
context.new_version_stamp = context.version_stamp.increment(user_id)
|
|
|
|
|
|
@then("the new version stamp version should be {expected:d}")
|
|
def step_then_new_stamp_version(context: Context, expected: int) -> None:
|
|
assert context.new_version_stamp.version == expected
|
|
|
|
|
|
@then('the new version stamp updated_by should be "{expected}"')
|
|
def step_then_new_stamp_updated_by(context: Context, expected: str) -> None:
|
|
assert context.new_version_stamp.updated_by == expected
|
|
|
|
|
|
@then("incrementing with empty user should raise ValueError")
|
|
def step_then_increment_empty_raises(context: Context) -> None:
|
|
raised = False
|
|
try:
|
|
context.version_stamp.increment("")
|
|
except ValueError:
|
|
raised = True
|
|
assert raised, "Expected ValueError"
|
|
|
|
|
|
# ── ConflictDetectionResult ────────────────────────────────────────
|
|
|
|
|
|
@given('a conflict detection result with no conflict for "{resource_id}"')
|
|
def step_given_no_conflict_result(context: Context, resource_id: str) -> None:
|
|
context.conflict_result = ConflictDetectionResult(
|
|
detected=False,
|
|
resource_id=resource_id,
|
|
resolved=True,
|
|
details="No conflict",
|
|
)
|
|
|
|
|
|
@then("the result detected should be False")
|
|
def step_then_result_not_detected(context: Context) -> None:
|
|
assert context.conflict_result.detected is False
|
|
|
|
|
|
@then('the result resource_id should be "{expected}"')
|
|
def step_then_result_resource_id(context: Context, expected: str) -> None:
|
|
assert context.conflict_result.resource_id == expected
|
|
|
|
|
|
# ── VersionConflictError ───────────────────────────────────────────
|
|
|
|
|
|
@when(
|
|
'a VersionConflictError is raised for "{resource_id}" '
|
|
"expected {expected:d} actual {actual:d}"
|
|
)
|
|
def step_when_version_conflict_error(
|
|
context: Context, resource_id: str, expected: int, actual: int
|
|
) -> None:
|
|
context.conflict_error = VersionConflictError(
|
|
resource_id=resource_id,
|
|
expected_version=expected,
|
|
actual_version=actual,
|
|
)
|
|
|
|
|
|
@then('the error resource_id should be "{expected}"')
|
|
def step_then_error_resource_id(context: Context, expected: str) -> None:
|
|
assert context.conflict_error.resource_id == expected
|
|
|
|
|
|
@then("the error expected_version should be {expected:d}")
|
|
def step_then_error_expected_version(context: Context, expected: int) -> None:
|
|
assert context.conflict_error.expected_version == expected
|
|
|
|
|
|
@then("the error actual_version should be {expected:d}")
|
|
def step_then_error_actual_version(context: Context, expected: int) -> None:
|
|
assert context.conflict_error.actual_version == expected
|
|
|
|
|
|
@then("creating VersionConflictError with empty resource_id raises ValueError")
|
|
def step_then_conflict_error_empty_raises(context: Context) -> None:
|
|
raised = False
|
|
try:
|
|
VersionConflictError(resource_id="", expected_version=1, actual_version=2)
|
|
except ValueError:
|
|
raised = True
|
|
assert raised, "Expected ValueError"
|
|
|
|
|
|
# ── SessionRegistry ────────────────────────────────────────────────
|
|
|
|
|
|
@given("a session registry")
|
|
def step_given_session_registry(context: Context) -> None:
|
|
context.session_registry = SessionRegistry()
|
|
context.sessions = {}
|
|
|
|
|
|
@given('a user session "{session_id}" for user "{user_id}" on project "{project_id}"')
|
|
def step_given_named_session(
|
|
context: Context, session_id: str, user_id: str, project_id: str
|
|
) -> None:
|
|
session = UserSession(
|
|
session_id=session_id,
|
|
user_id=user_id,
|
|
display_name=user_id.title(),
|
|
project_id=project_id,
|
|
)
|
|
if not hasattr(context, "sessions") or context.sessions is None:
|
|
context.sessions = {}
|
|
context.sessions[session_id] = session
|
|
|
|
|
|
@when("the session is registered")
|
|
def step_when_session_registered(context: Context) -> None:
|
|
for session in context.sessions.values():
|
|
context.session_registry.register(session)
|
|
|
|
|
|
@when("all sessions are registered")
|
|
def step_when_all_sessions_registered(context: Context) -> None:
|
|
for session in context.sessions.values():
|
|
context.session_registry.register(session)
|
|
|
|
|
|
@then('the registry should contain the session "{session_id}"')
|
|
def step_then_registry_contains(context: Context, session_id: str) -> None:
|
|
result = context.session_registry.get(session_id)
|
|
assert result is not None, f"Session '{session_id}' not found"
|
|
|
|
|
|
@then('the registry should not contain the session "{session_id}"')
|
|
def step_then_registry_not_contains(context: Context, session_id: str) -> None:
|
|
result = context.session_registry.get(session_id)
|
|
assert result is None, f"Session '{session_id}' should not be found"
|
|
|
|
|
|
@then("the registry active count should be {expected:d}")
|
|
def step_then_registry_active_count(context: Context, expected: int) -> None:
|
|
assert context.session_registry.active_count == expected, (
|
|
f"Expected {expected}, got {context.session_registry.active_count}"
|
|
)
|
|
|
|
|
|
@when('the session "{session_id}" is unregistered')
|
|
def step_when_session_unregistered(context: Context, session_id: str) -> None:
|
|
context.session_registry.unregister(session_id)
|
|
|
|
|
|
@then('the registry should have {count:d} sessions for user "{user_id}"')
|
|
def step_then_registry_user_sessions(
|
|
context: Context, count: int, user_id: str
|
|
) -> None:
|
|
sessions = context.session_registry.get_sessions_for_user(user_id)
|
|
assert len(sessions) == count, (
|
|
f"Expected {count} sessions for user '{user_id}', got {len(sessions)}"
|
|
)
|
|
|
|
|
|
@then('the registry should have {count:d} sessions for project "{project_id}"')
|
|
def step_then_registry_project_sessions(
|
|
context: Context, count: int, project_id: str
|
|
) -> None:
|
|
sessions = context.session_registry.get_sessions_for_project(project_id)
|
|
assert len(sessions) == count, (
|
|
f"Expected {count} sessions for project '{project_id}', got {len(sessions)}"
|
|
)
|
|
|
|
|
|
@when("the registry is cleared")
|
|
def step_when_registry_cleared(context: Context) -> None:
|
|
context.session_registry.clear()
|
|
|
|
|
|
@then("registering the same session again raises ValueError")
|
|
def step_then_duplicate_registration_raises(context: Context) -> None:
|
|
session = next(iter(context.sessions.values()))
|
|
raised = False
|
|
try:
|
|
context.session_registry.register(session)
|
|
except ValueError:
|
|
raised = True
|
|
assert raised, "Expected ValueError for duplicate registration"
|
|
|
|
|
|
@then("the generated session ID should be a valid ULID")
|
|
def step_then_generated_id_valid(context: Context) -> None:
|
|
sid = context.session_registry.generate_session_id()
|
|
assert ULID_RE.match(sid), f"'{sid}' is not a valid ULID"
|
|
|
|
|
|
# ── TeamCollaborationService — member management ───────────────────
|
|
|
|
|
|
@given("a team collaboration service")
|
|
def step_given_collab_service(context: Context) -> None:
|
|
context.collab_service = TeamCollaborationService()
|
|
context.last_session = None
|
|
context.last_conflict_result = None
|
|
|
|
|
|
@when('I add member "{user_id}" with role "{role}" and email "{email}"')
|
|
def step_when_add_member(context: Context, user_id: str, role: str, email: str) -> None:
|
|
context.collab_service.add_member(
|
|
user_id=user_id,
|
|
display_name=user_id.title(),
|
|
email=email,
|
|
role=TeamRole(role),
|
|
)
|
|
|
|
|
|
@then("the service should have {count:d} members")
|
|
def step_then_service_member_count(context: Context, count: int) -> None:
|
|
members = context.collab_service.list_members()
|
|
assert len(members) == count, f"Expected {count} members, got {len(members)}"
|
|
|
|
|
|
@then('getting member "{user_id}" should return the member')
|
|
def step_then_get_member(context: Context, user_id: str) -> None:
|
|
member = context.collab_service.get_member(user_id)
|
|
assert member.user_id == user_id
|
|
|
|
|
|
@then('adding member "{user_id}" again raises TeamMemberExistsError')
|
|
def step_then_add_duplicate_raises(context: Context, user_id: str) -> None:
|
|
raised = False
|
|
try:
|
|
context.collab_service.add_member(
|
|
user_id=user_id,
|
|
display_name="Dup",
|
|
email="dup@example.com",
|
|
role=TeamRole.MEMBER,
|
|
)
|
|
except TeamMemberExistsError:
|
|
raised = True
|
|
assert raised, "Expected TeamMemberExistsError"
|
|
|
|
|
|
@when('I remove member "{user_id}"')
|
|
def step_when_remove_member(context: Context, user_id: str) -> None:
|
|
context.collab_service.remove_member(user_id)
|
|
|
|
|
|
@then('removing member "{user_id}" raises TeamMemberNotFoundError')
|
|
def step_then_remove_not_found_raises(context: Context, user_id: str) -> None:
|
|
raised = False
|
|
try:
|
|
context.collab_service.remove_member(user_id)
|
|
except TeamMemberNotFoundError:
|
|
raised = True
|
|
assert raised, "Expected TeamMemberNotFoundError"
|
|
|
|
|
|
@when('I update member "{user_id}" role to "{role}"')
|
|
def step_when_update_role(context: Context, user_id: str, role: str) -> None:
|
|
context.collab_service.update_member_role(user_id, TeamRole(role))
|
|
|
|
|
|
@then('member "{user_id}" should have role "{role}"')
|
|
def step_then_member_has_role(context: Context, user_id: str, role: str) -> None:
|
|
member = context.collab_service.get_member(user_id)
|
|
assert member.role == TeamRole(role), f"Expected role {role}, got {member.role}"
|
|
|
|
|
|
# ── TeamCollaborationService — permission checks ──────────────────
|
|
|
|
|
|
@then('member "{user_id}" should have "{permission}" team permission')
|
|
def step_then_service_member_has_perm(
|
|
context: Context, user_id: str, permission: str
|
|
) -> None:
|
|
result = context.collab_service.check_permission(
|
|
user_id, TeamPermission(permission)
|
|
)
|
|
assert result is True, f"Expected {user_id} to have {permission}"
|
|
|
|
|
|
@then('member "{user_id}" should not have "{permission}" team permission')
|
|
def step_then_service_member_no_perm(
|
|
context: Context, user_id: str, permission: str
|
|
) -> None:
|
|
result = context.collab_service.check_permission(
|
|
user_id, TeamPermission(permission)
|
|
)
|
|
assert result is False, f"Expected {user_id} NOT to have {permission}"
|
|
|
|
|
|
@then('enforcing "{permission}" on "{user_id}" raises PermissionError')
|
|
def step_then_enforce_raises(context: Context, permission: str, user_id: str) -> None:
|
|
raised = False
|
|
try:
|
|
context.collab_service.enforce_permission(user_id, TeamPermission(permission))
|
|
except PermissionError:
|
|
raised = True
|
|
assert raised, "Expected PermissionError"
|
|
|
|
|
|
# ── TeamCollaborationService — session management ─────────────────
|
|
|
|
|
|
@when('I register a session for "{user_id}" on project "{project_id}"')
|
|
def step_when_register_service_session(
|
|
context: Context, user_id: str, project_id: str
|
|
) -> None:
|
|
session = context.collab_service.register_session(
|
|
user_id=user_id,
|
|
project_id=project_id,
|
|
)
|
|
context.last_session = session
|
|
|
|
|
|
@then('the project "{project_id}" should have {count:d} active sessions')
|
|
def step_then_project_active_sessions(
|
|
context: Context, project_id: str, count: int
|
|
) -> None:
|
|
sessions = context.collab_service.get_active_sessions_for_project(project_id)
|
|
assert len(sessions) == count, (
|
|
f"Expected {count} active sessions for '{project_id}', got {len(sessions)}"
|
|
)
|
|
|
|
|
|
@when("I unregister the last session")
|
|
def step_when_unregister_last_session(context: Context) -> None:
|
|
context.collab_service.unregister_session(context.last_session.session_id)
|
|
|
|
|
|
# ── TeamCollaborationService — conflict resolution ─────────────────
|
|
|
|
|
|
@when('I create a version stamp for "{resource_id}" by "{user_id}"')
|
|
def step_when_create_stamp(context: Context, resource_id: str, user_id: str) -> None:
|
|
context.collab_service.create_version_stamp(resource_id, user_id)
|
|
|
|
|
|
@then('the version stamp for "{resource_id}" should have version {version:d}')
|
|
def step_then_stamp_has_version(
|
|
context: Context, resource_id: str, version: int
|
|
) -> None:
|
|
stamp = context.collab_service.get_version_stamp(resource_id)
|
|
assert stamp is not None, f"No stamp for '{resource_id}'"
|
|
assert stamp.version == version, f"Expected version {version}, got {stamp.version}"
|
|
|
|
|
|
@when('I update version for "{resource_id}" expecting {version:d} by "{user_id}"')
|
|
def step_when_update_version(
|
|
context: Context, resource_id: str, version: int, user_id: str
|
|
) -> None:
|
|
context.last_conflict_result = context.collab_service.update_version(
|
|
resource_id=resource_id,
|
|
expected_version=version,
|
|
user_id=user_id,
|
|
)
|
|
|
|
|
|
@then(
|
|
'updating "{resource_id}" expecting {version:d} by "{user_id}" '
|
|
"with reject raises VersionConflictError"
|
|
)
|
|
def step_then_update_reject_raises(
|
|
context: Context, resource_id: str, version: int, user_id: str
|
|
) -> None:
|
|
raised = False
|
|
try:
|
|
context.collab_service.update_version(
|
|
resource_id=resource_id,
|
|
expected_version=version,
|
|
user_id=user_id,
|
|
strategy=ConflictResolutionStrategy.REJECT,
|
|
)
|
|
except VersionConflictError:
|
|
raised = True
|
|
assert raised, "Expected VersionConflictError"
|
|
|
|
|
|
@when(
|
|
'I update "{resource_id}" expecting {version:d} by "{user_id}" '
|
|
"with last_writer_wins"
|
|
)
|
|
def step_when_update_lww(
|
|
context: Context, resource_id: str, version: int, user_id: str
|
|
) -> None:
|
|
context.last_conflict_result = context.collab_service.update_version(
|
|
resource_id=resource_id,
|
|
expected_version=version,
|
|
user_id=user_id,
|
|
strategy=ConflictResolutionStrategy.LAST_WRITER_WINS,
|
|
)
|
|
|
|
|
|
@when('I update "{resource_id}" expecting {version:d} by "{user_id}" with merge')
|
|
def step_when_update_merge(
|
|
context: Context, resource_id: str, version: int, user_id: str
|
|
) -> None:
|
|
context.last_conflict_result = context.collab_service.update_version(
|
|
resource_id=resource_id,
|
|
expected_version=version,
|
|
user_id=user_id,
|
|
strategy=ConflictResolutionStrategy.MERGE,
|
|
)
|
|
|
|
|
|
@then("the conflict result should be detected and resolved")
|
|
def step_then_conflict_detected_resolved(context: Context) -> None:
|
|
result = context.last_conflict_result
|
|
assert result.detected is True, "Expected conflict detected"
|
|
assert result.resolved is True, "Expected conflict resolved"
|
|
|
|
|
|
@then('detecting conflict on "{resource_id}" expecting {version:d} should return True')
|
|
def step_then_detect_conflict_true(
|
|
context: Context, resource_id: str, version: int
|
|
) -> None:
|
|
result = context.collab_service.detect_conflict(resource_id, version)
|
|
assert result is True, "Expected conflict detected"
|
|
|
|
|
|
@then('detecting conflict on "{resource_id}" expecting {version:d} should return False')
|
|
def step_then_detect_conflict_false(
|
|
context: Context, resource_id: str, version: int
|
|
) -> None:
|
|
result = context.collab_service.detect_conflict(resource_id, version)
|
|
assert result is False, "Expected no conflict"
|
|
|
|
|
|
# ── Error hierarchy ────────────────────────────────────────────────
|
|
|
|
|
|
@then("TeamMemberNotFoundError should be a subclass of TeamCollaborationError")
|
|
def step_then_not_found_is_subclass(context: Context) -> None:
|
|
assert issubclass(TeamMemberNotFoundError, TeamCollaborationError)
|
|
|
|
|
|
@then("TeamMemberExistsError should be a subclass of TeamCollaborationError")
|
|
def step_then_exists_is_subclass(context: Context) -> None:
|
|
assert issubclass(TeamMemberExistsError, TeamCollaborationError)
|
|
|
|
|
|
@then("VersionConflictError should be a subclass of TeamCollaborationError")
|
|
def step_then_conflict_is_subclass(context: Context) -> None:
|
|
assert issubclass(VersionConflictError, TeamCollaborationError)
|
|
|
|
|
|
# ── Service edge-case validations ──────────────────────────────────
|
|
|
|
|
|
@given("a fresh TeamCollaborationService")
|
|
def step_fresh_service(context: Context) -> None:
|
|
context.service = TeamCollaborationService()
|
|
|
|
|
|
@then("the service session_registry should be a SessionRegistry instance")
|
|
def step_check_session_registry_type(context: Context) -> None:
|
|
assert isinstance(context.service.session_registry, SessionRegistry)
|
|
|
|
|
|
@when("I try to add a member with empty user_id via service")
|
|
def step_add_member_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.add_member(
|
|
user_id="", display_name="X", email="x@x.com", role=TeamRole.MEMBER
|
|
)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to add a member with empty display_name via service")
|
|
def step_add_member_empty_name(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.add_member(
|
|
user_id="u1", display_name="", email="x@x.com", role=TeamRole.MEMBER
|
|
)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to add a member with empty email via service")
|
|
def step_add_member_empty_email(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.add_member(
|
|
user_id="u1", display_name="X", email="", role=TeamRole.MEMBER
|
|
)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then('a collab ValueError should be raised mentioning "{fragment}"')
|
|
def step_check_value_error_msg(context: Context, fragment: str) -> None:
|
|
assert context.caught_exception is not None, "Expected ValueError"
|
|
assert isinstance(context.caught_exception, ValueError)
|
|
assert fragment in str(context.caught_exception), (
|
|
f"Expected '{fragment}' in '{context.caught_exception}'"
|
|
)
|
|
|
|
|
|
@when("I try to remove a member with empty user_id via service")
|
|
def step_remove_member_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.remove_member(user_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to get a member with empty user_id via service")
|
|
def step_get_member_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.get_member(user_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when('I try to get member "{uid}" via service')
|
|
def step_get_member_missing(context: Context, uid: str) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.get_member(user_id=uid)
|
|
except TeamMemberNotFoundError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then("a TeamMemberNotFoundError should be raised")
|
|
def step_check_not_found_error(context: Context) -> None:
|
|
assert context.caught_exception is not None, "Expected TeamMemberNotFoundError"
|
|
assert isinstance(context.caught_exception, TeamMemberNotFoundError)
|
|
|
|
|
|
@when("I try to update role with empty user_id via service")
|
|
def step_update_role_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.update_member_role(user_id="", new_role=TeamRole.ADMIN)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to register session with empty user_id via service")
|
|
def step_register_session_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.register_session(user_id="", project_id="p1")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@given('a team member "{uid}" with role "{role}"')
|
|
def step_add_team_member(context: Context, uid: str, role: str) -> None:
|
|
context.service.add_member(
|
|
user_id=uid,
|
|
display_name=f"User {uid}",
|
|
email=f"{uid}@test.com",
|
|
role=TeamRole(role),
|
|
)
|
|
|
|
|
|
@when('I try to register session with empty project_id for user "{uid}"')
|
|
def step_register_session_empty_pid(context: Context, uid: str) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.register_session(user_id=uid, project_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to unregister session with empty session_id")
|
|
def step_unregister_session_empty_sid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.unregister_session(session_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to get sessions for empty project_id")
|
|
def step_get_sessions_empty_pid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.get_active_sessions_for_project(project_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to create version stamp with empty resource_id")
|
|
def step_create_stamp_empty_rid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.create_version_stamp(resource_id="", user_id="u1")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to create version stamp with empty user_id")
|
|
def step_create_stamp_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.create_version_stamp(resource_id="r1", user_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to get version stamp with empty resource_id")
|
|
def step_get_stamp_empty_rid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.get_version_stamp(resource_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to update version with empty resource_id")
|
|
def step_update_version_empty_rid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.update_version(resource_id="", expected_version=1, user_id="u1")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to update version with empty user_id")
|
|
def step_update_version_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.update_version(resource_id="r1", expected_version=1, user_id="")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to update version with expected_version 0")
|
|
def step_update_version_zero(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.update_version(
|
|
resource_id="r1", expected_version=0, user_id="u1"
|
|
)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when(
|
|
'I update version for untracked resource "{rid}" by user "{uid}" expecting version {ver:d}'
|
|
)
|
|
def step_update_version_untracked(
|
|
context: Context, rid: str, uid: str, ver: int
|
|
) -> None:
|
|
context.conflict_result = context.service.update_version(
|
|
resource_id=rid, expected_version=ver, user_id=uid
|
|
)
|
|
|
|
|
|
@then("the result should indicate no conflict detected")
|
|
def step_check_no_conflict(context: Context) -> None:
|
|
assert not context.conflict_result.detected, "Expected no conflict"
|
|
|
|
|
|
@then('the result details should contain "{text}"')
|
|
def step_check_result_details(context: Context, text: str) -> None:
|
|
assert text in (context.conflict_result.details or ""), (
|
|
f"Expected '{text}' in '{context.conflict_result.details}'"
|
|
)
|
|
|
|
|
|
@when("I try to detect conflict with empty resource_id")
|
|
def step_detect_conflict_empty_rid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.detect_conflict(resource_id="", expected_version=1)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to detect conflict with expected_version 0")
|
|
def step_detect_conflict_zero(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.service.detect_conflict(resource_id="r1", expected_version=0)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when('I detect conflict for untracked resource "{rid}" expecting version {ver:d}')
|
|
def step_detect_conflict_untracked(context: Context, rid: str, ver: int) -> None:
|
|
context.conflict_bool = context.service.detect_conflict(
|
|
resource_id=rid, expected_version=ver
|
|
)
|
|
|
|
|
|
@then("the conflict result should be False")
|
|
def step_check_conflict_false(context: Context) -> None:
|
|
assert context.conflict_bool is False
|
|
|
|
|
|
# ── Domain model edge-case validations ─────────────────────────────
|
|
|
|
|
|
@when("I try to create VersionConflictError with expected_version -1")
|
|
def step_vce_negative_expected(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
VersionConflictError(resource_id="r1", expected_version=-1, actual_version=1)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to create VersionConflictError with actual_version -1")
|
|
def step_vce_negative_actual(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
VersionConflictError(resource_id="r1", expected_version=1, actual_version=-1)
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@given("a fresh SessionRegistry")
|
|
def step_fresh_registry(context: Context) -> None:
|
|
context.registry = SessionRegistry()
|
|
|
|
|
|
@when("I try to register a session with empty session_id")
|
|
def step_register_empty_sid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
session = UserSession(
|
|
session_id="",
|
|
user_id="u1",
|
|
display_name="User 1",
|
|
project_id="p1",
|
|
)
|
|
context.registry.register(session)
|
|
except (ValueError, ValidationError) as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to unregister a session with empty session_id")
|
|
def step_unregister_empty_sid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.registry.unregister("")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when('I unregister session "{sid}"')
|
|
def step_unregister_unknown(context: Context, sid: str) -> None:
|
|
context.unregister_result = context.registry.unregister(sid)
|
|
|
|
|
|
@then("the unregister result should be False")
|
|
def step_check_unregister_false(context: Context) -> None:
|
|
assert context.unregister_result is False
|
|
|
|
|
|
@when("I try to get session with empty session_id")
|
|
def step_get_session_empty_sid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.registry.get("")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@given('a registered session "{sid}" for user "{uid}" in project "{pid}"')
|
|
def step_register_session_in_registry(
|
|
context: Context, sid: str, uid: str, pid: str
|
|
) -> None:
|
|
session = UserSession(
|
|
session_id=sid,
|
|
user_id=uid,
|
|
display_name=f"User {uid}",
|
|
project_id=pid,
|
|
)
|
|
context.registry.register(session)
|
|
|
|
|
|
@when('I deactivate session "{sid}" in the registry')
|
|
def step_deactivate_session(context: Context, sid: str) -> None:
|
|
session = context.registry.get(sid)
|
|
session.deactivate()
|
|
|
|
|
|
@then("get_active_sessions should return {count:d} session")
|
|
def step_check_active_count(context: Context, count: int) -> None:
|
|
active = context.registry.get_active_sessions()
|
|
assert len(active) == count, f"Expected {count} active sessions, got {len(active)}"
|
|
|
|
|
|
@when("I try to get sessions for empty user_id")
|
|
def step_get_sessions_empty_uid(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.registry.get_sessions_for_user("")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when("I try to get sessions for empty project_id via registry")
|
|
def step_get_sessions_empty_pid_registry(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.registry.get_sessions_for_project("")
|
|
except ValueError as exc:
|
|
context.caught_exception = exc
|