""" Test script to verify directory permissions and file creation capabilities. This script should be run inside the container to verify the fixes. """ import os import tempfile import sys from pathlib import Path def test_directory_permissions(): """Test if we can create directories and files in the expected locations.""" print("=== Directory Permission Test ===") # Test directories that should be writable (use /data on Spaces) test_dirs = [ "/data/advisorai-data/test", "/data/merged/test", "/data/alpaca/test", "/data/crypto-bubbles/test", "/data/finnhub/test", "/data/finviz/test", "/data/marketaux/test" ] success_count = 0 for test_dir in test_dirs: try: # Try to create directory os.makedirs(test_dir, mode=0o755, exist_ok=True) # Try to create a test file test_file = os.path.join(test_dir, "test_write.txt") with open(test_file, 'w') as f: f.write(f"Test write successful at {test_dir}") # Try to read the file back with open(test_file, 'r') as f: content = f.read() # Clean up os.remove(test_file) os.rmdir(test_dir) print(f"āœ… SUCCESS: {test_dir}") success_count += 1 except Exception as e: print(f"āŒ FAILED: {test_dir} - {e}") print(f"\nšŸ“Š Results: {success_count}/{len(test_dirs)} directories passed the test") if success_count == len(test_dirs): print("šŸŽ‰ All directory permission tests PASSED!") return True else: print("āš ļø Some directory permission tests FAILED!") return False def test_user_info(): """Display current user and process information.""" print("\n=== User & Process Information ===") # Check if running on Windows or Unix if hasattr(os, 'getuid'): # Unix/Linux system print(f"Current UID: {os.getuid()}") print(f"Current GID: {os.getgid()}") print(f"Effective UID: {os.geteuid()}") print(f"Effective GID: {os.getegid()}") # Check if running as root if os.getuid() == 0: print("āœ… Running as root user") else: print("ā„¹ļø Running as non-root user") else: # Windows system print("ā„¹ļø Running on Windows system") print(f"Current user: {os.getenv('USERNAME', 'Unknown')}") print(f"Process ID: {os.getpid()}") print(f"Parent Process ID: {os.getppid()}") def test_filebase_connectivity(): """Test if we can load environment variables needed for Filebase.""" print("\n=== Environment Variables Test ===") required_vars = [ 'FILEBASE_ENDPOINT', 'FILEBASE_ACCESS_KEY', 'FILEBASE_SECRET_KEY', 'FILEBASE_BUCKET' ] missing_vars = [] for var in required_vars: value = os.getenv(var) if value: # Don't print sensitive values, just show they exist if 'KEY' in var: print(f"āœ… {var}: ***redacted*** (length: {len(value)})") else: print(f"āœ… {var}: {value}") else: print(f"āŒ {var}: NOT SET") missing_vars.append(var) if missing_vars: print(f"āš ļø Missing environment variables: {missing_vars}") return False else: print("šŸŽ‰ All required environment variables are set!") return True if __name__ == "__main__": print("Starting permission and environment tests...\n") test_user_info() perm_test = test_directory_permissions() env_test = test_filebase_connectivity() print(f"\n=== Final Results ===") if perm_test and env_test: print("šŸŽ‰ ALL TESTS PASSED! The container should work correctly.") sys.exit(0) else: print("āŒ SOME TESTS FAILED! Check the output above for details.") sys.exit(1)