88 lines
No EOL
2.8 KiB
Python
88 lines
No EOL
2.8 KiB
Python
import unittest
|
|
import subprocess
|
|
import requests
|
|
import time
|
|
import ssl
|
|
from urllib3.util.ssl_ import create_urllib3_context
|
|
|
|
class IntegrationTests(unittest.TestCase):
|
|
WEB_URL = "https://localhost:5000"
|
|
TEST_USER = "test_admin"
|
|
TEST_CERT = "test_cert.pem"
|
|
TEST_KEY = "test_key.pem"
|
|
|
|
def setUp(self):
|
|
# Configure TLS 1.3 context
|
|
self.ssl_context = create_urllib3_context()
|
|
self.ssl_context.options |= ssl.OP_NO_TLSv1_2
|
|
self.ssl_context.load_cert_chain(self.TEST_CERT, self.TEST_KEY)
|
|
|
|
def test_task_creation_equivalence(self):
|
|
"""Test task creation produces same result in CLI and web"""
|
|
# CLI
|
|
cli_result = subprocess.run(
|
|
["symphony", "add-task", "Test task"],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
# Web
|
|
web_result = requests.post(
|
|
f"{self.WEB_URL}/tasks",
|
|
json={"task": "Test task"},
|
|
headers={"X-Client-Cert-User": self.TEST_USER},
|
|
verify=False
|
|
)
|
|
|
|
self.assertEqual(cli_result.returncode, 0)
|
|
self.assertEqual(web_result.status_code, 200)
|
|
|
|
def test_rbac_enforcement(self):
|
|
"""Test RBAC is enforced consistently"""
|
|
# Test with invalid permission
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
subprocess.run(
|
|
["symphony", "add-task", "Unauthorized"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
web_result = requests.post(
|
|
f"{self.WEB_URL}/tasks",
|
|
json={"task": "Unauthorized"},
|
|
headers={"X-Client-Cert-User": "unauthorized_user"},
|
|
verify=False
|
|
)
|
|
self.assertEqual(web_result.status_code, 403)
|
|
|
|
def test_performance_requirements(self):
|
|
"""Test response times <500ms"""
|
|
start = time.time()
|
|
subprocess.run(["symphony", "next-task"], capture_output=True)
|
|
cli_time = time.time() - start
|
|
|
|
start = time.time()
|
|
requests.get(
|
|
f"{self.WEB_URL}/tasks/next",
|
|
headers={"X-Client-Cert-User": self.TEST_USER},
|
|
verify=False
|
|
)
|
|
web_time = time.time() - start
|
|
|
|
self.assertLess(cli_time, 0.5)
|
|
self.assertLess(web_time, 0.5)
|
|
|
|
def test_tls_1_3_requirement(self):
|
|
"""Test only TLS 1.3 connections accepted"""
|
|
# Try with TLS 1.2 (should fail)
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
|
|
with self.assertRaises(requests.exceptions.SSLError):
|
|
requests.get(
|
|
f"{self.WEB_URL}/tasks/next",
|
|
headers={"X-Client-Cert-User": self.TEST_USER},
|
|
verify=False
|
|
)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |