"""Tests for tier enforcement on connector endpoints."""
import json
import os
import sys
import unittest
# Ensure the app module is importable
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
os.environ.setdefault('FLASK_ENV', 'testing')
os.environ.setdefault('DATABASE_URL', 'sqlite:///test_tier.db')
os.environ.setdefault('SECRET_KEY', 'test-secret-key-for-tier-testing')
os.environ.setdefault('RATELIMIT_ENABLED', 'false')
from app import create_app
from app.models import db, User, Company, UserCompany, Connector
app = create_app()
app.config['TESTING'] = True
class BaseTest(unittest.TestCase):
"""Mixin with common setup for tier enforcement tests."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@classmethod
def setUpClass(cls):
cls.app = app
cls.app.config['TESTING'] = True
cls.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test_tier.db'
with cls.app.app_context():
db.drop_all()
db.create_all()
@classmethod
def tearDownClass(cls):
with cls.app.app_context():
db.drop_all()
def _create_tenant(self, email, company_name, tier='starter'):
"""Create a user + company with the given tier."""
with self.app.app_context():
user = User(email=email, full_name=email.split('@')[0])
user.set_password('password123')
db.session.add(user)
db.session.flush()
company = Company(name=company_name, settings_json={'tier': tier})
db.session.add(company)
db.session.flush()
uc = UserCompany(user_id=user.id, company_id=company.id)
db.session.add(uc)
db.session.commit()
return user, company
def _login(self, email):
"""Log in as the given user and return a test client."""
with self.app.test_client() as client:
resp = client.post('/api/auth/login', json={
'email': email,
'password': 'password123',
})
self.assertEqual(resp.status_code, 200, f"Login failed: {resp.get_json()}")
return client
def _create_connector(self, client, service='hubspot', sync_frequency='hourly'):
"""Create a connector and return the response."""
return client.post('/api/connectors', json={
'service': service,
'config': {},
'sync_frequency': sync_frequency,
})
class TestConnectorCountLimit(BaseTest):
"""POST /api/connectors should enforce per-tier connector limits."""
def _test_limit(self, tier, max_connectors):
with self.app.app_context():
user, company = self._create_tenant(
f'test-{tier}@example.com', f'{tier} Co', tier=tier
)
client = self._login(f'test-{tier}@example.com')
# Create connectors directly in DB to reach the limit (bypassing service check)
registered_services = ['hubspot', 'quickbooks', 'google_ads', 'jobnimbus',
'slack', 'facebook_ads', 'generic_rest',
'servicetitan', 'angi']
for i in range(max_connectors):
service = registered_services[i % len(registered_services)]
# If the service is already used, create directly in DB
if i >= len(registered_services):
with self.app.app_context():
conn = Connector(
company_id=company.id,
service=f'test_service_{i}',
status='inactive',
config_json={},
sync_frequency='hourly',
)
db.session.add(conn)
db.session.commit()
continue
resp = client.post('/api/connectors', json={
'service': service,
'config': {},
'sync_frequency': 'hourly',
})
# 201 or 409 (duplicate) are fine for the creation step
self.assertIn(resp.status_code, [201, 409],
f"Connector {i+1} creation failed for {tier}: {resp.get_json()}")
# The next one should be rejected with 403
next_service = registered_services[max_connectors % len(registered_services)]
resp = client.post('/api/connectors', json={
'service': next_service,
'config': {},
'sync_frequency': 'hourly',
})
data = resp.get_json()
if resp.status_code == 409:
# Could be a duplicate if the service was already created
# Try with a unique service name
resp = client.post('/api/connectors', json={
'service': 'unique_' + next_service,
'config': {},
'sync_frequency': 'hourly',
})
data = resp.get_json()
if resp.status_code == 400:
# Unknown service — that means the limit check didn't trigger
# because the service validation came first.
# The limit was hit, so the real test is that we can't create more.
self.fail(
f"Expected 403 with limit message for {tier} (max={max_connectors}), "
f"got {resp.status_code}: {data}. Service validation ran before tier check."
)
self.assertEqual(resp.status_code, 403,
f"Expected 403 for {tier} at limit, got {resp.status_code}: {data}")
self.assertIn('limit reached', data.get('error', '').lower(),
f"Wrong error message: {data}")
print(f" ✓ {tier}: limit of {max_connectors} enforced")
def test_starter_limit(self):
self._test_limit('starter', 1)
def test_growth_limit(self):
self._test_limit('growth', 5)
def test_command_no_limit(self):
"""Command tier should allow many connectors (limit=99)."""
with self.app.app_context():
self._create_tenant('cmd-test@example.com', 'Command Co', tier='command')
client = self._login('cmd-test@example.com')
# Create 3 connectors with registered services — well under the 99 limit
for service in ['hubspot', 'quickbooks', 'google_ads']:
resp = client.post('/api/connectors', json={
'service': service,
'config': {},
'sync_frequency': 'hourly',
})
self.assertEqual(resp.status_code, 201,
f"Command tier {service} connector creation failed: {resp.get_json()}")
print(f" ✓ command: 3 connectors created (limit=99)")
def test_enterprise_no_limit(self):
"""Enterprise tier should allow many connectors (limit=99)."""
with self.app.app_context():
self._create_tenant('ent-test@example.com', 'Enterprise Co', tier='enterprise')
client = self._login('ent-test@example.com')
for service in ['hubspot', 'quickbooks', 'google_ads']:
resp = client.post('/api/connectors', json={
'service': service,
'config': {},
'sync_frequency': 'hourly',
})
self.assertEqual(resp.status_code, 201,
f"Enterprise tier {service} connector creation failed: {resp.get_json()}")
print(f" ✓ enterprise: 3 connectors created (limit=99)")
class TestRealTimeSyncGate(BaseTest):
"""Real-time sync should require Command or Enterprise tier."""
def _test_realtime_blocked(self, tier):
with self.app.app_context():
self._create_tenant(f'rt-{tier}@example.com', f'{tier} Co', tier=tier)
client = self._login(f'rt-{tier}@example.com')
# Try to create with real_time
resp = client.post('/api/connectors', json={
'service': 'hubspot',
'config': {},
'sync_frequency': 'real_time',
})
self.assertEqual(resp.status_code, 403,
f"Expected 403 for {tier} real_time, got {resp.status_code}")
data = resp.get_json()
self.assertIn('real-time sync', data.get('error', '').lower(),
f"Wrong error message: {data}")
print(f" ✓ {tier}: real_time sync blocked on creation")
def test_starter_blocked(self):
self._test_realtime_blocked('starter')
def test_growth_blocked(self):
self._test_realtime_blocked('growth')
def test_realtime_allowed(self):
"""Command/Enterprise should allow real_time."""
for tier in ('command', 'enterprise'):
with self.app.app_context():
self._create_tenant(f'rt-{tier}@example.com', f'{tier} Co', tier=tier)
client = self._login(f'rt-{tier}@example.com')
resp = client.post('/api/connectors', json={
'service': 'hubspot',
'config': {},
'sync_frequency': 'real_time',
})
self.assertEqual(resp.status_code, 201,
f"{tier} real_time should be allowed: {resp.get_json()}")
print(f" ✓ {tier}: real_time sync allowed on creation")
def test_update_realtime_blocked(self):
"""PUT should also block real_time for starter/growth."""
with self.app.app_context():
self._create_tenant('update-test@example.com', 'Update Co', tier='growth')
client = self._login('update-test@example.com')
# First create a connector with hourly
resp = client.post('/api/connectors', json={
'service': 'hubspot',
'config': {},
'sync_frequency': 'hourly',
})
self.assertEqual(resp.status_code, 201)
connector_id = resp.get_json()['connector']['id']
# Now try to update to real_time
resp = client.put(f'/api/connectors/{connector_id}', json={
'sync_frequency': 'real_time',
})
self.assertEqual(resp.status_code, 403,
f"Expected 403 on PUT real_time, got {resp.status_code}")
data = resp.get_json()
self.assertIn('real-time sync', data.get('error', '').lower())
print(f" ✓ growth: real_time sync blocked on update (PUT)")
def test_update_realtime_allowed(self):
"""Command tier should allow updating to real_time."""
with self.app.app_context():
self._create_tenant('update-cmd@example.com', 'Cmd Update Co', tier='command')
client = self._login('update-cmd@example.com')
# Create with hourly
resp = client.post('/api/connectors', json={
'service': 'hubspot',
'config': {},
'sync_frequency': 'hourly',
})
self.assertEqual(resp.status_code, 201)
connector_id = resp.get_json()['connector']['id']
# Update to real_time — should succeed
resp = client.put(f'/api/connectors/{connector_id}', json={
'sync_frequency': 'real_time',
})
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data['connector']['sync_frequency'], 'real_time')
print(f" ✓ command: real_time sync allowed on update (PUT)")
class TestTierInfoInList(BaseTest):
"""GET /api/connectors should include tier info."""
def test_tier_info_present(self):
with self.app.app_context():
self._create_tenant('list-test@example.com', 'List Co', tier='growth')
client = self._login('list-test@example.com')
# Create one connector
resp = client.post('/api/connectors', json={
'service': 'hubspot',
'config': {},
'sync_frequency': 'hourly',
})
self.assertEqual(resp.status_code, 201)
# List connectors
resp = client.get('/api/connectors')
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertIn('tier', data)
self.assertIn('connector_limit', data)
self.assertIn('connectors_used', data)
self.assertEqual(data['tier'], 'growth')
self.assertEqual(data['connector_limit'], 5)
self.assertEqual(data['connectors_used'], 1)
print(f" ✓ GET /api/connectors includes tier info")
class TestPlansEndpoint(BaseTest):
"""GET /api/plans should return current tier for authenticated users."""
def test_plans_returns_tier(self):
with self.app.app_context():
self._create_tenant('plans-test@example.com', 'Plans Co', tier='starter')
client = self._login('plans-test@example.com')
resp = client.get('/api/plans')
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertIn('tier', data)
self.assertEqual(data['tier'], 'starter')
self.assertEqual(len(data['plans']), 4)
print(f" ✓ GET /api/plans returns current tier")
class TestOrganizationsEndpoint(BaseTest):
"""GET /api/partner/organizations should return dynamic tier."""
def test_dynamic_tier(self):
with self.app.app_context():
self._create_tenant('org-test@example.com', 'Org Co', tier='command')
client = self._login('org-test@example.com')
resp = client.get('/api/partner/organizations')
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertTrue(len(data['items']) > 0)
self.assertEqual(data['items'][0]['tier'], 'command')
print(f" ✓ GET /api/partner/organizations returns dynamic tier")
if __name__ == '__main__':
# Cleanup
try:
os.unlink('test_tier.db')
except FileNotFoundError:
pass
unittest.main()