#!/usr/bin/env python3
"""Test CRM models and HubSpot connector merge logic."""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from app import create_app
from app.models import db, CrmContact, CrmDeal, CrmCompany, Company
from app.connectors.hubspot import HubSpotConnector
app = create_app()
with app.app_context():
# Create a test company
company = Company.query.first()
if not company:
company = Company(id='test-company-1', name='Test Corp')
db.session.add(company)
db.session.commit()
print(f"Using company: {company.name} ({company.id})")
# Test merge on CrmContact
conn = HubSpotConnector(company_id=company.id, config={})
# First merge (should insert)
count = conn._merge(CrmContact, company.id, 'hs-123', {
'email': 'test@example.com',
'first_name': 'Test',
'last_name': 'User',
'properties_json': {'test': 'data'}
})
db.session.commit()
print(f" First merge returned: {count}")
# Check it was created
contact = CrmContact.query.filter_by(company_id=company.id, external_id='hs-123').first()
assert contact is not None, "Contact not found after merge"
assert contact.email == 'test@example.com', f"Wrong email: {contact.email}"
print(f" Contact created: {contact.full_name} <{contact.email}>")
# Second merge (should update)
count = conn._merge(CrmContact, company.id, 'hs-123', {
'email': 'updated@example.com',
'first_name': 'Updated',
'last_name': 'Name',
})
db.session.commit()
print(f" Second merge returned: {count}")
# Verify update
contact = CrmContact.query.filter_by(company_id=company.id, external_id='hs-123').first()
assert contact.email == 'updated@example.com', f"Email not updated: {contact.email}"
assert contact.first_name == 'Updated', f"First name not updated: {contact.first_name}"
print(f" Contact updated: {contact.full_name} <{contact.email}>")
# Test CrmDeal merge
count = conn._merge(CrmDeal, company.id, 'hs-deal-456', {
'name': 'Big Deal',
'amount': 50000.0,
'stage': 'appointments_set',
'probability': 0.3,
'contact_id': contact.id,
})
db.session.commit()
print(f" Deal merge returned: {count}")
deal = CrmDeal.query.filter_by(company_id=company.id, external_id='hs-deal-456').first()
assert deal is not None, "Deal not found"
assert deal.contact_id == contact.id, "Deal contact_id not set"
print(f" Deal created: {deal.name} stage: {deal.stage}")
# Test CrmCompany merge
count = conn._merge(CrmCompany, company.id, 'hs-company-789', {
'name': 'Acme Corp',
'domain': 'acme.com',
'industry': 'Technology',
'num_employees': 150,
})
db.session.commit()
print(f" Company merge returned: {count}")
crm_company = CrmCompany.query.filter_by(company_id=company.id, external_id='hs-company-789').first()
assert crm_company is not None, "CRM Company not found"
print(f" CRM Company created: {crm_company.name} ({crm_company.domain})")
# Cleanup test data
CrmContact.query.filter_by(company_id=company.id, external_id='hs-123').delete()
CrmDeal.query.filter_by(company_id=company.id, external_id='hs-deal-456').delete()
CrmCompany.query.filter_by(company_id=company.id, external_id='hs-company-789').delete()
db.session.commit()
print(" Test data cleaned up")
# Test indexes
from sqlalchemy import inspect
insp = inspect(db.engine)
for tbl in ['crm_contacts', 'crm_deals', 'crm_companies']:
indexes = insp.get_indexes(tbl)
names = [c["name"] for c in indexes]
print(f" {tbl} indexes: {names}")
# Verify unique constraint works by attempting duplicate
# (Should raise an IntegrityError on commit)
from sqlalchemy.exc import IntegrityError
conn._merge(CrmContact, company.id, 'dup-test', {'email': 'a@b.com'})
conn._merge(CrmContact, company.id, 'dup-test', {'email': 'c@d.com'})
try:
# This should succeed since external_id is the same (upsert pattern)
db.session.commit()
print(" Upsert on duplicate external_id: OK (update, not duplicate)")
except IntegrityError:
db.session.rollback()
print(" Note: Duplicate external_id raised IntegrityError (expected if using raw insert)")
# Cleanup
CrmContact.query.filter_by(company_id=company.id, external_id='dup-test').delete()
db.session.commit()
print()
print("All connector logic tests passed!")