100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""Tests for single server boot ups."""
|
|
|
|
from socket import gethostbyname, gethostname
|
|
from aiohttp import ClientSession
|
|
from .mtt_tc import MTTClusterTC, SESSION_KEY
|
|
|
|
|
|
class BootUpTC(MTTClusterTC):
|
|
"""Single server boot tests."""
|
|
|
|
async def test_default_boot(self):
|
|
"""Does the server default boot on http://localhost:3000?"""
|
|
await self.create_server_with_flags()
|
|
|
|
def tests(response):
|
|
"""Response tests."""
|
|
self.assertEqual(response.status, 200)
|
|
|
|
await self.run_tests("/", tests)
|
|
|
|
async def test_alt_port_boot(self):
|
|
"""Can the server boot off on alternate port?"""
|
|
port = 9025
|
|
await self.create_server_with_flags("-p", str(port))
|
|
|
|
def tests(response):
|
|
"""Response tests."""
|
|
self.assertEqual(response.status, 200)
|
|
|
|
await self.run_tests("/", tests)
|
|
|
|
async def test_alt_address_boot(self):
|
|
"""Can it boot off an alternate address?"""
|
|
addr = gethostbyname(gethostname())
|
|
await self.create_server_with_flags("-a", addr)
|
|
|
|
def tests(response):
|
|
"""Response tests."""
|
|
self.assertEqual(response.status, 200)
|
|
|
|
await self.run_tests("/", tests)
|
|
|
|
async def test_for_session_id(self):
|
|
"""Is there a session if?"""
|
|
await self.create_server()
|
|
|
|
def tests(response):
|
|
"""Response tests."""
|
|
self.assertIn(SESSION_KEY, response.cookies)
|
|
|
|
await self.run_tests("/", tests)
|
|
|
|
async def test_session_id_is_random(self):
|
|
"""Is the session id random?"""
|
|
await self.create_server()
|
|
async with ClientSession() as session:
|
|
async with session.get(f"{self.servers[0].host}/") as response:
|
|
result1 = response.cookies[SESSION_KEY].value
|
|
async with ClientSession() as session:
|
|
async with session.get(f"{self.servers[0].host}/") as response:
|
|
result2 = response.cookies[SESSION_KEY].value
|
|
self.assertNotEqual(result1, result2, "Session ids should be unique.")
|
|
|
|
async def test_session_does_not_reset_after_connection(self):
|
|
"""Does the session id remain constant during the session"""
|
|
await self.create_server()
|
|
ids = []
|
|
|
|
def tests(response):
|
|
"""tests"""
|
|
if SESSION_KEY in response.cookies:
|
|
ids.append(response.cookies[SESSION_KEY].value)
|
|
|
|
for _ in range(2):
|
|
await self.run_tests("/", tests)
|
|
self.assertEqual(len(ids), 1)
|
|
|
|
async def test_reset_bad_session_id(self):
|
|
"""Does the session id get reset if bad or expired?"""
|
|
await self.create_server()
|
|
value = "bad id"
|
|
async with ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.servers[0].host}/", cookies={SESSION_KEY: value}
|
|
) as response:
|
|
self.assertIn(SESSION_KEY, response.cookies)
|
|
self.assertNotEqual(response.cookies[SESSION_KEY].value, value)
|
|
|
|
async def test_sessions_are_shared_between_servers(self):
|
|
"""Does the session apply to the cluster."""
|
|
await self.create_cluster()
|
|
ids = []
|
|
|
|
def tests(response):
|
|
if SESSION_KEY in response.cookies:
|
|
ids.append(response.cookies[SESSION_KEY].value)
|
|
|
|
await self.run_tests("/", tests)
|
|
self.assertEqual(len(ids), 1, "Session info should be shared to the cluster.")
|