137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify the admin sync endpoint works correctly
|
|
This script tests the /admin/sync-seen endpoint.
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
from typing import List
|
|
|
|
# Configuration - CHANGE THIS TO YOUR FASTAPI HOST
|
|
API_BASE = "http://localhost:8000" # Change to your FastAPI host
|
|
API_ENDPOINT = f"{API_BASE}/admin/sync-seen"
|
|
|
|
def test_admin_sync():
|
|
"""Test the admin sync endpoint"""
|
|
|
|
print("🧪 Testing Admin Sync Endpoint")
|
|
print("=" * 50)
|
|
print(f"🌐 API endpoint: {API_ENDPOINT}")
|
|
print()
|
|
|
|
# Test data
|
|
test_cases = [
|
|
{
|
|
"name": "Single account, multiple posts",
|
|
"username": "test_user_1",
|
|
"ig_post_ids": ["1234567890123456789", "1234567890123456790", "1234567890123456791"]
|
|
},
|
|
{
|
|
"name": "Single account, single post",
|
|
"username": "test_user_2",
|
|
"ig_post_ids": ["9876543210987654321"]
|
|
},
|
|
{
|
|
"name": "Empty post list",
|
|
"username": "test_user_3",
|
|
"ig_post_ids": []
|
|
}
|
|
]
|
|
|
|
for i, case in enumerate(test_cases, 1):
|
|
print(f"📝 Test {i}: {case['name']}")
|
|
print(f" - Username: @{case['username']}")
|
|
print(f" - Post IDs: {len(case['ig_post_ids'])} posts")
|
|
|
|
try:
|
|
payload = {
|
|
"username": case["username"],
|
|
"ig_post_ids": case["ig_post_ids"]
|
|
}
|
|
|
|
print(f" 📤 Sending request...")
|
|
response = requests.post(API_ENDPOINT, json=payload, timeout=30)
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
marked_count = result.get("marked_seen", 0)
|
|
print(f" ✅ Success! marked_seen: {marked_count}")
|
|
|
|
if marked_count == len(case["ig_post_ids"]):
|
|
print(f" ✅ All posts marked correctly")
|
|
else:
|
|
print(f" ⚠️ Expected {len(case['ig_post_ids'])}, got {marked_count}")
|
|
else:
|
|
print(f" ❌ HTTP {response.status_code}: {response.text}")
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
print(f" ❌ Connection error. Is the API running at {API_BASE}?")
|
|
except requests.exceptions.Timeout:
|
|
print(f" ❌ Timeout error")
|
|
except requests.exceptions.RequestException as e:
|
|
print(f" ❌ Request error: {e}")
|
|
except Exception as e:
|
|
print(f" ❌ Unexpected error: {e}")
|
|
|
|
print()
|
|
|
|
print("🎉 Admin sync test completed!")
|
|
return True
|
|
|
|
def test_health_check():
|
|
"""Test if the API is running"""
|
|
|
|
print("🏥 Testing API Health Check")
|
|
print("=" * 30)
|
|
|
|
try:
|
|
health_url = f"{API_BASE}/health"
|
|
response = requests.get(health_url, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
print(f"✅ API is running at {API_BASE}")
|
|
return True
|
|
else:
|
|
print(f"❌ API health check failed: {response.status_code}")
|
|
return False
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
print(f"❌ Cannot connect to API at {API_BASE}")
|
|
print(" Make sure the FastAPI server is running:")
|
|
print(" docker-compose up")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Health check error: {e}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
print("🧪 Admin Sync Endpoint Test Suite")
|
|
print("=" * 60)
|
|
|
|
# First check if API is running
|
|
if not test_health_check():
|
|
print("\n❌ API is not running. Please start the FastAPI server first.")
|
|
print(" docker-compose up")
|
|
exit(1)
|
|
|
|
print()
|
|
|
|
# Test the admin sync endpoint
|
|
success = test_admin_sync()
|
|
|
|
print("=" * 60)
|
|
if success:
|
|
print("✅ Admin sync test passed!")
|
|
print("\n📝 Summary:")
|
|
print("- Admin endpoint is working correctly")
|
|
print("- Posts are being marked as seen in Redis")
|
|
print("- Endpoint handles various input sizes")
|
|
print("- Ready to use with sync_seen_from_table.py")
|
|
else:
|
|
print("❌ Admin sync test failed!")
|
|
print("\nTroubleshooting:")
|
|
print("1. Check that the API is running")
|
|
print("2. Verify Redis connection")
|
|
print("3. Check admin_sync.py is properly imported")
|
|
print("4. Ensure the endpoint is accessible") |