jpmschweitzerandClaude Fable 5
84e9185371
Remove the implicit jpmschweitzer default tenant (DEFAULT_USER) from src/core/multi_tenancy.py and every endpoint and request model that inherited it (~40 endpoints across /query, /wiki, /vector, /graph, /ingest, /volatile, /documents, /stats, /rag). - Add validate_required_user() + RequiredUser pydantic type in multi_tenancy and a shared require_user FastAPI dependency (RequiredUserQuery) that rejects missing, empty, and whitespace-only users with 422, following the /maintenance/* pattern. - Wiki page create / smart-create / dossier request models now require user (no fallback in wiki_service). - /maintenance/cleanup/test-data derives the tenant from the page path instead of using the production tenant collection. - Wiki.js change listener skips changes when no tenant user can be derived from the notification email instead of defaulting to the production tenant. - Consolidation service internal helpers no longer default to the production tenant. - Tool catalog marks user as required with honest descriptions. - OpenAPI descriptions updated honestly; CHANGELOG notes that callers (tatlock, Scheduler ingest tasks) must now send explicit user. - Offline tests: 422 coverage for query/body endpoints, required-user validator tests; updated legacy tests that assumed a default tenant. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
151 lines
4.6 KiB
Python
151 lines
4.6 KiB
Python
"""Tests for multi-tenancy helpers."""
|
|
|
|
import pytest
|
|
from src.core.multi_tenancy import (
|
|
sanitize_user_id,
|
|
get_qdrant_collection_name,
|
|
get_wikijs_namespace,
|
|
get_neo4j_user_label,
|
|
validate_user_id,
|
|
validate_required_user,
|
|
is_path_in_user_namespace,
|
|
)
|
|
|
|
|
|
class TestSanitizeUserId:
|
|
"""Test user ID sanitization."""
|
|
|
|
def test_lowercase_conversion(self):
|
|
assert sanitize_user_id("JohnDoe") == "johndoe"
|
|
|
|
def test_email_conversion(self):
|
|
assert sanitize_user_id("john@example.com") == "john_at_example_com"
|
|
|
|
def test_dot_conversion(self):
|
|
assert sanitize_user_id("john.doe") == "john_doe"
|
|
|
|
def test_space_conversion(self):
|
|
assert sanitize_user_id("John Doe") == "john_doe"
|
|
|
|
def test_special_chars_removal(self):
|
|
assert sanitize_user_id("john-doe!") == "john_doe"
|
|
|
|
def test_consecutive_underscores(self):
|
|
assert sanitize_user_id("john__doe") == "john_doe"
|
|
|
|
def test_leading_trailing_underscores(self):
|
|
assert sanitize_user_id("_john_") == "john"
|
|
|
|
|
|
class TestQdrantCollectionName:
|
|
"""Test Qdrant collection name generation."""
|
|
|
|
def test_simple_user(self):
|
|
assert get_qdrant_collection_name("jpmschweitzer") == "library_desk_jpmschweitzer"
|
|
|
|
def test_email_user(self):
|
|
assert get_qdrant_collection_name("john@example.com") == "library_desk_john_at_example_com"
|
|
|
|
def test_test_tenant(self):
|
|
assert get_qdrant_collection_name("llm_tester") == "library_desk_llm_tester"
|
|
|
|
|
|
class TestWikijsNamespace:
|
|
"""Test Wiki.js namespace generation."""
|
|
|
|
def test_simple_user(self):
|
|
assert get_wikijs_namespace("jpmschweitzer") == "/users/jpmschweitzer"
|
|
|
|
def test_email_user(self):
|
|
assert get_wikijs_namespace("john@example.com") == "/users/john_at_example_com"
|
|
|
|
def test_starts_with_slash(self):
|
|
namespace = get_wikijs_namespace("testuser")
|
|
assert namespace.startswith("/")
|
|
|
|
|
|
class TestNeo4jUserLabel:
|
|
"""Test Neo4j user label generation."""
|
|
|
|
def test_simple_user(self):
|
|
assert get_neo4j_user_label("jpmschweitzer") == "User_Jpmschweitzer_Document"
|
|
|
|
def test_email_user(self):
|
|
result = get_neo4j_user_label("john@example.com")
|
|
# Should be title case
|
|
assert result == "User_John_At_Example_Com_Document"
|
|
|
|
def test_title_case(self):
|
|
result = get_neo4j_user_label("john_doe")
|
|
assert result == "User_John_Doe_Document"
|
|
|
|
|
|
class TestValidateUserId:
|
|
"""Test user ID validation."""
|
|
|
|
def test_valid_simple(self):
|
|
assert validate_user_id("jpmschweitzer") is True
|
|
|
|
def test_valid_email(self):
|
|
assert validate_user_id("john@example.com") is True
|
|
|
|
def test_empty_invalid(self):
|
|
assert validate_user_id("") is False
|
|
|
|
def test_too_long_invalid(self):
|
|
assert validate_user_id("a" * 101) is False
|
|
|
|
def test_no_alphanumeric_invalid(self):
|
|
assert validate_user_id("___") is False
|
|
|
|
|
|
class TestValidateRequiredUser:
|
|
"""Test the required-user validator (no default tenant)."""
|
|
|
|
def test_no_default_user_constant(self):
|
|
"""The DEFAULT_USER escape hatch must not exist anymore."""
|
|
import src.core.multi_tenancy as mt
|
|
assert not hasattr(mt, "DEFAULT_USER")
|
|
|
|
def test_valid_user_returned(self):
|
|
assert validate_required_user("llm_tester") == "llm_tester"
|
|
|
|
def test_valid_user_stripped(self):
|
|
assert validate_required_user(" llm_tester ") == "llm_tester"
|
|
|
|
def test_empty_rejected(self):
|
|
with pytest.raises(ValueError):
|
|
validate_required_user("")
|
|
|
|
def test_whitespace_rejected(self):
|
|
with pytest.raises(ValueError):
|
|
validate_required_user(" ")
|
|
|
|
def test_none_rejected(self):
|
|
with pytest.raises(ValueError):
|
|
validate_required_user(None)
|
|
|
|
def test_no_alphanumeric_rejected(self):
|
|
with pytest.raises(ValueError):
|
|
validate_required_user("___")
|
|
|
|
def test_too_long_rejected(self):
|
|
with pytest.raises(ValueError):
|
|
validate_required_user("a" * 101)
|
|
|
|
|
|
class TestPathInNamespace:
|
|
"""Test path namespace checking."""
|
|
|
|
def test_path_in_namespace(self):
|
|
assert is_path_in_user_namespace("/users/jpmschweitzer/projects", "jpmschweitzer") is True
|
|
|
|
def test_path_not_in_namespace(self):
|
|
assert is_path_in_user_namespace("/users/other/projects", "jpmschweitzer") is False
|
|
|
|
def test_public_path_not_in_namespace(self):
|
|
assert is_path_in_user_namespace("/public/docs", "jpmschweitzer") is False
|
|
|
|
def test_root_path(self):
|
|
assert is_path_in_user_namespace("/users/test", "test") is True
|