1+ """
2+ Direct test of file tools with actual provider to verify core functionality.
3+ """
4+
5+ import asyncio
6+ import os
7+ import tempfile
8+ import shutil
9+ from pathlib import Path
10+ import sys
11+
12+ # Add the project root to sys.path
13+ project_root = Path (__file__ ).parent .parent
14+ sys .path .insert (0 , str (project_root ))
15+
16+ from tinyagent .code_agent .providers .modal_provider import ModalProvider
17+ from tinyagent .hooks .logging_manager import LoggingManager
18+
19+
20+ async def test_provider_file_operations_directly ():
21+ """Test provider file operations directly to verify they work."""
22+
23+ # Create temp directory for testing
24+ temp_dir = tempfile .mkdtemp ()
25+ dummy_path = os .path .join (temp_dir , "dummy.txt" )
26+
27+ try :
28+ print ("🧪 Testing provider file operations directly..." )
29+ print (f"📁 Test directory: { temp_dir } " )
30+ print (f"📄 Dummy file: { dummy_path } " )
31+
32+ # Create provider with local execution
33+ log_manager = LoggingManager ()
34+ provider = ModalProvider (log_manager = log_manager , local_execution = True )
35+
36+ print ("\n 1️⃣ Testing provider write_file..." )
37+ write_result = await provider .write_file (dummy_path , "Hello, World!\n This is a test file." )
38+ print (f"✅ Write result: { write_result } " )
39+
40+ if write_result .get ("success" ):
41+ # Verify file exists and has correct content
42+ assert os .path .exists (dummy_path ), "File should exist after write"
43+ with open (dummy_path , 'r' ) as f :
44+ content = f .read ()
45+ assert "Hello, World!" in content , "File should contain written content"
46+ print (f"✅ File created with content: { repr (content )} " )
47+
48+ print ("\n 2️⃣ Testing provider read_file..." )
49+ read_result = await provider .read_file (dummy_path )
50+ print (f"✅ Read result: { read_result } " )
51+ assert read_result .get ("success" ), "Read should succeed"
52+ assert "Hello, World!" in read_result .get ("content" , "" ), "Read should return file content"
53+
54+ print ("\n 3️⃣ Testing provider update_file..." )
55+ update_result = await provider .update_file (dummy_path , "Hello, World!" , "Hello, TinyAgent!" )
56+ print (f"✅ Update result: { update_result } " )
57+
58+ if update_result .get ("success" ):
59+ # Verify file was updated
60+ with open (dummy_path , 'r' ) as f :
61+ updated_content = f .read ()
62+ assert "Hello, TinyAgent!" in updated_content , "File should contain updated content"
63+ assert "Hello, World!" not in updated_content , "Old content should be replaced"
64+ print (f"✅ File updated with content: { repr (updated_content )} " )
65+
66+ print ("\n 4️⃣ Testing provider search_files..." )
67+ search_result = await provider .search_files ("TinyAgent" , temp_dir )
68+ print (f"✅ Search result: { search_result } " )
69+ assert search_result .get ("success" ), "Search should succeed"
70+
71+ print ("\n 🎉 All provider file operations work correctly!" )
72+ return True
73+ else :
74+ print (f"❌ Write operation failed: { write_result } " )
75+ return False
76+
77+ except Exception as e :
78+ print (f"❌ Test failed: { e } " )
79+ import traceback
80+ traceback .print_exc ()
81+ return False
82+ finally :
83+ # Clean up
84+ shutil .rmtree (temp_dir )
85+ print (f"🧹 Cleaned up test directory" )
86+
87+
88+ async def test_provider_current_directory ():
89+ """Test provider file operations in current directory."""
90+
91+ current_dir = os .getcwd ()
92+ dummy_path = os .path .join (current_dir , "dummy.txt" )
93+
94+ try :
95+ print ("\n 🌍 Testing provider file operations in current directory..." )
96+ print (f"📁 Current directory: { current_dir } " )
97+ print (f"📄 Dummy file: { dummy_path } " )
98+
99+ # Create provider with local execution
100+ log_manager = LoggingManager ()
101+ provider = ModalProvider (log_manager = log_manager , local_execution = True )
102+
103+ print ("\n 1️⃣ Testing provider write_file in current directory..." )
104+ write_result = await provider .write_file (dummy_path , "Test content from file tools\n This verifies file tools work out of the box!" )
105+ print (f"✅ Write result: { write_result } " )
106+
107+ if write_result .get ("success" ):
108+ # Verify file exists and has correct content
109+ assert os .path .exists (dummy_path ), "File should exist after write"
110+ with open (dummy_path , 'r' ) as f :
111+ content = f .read ()
112+ assert "Test content" in content , "File should contain written content"
113+ print (f"✅ File created successfully with content: { repr (content )} " )
114+
115+ print ("\n 2️⃣ Testing provider read_file in current directory..." )
116+ read_result = await provider .read_file (dummy_path )
117+ print (f"✅ Read result: { read_result } " )
118+ assert read_result .get ("success" ), "Read should succeed"
119+ assert "Test content" in read_result .get ("content" , "" ), "Read should return file content"
120+
121+ print ("\n 🎉 Provider file operations work out of the box in current directory!" )
122+ return True
123+ else :
124+ print (f"❌ Write operation failed: { write_result } " )
125+ return False
126+
127+ except Exception as e :
128+ print (f"❌ Test failed: { e } " )
129+ import traceback
130+ traceback .print_exc ()
131+ return False
132+ finally :
133+ # Clean up
134+ if os .path .exists (dummy_path ):
135+ os .remove (dummy_path )
136+ print (f"🧹 Cleaned up dummy.txt" )
137+
138+
139+ async def main ():
140+ """Run all direct provider tests."""
141+ print ("🚀 Starting direct provider file operations tests..." )
142+
143+ # Test 1: Direct provider testing
144+ test1_success = await test_provider_file_operations_directly ()
145+
146+ # Test 2: Current directory provider testing (real world scenario)
147+ test2_success = await test_provider_current_directory ()
148+
149+ if test1_success and test2_success :
150+ print ("\n ✅ All direct provider tests passed! File operations work at the provider level." )
151+ print ("🎯 Core file operations are functional ✓" )
152+ else :
153+ print ("\n ❌ Some direct provider tests failed. File operations need fixing at the provider level." )
154+
155+
156+ if __name__ == "__main__" :
157+ asyncio .run (main ())
0 commit comments