Update tests
This commit is contained in:
129
tests/frontend/conftest.py
Normal file
129
tests/frontend/conftest.py
Normal file
@ -0,0 +1,129 @@
|
||||
"""Test fixtures for the frontend."""
|
||||
|
||||
import asyncio
|
||||
import secrets
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
import uvicorn
|
||||
|
||||
|
||||
from sshecret_admin.core.app import create_admin_app
|
||||
from sshecret_admin.core.settings import AdminServerSettings
|
||||
from sshecret_backend.app import create_backend_app
|
||||
from sshecret_backend.settings import BackendSettings
|
||||
from sshecret_backend.testing import create_test_token
|
||||
|
||||
from tests.helpers import create_test_admin_user, in_tempdir
|
||||
from tests.types import PortFactory, TestPorts
|
||||
|
||||
|
||||
@pytest.fixture(name="ui_test_ports", scope="session")
|
||||
def generate_test_ports(unused_tcp_port_factory: PortFactory) -> TestPorts:
|
||||
"""Generate the test ports."""
|
||||
test_ports = TestPorts(
|
||||
backend=unused_tcp_port_factory(),
|
||||
admin=unused_tcp_port_factory(),
|
||||
sshd=unused_tcp_port_factory(),
|
||||
)
|
||||
print(f"{test_ports=!r}")
|
||||
return test_ports
|
||||
|
||||
|
||||
@pytest.fixture(scope="function", name="ui_backend_server")
|
||||
def run_backend_server(ui_test_ports: TestPorts):
|
||||
"""Run the backend server in a thread."""
|
||||
port = ui_test_ports.backend
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
backend_work_path = Path(tmp_dir)
|
||||
db_file = backend_work_path / "backend.db"
|
||||
backend_settings = BackendSettings(database=str(db_file.absolute()))
|
||||
backend_app = create_backend_app(backend_settings)
|
||||
token = create_test_token(backend_settings)
|
||||
|
||||
config = uvicorn.Config(
|
||||
app=backend_app, port=port, host="127.0.0.1", log_level="warning"
|
||||
)
|
||||
server = uvicorn.Server(config)
|
||||
|
||||
def run():
|
||||
asyncio.run(server.serve())
|
||||
|
||||
thread = threading.Thread(target=run)
|
||||
thread.start()
|
||||
|
||||
backend_url = f"http://127.0.0.1:{port}"
|
||||
for _ in range(30):
|
||||
try:
|
||||
r = requests.get(backend_url)
|
||||
if r.status_code < 500:
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(1)
|
||||
else:
|
||||
raise RuntimeError("Backend server did not start in time")
|
||||
|
||||
yield backend_url, token
|
||||
|
||||
server.should_exit = True
|
||||
thread.join()
|
||||
|
||||
|
||||
@pytest.fixture(scope="function", name="ui_admin_server")
|
||||
def run_admin_server(ui_test_ports: TestPorts, ui_backend_server: tuple[str, str]):
|
||||
"""Run the admin server in a thread."""
|
||||
backend_url, backend_token = ui_backend_server
|
||||
port = ui_test_ports.admin
|
||||
secret_key = secrets.token_urlsafe(32)
|
||||
|
||||
with in_tempdir() as admin_work_path:
|
||||
admin_db = admin_work_path / "ssh_admin.db"
|
||||
admin_settings = AdminServerSettings.model_validate(
|
||||
{
|
||||
"sshecret_backend_url": backend_url,
|
||||
"backend_token": backend_token,
|
||||
"secret_key": secret_key,
|
||||
"listen_address": "127.0.0.1",
|
||||
"port": port,
|
||||
"database": str(admin_db.absolute()),
|
||||
"password_manager_directory": str(admin_work_path.absolute()),
|
||||
}
|
||||
)
|
||||
|
||||
admin_app = create_admin_app(admin_settings)
|
||||
config = uvicorn.Config(
|
||||
app=admin_app, port=port, host="127.0.0.1", log_level="warning"
|
||||
)
|
||||
server = uvicorn.Server(config)
|
||||
|
||||
def run():
|
||||
asyncio.run(server.serve())
|
||||
|
||||
thread = threading.Thread(target=run)
|
||||
thread.start()
|
||||
|
||||
admin_url = f"http://127.0.0.1:{port}"
|
||||
admin_password = secrets.token_urlsafe(10)
|
||||
create_test_admin_user(admin_settings, "test", admin_password)
|
||||
|
||||
for _ in range(30):
|
||||
try:
|
||||
r = requests.get(admin_url)
|
||||
if r.status_code < 500:
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(1)
|
||||
else:
|
||||
raise RuntimeError("Admin server did not start in time")
|
||||
|
||||
yield admin_url, ("test", admin_password)
|
||||
|
||||
server.should_exit = True
|
||||
thread.join()
|
||||
Reference in New Issue
Block a user