MoreThanText/morethantext
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed old testing.
Some checks failed
Gitea Actions Demo / Explore-Gitea-Actions (push) Failing after 1s
Details

Browse Source
This commit is contained in:
Jeff Baskin 2025-06-24 17:08:38 -04:00
parent 5ce1dfc873
commit 17693947af
2 changed files with 0 additions and 213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
release_tests/mtt_tc.py
View File

@ -1,111 +0,0 @@
"""Base for MoreThanTest test cases."""
from asyncio import create_subprocess_exec, gather, sleep
from pathlib import Path
from socket import socket
from unittest import IsolatedAsyncioTestCase
from aiohttp import ClientSession
LOCALHOST = "127.0.0.1"
SESSION_KEY = "sessionid"
HOST = "example.com"
class Server:
"""Setup and run servers."""
def __init__(self, *args):
"""Initialize class"""
app = Path.cwd().joinpath("target", "release", "morethantext")
addr = "127.0.0.1"
port = 3000
if args:
self.cmd = list(args)
self.cmd.insert(0, app)
get_port = False
get_addr = False
for item in args:
if get_port:
port = item
get_port = False
if get_addr:
addr = item
get_addr = False
if item in ("-a", "--address"):
get_addr = True
if item in ("-p", "--port"):
get_port = True
else:
self.cmd = [app]
self.server = None
self.host = f"http://{addr}:{port}"
async def create(self):
"""Cerate the server"""
self.server = await create_subprocess_exec(*self.cmd)
await sleep(1)
async def destroy(self):
"""destroy servers"""
self.server.terminate()
await self.server.wait()
class MTTClusterTC(IsolatedAsyncioTestCase):
"""Test case for MoreThanTText."""
async def asyncSetUp(self):
"""Test setup"""
self.servers = []
self.cookies = {}
self.session = ClientSession()
async def asyncTearDown(self):
"""Test tear down."""
await self.session.close()
for server in self.servers:
await server.destroy()
@staticmethod
async def get_port():
"""Retrieve an unused port."""
sock = socket()
sock.bind((LOCALHOST, 0))
port = sock.getsockname()[1]
sock.close()
return port
async def create_server_with_flags(self, *args):
"""Create a single server with flags."""
server = Server(*args)
await server.create()
self.servers.append(server)
async def create_server(self):
"""Create a server on a random port."""
port = await self.get_port()
await self.create_server_with_flags("-p", str(port))
async def create_cluster(self, num=2):
"""Create a cluster of servers."""
ports = []
while len(ports) < num:
port = await self.get_port()
if port not in ports:
ports.append(port)
servers = []
for port in ports:
servers.append(self.create_server_with_flags("-p", str(port)))
cluster = gather(*servers)
await cluster
async def run_tests(self, uri, func):
"""Run the tests on each server."""
for server in self.servers:
async with self.session.get(
f"{server.host}{uri}", cookies=self.cookies
) as response:
if SESSION_KEY in response.cookies:
self.cookies[SESSION_KEY] = response.cookies[SESSION_KEY].value
func(response)

102
release_tests/no_test_single_boot.py
View File

@ -1,102 +0,0 @@
"""Tests for single server boot ups."""
from socket import gethostbyname, gethostname
from unittest import skip
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.")
@skip("Code not availaable yet.")
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)
@skip("Code not availaable yet.")
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.")