OverviewFilesCommitsIssuesDocsReleasesPeers

ollama_semantic_function_calling commits / e2b7d6e

updating with file ops
e2b7d6e41dcefa173f48263fd6a08f8c26725a05
by fleeky.ashatron <fleeky@7rnx.net> · 2025-04-10T12:48:20.000Z

Changed files (4) +453 -274

ollamatools-project/examples/file_operations.py +172 -0
……@@ -0,0 +1,172 @@
1#!/usr/bin/env python3
2"""
3Example script demonstrating file operations with OllamaTools.
4
5This script shows how to use the file operation functions with
6various examples and security features.
7"""
8
9import sys
10import os
11import tempfile
12import shutil
13
14# Add the parent directory to the path so we can import the package
15sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
16
17from ollamatools.tools.system_tools import read_file, write_file, find_files
18from ollamatools.security import PermissionManager
19
20
21def create_sample_files(directory):
22 """Create sample files for demonstration."""
23 # Create a few text files
24 write_file(os.path.join(directory, "notes.txt"),
25 "This is a sample text file for demonstration.\n"
26 "It contains multiple lines of text.\n"
27 "This can be used to test the read_file function.")
28
29 write_file(os.path.join(directory, "data.txt"),
30 "Sample data file with numbers:\n"
31 "1, 2, 3, 4, 5\n"
32 "10, 20, 30, 40, 50")
33
34 # Create a markdown file
35 write_file(os.path.join(directory, "readme.md"),
36 "# Sample Markdown File\n\n"
37 "This is a *markdown* file with some **formatting**.\n\n"
38 "- Item 1\n"
39 "- Item 2\n"
40 "- Item 3\n")
41
42 # Create a file in a subdirectory
43 os.makedirs(os.path.join(directory, "subdir"), exist_ok=True)
44 write_file(os.path.join(directory, "subdir", "nested.txt"),
45 "This file is in a subdirectory.")
46
47
48def main():
49 print("=== File Operations Example ===\n")
50
51 # Create a permission manager for secure file operations
52 pm = PermissionManager()
53
54 # Create a temporary directory for demonstration
55 temp_dir = tempfile.mkdtemp()
56 print(f"Created temporary directory: {temp_dir}")
57
58 try:
59 # Create sample files
60 print("\nCreating sample files...")
61 create_sample_files(temp_dir)
62 print("Sample files created successfully.")
63
64 # Example 1: Reading a file
65 print("\n--- Example 1: Reading a File ---")
66 file_path = os.path.join(temp_dir, "notes.txt")
67 print(f"Reading file: {file_path}")
68
69 try:
70 content = read_file(file_path, permission_manager=pm)
71 print("\nFile content:")
72 print("-" * 40)
73 print(content)
74 print("-" * 40)
75 except Exception as e:
76 print(f"Error reading file: {e}")
77
78 # Example 2: Writing to a file
79 print("\n--- Example 2: Writing to a File ---")
80 new_file_path = os.path.join(temp_dir, "new_file.txt")
81 new_content = "This is a new file created using write_file.\n" \
82 "It demonstrates how to write content to a file."
83
84 print(f"Writing to file: {new_file_path}")
85
86 try:
87 success = write_file(new_file_path, new_content, permission_manager=pm)
88 if success:
89 print("File written successfully.")
90
91 # Read back the file to verify
92 content = read_file(new_file_path, permission_manager=pm)
93 print("\nVerifying file content:")
94 print("-" * 40)
95 print(content)
96 print("-" * 40)
97 except Exception as e:
98 print(f"Error writing file: {e}")
99
100 # Example 3: Appending to a file
101 print("\n--- Example 3: Appending to a File ---")
102 append_content = "\nThis content was appended to the file."
103
104 print(f"Appending to file: {new_file_path}")
105
106 try:
107 success = write_file(new_file_path, append_content, mode='a', permission_manager=pm)
108 if success:
109 print("Content appended successfully.")
110
111 # Read back the file to verify
112 content = read_file(new_file_path, permission_manager=pm)
113 print("\nVerifying file content after append:")
114 print("-" * 40)
115 print(content)
116 print("-" * 40)
117 except Exception as e:
118 print(f"Error appending to file: {e}")
119
120 # Example 4: Finding files
121 print("\n--- Example 4: Finding Files ---")
122
123 try:
124 # Find all text files
125 print("Finding all .txt files (recursive):")
126 txt_files = find_files("*.txt", temp_dir, recursive=True, permission_manager=pm)
127 for file in txt_files:
128 print(f" - {file}")
129
130 # Find all markdown files
131 print("\nFinding all .md files:")
132 md_files = find_files("*.md", temp_dir, recursive=True, permission_manager=pm)
133 for file in md_files:
134 print(f" - {file}")
135
136 # Find files in the root directory only (non-recursive)
137 print("\nFinding files in the root directory only (non-recursive):")
138 root_files = find_files("*", temp_dir, recursive=False, permission_manager=pm)
139 for file in root_files:
140 print(f" - {file}")
141 except Exception as e:
142 print(f"Error finding files: {e}")
143
144 # Example 5: Handling errors
145 print("\n--- Example 5: Handling Errors ---")
146
147 print("Attempting to read a non-existent file:")
148 try:
149 content = read_file(os.path.join(temp_dir, "nonexistent.txt"), permission_manager=pm)
150 print("File content (should not reach here):", content)
151 except Exception as e:
152 print(f"Error correctly caught: {e}")
153
154 print("\nAttempting to read a file with size limit:")
155 try:
156 # Get the size of notes.txt and try to read with a smaller limit
157 file_path = os.path.join(temp_dir, "notes.txt")
158 file_size = os.path.getsize(file_path)
159 content = read_file(file_path, max_size=file_size - 10, permission_manager=pm)
160 print("File content (should not reach here):", content)
161 except Exception as e:
162 print(f"Error correctly caught: {e}")
163
164 finally:
165 # Clean up the temporary directory
166 print(f"\nCleaning up temporary directory: {temp_dir}")
167 shutil.rmtree(temp_dir)
168 print("Cleanup complete.")
169
170
171if __name__ == "__main__":
172 main()
ollamatools-project/ollamatools/tools/__init__.py +1 -0
……@@ -20,3 +20,4 @@ __all__ = [
2020 "write_file",
2121 "find_files"
2222]
23
ollamatools-project/ollamatools/tools/system_tools.py +160 -135
……@@ -10,7 +10,8 @@ import shlex
1010import os
1111import re
1212import glob
13from typing import Dict, Any, List, Optional, Union, Tuple
13from pathlib import Path
14from typing import Dict, Any, List, Optional, Union, Tuple, Set
1415
1516from ollamatools.security import PermissionManager
1617
……@@ -97,6 +98,106 @@ def execute_command(
9798 raise subprocess.SubprocessError(f"Command execution failed: {e}")
9899
99100
101def _sanitize_command(command: str) -> str:
102 """
103 Sanitize a command for security.
104
105 Args:
106 command: Command to sanitize
107
108 Returns:
109 str: Sanitized command
110
111 Raises:
112 ValueError: If command contains dangerous patterns
113 """
114 if not command or not command.strip():
115 raise ValueError("Empty command")
116
117 # Remove leading/trailing whitespace
118 command = command.strip()
119
120 # Disallow certain dangerous patterns
121 dangerous_patterns = [
122 r'[|&;]', # Shell control operators
123 r'>\s*\w+', # Redirections
124 r'<\s*\w+', # Input redirections
125 r'\$\(', # Command substitution
126 r'`', # Backtick command substitution
127 ]
128
129 for pattern in dangerous_patterns:
130 if re.search(pattern, command):
131 raise ValueError(f"Command contains dangerous pattern: {pattern}")
132
133 return command
134
135
136def _sanitize_argument(arg: str) -> str:
137 """
138 Sanitize a command argument for security.
139
140 Args:
141 arg: Argument to sanitize
142
143 Returns:
144 str: Sanitized argument
145
146 Raises:
147 ValueError: If argument contains dangerous patterns
148 """
149 if not isinstance(arg, str):
150 arg = str(arg)
151
152 # Remove leading/trailing whitespace
153 arg = arg.strip()
154
155 # Disallow certain dangerous patterns (similar to command sanitization)
156 dangerous_patterns = [
157 r'[|&;]', # Shell control operators
158 r'>\s*\w+', # Redirections
159 r'<\s*\w+', # Input redirections
160 r'\$\(', # Command substitution
161 r'`', # Backtick command substitution
162 ]
163
164 for pattern in dangerous_patterns:
165 if re.search(pattern, arg):
166 raise ValueError(f"Argument contains dangerous pattern: {pattern}")
167
168 return arg
169
170
171def _validate_path(path: str) -> str:
172 """
173 Validate and normalize a file path.
174
175 Args:
176 path: File path to validate
177
178 Returns:
179 str: Normalized path
180
181 Raises:
182 ValueError: If path is invalid or potentially dangerous
183 """
184 if not path or not path.strip():
185 raise ValueError("Empty path")
186
187 # Remove leading/trailing whitespace
188 path = path.strip()
189
190 # Resolve to absolute path and normalize
191 try:
192 # Use Path for better path handling
193 norm_path = Path(path).expanduser().resolve()
194
195 # Convert back to string for consistent return type
196 return str(norm_path)
197 except (ValueError, OSError) as e:
198 raise ValueError(f"Invalid path: {path}, error: {e}")
199
200
100201def read_file(
101202 path: str,
102203 max_size: Optional[int] = None,
……@@ -117,40 +218,44 @@ def read_file(
117218 FileNotFoundError: If file does not exist
118219 PermissionError: If permission is denied
119220 ValueError: If file exceeds max_size
221 IOError: If file reading fails
120222 """
121 # Sanitize and normalize path
122 path = os.path.normpath(os.path.abspath(path))
223 # Validate and normalize path
224 norm_path = _validate_path(path)
123225
124226 # Request permission if permission manager is provided
125227 if permission_manager:
126228 if not permission_manager.request_permission(
127229 "read_file",
128 {"path": path}
230 {"path": norm_path}
129231 ):
130 raise PermissionError(f"Permission denied for reading file: {path}")
232 raise PermissionError(f"Permission denied for reading file: {norm_path}")
131233
132234 # Check if file exists
133 if not os.path.isfile(path):
134 raise FileNotFoundError(f"File not found: {path}")
235 if not os.path.isfile(norm_path):
236 raise FileNotFoundError(f"File not found: {norm_path}")
135237
136238 # Check file size if max_size is specified
137239 if max_size is not None:
138 file_size = os.path.getsize(path)
240 file_size = os.path.getsize(norm_path)
139241 if file_size > max_size:
140 raise ValueError(f"File size ({file_size} bytes) exceeds maximum allowed size ({max_size} bytes)")
242 raise ValueError(
243 f"File size ({file_size} bytes) exceeds maximum allowed size ({max_size} bytes)"
244 )
141245
142 # Read and return file contents
143246 try:
144 with open(path, 'r', encoding='utf-8') as file:
247 # Read and return file contents
248 with open(norm_path, 'r', encoding='utf-8') as file:
145249 return file.read()
146250 except UnicodeDecodeError:
147 # Handle binary files by reading in binary mode and providing a warning
148 with open(path, 'rb') as file:
149 binary_content = file.read()
150 warning = f"[WARNING: File appears to be binary. Displaying as hex.]\n\n"
151 # Convert binary to hex representation for safe display
152 hex_content = binary_content.hex(' ', 16)
153 return warning + hex_content
251 # Try reading as binary if UTF-8 fails
252 try:
253 with open(norm_path, 'rb') as file:
254 return f"[Binary data, {os.path.getsize(norm_path)} bytes]"
255 except IOError as e:
256 raise IOError(f"Failed to read file: {norm_path}, error: {e}")
257 except IOError as e:
258 raise IOError(f"Failed to read file: {norm_path}, error: {e}")
154259
155260
156261def write_file(
……@@ -173,40 +278,35 @@ def write_file(
173278
174279 Raises:
175280 PermissionError: If permission is denied
176 ValueError: If mode is invalid
177 IOError: If file cannot be written
281 ValueError: If path or mode is invalid
282 IOError: If file writing fails
178283 """
284 # Validate and normalize path
285 norm_path = _validate_path(path)
286
179287 # Validate mode
180288 if mode not in ('w', 'a'):
181 raise ValueError(f"Invalid mode: {mode}. Must be 'w' (write) or 'a' (append)")
182
183 # Sanitize and normalize path
184 path = os.path.normpath(os.path.abspath(path))
289 raise ValueError(f"Invalid mode: {mode}, must be 'w' or 'a'")
185290
186291 # Request permission if permission manager is provided
187292 if permission_manager:
188 operation = "write_file" if mode == 'w' else "append_file"
189293 if not permission_manager.request_permission(
190 operation,
191 {"path": path, "mode": mode}
294 "write_file",
295 {"path": norm_path, "mode": mode}
192296 ):
193 raise PermissionError(f"Permission denied for writing to file: {path}")
297 raise PermissionError(f"Permission denied for writing to file: {norm_path}")
194298
195 # Ensure directory exists
196 directory = os.path.dirname(path)
197 if directory and not os.path.exists(directory):
198 try:
199 os.makedirs(directory, exist_ok=True)
200 except OSError as e:
201 raise IOError(f"Failed to create directory '{directory}': {e}")
202
203 # Write content to file
204299 try:
205 with open(path, mode, encoding='utf-8') as file:
300 # Create directory if it doesn't exist
301 os.makedirs(os.path.dirname(os.path.abspath(norm_path)), exist_ok=True)
302
303 # Write content to file
304 with open(norm_path, mode, encoding='utf-8') as file:
206305 file.write(content)
306
207307 return True
208308 except IOError as e:
209 raise IOError(f"Failed to write to file '{path}': {e}")
309 raise IOError(f"Failed to write to file: {norm_path}, error: {e}")
210310
211311
212312def find_files(
……@@ -229,114 +329,39 @@ def find_files(
229329
230330 Raises:
231331 PermissionError: If permission is denied
232 ValueError: If directory does not exist
233 ValueError: If pattern is invalid
332 ValueError: If directory is invalid
333 OSError: If directory traversal fails
234334 """
235 # Sanitize and normalize directory path
236 directory = os.path.normpath(os.path.abspath(directory))
335 # Validate and normalize directory path
336 norm_dir = _validate_path(directory)
237337
238 # Validate directory
239 if not os.path.isdir(directory):
240 raise ValueError(f"Directory not found: {directory}")
338 # Ensure directory exists
339 if not os.path.isdir(norm_dir):
340 raise ValueError(f"Directory not found: {norm_dir}")
241341
242342 # Request permission if permission manager is provided
243343 if permission_manager:
244344 if not permission_manager.request_permission(
245345 "find_files",
246 {"pattern": pattern, "directory": directory, "recursive": recursive}
346 {"directory": norm_dir, "pattern": pattern, "recursive": recursive}
247347 ):
248 raise PermissionError(f"Permission denied for finding files in: {directory}")
348 raise PermissionError(f"Permission denied for finding files in: {norm_dir}")
249349
250 # Find matching files
251350 try:
351 # Construct the full pattern with directory
252352 if recursive:
253 # Use ** for recursive search if not already in pattern
254 if '**' not in pattern:
255 search_pattern = os.path.join(directory, '**', pattern)
256 else:
257 search_pattern = os.path.join(directory, pattern)
258
259 # Use recursive glob for Python 3.5+
260 matching_files = glob.glob(search_pattern, recursive=True)
353 # Use ** for recursive glob
354 search_pattern = os.path.join(norm_dir, '**', pattern)
355 matching_paths = glob.glob(search_pattern, recursive=True)
261356 else:
262357 # Non-recursive search
263 search_pattern = os.path.join(directory, pattern)
264 matching_files = glob.glob(search_pattern)
265
266 # Sort results for consistent output
267 matching_files.sort()
268
269 return matching_files
270
271 except (ValueError, re.error) as e:
272 raise ValueError(f"Invalid pattern '{pattern}': {e}")
273
274
275def _sanitize_command(command: str) -> str:
276 """
277 Sanitize a command for security.
278
279 Args:
280 command: Command to sanitize
358 search_pattern = os.path.join(norm_dir, pattern)
359 matching_paths = glob.glob(search_pattern)
281360
282 Returns:
283 str: Sanitized command
361 # Filter to only include files (not directories)
362 matching_files = [path for path in matching_paths if os.path.isfile(path)]
284363
285 Raises:
286 ValueError: If command contains dangerous patterns
287 """
288 if not command or not command.strip():
289 raise ValueError("Empty command")
290
291 # Remove leading/trailing whitespace
292 command = command.strip()
293
294 # Disallow certain dangerous patterns
295 dangerous_patterns = [
296 r'[|&;]', # Shell control operators
297 r'>\s*\w+', # Redirections
298 r'<\s*\w+', # Input redirections
299 r'\$\(', # Command substitution
300 r'`', # Backtick command substitution
301 ]
302
303 for pattern in dangerous_patterns:
304 if re.search(pattern, command):
305 raise ValueError(f"Command contains dangerous pattern: {pattern}")
306
307 return command
308
309
310def _sanitize_argument(arg: str) -> str:
311 """
312 Sanitize a command argument for security.
313
314 Args:
315 arg: Argument to sanitize
316
317 Returns:
318 str: Sanitized argument
319
320 Raises:
321 ValueError: If argument contains dangerous patterns
322 """
323 if not isinstance(arg, str):
324 arg = str(arg)
325
326 # Remove leading/trailing whitespace
327 arg = arg.strip()
328
329 # Disallow certain dangerous patterns (similar to command sanitization)
330 dangerous_patterns = [
331 r'[|&;]', # Shell control operators
332 r'>\s*\w+', # Redirections
333 r'<\s*\w+', # Input redirections
334 r'\$\(', # Command substitution
335 r'`', # Backtick command substitution
336 ]
337
338 for pattern in dangerous_patterns:
339 if re.search(pattern, arg):
340 raise ValueError(f"Argument contains dangerous pattern: {pattern}")
341
342 return arg
364 # Convert to absolute paths and sort
365 return sorted([os.path.abspath(path) for path in matching_files])
366 except OSError as e:
367 raise OSError(f"Failed to search for files: {e}")
ollamatools-project/tests/test_system_tools.py +120 -139
……@@ -15,26 +15,16 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')
1515
1616from ollamatools.tools.system_tools import (
1717 execute_command,
18 _sanitize_command,
19 _sanitize_argument,
20 _validate_path,
1821 read_file,
1922 write_file,
20 find_files,
21 _sanitize_command,
22 _sanitize_argument
23 find_files
2324)
2425from ollamatools.security import PermissionManager
2526
2627
27# Setup and teardown for temporary test directory
28@pytest.fixture
29def temp_dir():
30 """Create a temporary directory for file operations tests."""
31 tmp_dir = tempfile.mkdtemp()
32 try:
33 yield tmp_dir
34 finally:
35 shutil.rmtree(tmp_dir)
36
37
3828def test_sanitize_command():
3929 """Test the command sanitization function."""
4030 # Valid commands
……@@ -168,177 +158,168 @@ def test_execute_command_sudo():
168158 )
169159
170160
161def test_validate_path():
162 """Test path validation."""
163 # Test valid paths
164 assert os.path.isabs(_validate_path("test.txt")) # Should convert to absolute path
165 assert os.path.isabs(_validate_path("/tmp/test.txt"))
166 assert os.path.isabs(_validate_path("./test.txt"))
167 assert os.path.isabs(_validate_path("~/test.txt")) # Should expand user
168
169 # Test invalid paths
170 with pytest.raises(ValueError) as excinfo:
171 _validate_path("")
172 assert "Empty path" in str(excinfo.value)
173
174
175# Create a fixture for temporary test directory
176@pytest.fixture
177def temp_dir():
178 """Create a temporary directory for testing file operations."""
179 temp_dir = tempfile.mkdtemp()
180 try:
181 yield temp_dir
182 finally:
183 shutil.rmtree(temp_dir)
184
185
171186def test_read_file(temp_dir):
172 """Test reading file contents."""
187 """Test reading a file."""
173188 # Create a test file
174 test_file_path = os.path.join(temp_dir, "test_file.txt")
175 test_content = "Hello, world!\nThis is a test file."
189 test_file = os.path.join(temp_dir, "test.txt")
190 test_content = "This is a test file.\nWith multiple lines."
176191
177 with open(test_file_path, 'w') as f:
192 with open(test_file, 'w') as f:
178193 f.write(test_content)
179194
180195 # Test reading the file
181 content = read_file(test_file_path)
182 assert content == test_content
196 assert read_file(test_file) == test_content
183197
184 # Test with non-existent file
185 with pytest.raises(FileNotFoundError):
186 read_file(os.path.join(temp_dir, "nonexistent.txt"))
187
188
189def test_read_file_with_max_size(temp_dir):
190 """Test reading file with max size limit."""
191 # Create a test file
192 test_file_path = os.path.join(temp_dir, "large_file.txt")
193 test_content = "X" * 1000 # 1000 bytes
198 # Test with permission manager (granted)
199 mock_pm = MagicMock(spec=PermissionManager)
200 mock_pm.request_permission.return_value = True
194201
195 with open(test_file_path, 'w') as f:
196 f.write(test_content)
202 assert read_file(test_file, permission_manager=mock_pm) == test_content
203 assert mock_pm.request_permission.called
204
205 # Test with permission manager (denied)
206 mock_pm.request_permission.return_value = False
197207
198 # Test reading with sufficient max_size
199 content = read_file(test_file_path, max_size=2000)
200 assert content == test_content
208 with pytest.raises(PermissionError) as excinfo:
209 read_file(test_file, permission_manager=mock_pm)
210 assert "Permission denied" in str(excinfo.value)
201211
202 # Test reading with insufficient max_size
212 # Test with max_size (within limit)
213 file_size = os.path.getsize(test_file)
214 assert read_file(test_file, max_size=file_size + 1) == test_content
215
216 # Test with max_size (exceeding limit)
203217 with pytest.raises(ValueError) as excinfo:
204 read_file(test_file_path, max_size=500)
218 read_file(test_file, max_size=file_size - 1)
205219 assert "exceeds maximum allowed size" in str(excinfo.value)
220
221 # Test with non-existent file
222 with pytest.raises(FileNotFoundError) as excinfo:
223 read_file(os.path.join(temp_dir, "nonexistent.txt"))
224 assert "File not found" in str(excinfo.value)
206225
207226
208def test_read_file_with_permission_manager(temp_dir):
209 """Test reading file with permission manager."""
210 # Create a test file
211 test_file_path = os.path.join(temp_dir, "test_file.txt")
212 test_content = "Hello, world!"
227def test_write_file(temp_dir):
228 """Test writing to a file."""
229 # Test writing to a new file
230 test_file = os.path.join(temp_dir, "write_test.txt")
231 test_content = "This is content to write."
213232
214 with open(test_file_path, 'w') as f:
215 f.write(test_content)
233 assert write_file(test_file, test_content) is True
216234
217 # Create a mock permission manager
218 mock_pm = MagicMock(spec=PermissionManager)
235 # Verify the content was written
236 with open(test_file, 'r') as f:
237 assert f.read() == test_content
219238
220 # Test with permission granted
239 # Test with permission manager (granted)
240 mock_pm = MagicMock(spec=PermissionManager)
221241 mock_pm.request_permission.return_value = True
222 content = read_file(test_file_path, permission_manager=mock_pm)
223 assert content == test_content
242
243 assert write_file(test_file, test_content, permission_manager=mock_pm) is True
224244 assert mock_pm.request_permission.called
225245
226 # Test with permission denied
246 # Test with permission manager (denied)
227247 mock_pm.request_permission.return_value = False
248
228249 with pytest.raises(PermissionError) as excinfo:
229 read_file(test_file_path, permission_manager=mock_pm)
250 write_file(test_file, test_content, permission_manager=mock_pm)
230251 assert "Permission denied" in str(excinfo.value)
231 assert mock_pm.request_permission.called
232
233
234def test_write_file(temp_dir):
235 """Test writing content to a file."""
236 # Define test file path and content
237 test_file_path = os.path.join(temp_dir, "write_test.txt")
238 test_content = "This is test content to write to a file."
239
240 # Test writing to file
241 result = write_file(test_file_path, test_content)
242 assert result is True
243
244 # Verify file content
245 with open(test_file_path, 'r') as f:
246 assert f.read() == test_content
247252
248253 # Test append mode
249 additional_content = "\nThis is additional content."
250 result = write_file(test_file_path, additional_content, mode='a')
251 assert result is True
254 append_content = "\nThis is appended content."
255 assert write_file(test_file, append_content, mode='a') is True
252256
253 # Verify appended content
254 with open(test_file_path, 'r') as f:
255 assert f.read() == test_content + additional_content
257 # Verify content was appended
258 with open(test_file, 'r') as f:
259 assert f.read() == test_content + append_content
256260
257 # Test invalid mode
261 # Test with invalid mode
258262 with pytest.raises(ValueError) as excinfo:
259 write_file(test_file_path, test_content, mode='x')
263 write_file(test_file, test_content, mode='x')
260264 assert "Invalid mode" in str(excinfo.value)
261
262
263def test_write_file_with_new_directory(temp_dir):
264 """Test writing to a file in a non-existent directory."""
265 # Define test file path in a new directory
266 new_dir = os.path.join(temp_dir, "new_directory")
267 test_file_path = os.path.join(new_dir, "test_file.txt")
268 test_content = "This is a test file in a new directory."
269265
270 # Write to file (should create directory)
271 result = write_file(test_file_path, test_content)
272 assert result is True
266 # Test writing to a subdirectory (should create the directory)
267 subdir_file = os.path.join(temp_dir, "subdir", "test.txt")
268 assert write_file(subdir_file, test_content) is True
273269
274 # Verify directory and file were created
275 assert os.path.exists(new_dir)
276 assert os.path.isfile(test_file_path)
277
278 # Verify file content
279 with open(test_file_path, 'r') as f:
270 # Verify the file was created
271 assert os.path.isfile(subdir_file)
272 with open(subdir_file, 'r') as f:
280273 assert f.read() == test_content
281274
282275
283def test_write_file_with_permission_manager(temp_dir):
284 """Test writing file with permission manager."""
285 # Define test file path and content
286 test_file_path = os.path.join(temp_dir, "permission_test.txt")
287 test_content = "Test content for permission test."
288
289 # Create a mock permission manager
290 mock_pm = MagicMock(spec=PermissionManager)
291
292 # Test with permission granted
293 mock_pm.request_permission.return_value = True
294 result = write_file(test_file_path, test_content, permission_manager=mock_pm)
295 assert result is True
296 assert os.path.exists(test_file_path)
297 assert mock_pm.request_permission.called
298
299 # Test with permission denied
300 mock_pm.request_permission.return_value = False
301 with pytest.raises(PermissionError) as excinfo:
302 write_file(test_file_path, test_content, permission_manager=mock_pm)
303 assert "Permission denied" in str(excinfo.value)
304 assert mock_pm.request_permission.called
305
306
307276def test_find_files(temp_dir):
308 """Test finding files matching a pattern."""
277 """Test finding files with a pattern."""
309278 # Create test files
310 test_files = [
311 "file1.txt",
312 "file2.txt",
313 "image1.jpg",
314 "image2.jpg",
315 "nested/file3.txt",
316 "nested/file4.txt",
317 "nested/deep/file5.txt"
279 files = [
280 "test1.txt",
281 "test2.txt",
282 "report.pdf",
283 "subdir/test3.txt",
284 "subdir/nested/test4.txt"
318285 ]
319286
320 # Create directories and files
321 for file_path in test_files:
287 for file_path in files:
322288 full_path = os.path.join(temp_dir, file_path)
323289 os.makedirs(os.path.dirname(full_path), exist_ok=True)
324290 with open(full_path, 'w') as f:
325291 f.write(f"Content of {file_path}")
326292
327 # Test non-recursive search for txt files
328 txt_files = find_files("*.txt", directory=temp_dir, recursive=False)
329 assert len(txt_files) == 2 # Only top-level txt files
330 assert all(os.path.basename(f) in ["file1.txt", "file2.txt"] for f in txt_files)
293 # Test finding all .txt files (recursive)
294 txt_files = find_files("*.txt", temp_dir)
295 assert len(txt_files) == 4
296 assert all(os.path.basename(f).endswith(".txt") for f in txt_files)
297
298 # Test finding all files in the root (non-recursive)
299 root_files = find_files("*", temp_dir, recursive=False)
300 assert len(root_files) == 3 # test1.txt, test2.txt, report.pdf
301 assert all(os.path.dirname(f) == temp_dir for f in root_files)
331302
332 # Test recursive search for txt files
333 all_txt_files = find_files("*.txt", directory=temp_dir, recursive=True)
334 assert len(all_txt_files) == 5 # All txt files
303 # Test with permission manager (granted)
304 mock_pm = MagicMock(spec=PermissionManager)
305 mock_pm.request_permission.return_value = True
306
307 txt_files = find_files("*.txt", temp_dir, permission_manager=mock_pm)
308 assert len(txt_files) == 4
309 assert mock_pm.request_permission.called
310
311 # Test with permission manager (denied)
312 mock_pm.request_permission.return_value = False
335313
336 # Test search for jpg files
337 jpg_files = find_files("*.jpg", directory=temp_dir, recursive=True)
338 assert len(jpg_files) == 2
339 assert all(os.path.basename(f) in ["image1.jpg", "image2.jpg"] for f in jpg_files)
314 with pytest.raises(PermissionError) as excinfo:
315 find_files("*.txt", temp_dir, permission_manager=mock_pm)
316 assert "Permission denied" in str(excinfo.value)
340317
341 # Test search"""
318 # Test with non-existent directory
319 with pytest.raises(ValueError) as excinfo:
320 find_files("*.txt", os.path.join(temp_dir, "nonexistent"))
321 assert "Directory not found" in str(excinfo.value)
322"""
342323Tests for the system tools module.
343324"""
344325import pytest