"""Tests for Jobber connector and Jobber entity models."""
import os
os.environ['DISABLE_SCHEDULER'] = '1'
os.environ.setdefault('SECRET_KEY', 'test-secret-key')
# Isolated in-memory DB per test file to avoid lock contention
os.environ["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
import unittest
import json
import uuid
from datetime import datetime, timezone, timedelta
from unittest.mock import patch, MagicMock
from app import create_app, db
from app.models import Company, JobberClient, JobberQuote, JobberJob, JobberInvoice, Estimate
class TestJobberModels(unittest.TestCase):
"""Test Jobber entity models."""
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app_context = cls.app.app_context()
cls.app_context.push()
cls.client = cls.app.test_client()
db.create_all()
@classmethod
def tearDownClass(cls):
db.drop_all()
cls.app_context.pop()
def setUp(self):
try:
db.session.query(JobberInvoice).delete()
db.session.query(JobberJob).delete()
db.session.query(JobberQuote).delete()
db.session.query(JobberClient).delete()
db.session.query(Estimate).delete()
db.session.commit()
except Exception:
db.session.rollback()
db.session.commit()
def tearDown(self):
Company.query.filter(Company.name.like("JobberModelCo_%")).delete()
db.session.commit()
def _make_company(self):
c = Company(name=f"JobberModelCo_{uuid.uuid4().hex[:8]}", stripe_customer_id=f"cus_jobber_{uuid.uuid4().hex[:8]}")
db.session.add(c)
db.session.commit()
return c
def test_create_client(self):
company = self._make_company()
now = datetime.now(timezone.utc)
client = JobberClient(
company_id=company.id,
external_id="j-client-001",
first_name="John",
last_name="Doe",
email="john@example.com",
phone="555-1234",
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(client)
db.session.commit()
self.assertIsNotNone(client.id)
self.assertEqual(client.full_name, "John Doe")
self.assertEqual(client.email, "john@example.com")
def test_create_quote(self):
company = self._make_company()
now = datetime.now(timezone.utc)
quote = JobberQuote(
company_id=company.id,
external_id="j-quote-001",
name="Roof Replacement",
status="pending",
amount=15000.0,
currency="USD",
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(quote)
db.session.commit()
self.assertEqual(quote.status, "pending")
self.assertEqual(quote.amount, 15000.0)
def test_create_job(self):
company = self._make_company()
now = datetime.now(timezone.utc)
job = JobberJob(
company_id=company.id,
external_id="j-job-001",
name="Roof Install",
status="in_progress",
project_type="roofing",
estimated_revenue=20000.0,
actual_revenue=18500.0,
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(job)
db.session.commit()
self.assertEqual(job.project_type, "roofing")
self.assertIsNotNone(job.id)
def test_create_invoice(self):
company = self._make_company()
now = datetime.now(timezone.utc)
invoice = JobberInvoice(
company_id=company.id,
external_id="j-inv-001",
invoice_number="INV-001",
status="paid",
amount=5000.0,
paid_amount=5000.0,
issued_at=now - timedelta(days=10),
paid_at=now - timedelta(days=5),
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(invoice)
db.session.commit()
self.assertEqual(invoice.status, "paid")
self.assertEqual(invoice.paid_amount, 5000.0)
def test_client_uniqueness(self):
company = self._make_company()
now = datetime.now(timezone.utc)
c1 = JobberClient(
company_id=company.id,
external_id="j-unique-001",
first_name="A",
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(c1)
db.session.commit()
c2 = JobberClient(
company_id=company.id,
external_id="j-unique-001",
first_name="B",
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(c2)
with self.assertRaises(Exception):
db.session.commit()
db.session.rollback()
def test_client_full_name_property(self):
company = self._make_company()
now = datetime.now(timezone.utc)
client = JobberClient(
company_id=company.id,
external_id="j-name-001",
first_name="Jane",
last_name="Smith",
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(client)
db.session.commit()
self.assertEqual(client.full_name, "Jane Smith")
def test_client_full_name_no_last(self):
company = self._make_company()
now = datetime.now(timezone.utc)
client = JobberClient(
company_id=company.id,
external_id="j-name-002",
first_name="Jane",
synced_at=now,
created_at=now,
updated_at=now,
)
db.session.add(client)
db.session.commit()
self.assertEqual(client.full_name, "Jane")
class TestJobberConnector(unittest.TestCase):
"""Test JobberConnector GraphQL and OAuth methods."""
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app_context = cls.app.app_context()
cls.app_context.push()
cls.client = cls.app.test_client()
db.create_all()
@classmethod
def tearDownClass(cls):
db.drop_all()
cls.app_context.pop()
def setUp(self):
self.company = Company(name="JobberConnCo", stripe_customer_id=f"cus_jobber_conn_{uuid.uuid4().hex[:8]}")
db.session.add(self.company)
db.session.commit()
def tearDown(self):
db.session.query(JobberInvoice).delete()
db.session.query(JobberJob).delete()
db.session.query(JobberQuote).delete()
db.session.query(JobberClient).delete()
db.session.query(Estimate).delete()
Company.query.filter_by(id=self.company.id).delete()
db.session.commit()
def test_connector_instantiation(self):
from app.connectors.jobber import JobberConnector
conn = JobberConnector(
company_id=self.company.id,
config={
"client_id": "test_client",
"client_secret": "test_secret",
"access_token": "test_token",
"refresh_token": "test_refresh",
},
)
self.assertEqual(conn._SERVICE, "jobber")
self.assertEqual(conn._access_token, "test_token")
def test_connector_missing_fields(self):
from app.connectors.jobber import JobberConnector
conn = JobberConnector(
company_id=self.company.id,
config={"client_id": "test_client"},
)
result = conn.connect()
self.assertEqual(result["status"], "error")
self.assertIn("Missing required fields", result["error"])
@patch("app.connectors.jobber.requests.post")
def test_graphql_request(self, mock_post):
from app.connectors.jobber import JobberConnector
mock_resp = MagicMock()
mock_resp.json.return_value = {"data": {"clients": {"pageInfo": {"hasNextPage": False}, "edges": []}}}
mock_resp.raise_for_status = MagicMock()
mock_post.return_value = mock_resp
conn = JobberConnector(
company_id=self.company.id,
config={
"client_id": "test_client",
"client_secret": "test_secret",
"access_token": "test_token",
"refresh_token": "test_refresh",
},
)
from app.connectors.jobber import CLIENTS_QUERY
result = conn._graphql_request(CLIENTS_QUERY, {})
self.assertIn("clients", result)
@patch("app.connectors.jobber.requests.post")
def test_graphql_request_error(self, mock_post):
from app.connectors.jobber import JobberConnector
mock_resp = MagicMock()
mock_resp.json.return_value = {"errors": [{"message": "Unauthorized"}]}
mock_resp.raise_for_status = MagicMock()
mock_post.return_value = mock_resp
conn = JobberConnector(
company_id=self.company.id,
config={
"client_id": "test_client",
"client_secret": "test_secret",
"access_token": "test_token",
"refresh_token": "test_refresh",
},
)
from app.connectors.jobber import CLIENTS_QUERY
with self.assertRaises(ValueError) as ctx:
conn._graphql_request(CLIENTS_QUERY, {})
self.assertIn("Unauthorized", str(ctx.exception))
def test_oauth_authorize_url(self):
from app.connectors.jobber import JobberConnector
conn = JobberConnector(
company_id=self.company.id,
config={
"client_id": "app123",
"client_secret": "secret",
"redirect_uri": "https://example.com/callback",
},
)
url = conn.oauth_authorize_url("state123")
self.assertIn("response_type=code", url)
self.assertIn("client_id=app123", url)
self.assertIn("state=state123", url)
def test_status_disconnected(self):
from app.connectors.jobber import JobberConnector
conn = JobberConnector(
company_id=self.company.id,
config={"client_id": "test", "client_secret": "secret"},
)
status = conn.status()
self.assertEqual(status["status"], "disconnected")
self.assertFalse(status["connected"])
self.assertFalse(status["has_token"])
@patch("app.connectors.jobber.requests.post")
def test_exchange_code(self, mock_post):
from app.connectors.jobber import JobberConnector
mock_resp = MagicMock()
mock_resp.json.return_value = {
"access_token": "new_access",
"refresh_token": "new_refresh",
"expires_in": 3600,
}
mock_resp.raise_for_status = MagicMock()
mock_post.return_value = mock_resp
conn = JobberConnector(
company_id=self.company.id,
config={
"client_id": "app123",
"client_secret": "secret",
"redirect_uri": "https://example.com/callback",
},
)
tokens = conn.exchange_code_for_tokens("auth_code_123")
self.assertEqual(tokens["access_token"], "new_access")
self.assertEqual(tokens["expires_in"], 3600)
def test_connector_registered(self):
from app.connectors import _AVAILABLE_INFO
self.assertIn("jobber", _AVAILABLE_INFO)
info = _AVAILABLE_INFO["jobber"]
self.assertEqual(info["service"], "jobber")
self.assertEqual(info["auth_type"], "oauth2")
@patch("app.connectors.jobber.JobberConnector._paginated_query")
def test_sync_clients(self, mock_query):
from app.connectors.jobber import JobberConnector
mock_query.return_value = [
{
"id": "sync-client-1",
"firstName": "Alice",
"lastName": "Wonder",
"companyName": "",
"defaultAddress": {"line1": "123 Main St", "city": "Austin", "province": "TX", "country": "US", "postalCode": "78701", "line2": ""},
"contacts": [{"id": "c1", "email": "alice@example.com", "mobilePhone": "555-9999"}],
"timezones": [{"id": "America/Chicago"}],
"languageCode": "en",
}
]
conn = JobberConnector(
company_id=self.company.id,
config={
"client_id": "app",
"client_secret": "secret",
"access_token": "token",
"refresh_token": "refresh",
},
)
conn._sync_clients()
client = JobberClient.query.filter_by(
company_id=self.company.id,
external_id="sync-client-1",
).first()
self.assertIsNotNone(client)
self.assertEqual(client.first_name, "Alice")
self.assertEqual(client.city, "Austin")
if __name__ == "__main__":
unittest.main()