Maxim Snesarev
33d6bb7ebd
- Updated database configuration to switch from SQLite to PostgreSQL, including changes to alembic.ini, Docker Compose, and environment settings. - Refactored application code to utilize PostgreSQL database adapters, ensuring compatibility with the new database structure. - Enhanced API routes and data handling to support the new database, including adjustments in data models and query logic. - Introduced new job processing mechanisms for full synchronization of AMO CRM entities, leveraging FastStream for background tasks. - Improved logging and error handling across the application to facilitate better monitoring and debugging. - Removed obsolete SQLite adapter files and migrations, streamlining the project structure for PostgreSQL integration.
95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for AMO CRM import functionality
|
|
Verifies that the import script works correctly
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add project root to Python path
|
|
project_root = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from scripts.import_amocrm_data import AMOCRMImporter
|
|
from adapters.postgres.database import SessionLocal, init_db
|
|
from adapters.postgres.models import User, Deal, CustomField
|
|
from utils.config import settings
|
|
|
|
|
|
async def test_import():
|
|
"""Test the import functionality"""
|
|
|
|
print("🧪 Testing AMO CRM Import Script")
|
|
print("=" * 50)
|
|
|
|
# Check configuration
|
|
if not settings.AMO_CRM_ACCESS_TOKEN:
|
|
print("❌ AMO_CRM_ACCESS_TOKEN not set - cannot test import")
|
|
return False
|
|
|
|
try:
|
|
# Test dry run import
|
|
print("Testing dry run import...")
|
|
importer = AMOCRMImporter(
|
|
limit_per_entity=5, # Very small limit for testing
|
|
batch_size=2,
|
|
dry_run=True,
|
|
verbose=True
|
|
)
|
|
|
|
# Import just users for testing
|
|
stats = await importer.import_all_data(['users'])
|
|
|
|
if 'users' in stats:
|
|
user_stats = stats['users']
|
|
print(f"✅ Dry run successful: {user_stats}")
|
|
|
|
if user_stats.fetched > 0:
|
|
print("✅ Successfully fetched data from AMO CRM")
|
|
else:
|
|
print("⚠️ No data fetched - check AMO CRM connection")
|
|
return False
|
|
else:
|
|
print("❌ No stats returned for users")
|
|
return False
|
|
|
|
# Test database connection
|
|
print("\nTesting database connection...")
|
|
db = SessionLocal()
|
|
try:
|
|
# Count existing users
|
|
user_count = db.query(User).count()
|
|
print(f"✅ Database connection OK - {user_count} existing users")
|
|
except Exception as e:
|
|
print(f"❌ Database connection failed: {e}")
|
|
return False
|
|
finally:
|
|
db.close()
|
|
|
|
print("\n✅ All tests passed!")
|
|
print("\nYou can now run the full import with:")
|
|
print(" python scripts/run_import.py test")
|
|
print(" python scripts/run_import.py quick")
|
|
print(" python scripts/import_amocrm_data.py --help")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Test failed: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""Main function"""
|
|
success = asyncio.run(test_import())
|
|
return 0 if success else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit_code = main()
|
|
sys.exit(exit_code)
|