diff --git a/features/steps/actor_cli_steps.py b/features/steps/actor_cli_steps.py index 561d333a..57c27012 100644 --- a/features/steps/actor_cli_steps.py +++ b/features/steps/actor_cli_steps.py @@ -1088,7 +1088,9 @@ def step_impl(context): assert context.result.exit_code != 0 expected = getattr(context, "expected_error", None) if expected: - assert expected in context.result.output + # Normalize whitespace to handle Rich text wrapping in narrow terminals + normalized = " ".join(context.result.output.split()) + assert expected in normalized, f"Expected '{expected}' in output: {normalized}" @then("the actor update should set safe and merge config") diff --git a/features/steps/auto_debug_cli_coverage_steps.py b/features/steps/auto_debug_cli_coverage_steps.py index aff34ce6..8d72e032 100644 --- a/features/steps/auto_debug_cli_coverage_steps.py +++ b/features/steps/auto_debug_cli_coverage_steps.py @@ -320,7 +320,9 @@ def step_assert_command_aborted(context): @then('the output should contain "{text}"') def step_assert_output_contains(context, text): output = _capture_output(context) - assert text in output, f"Expected to find '{text}' in CLI output: {output}" + # Normalize whitespace to handle Rich text wrapping in narrow terminals + normalized = " ".join(output.split()) + assert text in normalized, f"Expected to find '{text}' in CLI output: {output}" @given( diff --git a/features/steps/edge_case_plan_steps.py b/features/steps/edge_case_plan_steps.py index 68686067..08dd8210 100644 --- a/features/steps/edge_case_plan_steps.py +++ b/features/steps/edge_case_plan_steps.py @@ -2,12 +2,12 @@ from __future__ import annotations -import contextlib import os import shutil import tempfile from datetime import datetime from pathlib import Path +from unittest.mock import patch from behave import given, then, when from behave.runner import Context @@ -327,8 +327,7 @@ def step_create_edge_project_with_plan(context: Context) -> None: def step_add_readonly_modify_change(context: Context) -> None: test_file = context.edge_temp_dir / "readonly_edge.py" test_file.write_text("# original content") - os.chmod(test_file, 0o444) - context.edge_readonly_file = test_file + context.readonly_target = str(test_file) with context.edge_uow.transaction() as ctx: change = Change( @@ -348,19 +347,23 @@ def step_add_readonly_modify_change(context: Context) -> None: @when("I try to apply the edge case changes") def step_try_apply_edge_changes(context: Context) -> None: + target = getattr(context, "readonly_target", None) + + original_write_text = Path.write_text + + def guarded_write_text(self, *args, **kwargs): + if target and str(self) == target: + raise PermissionError(f"[Errno 13] Permission denied: '{self}'") + return original_write_text(self, *args, **kwargs) + try: - context.edge_applied_count = context.edge_plan_service.apply_changes( - context.edge_project - ) + with patch.object(Path, "write_text", guarded_write_text): + context.edge_applied_count = context.edge_plan_service.apply_changes( + context.edge_project + ) context.edge_exception = None except Exception as exc: context.edge_exception = exc - finally: - # Clean up read-only files - readonly = getattr(context, "edge_readonly_file", None) - if readonly and readonly.exists(): - with contextlib.suppress(Exception): - os.chmod(readonly, 0o644) @then("a PlanError should be raised mentioning file apply failure") @@ -813,11 +816,10 @@ def step_add_create_then_failing_modify(context: Context) -> None: ) ctx.changes.add(create_change) - # Create a read-only file so MODIFY will fail + # Create the file that MODIFY will target; mock will make writes fail fail_file = context.edge_temp_dir / "fail_target.py" fail_file.write_text("# original fail target content") - os.chmod(fail_file, 0o444) - context.edge_readonly_file = fail_file + context.readonly_target = str(fail_file) with context.edge_uow.transaction() as ctx: modify_change = Change( @@ -844,15 +846,14 @@ def step_check_first_file_exists(context: Context) -> None: @then("the second file should remain unchanged on disk") def step_check_second_file_unchanged(context: Context) -> None: - # Clean up read-only - readonly = getattr(context, "edge_readonly_file", None) - if readonly and readonly.exists(): - with contextlib.suppress(Exception): - os.chmod(readonly, 0o644) - content = readonly.read_text() - assert content == "# original fail target content", ( - f"Expected original content, got: {content}" - ) + target = getattr(context, "readonly_target", None) + if target: + target_path = Path(target) + if target_path.exists(): + content = target_path.read_text() + assert content == "# original fail target content", ( + f"Expected original content, got: {content}" + ) @then("the plan cannot be restarted from errored state") diff --git a/features/steps/plan_service_steps.py b/features/steps/plan_service_steps.py index ae442636..7a7d6f8c 100644 --- a/features/steps/plan_service_steps.py +++ b/features/steps/plan_service_steps.py @@ -1,7 +1,6 @@ """Step definitions for plan service coverage tests.""" import asyncio -import builtins import contextlib import os import tempfile @@ -2392,14 +2391,10 @@ def step_check_changes_marked_applied(context: Context) -> None: @given("the plan has a change that will fail on apply") def step_add_failing_change(context: Context) -> None: """Add a change that will fail when applied.""" - import os - with context.unit_of_work.transaction() as ctx: - # Create a file and make it readonly, then try to modify it + # Create a file then simulate permission denial via mock test_file = context.temp_dir / "readonly_file.py" test_file.write_text("original content") - # Make the file read-only - os.chmod(test_file, 0o444) # Try to modify a read-only file - this should fail change = Change( @@ -2415,17 +2410,31 @@ def step_add_failing_change(context: Context) -> None: created_at=datetime.now(), ) ctx.changes.add(change) - # Store for cleanup - context.readonly_file = test_file + # Store path for the mock in the apply step + context.readonly_file = str(test_file) @when("I try to apply the changes") def step_try_apply_changes(context: Context) -> None: - """Try to apply changes.""" + """Try to apply changes, simulating PermissionError for readonly targets.""" + target = getattr(context, "readonly_file", None) + + original_write_text = Path.write_text + + def guarded_write_text(self, *args, **kwargs): + if target and str(self) == target: + raise PermissionError(f"[Errno 13] Permission denied: '{self}'") + return original_write_text(self, *args, **kwargs) + try: - result = context.plan_service.apply_changes(context.project) - context.exception = None - context.apply_result = result + with ( + patch.object(Path, "write_text", guarded_write_text) + if target + else contextlib.nullcontext() + ): + result = context.plan_service.apply_changes(context.project) + context.exception = None + context.apply_result = result except Exception as e: context.exception = e @@ -2433,13 +2442,6 @@ def step_try_apply_changes(context: Context) -> None: @then("a PlanError should be raised with file operation details") def step_check_plan_error_with_details(context: Context) -> None: """Check that PlanError was raised with file operation details.""" - import os - - # Clean up the readonly file if it was created - if hasattr(context, "readonly_file"): - with contextlib.suppress(builtins.BaseException): - os.chmod(context.readonly_file, 0o644) - assert context.exception is not None, ( f"Expected an exception but got result: {getattr(context, 'apply_result', 'none')}" ) diff --git a/features/steps/project_service_steps.py b/features/steps/project_service_steps.py index 2996118e..3ff2260e 100644 --- a/features/steps/project_service_steps.py +++ b/features/steps/project_service_steps.py @@ -68,23 +68,28 @@ def step_create_readonly_directory(context: Context) -> None: """Create a read-only directory.""" context.readonly_dir = context.temp_dir / "readonly" context.readonly_dir.mkdir() - os.chmod(context.readonly_dir, 0o444) @when("I try to create a project in the read-only directory") def step_try_create_project_readonly(context: Context) -> None: """Try to create a project in a read-only directory.""" + readonly_dir = context.readonly_dir + + original_mkdir = Path.mkdir + + def guarded_mkdir(self, *args, **kwargs): + if str(self).startswith(str(readonly_dir)): + raise PermissionError(f"[Errno 13] Permission denied: '{self}'") + return original_mkdir(self, *args, **kwargs) + try: - context.project_service.initialize_project( - name="test-project", path=context.readonly_dir / "test-project", force=False - ) - context.exception = None + with patch.object(Path, "mkdir", guarded_mkdir): + context.project_service.initialize_project( + name="test-project", path=readonly_dir / "test-project", force=False + ) + context.exception = None except Exception as e: context.exception = e - finally: - # Clean up - restore permissions - if hasattr(context, "readonly_dir"): - os.chmod(context.readonly_dir, 0o755) @then('a FileSystemError should be raised with message "{message}"') @@ -686,25 +691,30 @@ def step_create_no_write_permissions_directory(context: Context) -> None: """Create a directory without write permissions.""" context.no_write_dir = context.temp_dir / "no-write" context.no_write_dir.mkdir() - os.chmod(context.no_write_dir, 0o444) @when("I try to create a project there") def step_try_create_project_no_permissions(context: Context) -> None: """Try to create a project in directory without permissions.""" + no_write_dir = context.no_write_dir + + original_mkdir = Path.mkdir + + def guarded_mkdir(self, *args, **kwargs): + if str(self).startswith(str(no_write_dir)): + raise PermissionError(f"[Errno 13] Permission denied: '{self}'") + return original_mkdir(self, *args, **kwargs) + try: - context.project_service.initialize_project( - name="permission-test", - path=context.no_write_dir / "permission-test", - force=False, - ) - context.exception = None + with patch.object(Path, "mkdir", guarded_mkdir): + context.project_service.initialize_project( + name="permission-test", + path=no_write_dir / "permission-test", + force=False, + ) + context.exception = None except Exception as e: context.exception = e - finally: - # Clean up - restore permissions - if hasattr(context, "no_write_dir"): - os.chmod(context.no_write_dir, 0o755) @then("a FileSystemError should be raised with permission details")