JDSS/mem0
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: drop test scaffold and CI workflow to keep PR minimal

Browse Source
This commit is contained in:
Mgeeeek
2026-05-13 21:46:56 +05:30
parent e2fd8dc7a8
commit 8e00815589
7 changed files with 0 additions and 634 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/server-ci.yml
-38
View File
@@ -1,38 +0,0 @@
name: Server CI
on:
workflow_dispatch:
push:
branches: [main]
paths:
- 'server/**'
- '.github/workflows/server-ci.yml'
pull_request:
paths:
- 'server/**'
- '.github/workflows/server-ci.yml'
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.10", "3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dev dependencies
working-directory: server
run: pip install -r requirements-dev.txt
- name: Run tests
working-directory: server
env:
JWT_SECRET: ci-test-secret-do-not-use-in-prod-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_API_KEY: ci-test-key
run: pytest tests/ -v
server/requirements-dev.txt
-5
View File
@@ -1,5 +0,0 @@
-r requirements.txt
pytest>=8.0,<9.0
pytest-asyncio>=0.23,<1.0
httpx>=0.27,<1.0
server/tests/__init__.py
View File
server/tests/conftest.py
-222
View File
@@ -1,222 +0,0 @@
"""Shared pytest fixtures for server/ tests.
Strategy:
- SQLite in-memory engine for the auth DB (Users, APIKey, RefreshTokenJti, etc).
- get_memory_instance() is patched to return a MagicMock so routes that touch
the memory backend don't need pgvector.
- The FastAPI app is imported lazily after env vars are set so module-level
constants in server.auth (JWT_SECRET, AUTH_DISABLED, ADMIN_API_KEY) bind
to the test values, not whatever the host shell has.
- SQLAlchemy's generic Uuid type decorator rejects string values on SQLite
(it calls value.hex on bind); production uses Postgres which has native
UUID handling that accepts strings transparently. Patch the bind processor
to coerce str -> UUID for the test session.
"""
from __future__ import annotations
import os
import sys
import uuid
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from sqlalchemy.sql import sqltypes
# Make server/ importable as a top-level package (matches uvicorn's CWD).
SERVER_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(SERVER_DIR))
# Required for JWT issuance during tests.
os.environ.setdefault("JWT_SECRET", "test-secret-do-not-use-in-prod-" + "x" * 32)
os.environ.setdefault("OPENAI_API_KEY", "test-key")
os.environ.setdefault("POSTGRES_HOST", "localhost") # silence db url builder
# JWT 'sub' is always a string per spec, so _resolve_user_from_jwt looks up
# User by string PK. SQLite's generic Uuid type-decorator fails on strings;
# Postgres' native UUID accepts them. Coerce at the bind layer once at import.
_orig_uuid_bind_processor = sqltypes.Uuid.bind_processor
def _patched_uuid_bind_processor(self, dialect):
proc = _orig_uuid_bind_processor(self, dialect)
if proc is None:
return proc
def wrapper(value):
if isinstance(value, str):
value = uuid.UUID(value)
return proc(value)
return wrapper
sqltypes.Uuid.bind_processor = _patched_uuid_bind_processor
@pytest.fixture
def test_engine():
"""Fresh in-memory SQLite engine per test (no cross-test pollution).
StaticPool is non-negotiable: in-memory SQLite is scoped to a single
connection, so every Session must share it. Without StaticPool, rows
inserted via db_session would be invisible to the TestClient's overridden
get_db (each Session would get its own empty in-memory DB).
"""
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
# Import Base AFTER env vars are set.
from db import Base # noqa: E402
# Importing models registers them on Base.metadata.
import models # noqa: F401, E402
Base.metadata.create_all(engine)
return engine
@pytest.fixture
def test_session_factory(test_engine):
return sessionmaker(bind=test_engine, autoflush=False, expire_on_commit=False)
@pytest.fixture
def mock_memory():
"""A MagicMock standing in for the global Memory instance."""
mock = MagicMock()
mock.get_all.return_value = {"results": []}
mock.search.return_value = {"results": []}
mock.add.return_value = {"results": [], "events": []}
mock.get.return_value = {"id": "memory-id", "memory": "stub"}
mock.history.return_value = []
mock.delete.return_value = None
mock.delete_all.return_value = None
mock.reset.return_value = None
mock.vector_store.list.return_value = [[]]
return mock
@pytest.fixture
def client(test_session_factory, mock_memory, monkeypatch):
"""FastAPI TestClient with overridden DB session and mocked memory instance."""
from db import get_db # noqa: E402
import server_state # noqa: E402
# Neutralize the real Memory.from_config() call that initialize_state()
# makes during main.py import — pgvector and the history sqlite path are
# both unavailable in tests. update_config takes the same path on POST /configure.
monkeypatch.setattr(server_state, "initialize_state", lambda *_args, **_kwargs: None)
monkeypatch.setattr(server_state, "get_memory_instance", lambda: mock_memory)
monkeypatch.setattr(server_state, "update_config", lambda updates: {"config": updates})
from main import app # noqa: E402
# Also patch the bindings inside main.py since it imported by name.
monkeypatch.setattr("main.get_memory_instance", lambda: mock_memory)
monkeypatch.setattr("main.update_config", lambda updates: {"config": updates})
# Async request-log persistence uses the real SessionLocal (Postgres) — noop it.
monkeypatch.setattr("main._persist_request_log", lambda *_args, **_kwargs: None)
def _override_get_db():
db = test_session_factory()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = _override_get_db
try:
with TestClient(app) as c:
yield c
finally:
app.dependency_overrides.pop(get_db, None)
@pytest.fixture
def db_session(test_session_factory):
"""A direct DB session for tests that need to insert User rows."""
session = test_session_factory()
try:
yield session
finally:
session.close()
def _make_user(db_session, *, role: str, email: str | None = None):
from auth import hash_password
from models import User
user = User(
id=uuid.uuid4(),
name=f"{role}-user",
email=email or f"{role}-{uuid.uuid4().hex[:8]}@example.com",
password_hash=hash_password("test-password-123"),
role=role,
)
db_session.add(user)
db_session.commit()
db_session.refresh(user)
return user
@pytest.fixture
def admin_user(db_session):
return _make_user(db_session, role="admin")
@pytest.fixture
def member_user(db_session):
"""Non-admin User. No public endpoint produces these today; we insert directly."""
return _make_user(db_session, role="member")
@pytest.fixture
def admin_jwt(admin_user):
from auth import create_access_token
return create_access_token(str(admin_user.id), admin_user.role)
@pytest.fixture
def member_jwt(member_user):
from auth import create_access_token
return create_access_token(str(member_user.id), member_user.role)
@pytest.fixture
def auth_admin_header(admin_jwt):
return {"Authorization": f"Bearer {admin_jwt}"}
@pytest.fixture
def auth_member_header(member_jwt):
return {"Authorization": f"Bearer {member_jwt}"}
@pytest.fixture
def admin_api_key_env(monkeypatch):
"""Activates the legacy ADMIN_API_KEY escape hatch."""
import auth
key = "admin-api-key-test-value-" + "y" * 16
monkeypatch.setattr(auth, "ADMIN_API_KEY", key)
return {"X-API-Key": key}
@pytest.fixture
def auth_disabled_env(monkeypatch):
"""Activates AUTH_DISABLED=true."""
import auth
monkeypatch.setattr(auth, "AUTH_DISABLED", True)
return {} # no headers needed
server/tests/test_admin_gating.py
-222
View File
@@ -1,222 +0,0 @@
"""Integration tests for admin-gated routes.
Covers, for each affected route, the regression matrix:
- admin JWT -> success
- ADMIN_API_KEY -> success (no DB user required)
- AUTH_DISABLED -> success
- member JWT -> 403
- no auth -> 401
"""
from __future__ import annotations
# --- GET /configure ---
def test_get_configure_admin_jwt(client, auth_admin_header):
response = client.get("/configure", headers=auth_admin_header)
assert response.status_code == 200
def test_get_configure_admin_api_key(client, admin_api_key_env):
response = client.get("/configure", headers=admin_api_key_env)
assert response.status_code == 200
def test_get_configure_auth_disabled(client, auth_disabled_env):
response = client.get("/configure")
assert response.status_code == 200
def test_get_configure_member_forbidden(client, auth_member_header):
response = client.get("/configure", headers=auth_member_header)
assert response.status_code == 403
def test_get_configure_no_auth_unauthorized(client):
response = client.get("/configure")
assert response.status_code == 401
# --- POST /configure ---
def _valid_config():
return {
"vector_store": {"provider": "pgvector", "config": {"host": "h"}},
"llm": {"provider": "openai", "config": {"api_key": "x", "model": "m"}},
"embedder": {"provider": "openai", "config": {"api_key": "x", "model": "e"}},
}
def test_post_configure_admin_jwt(client, auth_admin_header):
response = client.post("/configure", headers=auth_admin_header, json=_valid_config())
assert response.status_code == 200
def test_post_configure_admin_api_key(client, admin_api_key_env):
response = client.post("/configure", headers=admin_api_key_env, json=_valid_config())
assert response.status_code == 200
def test_post_configure_auth_disabled(client, auth_disabled_env):
response = client.post("/configure", json=_valid_config())
assert response.status_code == 200
def test_post_configure_member_forbidden(client, auth_member_header):
response = client.post("/configure", headers=auth_member_header, json=_valid_config())
assert response.status_code == 403
def test_post_configure_no_auth_unauthorized(client):
response = client.post("/configure", json=_valid_config())
assert response.status_code == 401
# --- POST /reset ---
def test_post_reset_admin_jwt(client, auth_admin_header):
response = client.post("/reset", headers=auth_admin_header)
assert response.status_code == 200
def test_post_reset_admin_api_key(client, admin_api_key_env):
response = client.post("/reset", headers=admin_api_key_env)
assert response.status_code == 200
def test_post_reset_auth_disabled(client, auth_disabled_env):
response = client.post("/reset")
assert response.status_code == 200
def test_post_reset_member_forbidden(client, auth_member_header):
response = client.post("/reset", headers=auth_member_header)
assert response.status_code == 403
def test_post_reset_no_auth_unauthorized(client):
response = client.post("/reset")
assert response.status_code == 401
# --- GET /entities ---
def test_get_entities_admin_jwt(client, auth_admin_header):
response = client.get("/entities", headers=auth_admin_header)
assert response.status_code == 200
def test_get_entities_admin_api_key(client, admin_api_key_env):
response = client.get("/entities", headers=admin_api_key_env)
assert response.status_code == 200
def test_get_entities_auth_disabled(client, auth_disabled_env):
response = client.get("/entities")
assert response.status_code == 200
def test_get_entities_member_forbidden(client, auth_member_header):
response = client.get("/entities", headers=auth_member_header)
assert response.status_code == 403
def test_get_entities_no_auth_unauthorized(client):
response = client.get("/entities")
assert response.status_code == 401
# --- DELETE /entities/{type}/{id} ---
def test_delete_entity_admin_jwt(client, auth_admin_header):
response = client.delete("/entities/user/alice", headers=auth_admin_header)
assert response.status_code == 200
def test_delete_entity_admin_api_key(client, admin_api_key_env):
response = client.delete("/entities/user/alice", headers=admin_api_key_env)
assert response.status_code == 200
def test_delete_entity_auth_disabled(client, auth_disabled_env):
response = client.delete("/entities/user/alice")
assert response.status_code == 200
def test_delete_entity_member_forbidden(client, auth_member_header):
response = client.delete("/entities/user/alice", headers=auth_member_header)
assert response.status_code == 403
def test_delete_entity_no_auth_unauthorized(client):
response = client.delete("/entities/user/alice")
assert response.status_code == 401
# --- GET /requests ---
def test_get_requests_admin_jwt(client, auth_admin_header):
response = client.get("/requests", headers=auth_admin_header)
assert response.status_code == 200
def test_get_requests_admin_api_key(client, admin_api_key_env):
response = client.get("/requests", headers=admin_api_key_env)
assert response.status_code == 200
def test_get_requests_auth_disabled(client, auth_disabled_env):
response = client.get("/requests")
assert response.status_code == 200
def test_get_requests_member_forbidden(client, auth_member_header):
response = client.get("/requests", headers=auth_member_header)
assert response.status_code == 403
def test_get_requests_no_auth_unauthorized(client):
response = client.get("/requests")
assert response.status_code == 401
# --- GET /memories (no filter branch only) ---
def test_get_memories_no_filter_admin_jwt(client, auth_admin_header):
response = client.get("/memories", headers=auth_admin_header)
assert response.status_code == 200
def test_get_memories_no_filter_admin_api_key(client, admin_api_key_env):
response = client.get("/memories", headers=admin_api_key_env)
assert response.status_code == 200
def test_get_memories_no_filter_auth_disabled(client, auth_disabled_env):
response = client.get("/memories")
assert response.status_code == 200
def test_get_memories_no_filter_member_forbidden(client, auth_member_header):
"""The info-disclosure branch (_list_all_memories) is admin-only."""
response = client.get("/memories", headers=auth_member_header)
assert response.status_code == 403
def test_get_memories_filtered_member_succeeds(client, auth_member_header):
"""Filtered queries are unchanged — member can still call them."""
response = client.get("/memories?user_id=alice", headers=auth_member_header)
assert response.status_code == 200
def test_get_memories_no_auth_unauthorized(client):
response = client.get("/memories")
assert response.status_code == 401
server/tests/test_auth_helpers.py
-56
View File
@@ -1,56 +0,0 @@
"""Unit tests for _ensure_admin and require_admin in server/auth.py."""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from fastapi import HTTPException
def _fake_request(auth_type: str = "none"):
return SimpleNamespace(state=SimpleNamespace(auth_type=auth_type))
def test_ensure_admin_passes_for_admin_user(admin_user):
from auth import _ensure_admin
_ensure_admin(_fake_request("bearer"), admin_user) # must not raise
def test_ensure_admin_passes_for_admin_api_key_auth_type():
from auth import _ensure_admin
_ensure_admin(_fake_request("admin_api_key"), None) # must not raise
def test_ensure_admin_passes_for_auth_disabled():
from auth import _ensure_admin
_ensure_admin(_fake_request("disabled"), None) # must not raise
def test_ensure_admin_rejects_non_admin_user(member_user):
from auth import _ensure_admin
with pytest.raises(HTTPException) as exc:
_ensure_admin(_fake_request("bearer"), member_user)
assert exc.value.status_code == 403
assert "Admin role required" in exc.value.detail
def test_ensure_admin_rejects_no_user_with_no_legacy_path():
from auth import _ensure_admin
with pytest.raises(HTTPException) as exc:
_ensure_admin(_fake_request("none"), None)
assert exc.value.status_code == 403
def test_ensure_admin_rejects_bearer_with_no_user():
"""Defense-in-depth: a 'bearer' auth_type without a User should not pass."""
from auth import _ensure_admin
with pytest.raises(HTTPException) as exc:
_ensure_admin(_fake_request("bearer"), None)
assert exc.value.status_code == 403
server/tests/test_unchanged_routes.py
-91
View File
@@ -1,91 +0,0 @@
"""Regression tests for routes deliberately left untouched by the authz fix.
These prove the per-route changes did not accidentally regress memory CRUD,
search, or the API-key management endpoints."""
from __future__ import annotations
def test_post_memories_admin_succeeds(client, auth_admin_header):
response = client.post(
"/memories",
headers=auth_admin_header,
json={
"messages": [{"role": "user", "content": "I love pizza."}],
"user_id": "alice",
},
)
assert response.status_code == 200
def test_post_memories_requires_identifier(client, auth_admin_header):
response = client.post(
"/memories",
headers=auth_admin_header,
json={"messages": [{"role": "user", "content": "hi"}]},
)
assert response.status_code == 400
def test_get_memories_filtered_admin_succeeds(client, auth_admin_header):
response = client.get("/memories?user_id=alice", headers=auth_admin_header)
assert response.status_code == 200
def test_post_search_admin_succeeds(client, auth_admin_header):
response = client.post(
"/search",
headers=auth_admin_header,
json={"query": "pizza", "user_id": "alice"},
)
assert response.status_code == 200
def test_get_memory_by_id_admin_succeeds(client, auth_admin_header):
response = client.get("/memories/some-id", headers=auth_admin_header)
assert response.status_code == 200
def test_delete_memories_filtered_admin_succeeds(client, auth_admin_header):
response = client.delete("/memories?user_id=alice", headers=auth_admin_header)
assert response.status_code == 200
def test_delete_memories_no_filter_returns_400(client, auth_admin_header):
"""Even admin must supply an identifier for filtered delete."""
response = client.delete("/memories", headers=auth_admin_header)
assert response.status_code == 400
def test_get_configure_providers_admin_succeeds(client, auth_admin_header):
"""GET /configure/providers stays on verify_auth (not admin-gated)."""
response = client.get("/configure/providers", headers=auth_admin_header)
assert response.status_code == 200
def test_get_configure_providers_member_succeeds(client, auth_member_header):
"""Members can still query bundled providers — not admin-gated."""
response = client.get("/configure/providers", headers=auth_member_header)
assert response.status_code == 200
def test_api_keys_list_admin_succeeds(client, auth_admin_header):
response = client.get("/api-keys", headers=auth_admin_header)
assert response.status_code == 200
def test_api_keys_list_member_succeeds(client, auth_member_header):
"""API-key management remains scoped to the caller (require_auth, not require_admin)."""
response = client.get("/api-keys", headers=auth_member_header)
assert response.status_code == 200
def test_auth_me_admin_succeeds(client, auth_admin_header):
response = client.get("/auth/me", headers=auth_admin_header)
assert response.status_code == 200
def test_setup_status_no_auth(client):
"""Open route — no auth required even after the fix."""
response = client.get("/auth/setup-status")
assert response.status_code == 200