"""ServiceTitan connector tests.
Tests OAuth2 connect, sync (customers → CrmContact, jobs → CrmDeal,
companies → CrmCompany), status, and disconnect with mocked HTTP calls.
"""
import os
import sys
import unittest
from unittest.mock import patch, MagicMock
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
os.environ.setdefault("DATABASE_URL", "sqlite:///test_servicetitan.db")
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing")
from app import create_app
from app.models import db, Company, CrmContact, CrmDeal, CrmCompany
from app.connectors.servicetitan import ServiceTitanConnector
class ServiceTitanConnectorTests(unittest.TestCase):
"""Tests for ServiceTitanConnector with mocked requests."""
@classmethod
def setUpClass(cls) -> None:
cls.app = create_app()
cls.app.config["TESTING"] = True
cls.app.config["RATELIMIT_ENABLED"] = False
with cls.app.app_context():
db.drop_all()
db.create_all()
company = Company(name="Test Roofing Co")
db.session.add(company)
db.session.commit()
cls.company_id = company.id
def setUp(self) -> None:
"""Clear CRM tables before each test."""
with self.app.app_context():
CrmContact.query.delete()
CrmDeal.query.delete()
CrmCompany.query.delete()
db.session.commit()
@classmethod
def tearDownClass(cls) -> None:
with cls.app.app_context():
db.session.remove()
db.drop_all()
try:
os.remove("test_servicetitan.db")
except FileNotFoundError:
pass
# -- Helper factories -----------------------------------------------------
def _make_connector(self, **config_overrides):
"""Create a connector instance with default test config."""
from app.connectors.servicetitan import ServiceTitanConnector
config = {
"client_id": "test_client_id",
"client_secret": "test_client_secret",
"app_key": "test_app_key",
"tenant_id": "tenant-jpm-v2",
}
config.update(config_overrides)
return ServiceTitanConnector(
company_id=self.company_id, config=config, connector_id="st-1"
)
# -- Registration tests --------------------------------------------------
def test_connector_in_registry(self):
"""ServiceTitan is registered in the connector registry."""
from app.connectors import get_available_connectors
services = [c["service"] for c in get_available_connectors()]
self.assertIn("servicetitan", services)
def test_connector_metadata(self):
"""Connector metadata has correct fields."""
from app.connectors import get_available_connectors
st = next(
(c for c in get_available_connectors() if c["service"] == "servicetitan"),
None,
)
self.assertIsNotNone(st)
self.assertEqual(st["name"], "ServiceTitan")
self.assertEqual(st["category"], "crm")
self.assertIn("client_id", st["auth_fields"])
self.assertIn("client_secret", st["auth_fields"])
self.assertIn("app_key", st["auth_fields"])
def test_connector_class_available(self):
"""Connector class is retrievable from registry."""
from app.connectors import get_connector_class
cls = get_connector_class("servicetitan")
self.assertIsNotNone(cls)
self.assertEqual(cls._SERVICE, "servicetitan")
# -- connect() tests -----------------------------------------------------
@patch("app.connectors.servicetitan.requests.post")
def test_connect_success(self, mock_post):
"""connect() obtains token and validates API."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {
"access_token": "new_token",
"expires_in": 3600,
"token_type": "Bearer",
}
connector = self._make_connector()
with self.app.app_context():
with patch.object(connector, "_st_request", return_value={}):
result = connector.connect()
self.assertEqual(result["status"], "connected")
self.assertTrue(connector._connected)
@patch("app.connectors.servicetitan.requests.post")
def test_connect_missing_client_id(self, mock_post):
"""connect() fails when client_id is missing."""
connector = self._make_connector(client_id="")
with self.app.app_context():
result = connector.connect()
self.assertEqual(result["status"], "error")
self.assertIn("client_id", result["error"])
@patch("app.connectors.servicetitan.requests.post")
def test_connect_missing_app_key(self, mock_post):
"""connect() fails when app_key is missing."""
connector = self._make_connector(app_key="")
with self.app.app_context():
result = connector.connect()
self.assertEqual(result["status"], "error")
self.assertIn("app_key", result["error"])
@patch("app.connectors.servicetitan.requests.post")
def test_connect_token_error(self, mock_post):
"""connect() returns error when token endpoint fails."""
mock_post.return_value.status_code = 401
mock_post.return_value.raise_for_status.side_effect = Exception("Unauthorized")
connector = self._make_connector()
with self.app.app_context():
result = connector.connect()
self.assertEqual(result["status"], "error")
self.assertFalse(connector._connected)
# -- disconnect() tests --------------------------------------------------
def test_disconnect_clears_state(self):
"""disconnect() clears token cache and sets _connected=False."""
connector = self._make_connector()
connector._connected = True
connector._access_token = "some_token"
connector._token_expires_at = None
connector.config["access_token"] = "some_token"
connector.config["refresh_token"] = "refresh_val"
with self.app.app_context():
result = connector.disconnect()
self.assertEqual(result["status"], "disconnected")
self.assertFalse(connector._connected)
self.assertIsNone(connector._access_token)
self.assertNotIn("access_token", connector.config)
self.assertNotIn("refresh_token", connector.config)
# -- sync() tests --------------------------------------------------------
@patch("app.connectors.servicetitan.requests.post")
def test_sync_creates_contacts(self, mock_post):
"""sync() merges ServiceTitan customers into CrmContact."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {
"access_token": "tok",
"expires_in": 3600,
}
connector = self._make_connector()
mock_customers = {
"data": [
{
"id": "cust_1",
"first_name": "John",
"last_name": "Doe",
"email": "john@example.com",
"phone": "555-1234",
"company_name": "Doe Roofing",
},
{
"id": "cust_2",
"first_name": "Jane",
"last_name": "Smith",
"email": "jane@example.com",
"phone": "555-5678",
"company_name": "Smith HVAC",
},
]
}
mock_jobs = {"data": []}
mock_companies = {"data": []}
with self.app.app_context():
connector._connected = True
connector._access_token = "tok"
# Patch _fetch_all so each endpoint returns the right data
def fake_fetch(endpoint, extra_params=None):
if endpoint == "/customers":
return mock_customers["data"]
elif endpoint == "/jobs":
return mock_jobs["data"]
elif endpoint == "/companies":
return mock_companies["data"]
return []
with patch.object(connector, "_fetch_all", side_effect=fake_fetch):
result = connector.sync()
self.assertEqual(result["status"], "success")
self.assertEqual(result["record_count"], 2)
self.assertEqual(result["details"]["contacts"], 2)
with self.app.app_context():
contacts = CrmContact.query.filter_by(
company_id=self.company_id
).all()
self.assertEqual(len(contacts), 2)
john = next((c for c in contacts if c.external_id == "cust_1"), None)
self.assertIsNotNone(john)
self.assertEqual(john.first_name, "John")
self.assertEqual(john.last_name, "Doe")
self.assertEqual(john.email, "john@example.com")
@patch("app.connectors.servicetitan.requests.post")
def test_sync_creates_deals(self, mock_post):
"""sync() merges ServiceTitan jobs into CrmDeal."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {
"access_token": "tok",
"expires_in": 3600,
}
connector = self._make_connector()
mock_jobs = {
"data": [
{
"id": "job_1",
"job_name": "Roof Replacement",
"total": "15000.00",
"status": "scheduled",
"customer_id": "cust_1",
"expected_date": "2026-08-15T10:00:00Z",
},
]
}
# Seed a contact so the deal can link to it
with self.app.app_context():
contact = CrmContact(
company_id=self.company_id,
external_id="cust_1",
first_name="John",
last_name="Doe",
email="john@example.com",
)
db.session.add(contact)
db.session.commit()
connector._connected = True
connector._access_token = "tok"
def fake_fetch(endpoint, extra_params=None):
if endpoint == "/customers":
return []
elif endpoint == "/jobs":
return mock_jobs["data"]
elif endpoint == "/companies":
return []
return []
with patch.object(connector, "_fetch_all", side_effect=fake_fetch):
result = connector.sync()
self.assertEqual(result["status"], "success")
self.assertEqual(result["details"]["deals"], 1)
with self.app.app_context():
deals = CrmDeal.query.filter_by(company_id=self.company_id).all()
self.assertEqual(len(deals), 1)
deal = deals[0]
self.assertEqual(deal.external_id, "job_1")
self.assertEqual(deal.name, "Roof Replacement")
self.assertEqual(deal.amount, 15000.0)
self.assertEqual(deal.stage, "scheduled")
self.assertIsNotNone(deal.contact_id)
@patch("app.connectors.servicetitan.requests.post")
def test_sync_upsert_contacts(self, mock_post):
"""sync() updates existing contacts instead of creating duplicates."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {
"access_token": "tok",
"expires_in": 3600,
}
# Seed an existing contact
with self.app.app_context():
CrmContact(
company_id=self.company_id,
external_id="cust_1",
first_name="Old",
last_name="Name",
email="old@example.com",
)
db.session.commit()
connector = self._make_connector()
mock_customers = {
"data": [
{
"id": "cust_1",
"first_name": "John",
"last_name": "Updated",
"email": "john@updated.com",
"phone": "555-9999",
},
]
}
with self.app.app_context():
connector._connected = True
connector._access_token = "tok"
def fake_fetch(endpoint, extra_params=None):
if endpoint == "/customers":
return mock_customers["data"]
return []
with patch.object(connector, "_fetch_all", side_effect=fake_fetch):
result = connector.sync()
# Should still be 1 record, not 2
with self.app.app_context():
contacts = CrmContact.query.filter_by(company_id=self.company_id).all()
self.assertEqual(len(contacts), 1)
self.assertEqual(contacts[0].first_name, "John")
self.assertEqual(contacts[0].last_name, "Updated")
self.assertEqual(contacts[0].email, "john@updated.com")
@patch("app.connectors.servicetitan.requests.post")
def test_sync_not_connected(self, mock_post):
"""sync() fails when not connected."""
connector = self._make_connector()
with self.app.app_context():
result = connector.sync()
self.assertEqual(result["status"], "error")
self.assertIn("Not connected", result["error"])
# -- status() tests ------------------------------------------------------
def test_status_not_connected(self):
"""status() returns not-connected when _connected is False."""
connector = self._make_connector()
with self.app.app_context():
result = connector.status()
self.assertFalse(result["connected"])
self.assertTrue(result["config_present"])
@patch.object(
ServiceTitanConnector, "_st_request", return_value={}
)
def test_status_connected(self, mock_req):
"""status() returns connected when _connected is True."""
connector = self._make_connector()
connector._connected = True
with self.app.app_context():
result = connector.status()
self.assertTrue(result["connected"])
def test_status_service_name(self):
"""status() includes the correct service name."""
connector = self._make_connector()
with self.app.app_context():
result = connector.status()
self.assertEqual(result["service"], "servicetitan")
# -- _merge() tests ------------------------------------------------------
def test_merge_creates_new(self):
"""_merge() creates a new record when external_id doesn't exist."""
from app.connectors.servicetitan import ServiceTitanConnector
with self.app.app_context():
count = ServiceTitanConnector._merge(
CrmContact,
self.company_id,
"new_id",
{"first_name": "New", "last_name": "User"},
)
db.session.commit()
self.assertEqual(count, 1)
with self.app.app_context():
record = CrmContact.query.filter_by(
company_id=self.company_id, external_id="new_id"
).first()
self.assertIsNotNone(record)
self.assertEqual(record.first_name, "New")
def test_merge_updates_existing(self):
"""_merge() updates an existing record when external_id matches."""
from app.connectors.servicetitan import ServiceTitanConnector
# Seed an existing record and commit in one context
with self.app.app_context():
existing = CrmContact(
company_id=self.company_id,
external_id="existing_id",
first_name="Old",
last_name="Name",
)
db.session.add(existing)
db.session.commit()
# Verify it's actually there
self.assertIsNotNone(
CrmContact.query.filter_by(
company_id=self.company_id, external_id="existing_id"
).first()
)
# Merge in the same context — must find and update the record
with self.app.app_context():
count = ServiceTitanConnector._merge(
CrmContact,
self.company_id,
"existing_id",
{"first_name": "Updated"},
)
db.session.commit()
record = CrmContact.query.filter_by(
company_id=self.company_id, external_id="existing_id"
).first()
self.assertEqual(count, 1)
self.assertIsNotNone(record)
self.assertEqual(record.first_name, "Updated")
# last_name should be preserved since we didn't update it
self.assertEqual(record.last_name, "Name")
# -- _safe_int() tests ---------------------------------------------------
def test_safe_int_valid(self):
self.assertEqual(
ServiceTitanConnector._safe_int("42"), 42
)
self.assertEqual(ServiceTitanConnector._safe_int(42), 42)
def test_safe_int_none(self):
self.assertIsNone(ServiceTitanConnector._safe_int(None))
def test_safe_int_invalid(self):
self.assertIsNone(ServiceTitanConnector._safe_int("not_a_number"))
self.assertIsNone(ServiceTitanConnector._safe_int(""))
if __name__ == "__main__":
unittest.main()