nodes/machinepainting-ComfyUI_DropSendNode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cleanup

Browse Source
This commit is contained in:
adamhaley
2025-08-20 18:33:56 -07:00
parent 66d4104f6d
commit c7665e8d6b
8 changed files with 55 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files
RUNPOD_SETUP.md
-67
View File
@@ -1,67 +0,0 @@
# RunPod OAuth Setup Guide
## Quick Setup for RunPod
The DropSend node now automatically detects RunPod environments, but for OAuth to work properly, you may need to configure your public URL.
### Option 1: Automatic Detection (Try This First)
The node will try to automatically detect your RunPod URL. Just run the OAuth flow and check the console for messages like:
```
[OAuthHandler] Detected server URL from request: https://your-runpod-url.runpod.net
```
### Option 2: Manual Configuration (If Auto-Detection Fails)
If the automatic detection doesn't work, set one of these environment variables:
#### Method A: Set PUBLIC_URL Environment Variable
```bash
export PUBLIC_URL="https://your-runpod-id.runpod.net:8188"
```
#### Method B: Set COMFYUI_PUBLIC_URL Environment Variable
```bash
export COMFYUI_PUBLIC_URL="https://your-runpod-id.runpod.net:8188"
```
### How to Find Your RunPod URL
1. **From RunPod Dashboard**: Look for your pod's public URL
2. **From Browser**: Copy the URL you use to access ComfyUI
3. **From Console**: The node will log detected URLs during OAuth
### Example URLs
- `https://abc123-8188.proxy.runpod.net`
- `https://xyz789.runpod.net:8188`
- `https://your-custom-domain.com:8188`
### Testing the Configuration
1. Run the DropboxSetupNode
2. Check console output for detected URL
3. If URL looks wrong, set the environment variable and restart ComfyUI
### Troubleshooting
**Problem**: OAuth callback fails with "redirect_uri mismatch"
**Solution**: Check that the detected URL in the console matches how you access ComfyUI
**Problem**: Automatic detection shows localhost
**Solution**: Set PUBLIC_URL or COMFYUI_PUBLIC_URL environment variable
**Problem**: Still getting localhost after setting environment variables
**Solution**: Restart ComfyUI after setting environment variables
### Storage Method Recommendations for RunPod
1. **`keyring`**: Now works with encrypted file storage (recommended)
2. **`display_only`**: Best for temporary pods - copy credentials to environment
3. **`env_file`**: Good for persistent storage volumes
## Tips
- Use `display_only` for maximum flexibility in cloud environments
- Set environment variables in your RunPod template for automatic configuration
generate_base64.py
-10
View File
@@ -1,10 +0,0 @@
# ComfyUI_DropSendNode/generate_base64.py
import base64
app_key = input("Enter your App Key: ")
app_secret = input("Enter your App Secret: ")
credentials = f"{app_key}:{app_secret}"
encoded = base64.b64encode(credentials.encode()).decode()
print(f"\nBasic {encoded}")
get_refresh_token.py
-27
View File
@@ -1,27 +0,0 @@
# ComfyUI_DropSendNode/get_refresh_token.py
import base64
import requests
APP_KEY = input("Enter your APP KEY: ")
APP_SECRET = input("Enter your APP SECRET: ")
AUTH_CODE = input("Paste the Dropbox authorization code: ")
auth_header = base64.b64encode(f"{APP_KEY}:{APP_SECRET}".encode()).decode()
res = requests.post(
"https://api.dropbox.com/oauth2/token",
data={
"code": AUTH_CODE,
"grant_type": "authorization_code",
"redirect_uri": "https://localhost"
},
headers={
"Authorization": f"Basic {auth_header}",
"Content-Type": "application/x-www-form-urlencoded"
}
)
if res.ok:
print("\nYour refresh token is:\n", res.json().get("refresh_token"))
else:
print("Error:", res.text)
oauth_handler.py
+1 -5
View File
@@ -70,12 +70,8 @@ class OAuthCallbackHandler:
success_message = "Dropbox connected successfully! Credentials saved to .env file."
elif self.storage_method == "display_only":
# Display credentials for manual copying and create completion marker
# Display credentials for manual copying
import os
node_dir = os.path.dirname(__file__)
display_marker_path = os.path.join(node_dir, ".dropbox_display_complete")
with open(display_marker_path, "w") as f:
f.write("display_only_setup_completed")
# Print credentials to console for easy copy/paste
print("\n" + "=" * 80)
requirements.txt
-2
View File
@@ -2,5 +2,3 @@ requests
python-dotenv
watchdog
dropbox
keyring
keyrings.alt
setup_dropbox_node.py
+10 -33
View File
@@ -68,13 +68,6 @@ class DropboxSetupNode:
print(f"[DropboxSetup] Removing .env file: {env_path}")
os.remove(env_path)
# Clear display_only marker if it exists
display_marker_path = os.path.join(node_dir, ".dropbox_display_complete")
if os.path.exists(display_marker_path):
print(f"[DropboxSetup] Removing display_only marker: {display_marker_path}")
os.remove(display_marker_path)
# Send WebSocket message to trigger ComfyUI refresh after clearing credentials
try:
from server import PromptServer
@@ -95,31 +88,19 @@ class DropboxSetupNode:
"result": (message,)
}
# Check for display_only environment variables (user set them up after OAuth flow)
display_only_env_set = all([
# Check for environment variables (from any source)
env_vars_set = all([
os.getenv("DROPBOX_APP_KEY"),
os.getenv("DROPBOX_APP_SECRET"),
os.getenv("DROPBOX_REFRESH_TOKEN")
])
if display_only_env_set:
# Check if this came from display_only flow
node_dir = os.path.dirname(__file__)
display_marker_path = os.path.join(node_dir, ".dropbox_display_complete")
if os.path.exists(display_marker_path):
message = "Dropbox connected using environment variables (display_only setup). Ready to upload files."
print(f"[DropboxSetup] {message}")
return {
"ui": {"text": [message]},
"result": (message,)
}
else:
# Legacy environment variables (not from display_only)
message = "Dropbox credentials found in system environment variables. Ready to upload files."
print(f"[DropboxSetup] {message}")
return {
"ui": {"text": [message]},
"result": (message,)
}
if env_vars_set:
message = "Dropbox credentials found in system environment variables. Ready to upload files."
print(f"[DropboxSetup] {message}")
return {
"ui": {"text": [message]},
"result": (message,)
}
# Check for RunPod secrets
runpod_env_set = all([
@@ -199,11 +180,7 @@ class DropboxSetupNode:
message = "Dropbox connected successfully! Credentials saved to .env file."
elif storage_method == "display_only":
# Display credentials for manual copying and create completion marker
node_dir = os.path.dirname(__file__)
display_marker_path = os.path.join(node_dir, ".dropbox_display_complete")
with open(display_marker_path, "w") as f:
f.write("display_only_setup_completed")
# Display credentials for manual copying
message = f"""Dropbox Connected Successfully!
test_dropbox_auth.py
-140
View File
@@ -1,140 +0,0 @@
#!/usr/bin/env python3
# test_dropbox_auth_simple.py - Non-interactive testing script for DropboxAuthManager
from dropbox_auth_manager import DropboxAuthManager
def test_oauth_url_generation():
"""Test OAuth URL generation"""
print("🧪 TEST 1: OAuth URL generation")
print("-" * 40)
# Use a dummy app key for testing
test_app_key = "test_app_key_12345"
try:
auth = DropboxAuthManager(app_key=test_app_key)
oauth_url = auth.get_oauth_url()
print(f"✅ Generated OAuth URL:")
print(f" {oauth_url}")
expected_url = f"https://www.dropbox.com/oauth2/authorize?response_type=code&client_id={test_app_key}&token_access_type=offline"
if oauth_url == expected_url:
print("✅ URL format is correct")
return True
else:
print("❌ URL format mismatch")
return False
except Exception as e:
print(f"❌ Error generating OAuth URL: {e}")
return False
def test_stored_credentials():
"""Test retrieving stored credentials"""
print("\n🧪 TEST 2: Check for stored credentials")
print("-" * 40)
try:
auth = DropboxAuthManager()
print(f"App key stored: {bool(auth.app_key)}")
print(f"App secret stored: {bool(auth.app_secret)}")
print(f"Refresh token stored: {bool(auth.refresh_token)}")
print(f"Is connected: {auth.is_connected()}")
if auth.is_connected():
print("✅ Stored credentials found - testing access token...")
try:
access_token = auth.get_access_token()
print(f"✅ Got access token: {access_token[:10]}...")
return True
except Exception as e:
print(f"❌ Error getting access token: {e}")
return False
else:
print("ℹ️ No stored credentials found (this is expected for first run)")
return True
except Exception as e:
print(f"❌ Error checking stored credentials: {e}")
return False
def test_auth_code_exchange():
"""Test auth code exchange with placeholder values"""
print("\n🧪 TEST 3: Auth code exchange test (with dummy values)")
print("-" * 40)
# These are dummy values - replace with real ones to test
test_app_key = "your_app_key_here"
test_app_secret = "your_app_secret_here"
test_auth_code = "your_auth_code_here"
if test_app_key == "your_app_key_here":
print("ℹ️ Skipping auth code exchange test - placeholder values detected")
print("ℹ️ To test this, replace the dummy values in this function with real credentials")
return True
try:
auth = DropboxAuthManager(test_app_key, test_app_secret)
print(f"Created DropboxAuthManager with app_key: {bool(auth.app_key)}, app_secret: {bool(auth.app_secret)}")
auth.exchange_auth_code(test_auth_code)
print("✅ Auth code exchange successful")
print("✅ Credentials stored successfully")
return True
except Exception as e:
print(f"❌ Error during auth code exchange: {e}")
return False
def test_manager_initialization():
"""Test DropboxAuthManager initialization with different parameters"""
print("\n🧪 TEST 4: Manager initialization tests")
print("-" * 40)
try:
# Test 1: Empty initialization
auth1 = DropboxAuthManager()
print(f"✅ Empty init - app_key: {auth1.app_key}, app_secret: {auth1.app_secret}")
# Test 2: With parameters
auth2 = DropboxAuthManager("test_key", "test_secret")
print(f"✅ With params - app_key: {auth2.app_key}, app_secret: {auth2.app_secret}")
# Test 3: With empty strings (this was causing the bug)
auth3 = DropboxAuthManager("", "")
print(f"✅ Empty strings - app_key: '{auth3.app_key}', app_secret: '{auth3.app_secret}'")
return True
except Exception as e:
print(f"❌ Error during initialization tests: {e}")
return False
def main():
print("🚀 Dropbox Auth Manager Test Suite (Non-Interactive)")
print("=" * 60)
# Run all tests
tests = [
test_oauth_url_generation,
test_stored_credentials,
test_manager_initialization,
test_auth_code_exchange
]
results = []
for test in tests:
result = test()
results.append(result)
print("\n" + "=" * 60)
print("📊 TEST SUMMARY:")
passed = sum(results)
total = len(results)
print(f"✅ Passed: {passed}/{total}")
if passed == total:
print("🎉 All tests passed!")
else:
print("⚠️ Some tests failed - check output above")
if __name__ == "__main__":
main()
test_dropbox_auth_simple.py → tests/test_dropbox_auth.py
+44 -42
View File
@@ -8,8 +8,11 @@ Usage:
This script tests the DropboxAuthManager without requiring user input.
"""
from dropbox_auth_manager import DropboxAuthManager
import sys
import os
# Add parent directory to path so we can import from the main module
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from dropbox_auth_manager import DropboxAuthManager
def test_basic_functionality():
"""Test basic DropboxAuthManager functionality"""
@@ -21,44 +24,43 @@ def test_basic_functionality():
# Test initialization without credentials
print("\n1. Testing initialization without credentials...")
auth = DropboxAuthManager()
print(f"✅ DropboxAuthManager initialized")
print(f"PASS: DropboxAuthManager initialized")
print(f" is_connected(): {auth.is_connected()}")
# Test initialization with test credentials
print("\n2. Testing initialization with test credentials...")
test_auth = DropboxAuthManager(app_key="test_key_123", app_secret="test_secret_456")
print(f"✅ DropboxAuthManager initialized with test credentials")
print(f"PASS: DropboxAuthManager initialized with test credentials")
print(f" app_key: {test_auth.app_key}")
print(f" is_connected(): {test_auth.is_connected()}")
# Test OAuth URL generation
print("\n3. Testing OAuth URL generation...")
oauth_url = test_auth.get_oauth_url()
print(f"✅ Basic OAuth URL: {oauth_url[:80]}...")
print(f"PASS: Basic OAuth URL: {oauth_url[:80]}...")
# Test OAuth URL with redirect_uri
callback_url = "http://localhost:8188/oauth/dropbox/callback"
oauth_url_with_redirect = test_auth.get_oauth_url(redirect_uri=callback_url, state="test-session-123")
print(f"✅ OAuth URL with redirect: {oauth_url_with_redirect[:100]}...")
# Test OAuth URL with manual flow (no redirect_uri needed)
oauth_url_manual = test_auth.get_oauth_url(state="test-session-123", force_reapprove=True)
print(f"PASS: Manual OAuth URL: {oauth_url_manual[:100]}...")
# Test that redirect_uri is included
if "redirect_uri=" in oauth_url_with_redirect and callback_url in oauth_url_with_redirect:
print("✅ redirect_uri properly included in OAuth URL")
# Test that manual OAuth URL is properly formatted
if "response_type=code" in oauth_url_manual and "token_access_type=offline" in oauth_url_manual:
print("PASS: OAuth URL properly formatted for manual flow")
else:
print("❌ redirect_uri not found in OAuth URL")
print("FAIL: OAuth URL missing required parameters")
# Test that state is included
if "state=test-session-123" in oauth_url_with_redirect:
print("✅ state parameter properly included in OAuth URL")
# Test that state is included when provided
if "state=test-session-123" in oauth_url_manual:
print("PASS: state parameter properly included in OAuth URL")
else:
print("❌ state parameter not found in OAuth URL")
print("FAIL: state parameter not found in OAuth URL")
print("\n" + "=" * 50)
print("✅ All basic functionality tests passed!")
print("PASS: All basic functionality tests passed!")
print("=" * 50)
except Exception as e:
print(f"❌ Error during basic functionality test: {e}")
print(f"FAIL: Error during basic functionality test: {e}")
raise
def test_error_handling():
@@ -73,24 +75,24 @@ def test_error_handling():
auth_no_key = DropboxAuthManager()
try:
auth_no_key.get_oauth_url()
print("❌ Should have raised ValueError")
print("FAIL: Should have raised ValueError")
except ValueError as e:
print(f"✅ Correctly raised ValueError: {e}")
print(f"PASS: Correctly raised ValueError: {e}")
# Test auth code exchange without credentials
print("\n2. Testing auth code exchange without credentials...")
try:
auth_no_key.exchange_auth_code_raw("test_code")
print("❌ Should have raised ValueError")
print("FAIL: Should have raised ValueError")
except ValueError as e:
print(f"✅ Correctly raised ValueError: {e}")
print(f"PASS: Correctly raised ValueError: {e}")
print("\n" + "=" * 50)
print("✅ All error handling tests passed!")
print("PASS: All error handling tests passed!")
print("=" * 50)
except Exception as e:
print(f"❌ Unexpected error during error handling test: {e}")
print(f"FAIL: Unexpected error during error handling test: {e}")
raise
def test_token_exchange_retry_logic():
@@ -107,46 +109,46 @@ def test_token_exchange_retry_logic():
print("\n1. Testing exchange_auth_code_raw method signature...")
# Test method exists and accepts parameters (will fail with API call, but that's expected)
try:
test_auth.exchange_auth_code_raw("fake_code", redirect_uri="http://localhost:8188/oauth/dropbox/callback")
except Exception as e:
if "400" in str(e) or "requests" in str(e) or "HTTPSConnectionPool" in str(e):
print(f"✅ Method executed correctly (expected API failure): {type(e).__name__}")
else:
print(f"❌ Unexpected error type: {e}")
# Test method without redirect_uri
try:
test_auth.exchange_auth_code_raw("fake_code")
except Exception as e:
if "400" in str(e) or "requests" in str(e) or "HTTPSConnectionPool" in str(e):
print(f"✅ Method executed correctly without redirect_uri (expected API failure): {type(e).__name__}")
print(f"PASS: Method executed correctly (expected API failure): {type(e).__name__}")
else:
print(f"❌ Unexpected error type: {e}")
print(f"FAIL: Unexpected error type: {e}")
# Test method with redirect_uri (for backward compatibility)
try:
test_auth.exchange_auth_code_raw("fake_code", redirect_uri="http://localhost:8188/oauth/dropbox/callback")
except Exception as e:
if "400" in str(e) or "requests" in str(e) or "HTTPSConnectionPool" in str(e):
print(f"PASS: Method with redirect_uri executed correctly (expected API failure): {type(e).__name__}")
else:
print(f"FAIL: Unexpected error type: {e}")
print("\n" + "=" * 50)
print("✅ Token exchange retry logic tests completed!")
print("PASS: Token exchange retry logic tests completed!")
print("=" * 50)
except Exception as e:
print(f"❌ Error during retry logic test: {e}")
print(f"FAIL: Error during retry logic test: {e}")
raise
def main():
"""Run all tests"""
print("🧪 Starting DropboxAuthManager Tests")
print("📝 This script tests the auth manager without making real API calls")
print("Starting DropboxAuthManager Tests")
print("This script tests the auth manager without making real API calls")
try:
test_basic_functionality()
test_error_handling()
test_token_exchange_retry_logic()
print("\n🎉 All tests completed successfully!")
print("💡 To test with real Dropbox credentials, use the interactive flow in ComfyUI")
print("\nAll tests completed successfully!")
print("To test with real Dropbox credentials, use the interactive flow in ComfyUI")
except Exception as e:
print(f"\n💥 Test failed: {e}")
print(f"\nTest failed: {e}")
return 1
return 0