130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
"""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()
|