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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
"""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, recovery_key=None):
pass
monkeypatch.setattr(
"meshbay_hub.api.users._create_and_send_verification", _noop)
# `**_` because `_send` takes a required keyword `purpose` — the gate
# that keeps the hub off the open-relay list. A stub with the old
# signature turns every send into a TypeError, which looks like a bug
# in the handler. `test_mail_is_not_a_relay.py` opts out of this and
# drives the real thing.
monkeypatch.setattr("meshbay_hub.mail._send", lambda msg, **_: True)
@pytest.fixture(autouse=True)
def _no_cleanup_task(monkeypatch):
"""
Do not run the maintenance loop under test.
`create_app`'s lifespan starts `cleanup_loop` as an asyncio task, so every
test — each of which enters that lifespan — ran a purge pass concurrently
with its own requests. On SQLite `:memory:` that is not merely noisy: the
engine uses a **StaticPool**, one connection for the whole process, so the
request's session and the cleanup task's session interleave their
transactions on the *same* connection. A registration could commit and then
not be visible to the login three lines later, which surfaced as
`401 Invalid credentials` for an account created moments before, in about
one run of `test_node_ws_auth.py` in four.
The purge itself is not at fault and this is not a production condition:
the DELETE was measured removing 0 rows, and PostgreSQL gives every session
its own connection. What is removed here is the second user of the shared
one. Tests that want the maintenance behaviour call the `purge_*` functions
directly, which is how they are covered.
"""
async def _noop(get_session):
return
monkeypatch.setattr("meshbay_hub.tasks.cleanup.cleanup_loop", _noop)
|