"""Security tests for Command Sovereignty."""
import os
import sys
import unittest
import tempfile
import json
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
os.environ.setdefault('DATABASE_URL', 'sqlite:///test_security.db')
os.environ.setdefault('SECRET_KEY', 'test-secret-key-for-testing')
from app import create_app
from app.models import db
class SecurityTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app.config['TESTING'] = True
cls.app.config['RATELIMIT_ENABLED'] = False
cls.client = cls.app.test_client()
with cls.app.app_context():
db.create_all()
def test_health_check(self):
"""Health endpoint returns 200."""
resp = self.client.get('/health')
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertEqual(data['status'], 'healthy')
def test_no_debug_in_production(self):
"""App debug mode is off."""
self.assertFalse(self.app.debug)
def test_session_cookie_httponly(self):
"""Session cookie is HttpOnly."""
self.assertTrue(self.app.config.get('SESSION_COOKIE_HTTPONLY', False))
def test_session_cookie_samesite(self):
"""Session cookie has SameSite attribute."""
self.assertEqual(self.app.config.get('SESSION_COOKIE_SAMESITE'), 'Lax')
def test_login_rate_limited(self):
"""Login endpoint is rate limited (returns 429 after too many attempts).
Note: Rate limiting is disabled in test mode, so this test verifies
that repeated invalid logins return 401 (auth error).
"""
# Try many login attempts
last_resp = None
for _ in range(20):
last_resp = self.client.post('/api/auth/login',
data=json.dumps({'email': 'test@test.com', 'password': 'wrong'}),
content_type='application/json')
# Should return auth error (401) for invalid credentials
# In production with rate limiting enabled, would return 429
self.assertTrue(last_resp.status_code in (401, 429))
def test_demo_request_requires_valid_email(self):
"""Demo request rejects invalid email."""
resp = self.client.post('/api/leads/demo-request',
data=json.dumps({'name': 'Test', 'email': 'not-an-email'}),
content_type='application/json')
self.assertEqual(resp.status_code, 400)
def test_demo_request_requires_name_and_email(self):
"""Demo request rejects missing required fields."""
resp = self.client.post('/api/leads/demo-request',
data=json.dumps({'name': '', 'email': ''}),
content_type='application/json')
self.assertEqual(resp.status_code, 400)
def test_api_returns_json_not_html(self):
"""API errors return JSON, not HTML."""
resp = self.client.get('/api/nonexistent')
self.assertEqual(resp.status_code, 404)
self.assertEqual(resp.content_type, 'application/json')
def test_webhook_requires_json(self):
"""Webhook endpoint rejects requests without API key."""
resp = self.client.post('/api/connectors/webhooks/angi')
self.assertEqual(resp.status_code, 401)
data = resp.get_json()
self.assertFalse(data['success'])
self.assertIn('api-key', data['error'].lower())
def test_zapier_webhook_requires_json(self):
"""Zapier webhook endpoint rejects non-JSON."""
resp = self.client.post('/api/connectors/webhooks/zapier')
self.assertEqual(resp.status_code, 400)
def test_zapier_webhook_no_active_connector(self):
"""Zapier webhook returns 200 when no active connector configured."""
resp = self.client.post(
'/api/connectors/webhooks/zapier',
data=json.dumps({"action": "create_lead", "data": {"id": "test"}}),
content_type='application/json'
)
self.assertEqual(resp.status_code, 200)
data = resp.get_json()
self.assertFalse(data['success'])
self.assertIn('connector', data['error'].lower())
if __name__ == '__main__':
unittest.main()