diff options
Diffstat (limited to 'packages/meshbay-hub/tests/conftest.py')
| -rw-r--r-- | packages/meshbay-hub/tests/conftest.py | 64 |
1 files changed, 64 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/conftest.py b/packages/meshbay-hub/tests/conftest.py new file mode 100644 index 0000000..b42aba1 --- /dev/null +++ b/packages/meshbay-hub/tests/conftest.py @@ -0,0 +1,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 |