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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
"""Shared pytest fixtures for hub tests."""
import os
import pytest
import pytest_asyncio
from pathlib import Path
from unittest.mock import AsyncMock
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)
# Disable rate limiting in tests
from meshbay_hub.api.middleware import limiter
limiter.enabled = False
# Run lifespan startup manually
async with application.router.lifespan_context(application):
yield application
# Reset admin usernames after each test
from meshbay_hub.api.deps import set_admin_usernames
set_admin_usernames([])
@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
@pytest_asyncio.fixture
async def db_session(app):
"""
A session on the same in-memory database the app is using.
For assertions that cannot be made through the API — what a deletion left
behind, for instance, which is exactly the sort of thing worth checking
directly rather than inferring.
"""
from meshbay_hub.db.engine import get_session_factory
async with get_session_factory()() as session:
yield session
@pytest.fixture(autouse=True)
def _skip_email_verification(monkeypatch):
"""Skip email verification in tests — users are active immediately."""
async def _noop(db, user, email, eh):
pass
monkeypatch.setattr(
"meshbay_hub.api.users._create_and_send_verification", _noop)
monkeypatch.setattr("meshbay_hub.mail._send", lambda msg: True)
|