1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
"""Shared pytest fixtures for hub tests."""
import os
import pytest
import pytest_asyncio
from pathlib import Path
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
# Force SQLite in-memory for all hub tests
os.environ.setdefault("MESHBAY_DATABASE_URL", "sqlite+aiosqlite:///:memory:")
@pytest.fixture(scope="session")
def hub_key_path(tmp_path_factory) -> Path:
"""Generate a hub keypair PEM for the test session."""
d = tmp_path_factory.mktemp("hub_keys")
path = d / "hub_private.pem"
sk = Ed25519PrivateKey.generate()
path.write_bytes(sk.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
))
return path
@pytest.fixture
def hub_config(hub_key_path, tmp_path):
from meshbay_hub.config import HubConfig, DatabaseConfig, ServerConfig, HubIdentityConfig, JWTConfig
cfg = HubConfig(
db=DatabaseConfig(url="sqlite+aiosqlite:///:memory:"),
server=ServerConfig(host="127.0.0.1", port=8000),
identity=HubIdentityConfig(id="test-hub", private_key_path=hub_key_path),
jwt=JWTConfig(access_token_ttl=3600, refresh_token_ttl=86400),
)
return cfg
@pytest_asyncio.fixture
async def app(hub_config):
"""Create a fresh FastAPI app with in-memory DB for each test."""
# Reset module-level engine state
from meshbay_hub.db import engine as eng_mod
eng_mod._engine = None
eng_mod._session_factory = None
from meshbay_hub.app import create_app
application = create_app(hub_config)
# Run lifespan startup manually
async with application.router.lifespan_context(application):
yield application
@pytest_asyncio.fixture
async def client(app):
"""httpx.AsyncClient pointing at the test app — no network."""
import httpx
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app),
base_url="http://test",
) as c:
yield c
|