forked from cleveragents/cleveragents-core
test(features/steps): fix features that depend on running as user not superuser
This commit is contained in:
@@ -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")
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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')}"
|
||||
)
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user