112 lines
4.4 KiB
Python
112 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test Billing Tools with real configuration from config.json
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import json
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from src.billing import BillingDatabase
|
|
|
|
def load_real_config():
|
|
config_path = os.path.join(os.path.dirname(__file__), 'config.json')
|
|
if not os.path.exists(config_path):
|
|
return None
|
|
with open(config_path, 'r') as f:
|
|
return json.load(f)
|
|
|
|
def test_billing_with_real_config():
|
|
print("Testing BillingDatabase with real configuration...")
|
|
|
|
try:
|
|
config = load_real_config()
|
|
if not config:
|
|
print("❌ config.json not found")
|
|
return False
|
|
|
|
db_configs = {}
|
|
if 'isps' in config:
|
|
for isp_id, isp_data in config['isps'].items():
|
|
if 'billing' in isp_data:
|
|
conf = isp_data['billing']
|
|
if 'alias' not in conf:
|
|
conf['alias'] = isp_data.get('alias', isp_id)
|
|
db_configs[isp_id] = conf
|
|
elif 'billing_databases' in config:
|
|
db_configs = config.get("billing_databases", {})
|
|
|
|
if not db_configs:
|
|
print("❌ No billing configuration (isps or billing_databases) found in config.json")
|
|
return False
|
|
|
|
# Create instance
|
|
billing = BillingDatabase(db_configs)
|
|
|
|
# Test server loading
|
|
servers = billing.list_servers()
|
|
print(f"Loaded servers: {len(servers)}")
|
|
for server in servers:
|
|
print(f" Server {server['server_id']}: {server['host']} ({server['database']})")
|
|
|
|
if not servers:
|
|
print("❌ No servers configured")
|
|
return False
|
|
|
|
# Test connection to first server
|
|
server_id = servers[0]['server_id']
|
|
print(f"\nTesting connection to server {server_id}...")
|
|
|
|
result = billing.check_connection(server_id)
|
|
|
|
if result.get('success'):
|
|
print(f"✅ Connection successful!")
|
|
print(f" Host: {result.get('host')}")
|
|
print(f" Database: {result.get('database')}")
|
|
print(f" Version: {result.get('version')}")
|
|
print(f" Customer count: {result.get('customer_count')}")
|
|
|
|
# Test search if connection successful
|
|
print("\nTesting search customers (first 5)...")
|
|
# Note: Search uses snapshot, so we might need to refresh first if no snapshot exists
|
|
# For test purposes, we will try to refresh snapshot first briefly
|
|
print(" Refusing snapshot first (may take a moment)...")
|
|
billing.refresh_snapshot(server_id)
|
|
|
|
search_result = billing.search_customers('', server_id, 5, 0)
|
|
if search_result.get('success'):
|
|
print(f"✅ Search successful: {len(search_result.get('customers', []))} customers")
|
|
for i, customer in enumerate(search_result.get('customers', [])[:3]): # Show first 3
|
|
print(f" {i+1}. {customer.get('name', 'N/A')} - {customer.get('no_wa', 'N/A')}")
|
|
else:
|
|
print(f"❌ Search failed: {search_result.get('error', 'Unknown error')}")
|
|
|
|
# Test summary
|
|
print("\nTesting customer summary...")
|
|
summary_result = billing.get_customer_summary(server_id)
|
|
if summary_result.get('success'):
|
|
print(f"✅ Summary successful:")
|
|
print(f" Total customers: {summary_result.get('total_customers', 0)}")
|
|
if summary_result.get('status_distribution'):
|
|
for status in summary_result.get('status_distribution', [])[:5]:
|
|
print(f" - {status.get('c_status', 'N/A')}: {status.get('count', 0)} ({status.get('percentage', 0)}%)")
|
|
else:
|
|
print(f"❌ Summary failed: {summary_result.get('error', 'Unknown error')}")
|
|
|
|
else:
|
|
print(f"❌ Connection failed: {result.get('error', 'Unknown error')}")
|
|
|
|
return result.get('success', False)
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error testing billing module: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
if __name__ == '__main__':
|
|
success = test_billing_with_real_config()
|
|
sys.exit(0 if success else 1) |