67 lines
2.4 KiB
Python
67 lines
2.4 KiB
Python
|
import asyncio
|
||
|
|
import aiohttp
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from dotenv import load_dotenv
|
||
|
|
|
||
|
|
load_dotenv()
|
||
|
|
|
||
|
|
async def test_vtiger():
|
||
|
|
base_url = os.getenv('VTIGER_URL')
|
||
|
|
username = os.getenv('VTIGER_USERNAME')
|
||
|
|
api_key = os.getenv('VTIGER_API_KEY')
|
||
|
|
|
||
|
|
print(f"🔑 Testing vTiger connection...")
|
||
|
|
print(f"URL: {base_url}")
|
||
|
|
print(f"Username: {username}")
|
||
|
|
|
||
|
|
auth = aiohttp.BasicAuth(username, api_key)
|
||
|
|
|
||
|
|
# Test 1: Connection
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
async with session.get(f"{base_url}/restapi/v1/vtiger/default/me", auth=auth) as response:
|
||
|
|
text = await response.text()
|
||
|
|
print(f"\n✅ Connection test: {response.status}")
|
||
|
|
data = json.loads(text)
|
||
|
|
print(json.dumps(data, indent=2))
|
||
|
|
|
||
|
|
# Test 2: List all modules
|
||
|
|
async with session.get(f"{base_url}/restapi/v1/vtiger/default/listtypes", auth=auth) as response:
|
||
|
|
text = await response.text()
|
||
|
|
data = json.loads(text)
|
||
|
|
if data.get('success'):
|
||
|
|
modules = data.get('result', {}).get('types', [])
|
||
|
|
print(f"\n📋 Available modules ({len(modules)}):")
|
||
|
|
for mod in sorted(modules):
|
||
|
|
if 'sub' in mod.lower() or 'invoice' in mod.lower() or 'order' in mod.lower():
|
||
|
|
print(f" - {mod}")
|
||
|
|
|
||
|
|
# Test 3: Query Subscriptions
|
||
|
|
vtiger_id = '3x760'
|
||
|
|
query = f"SELECT * FROM Subscriptions WHERE account_id='{vtiger_id}';"
|
||
|
|
print(f"\n🔍 Testing query: {query}")
|
||
|
|
async with session.get(
|
||
|
|
f"{base_url}/restapi/v1/vtiger/default/query",
|
||
|
|
params={"query": query},
|
||
|
|
auth=auth
|
||
|
|
) as response:
|
||
|
|
text = await response.text()
|
||
|
|
print(f"Status: {response.status}")
|
||
|
|
data = json.loads(text)
|
||
|
|
print(json.dumps(data, indent=2))
|
||
|
|
|
||
|
|
# Test 4: Query Invoice
|
||
|
|
query2 = f"SELECT * FROM Invoice WHERE account_id='{vtiger_id}';"
|
||
|
|
print(f"\n🔍 Testing Invoice query: {query2}")
|
||
|
|
async with session.get(
|
||
|
|
f"{base_url}/restapi/v1/vtiger/default/query",
|
||
|
|
params={"query": query2},
|
||
|
|
auth=auth
|
||
|
|
) as response:
|
||
|
|
text = await response.text()
|
||
|
|
print(f"Status: {response.status}")
|
||
|
|
data = json.loads(text)
|
||
|
|
print(json.dumps(data, indent=2))
|
||
|
|
|
||
|
|
asyncio.run(test_vtiger())
|