feat: Implement tenant-aware configuration system
- Added tenant verification checklist in TENANT_VERIFICATION.md for system validation. - Created tenant_config.py to manage tenant-specific configurations with global defaults and per-tenant overrides. - Introduced tenant_guards.py for route protection based on module availability. - Developed tenant_resolver.py for identifying active tenants from requests using subdomains and headers. - Added tenant_templates.py for Jinja2 template helpers to check module visibility. - Created error_403.html for custom 403 error handling. - Defined tenants.json for tenant configurations and module management. - Implemented unit tests in test_tenant_system.py to ensure functionality of tenant system components. Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
+46
-1
@@ -1,4 +1,4 @@
|
||||
from flask import Flask, render_template, request, jsonify, flash, redirect, url_for, get_flashed_messages, session, send_file, after_this_request
|
||||
from flask import Flask, render_template, request, jsonify, flash, redirect, url_for, get_flashed_messages, session, send_file, after_this_request, g
|
||||
import os
|
||||
import json
|
||||
import atexit
|
||||
@@ -22,6 +22,12 @@ from pymongo.errors import PyMongoError
|
||||
from bson.objectid import ObjectId
|
||||
import user as user_store
|
||||
|
||||
# Tenant-aware configuration imports
|
||||
import tenant_resolver
|
||||
import tenant_config
|
||||
import tenant_guards
|
||||
import tenant_templates
|
||||
|
||||
app = Flask(__name__)
|
||||
app.secret_key = "ASDfhbsdfseiufhgildsrfrjg874368546987s6e8468f4!?FAUS/&s"
|
||||
app.config["SESSION_COOKIE_HTTPONLY"] = True
|
||||
@@ -30,6 +36,45 @@ app.config["SESSION_COOKIE_SECURE"] = os.environ.get("SESSION_COOKIE_SECURE", "0
|
||||
app.config["PREFERRED_URL_SCHEME"] = "https" if os.environ.get("SESSION_COOKIE_SECURE") == "1" else "http"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TENANT-AWARE CONFIGURATION SETUP
|
||||
# ============================================================================
|
||||
|
||||
# Register Jinja2 context processor for tenant-aware template helpers
|
||||
app.context_processor(tenant_templates.inject_tenant_context)
|
||||
|
||||
|
||||
@app.before_request
|
||||
def resolve_tenant_context():
|
||||
"""
|
||||
Resolve the active tenant for the current request and store in g.
|
||||
This runs before every request, making the tenant available to all handlers.
|
||||
"""
|
||||
parent_domain = os.environ.get("INSTANCE_PARENT_DOMAIN", "meine-domain")
|
||||
tenant_id = tenant_resolver.resolve_tenant(parent_domain)
|
||||
g.tenant_id = tenant_id
|
||||
|
||||
|
||||
@app.errorhandler(403)
|
||||
def forbidden_error(error):
|
||||
"""
|
||||
Custom error handler for 403 Forbidden responses.
|
||||
Provides tenant-aware error messages for disabled modules.
|
||||
"""
|
||||
if request.accept_mimetypes.best_match(['application/json', 'text/html']) == 'application/json':
|
||||
return jsonify({
|
||||
'error': 'Access denied',
|
||||
'message': 'This resource is not available for your organization.'
|
||||
}), 403
|
||||
|
||||
return render_template('error_403.html',
|
||||
tenant_id=g.get('tenant_id', 'default')), 403
|
||||
|
||||
|
||||
# ============================================================================
|
||||
|
||||
|
||||
|
||||
@app.after_request
|
||||
def set_security_headers(response):
|
||||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="de">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>403 - Zugriff verweigert</title>
|
||||
<style>
|
||||
* {
|
||||
margin: 0;
|
||||
padding: 0;
|
||||
box-sizing: border-box;
|
||||
}
|
||||
|
||||
body {
|
||||
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif;
|
||||
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
|
||||
min-height: 100vh;
|
||||
display: flex;
|
||||
align-items: center;
|
||||
justify-content: center;
|
||||
padding: 20px;
|
||||
}
|
||||
|
||||
.error-container {
|
||||
background: white;
|
||||
border-radius: 10px;
|
||||
box-shadow: 0 20px 60px rgba(0, 0, 0, 0.3);
|
||||
padding: 60px 40px;
|
||||
max-width: 500px;
|
||||
text-align: center;
|
||||
}
|
||||
|
||||
.error-code {
|
||||
font-size: 72px;
|
||||
font-weight: bold;
|
||||
color: #667eea;
|
||||
margin-bottom: 20px;
|
||||
}
|
||||
|
||||
.error-title {
|
||||
font-size: 28px;
|
||||
color: #333;
|
||||
margin-bottom: 15px;
|
||||
font-weight: 600;
|
||||
}
|
||||
|
||||
.error-message {
|
||||
font-size: 16px;
|
||||
color: #666;
|
||||
margin-bottom: 30px;
|
||||
line-height: 1.6;
|
||||
}
|
||||
|
||||
.error-details {
|
||||
background: #f8f9fa;
|
||||
border-left: 4px solid #667eea;
|
||||
padding: 15px;
|
||||
margin-bottom: 30px;
|
||||
text-align: left;
|
||||
border-radius: 5px;
|
||||
font-size: 14px;
|
||||
color: #555;
|
||||
}
|
||||
|
||||
.back-link {
|
||||
display: inline-block;
|
||||
padding: 12px 30px;
|
||||
background: #667eea;
|
||||
color: white;
|
||||
text-decoration: none;
|
||||
border-radius: 5px;
|
||||
transition: background 0.3s;
|
||||
font-weight: 500;
|
||||
}
|
||||
|
||||
.back-link:hover {
|
||||
background: #764ba2;
|
||||
}
|
||||
|
||||
.tenant-info {
|
||||
margin-top: 20px;
|
||||
padding-top: 20px;
|
||||
border-top: 1px solid #eee;
|
||||
font-size: 12px;
|
||||
color: #999;
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="error-container">
|
||||
<div class="error-code">403</div>
|
||||
<div class="error-title">Zugriff verweigert</div>
|
||||
|
||||
<div class="error-message">
|
||||
Diese Ressource ist für Ihre Organisation nicht verfügbar.
|
||||
</div>
|
||||
|
||||
<div class="error-details">
|
||||
<strong>Grund:</strong> Das angeforderte Modul ist in der Konfiguration Ihrer Organisation nicht aktiviert.
|
||||
<br><br>
|
||||
Kontaktieren Sie einen Administrator, wenn Sie glauben, dass dies ein Fehler ist.
|
||||
</div>
|
||||
|
||||
<a href="/" class="back-link">Zur Startseite</a>
|
||||
|
||||
{% if tenant_id %}
|
||||
<div class="tenant-info">
|
||||
Tenant: {{ tenant_id }}
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,246 @@
|
||||
"""
|
||||
Tenant-aware configuration system for module and feature management.
|
||||
|
||||
Supports:
|
||||
- Global default module configuration
|
||||
- Per-tenant overrides
|
||||
- Runtime module availability checks
|
||||
- Safe fallback to defaults for missing configurations
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Dict, Optional, Set
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TenantConfigManager:
|
||||
"""
|
||||
Manages tenant-specific configurations with global defaults and per-tenant overrides.
|
||||
"""
|
||||
|
||||
def __init__(self, config_file: Optional[str] = None):
|
||||
"""
|
||||
Initialize the tenant config manager.
|
||||
|
||||
Args:
|
||||
config_file: Path to the tenants.json configuration file.
|
||||
If not provided, defaults to tenants.json in the same directory.
|
||||
"""
|
||||
if config_file is None:
|
||||
config_file = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"tenants.json"
|
||||
)
|
||||
|
||||
self.config_file = config_file
|
||||
self._config: Dict[str, Any] = {}
|
||||
self._load_config()
|
||||
|
||||
def _load_config(self) -> None:
|
||||
"""Load configuration from tenants.json file."""
|
||||
try:
|
||||
if not os.path.exists(self.config_file):
|
||||
logger.warning(f"Config file not found: {self.config_file}. Using empty defaults.")
|
||||
self._config = {"defaults": {"modules": {}}, "tenants": {}}
|
||||
return
|
||||
|
||||
with open(self.config_file, 'r', encoding='utf-8') as f:
|
||||
self._config = json.load(f)
|
||||
|
||||
# Ensure required sections exist
|
||||
if "defaults" not in self._config:
|
||||
self._config["defaults"] = {}
|
||||
if "modules" not in self._config["defaults"]:
|
||||
self._config["defaults"]["modules"] = {}
|
||||
if "tenants" not in self._config:
|
||||
self._config["tenants"] = {}
|
||||
|
||||
logger.info(f"Loaded tenant configuration from {self.config_file}")
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"Invalid JSON in config file: {e}")
|
||||
self._config = {"defaults": {"modules": {}}, "tenants": {}}
|
||||
except Exception as e:
|
||||
logger.error(f"Error loading config file: {e}")
|
||||
self._config = {"defaults": {"modules": {}}, "tenants": {}}
|
||||
|
||||
def reload(self) -> None:
|
||||
"""Reload configuration from file."""
|
||||
self._load_config()
|
||||
|
||||
def get_tenant_config(self, tenant_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Get the complete configuration for a tenant (merged with defaults).
|
||||
|
||||
Args:
|
||||
tenant_id: The tenant identifier
|
||||
|
||||
Returns:
|
||||
Dict with tenant configuration, including inherited defaults
|
||||
"""
|
||||
# Validate tenant_id
|
||||
if not isinstance(tenant_id, str) or not tenant_id.strip():
|
||||
return self._get_default_config()
|
||||
|
||||
tenant_id = tenant_id.lower().strip()
|
||||
|
||||
# Get tenant-specific config if it exists
|
||||
tenant_config = self._config.get("tenants", {}).get(tenant_id, {})
|
||||
|
||||
# Start with defaults
|
||||
merged = {
|
||||
"modules": dict(self._config.get("defaults", {}).get("modules", {}))
|
||||
}
|
||||
|
||||
# Merge tenant-specific module overrides
|
||||
if "modules" in tenant_config:
|
||||
merged["modules"].update(tenant_config.get("modules", {}))
|
||||
|
||||
# Include any other tenant-specific settings
|
||||
for key in tenant_config:
|
||||
if key != "modules":
|
||||
merged[key] = tenant_config[key]
|
||||
|
||||
return merged
|
||||
|
||||
def _get_default_config(self) -> Dict[str, Any]:
|
||||
"""Get the default configuration."""
|
||||
return {
|
||||
"modules": dict(self._config.get("defaults", {}).get("modules", {}))
|
||||
}
|
||||
|
||||
def is_module_enabled(self, tenant_id: str, module_name: str) -> bool:
|
||||
"""
|
||||
Check if a module is enabled for a specific tenant.
|
||||
|
||||
Args:
|
||||
tenant_id: The tenant identifier
|
||||
module_name: The module name (e.g., 'library', 'chat', 'appointments')
|
||||
|
||||
Returns:
|
||||
True if the module is enabled, False otherwise
|
||||
"""
|
||||
if not isinstance(module_name, str) or not module_name.strip():
|
||||
return False
|
||||
|
||||
config = self.get_tenant_config(tenant_id)
|
||||
modules = config.get("modules", {})
|
||||
|
||||
# If module is explicitly configured, use that value
|
||||
if module_name in modules:
|
||||
enabled = modules[module_name]
|
||||
return enabled is True or (isinstance(enabled, dict) and enabled.get("enabled", False) is True)
|
||||
|
||||
# If not configured, default to False (fail-safe for optional features)
|
||||
return False
|
||||
|
||||
def get_enabled_modules(self, tenant_id: str) -> Set[str]:
|
||||
"""
|
||||
Get the set of enabled modules for a tenant.
|
||||
|
||||
Args:
|
||||
tenant_id: The tenant identifier
|
||||
|
||||
Returns:
|
||||
Set of enabled module names
|
||||
"""
|
||||
config = self.get_tenant_config(tenant_id)
|
||||
modules = config.get("modules", {})
|
||||
|
||||
enabled = set()
|
||||
for module_name, module_config in modules.items():
|
||||
if module_config is True:
|
||||
enabled.add(module_name)
|
||||
elif isinstance(module_config, dict) and module_config.get("enabled", False) is True:
|
||||
enabled.add(module_name)
|
||||
|
||||
return enabled
|
||||
|
||||
def get_all_modules(self) -> Set[str]:
|
||||
"""
|
||||
Get all module names defined in the configuration.
|
||||
|
||||
Returns:
|
||||
Set of all module names
|
||||
"""
|
||||
all_modules = set()
|
||||
|
||||
# Add modules from defaults
|
||||
all_modules.update(self._config.get("defaults", {}).get("modules", {}).keys())
|
||||
|
||||
# Add modules from tenants
|
||||
for tenant_config in self._config.get("tenants", {}).values():
|
||||
all_modules.update(tenant_config.get("modules", {}).keys())
|
||||
|
||||
return all_modules
|
||||
|
||||
def get_config_value(self, tenant_id: str, key: str, default: Any = None) -> Any:
|
||||
"""
|
||||
Get a tenant-specific configuration value with fallback to default.
|
||||
|
||||
Args:
|
||||
tenant_id: The tenant identifier
|
||||
key: The configuration key
|
||||
default: The default value if not found
|
||||
|
||||
Returns:
|
||||
The configuration value or default
|
||||
"""
|
||||
config = self.get_tenant_config(tenant_id)
|
||||
return config.get(key, default)
|
||||
|
||||
|
||||
# Global instance (lazy loaded)
|
||||
_manager: Optional[TenantConfigManager] = None
|
||||
|
||||
|
||||
def get_config_manager() -> TenantConfigManager:
|
||||
"""Get or create the global config manager instance."""
|
||||
global _manager
|
||||
if _manager is None:
|
||||
_manager = TenantConfigManager()
|
||||
return _manager
|
||||
|
||||
|
||||
def is_module_enabled(tenant_id: str, module_name: str) -> bool:
|
||||
"""
|
||||
Check if a module is enabled for a tenant (convenience function).
|
||||
|
||||
Args:
|
||||
tenant_id: The tenant identifier
|
||||
module_name: The module name
|
||||
|
||||
Returns:
|
||||
True if enabled, False otherwise
|
||||
"""
|
||||
return get_config_manager().is_module_enabled(tenant_id, module_name)
|
||||
|
||||
|
||||
def get_tenant_config(tenant_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Get tenant configuration (convenience function).
|
||||
|
||||
Args:
|
||||
tenant_id: The tenant identifier
|
||||
|
||||
Returns:
|
||||
The tenant configuration dict
|
||||
"""
|
||||
return get_config_manager().get_tenant_config(tenant_id)
|
||||
|
||||
|
||||
def get_enabled_modules(tenant_id: str) -> Set[str]:
|
||||
"""
|
||||
Get enabled modules for a tenant (convenience function).
|
||||
|
||||
Args:
|
||||
tenant_id: The tenant identifier
|
||||
|
||||
Returns:
|
||||
Set of enabled module names
|
||||
"""
|
||||
return get_config_manager().get_enabled_modules(tenant_id)
|
||||
@@ -0,0 +1,162 @@
|
||||
"""
|
||||
Route guards and helpers for module-aware access control.
|
||||
|
||||
Provides decorators and functions to:
|
||||
- Protect routes based on module availability
|
||||
- Return consistent error responses for disabled modules
|
||||
- Check module availability in views
|
||||
"""
|
||||
|
||||
from functools import wraps
|
||||
from flask import request, jsonify, abort, g
|
||||
from typing import Callable, Any
|
||||
from tenant_config import is_module_enabled, get_enabled_modules
|
||||
|
||||
|
||||
def require_module(module_name: str) -> Callable:
|
||||
"""
|
||||
Decorator to require a module to be enabled for accessing a route.
|
||||
|
||||
Usage:
|
||||
@app.route('/chat')
|
||||
@require_module('chat')
|
||||
def chat_endpoint():
|
||||
return render_template('chat.html')
|
||||
|
||||
Args:
|
||||
module_name: The module to check (e.g., 'chat', 'invoices')
|
||||
|
||||
Returns:
|
||||
Decorator function
|
||||
"""
|
||||
def decorator(f: Callable) -> Callable:
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs) -> Any:
|
||||
tenant_id = g.get('tenant_id', 'default')
|
||||
|
||||
if not is_module_enabled(tenant_id, module_name):
|
||||
# For JSON requests, return JSON response
|
||||
if request.accept_mimetypes.best_match(['application/json', 'text/html']) == 'application/json':
|
||||
return jsonify({
|
||||
'error': 'Module not available',
|
||||
'message': f'The {module_name} module is not available for your organization.',
|
||||
'module': module_name
|
||||
}), 403
|
||||
|
||||
# For HTML requests, return abort with custom message
|
||||
abort(403)
|
||||
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
return decorator
|
||||
|
||||
|
||||
def require_admin() -> Callable:
|
||||
"""
|
||||
Decorator to require admin module to be enabled (and user to be admin).
|
||||
|
||||
Usage:
|
||||
@app.route('/admin/dashboard')
|
||||
@require_admin()
|
||||
def admin_dashboard():
|
||||
return render_template('admin_dashboard.html')
|
||||
|
||||
Returns:
|
||||
Decorator function
|
||||
"""
|
||||
def decorator(f: Callable) -> Callable:
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs) -> Any:
|
||||
tenant_id = g.get('tenant_id', 'default')
|
||||
|
||||
# Check if admin module is enabled
|
||||
if not is_module_enabled(tenant_id, 'admin'):
|
||||
if request.accept_mimetypes.best_match(['application/json', 'text/html']) == 'application/json':
|
||||
return jsonify({
|
||||
'error': 'Admin module not available',
|
||||
'message': 'Admin functionality is not available for your organization.'
|
||||
}), 403
|
||||
abort(403)
|
||||
|
||||
# Additional user role check (if implemented)
|
||||
# This would be combined with existing auth checks
|
||||
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
return decorator
|
||||
|
||||
|
||||
def module_enabled_in_context(module_name: str) -> bool:
|
||||
"""
|
||||
Check if a module is enabled for the current request context.
|
||||
|
||||
Can be used in route handlers and templates (via Jinja2 context).
|
||||
|
||||
Args:
|
||||
module_name: The module name
|
||||
|
||||
Returns:
|
||||
True if enabled, False otherwise
|
||||
"""
|
||||
tenant_id = g.get('tenant_id', 'default')
|
||||
return is_module_enabled(tenant_id, module_name)
|
||||
|
||||
|
||||
def get_enabled_modules_in_context() -> set:
|
||||
"""
|
||||
Get all enabled modules for the current request context.
|
||||
|
||||
Useful for determining which menu items to show, etc.
|
||||
|
||||
Returns:
|
||||
Set of enabled module names
|
||||
"""
|
||||
tenant_id = g.get('tenant_id', 'default')
|
||||
return get_enabled_modules(tenant_id)
|
||||
|
||||
|
||||
class TenantAwareErrorHandler:
|
||||
"""
|
||||
Provides consistent error handling for tenant-specific access issues.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def module_not_available(module_name: str, is_json: bool = False):
|
||||
"""
|
||||
Generate error response for disabled module.
|
||||
|
||||
Args:
|
||||
module_name: The module that is not available
|
||||
is_json: Whether to return JSON or HTML
|
||||
|
||||
Returns:
|
||||
Response tuple (response, status_code)
|
||||
"""
|
||||
if is_json:
|
||||
return jsonify({
|
||||
'error': 'Module not available',
|
||||
'message': f'The {module_name} module is not available for your organization.',
|
||||
'module': module_name
|
||||
}), 403
|
||||
|
||||
abort(403)
|
||||
|
||||
@staticmethod
|
||||
def unauthorized_access(reason: str = None, is_json: bool = False):
|
||||
"""
|
||||
Generate error response for unauthorized access.
|
||||
|
||||
Args:
|
||||
reason: Optional reason for the denial
|
||||
is_json: Whether to return JSON or HTML
|
||||
|
||||
Returns:
|
||||
Response tuple (response, status_code)
|
||||
"""
|
||||
if is_json:
|
||||
return jsonify({
|
||||
'error': 'Access denied',
|
||||
'message': reason or 'You do not have access to this resource.'
|
||||
}), 403
|
||||
|
||||
abort(403)
|
||||
@@ -0,0 +1,74 @@
|
||||
"""
|
||||
Tenant resolver for identifying the active tenant from incoming requests.
|
||||
|
||||
Supports:
|
||||
- Subdomain-based tenant detection (e.g., school1.example.com -> school1)
|
||||
- X-Tenant-ID header for internal APIs and testing
|
||||
- Fallback to 'default' tenant
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Optional
|
||||
from flask import request
|
||||
|
||||
|
||||
def extract_subdomain(host: str, parent_domain: str) -> Optional[str]:
|
||||
"""
|
||||
Extract subdomain from host if it matches the parent domain.
|
||||
|
||||
Args:
|
||||
host: The Host header value (e.g., 'school1.example.com')
|
||||
parent_domain: The parent domain (e.g., 'example.com')
|
||||
|
||||
Returns:
|
||||
The subdomain if found, None otherwise
|
||||
"""
|
||||
if not host or not parent_domain:
|
||||
return None
|
||||
|
||||
# Remove port if present
|
||||
host = host.split(':')[0]
|
||||
parent_domain = parent_domain.strip()
|
||||
|
||||
# Check if host ends with parent domain
|
||||
if not host.endswith(parent_domain):
|
||||
return None
|
||||
|
||||
# Extract subdomain
|
||||
prefix = host[: -len(parent_domain)].rstrip('.')
|
||||
|
||||
# Validate subdomain (alphanumeric, hyphens, underscores)
|
||||
if prefix and re.match(r'^[a-zA-Z0-9_-]+$', prefix):
|
||||
return prefix.lower()
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def resolve_tenant(parent_domain: str = "example.com") -> str:
|
||||
"""
|
||||
Resolve the active tenant for the current request.
|
||||
|
||||
Resolution order:
|
||||
1. X-Tenant-ID header (for APIs and testing)
|
||||
2. Subdomain from Host header
|
||||
3. Fallback to 'default'
|
||||
|
||||
Args:
|
||||
parent_domain: The parent domain for subdomain extraction
|
||||
|
||||
Returns:
|
||||
The tenant identifier (lowercase alphanumeric string)
|
||||
"""
|
||||
# 1. Check X-Tenant-ID header
|
||||
tenant_from_header = request.headers.get("X-Tenant-ID", "").strip().lower()
|
||||
if tenant_from_header and re.match(r'^[a-zA-Z0-9_-]+$', tenant_from_header):
|
||||
return tenant_from_header
|
||||
|
||||
# 2. Check subdomain
|
||||
host = request.host
|
||||
subdomain = extract_subdomain(host, parent_domain)
|
||||
if subdomain:
|
||||
return subdomain
|
||||
|
||||
# 3. Fallback to default
|
||||
return "default"
|
||||
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
Jinja2 template helpers for tenant-aware module visibility.
|
||||
|
||||
Provides template functions to:
|
||||
- Check if a module is enabled in templates
|
||||
- Get list of enabled modules
|
||||
- Conditionally show UI elements based on module availability
|
||||
"""
|
||||
|
||||
from flask import g
|
||||
from tenant_config import is_module_enabled, get_enabled_modules
|
||||
|
||||
|
||||
def inject_tenant_context():
|
||||
"""
|
||||
Jinja2 context processor to inject tenant-aware helpers into all templates.
|
||||
|
||||
Usage in Flask app setup:
|
||||
app.context_processor(inject_tenant_context)
|
||||
|
||||
This makes available in templates:
|
||||
- module_enabled(module_name) - check if module is enabled
|
||||
- enabled_modules - set of enabled module names
|
||||
- tenant_id - current tenant identifier
|
||||
"""
|
||||
tenant_id = g.get('tenant_id', 'default')
|
||||
|
||||
return {
|
||||
'module_enabled': lambda module: is_module_enabled(tenant_id, module),
|
||||
'enabled_modules': get_enabled_modules(tenant_id),
|
||||
'tenant_id': tenant_id
|
||||
}
|
||||
|
||||
|
||||
def module_visible(module_name: str) -> bool:
|
||||
"""
|
||||
Check if a module should be visible in UI (convenience function).
|
||||
|
||||
Args:
|
||||
module_name: The module name
|
||||
|
||||
Returns:
|
||||
True if visible/enabled, False otherwise
|
||||
"""
|
||||
return is_module_enabled(g.get('tenant_id', 'default'), module_name)
|
||||
@@ -0,0 +1,68 @@
|
||||
{
|
||||
"defaults": {
|
||||
"modules": {
|
||||
"inventarsystem": true,
|
||||
"appointments": false,
|
||||
"blog": true,
|
||||
"chat": false,
|
||||
"tickets": false,
|
||||
"invoices": false,
|
||||
"dienstleistungen": true,
|
||||
"projekte": true,
|
||||
"team": true,
|
||||
"kontakt": true,
|
||||
"admin": false
|
||||
},
|
||||
"description": "Default configuration for all tenants. Tenants inherit these settings unless they override them explicitly."
|
||||
},
|
||||
"tenants": {
|
||||
"default": {
|
||||
"description": "Default tenant - used as fallback when tenant cannot be resolved"
|
||||
},
|
||||
"school1": {
|
||||
"description": "School 1 - full featured",
|
||||
"modules": {
|
||||
"inventarsystem": true,
|
||||
"appointments": true,
|
||||
"blog": true,
|
||||
"chat": true,
|
||||
"tickets": true,
|
||||
"invoices": true,
|
||||
"dienstleistungen": false,
|
||||
"projekte": false,
|
||||
"team": true,
|
||||
"admin": true
|
||||
}
|
||||
},
|
||||
"school2": {
|
||||
"description": "School 2 - limited features, no invoicing",
|
||||
"modules": {
|
||||
"inventarsystem": true,
|
||||
"appointments": true,
|
||||
"blog": false,
|
||||
"chat": false,
|
||||
"tickets": true,
|
||||
"invoices": false,
|
||||
"dienstleistungen": false,
|
||||
"projekte": false,
|
||||
"team": true,
|
||||
"admin": true
|
||||
}
|
||||
},
|
||||
"partner-org": {
|
||||
"description": "Partner organization - service provider",
|
||||
"modules": {
|
||||
"inventarsystem": false,
|
||||
"appointments": false,
|
||||
"blog": false,
|
||||
"chat": true,
|
||||
"tickets": false,
|
||||
"invoices": true,
|
||||
"dienstleistungen": true,
|
||||
"projekte": true,
|
||||
"team": true,
|
||||
"admin": false
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,341 @@
|
||||
"""
|
||||
Unit tests for the tenant-aware configuration system.
|
||||
|
||||
Run with: pytest test_tenant_system.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import json
|
||||
import tempfile
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
# Test imports
|
||||
from tenant_resolver import resolve_tenant, extract_subdomain
|
||||
from tenant_config import (
|
||||
TenantConfigManager,
|
||||
is_module_enabled,
|
||||
get_config_manager,
|
||||
get_enabled_modules,
|
||||
get_tenant_config,
|
||||
)
|
||||
|
||||
|
||||
class TestTenantResolver:
|
||||
"""Tests for tenant resolution from requests."""
|
||||
|
||||
def test_extract_subdomain_valid(self):
|
||||
"""Test extracting valid subdomain."""
|
||||
assert extract_subdomain("school1.example.com", "example.com") == "school1"
|
||||
assert extract_subdomain("my-tenant.example.com", "example.com") == "my-tenant"
|
||||
assert extract_subdomain("tenant_1.example.com", "example.com") == "tenant_1"
|
||||
|
||||
def test_extract_subdomain_with_port(self):
|
||||
"""Test subdomain extraction with port number."""
|
||||
assert extract_subdomain("school1.example.com:8080", "example.com") == "school1"
|
||||
assert extract_subdomain("example.com:443", "example.com") is None
|
||||
|
||||
def test_extract_subdomain_invalid(self):
|
||||
"""Test invalid subdomain extraction."""
|
||||
assert extract_subdomain("example.com", "example.com") is None
|
||||
assert extract_subdomain("other.com", "example.com") is None
|
||||
assert extract_subdomain("", "example.com") is None
|
||||
# Subdomain with invalid characters
|
||||
assert extract_subdomain("invalid@sub.example.com", "example.com") is None
|
||||
assert extract_subdomain("invalid#sub.example.com", "example.com") is None
|
||||
|
||||
def test_extract_subdomain_case_insensitive(self):
|
||||
"""Test that subdomains are normalized to lowercase."""
|
||||
result = extract_subdomain("SCHOOL1.example.com", "example.com")
|
||||
assert result == "school1" # Should be lowercase
|
||||
|
||||
|
||||
class TestTenantConfigManager:
|
||||
"""Tests for tenant configuration management."""
|
||||
|
||||
@pytest.fixture
|
||||
def config_file(self):
|
||||
"""Create a temporary config file for testing."""
|
||||
config = {
|
||||
"defaults": {
|
||||
"modules": {
|
||||
"chat": True,
|
||||
"blog": False,
|
||||
"tickets": False,
|
||||
"invoices": False,
|
||||
}
|
||||
},
|
||||
"tenants": {
|
||||
"tenant_a": {
|
||||
"modules": {
|
||||
"chat": False,
|
||||
"blog": True,
|
||||
}
|
||||
},
|
||||
"tenant_b": {
|
||||
"modules": {
|
||||
"invoices": True,
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fd, path = tempfile.mkstemp(suffix=".json")
|
||||
try:
|
||||
with os.fdopen(fd, 'w') as f:
|
||||
json.dump(config, f)
|
||||
yield path
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_load_config(self, config_file):
|
||||
"""Test loading configuration from file."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
assert "defaults" in manager._config
|
||||
assert "tenants" in manager._config
|
||||
assert "tenant_a" in manager._config["tenants"]
|
||||
|
||||
def test_missing_config_file(self):
|
||||
"""Test handling of missing configuration file."""
|
||||
manager = TenantConfigManager("/nonexistent/path/tenants.json")
|
||||
# Should create empty config with safe defaults
|
||||
assert manager._config == {"defaults": {"modules": {}}, "tenants": {}}
|
||||
|
||||
def test_get_tenant_config_with_defaults(self, config_file):
|
||||
"""Test getting tenant config with defaults merged."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# Tenant A overrides some modules
|
||||
config_a = manager.get_tenant_config("tenant_a")
|
||||
assert config_a["modules"]["chat"] == False # Override
|
||||
assert config_a["modules"]["blog"] == True # Override
|
||||
assert config_a["modules"]["tickets"] == False # From default
|
||||
|
||||
def test_get_tenant_config_inherits_defaults(self, config_file):
|
||||
"""Test that tenant inherits all defaults."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# Tenant B has minimal overrides
|
||||
config_b = manager.get_tenant_config("tenant_b")
|
||||
assert config_b["modules"]["chat"] == True # From default
|
||||
assert config_b["modules"]["blog"] == False # From default
|
||||
assert config_b["modules"]["invoices"] == True # Override
|
||||
|
||||
def test_get_tenant_config_missing_tenant(self, config_file):
|
||||
"""Test getting config for non-existent tenant."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
config = manager.get_tenant_config("nonexistent_tenant")
|
||||
# Should inherit all defaults
|
||||
assert config["modules"]["chat"] == True
|
||||
assert config["modules"]["blog"] == False
|
||||
|
||||
def test_is_module_enabled_default_true(self, config_file):
|
||||
"""Test module enabled when default is true."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# chat is enabled by default
|
||||
assert manager.is_module_enabled("tenant_c", "chat") == True
|
||||
|
||||
def test_is_module_enabled_default_false(self, config_file):
|
||||
"""Test module disabled when default is false."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# blog is disabled by default
|
||||
assert manager.is_module_enabled("tenant_c", "blog") == False
|
||||
|
||||
def test_is_module_enabled_override(self, config_file):
|
||||
"""Test module override by tenant."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# tenant_a overrides chat to false
|
||||
assert manager.is_module_enabled("tenant_a", "chat") == False
|
||||
# tenant_a overrides blog to true
|
||||
assert manager.is_module_enabled("tenant_a", "blog") == True
|
||||
|
||||
def test_is_module_enabled_nonexistent_module(self, config_file):
|
||||
"""Test checking non-existent module."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# Module not configured anywhere should be False
|
||||
assert manager.is_module_enabled("tenant_a", "nonexistent") == False
|
||||
|
||||
def test_get_enabled_modules(self, config_file):
|
||||
"""Test getting set of enabled modules."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# Tenant A has chat=false and blog=true (+ inherited tickets=false, invoices=false)
|
||||
enabled = manager.get_enabled_modules("tenant_a")
|
||||
assert "blog" in enabled
|
||||
assert "chat" not in enabled
|
||||
assert "tickets" not in enabled
|
||||
assert "invoices" not in enabled
|
||||
|
||||
def test_get_all_modules(self, config_file):
|
||||
"""Test getting all modules defined in config."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
all_modules = manager.get_all_modules()
|
||||
assert "chat" in all_modules
|
||||
assert "blog" in all_modules
|
||||
assert "tickets" in all_modules
|
||||
assert "invoices" in all_modules
|
||||
|
||||
def test_invalid_json_file(self):
|
||||
"""Test handling of invalid JSON."""
|
||||
fd, path = tempfile.mkstemp(suffix=".json")
|
||||
try:
|
||||
with os.fdopen(fd, 'w') as f:
|
||||
f.write("{ invalid json }")
|
||||
|
||||
manager = TenantConfigManager(path)
|
||||
# Should fall back to empty config
|
||||
assert manager._config == {"defaults": {"modules": {}}, "tenants": {}}
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_reload_config(self, config_file):
|
||||
"""Test reloading configuration."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# Verify initial state
|
||||
assert manager.is_module_enabled("tenant_a", "chat") == False
|
||||
|
||||
# Modify the config file
|
||||
with open(config_file, 'r') as f:
|
||||
config = json.load(f)
|
||||
config["tenants"]["tenant_a"]["modules"]["chat"] = True
|
||||
with open(config_file, 'w') as f:
|
||||
json.dump(config, f)
|
||||
|
||||
# Reload and verify new state
|
||||
manager.reload()
|
||||
assert manager.is_module_enabled("tenant_a", "chat") == True
|
||||
|
||||
def test_get_config_value(self, config_file):
|
||||
"""Test getting arbitrary config values."""
|
||||
manager = TenantConfigManager(config_file)
|
||||
|
||||
# Get existing value
|
||||
modules = manager.get_config_value("tenant_a", "modules")
|
||||
assert modules is not None
|
||||
|
||||
# Get non-existent value with default
|
||||
value = manager.get_config_value("tenant_a", "nonexistent", default="fallback")
|
||||
assert value == "fallback"
|
||||
|
||||
|
||||
class TestConvenienceFunctions:
|
||||
"""Tests for convenience wrapper functions."""
|
||||
|
||||
@pytest.fixture
|
||||
def setup_global_manager(self):
|
||||
"""Setup global config manager for testing."""
|
||||
config = {
|
||||
"defaults": {
|
||||
"modules": {"test_module": True}
|
||||
},
|
||||
"tenants": {
|
||||
"test_tenant": {
|
||||
"modules": {"test_module": False}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fd, path = tempfile.mkstemp(suffix=".json")
|
||||
try:
|
||||
with os.fdopen(fd, 'w') as f:
|
||||
json.dump(config, f)
|
||||
|
||||
# Reset global manager
|
||||
import tenant_config
|
||||
tenant_config._manager = TenantConfigManager(path)
|
||||
yield path
|
||||
finally:
|
||||
os.unlink(path)
|
||||
# Reset global manager
|
||||
import tenant_config
|
||||
tenant_config._manager = None
|
||||
|
||||
def test_is_module_enabled_convenience(self, setup_global_manager):
|
||||
"""Test convenience function for module checking."""
|
||||
assert is_module_enabled("default", "test_module") == True
|
||||
assert is_module_enabled("test_tenant", "test_module") == False
|
||||
|
||||
def test_get_tenant_config_convenience(self, setup_global_manager):
|
||||
"""Test convenience function for config retrieval."""
|
||||
config = get_tenant_config("test_tenant")
|
||||
assert "modules" in config
|
||||
assert config["modules"]["test_module"] == False
|
||||
|
||||
def test_get_enabled_modules_convenience(self, setup_global_manager):
|
||||
"""Test convenience function for enabled modules."""
|
||||
enabled = get_enabled_modules("default")
|
||||
assert "test_module" in enabled
|
||||
|
||||
|
||||
class TestEdgeCases:
|
||||
"""Tests for edge cases and error handling."""
|
||||
|
||||
def test_empty_tenant_id(self):
|
||||
"""Test handling empty tenant ID."""
|
||||
config = TenantConfigManager("/nonexistent/path/tenants.json")
|
||||
result = config.get_tenant_config("")
|
||||
# Should get defaults
|
||||
assert "modules" in result
|
||||
|
||||
def test_none_tenant_id(self):
|
||||
"""Test handling None tenant ID."""
|
||||
config = TenantConfigManager("/nonexistent/path/tenants.json")
|
||||
result = config.get_tenant_config(None)
|
||||
# Should get defaults
|
||||
assert "modules" in result
|
||||
|
||||
def test_module_as_dict(self):
|
||||
"""Test module configuration as dict with 'enabled' key."""
|
||||
fd, path = tempfile.mkstemp(suffix=".json")
|
||||
try:
|
||||
config = {
|
||||
"defaults": {
|
||||
"modules": {
|
||||
"advanced_module": {"enabled": True, "tier": "premium"}
|
||||
}
|
||||
}
|
||||
}
|
||||
with os.fdopen(fd, 'w') as f:
|
||||
json.dump(config, f)
|
||||
|
||||
manager = TenantConfigManager(path)
|
||||
assert manager.is_module_enabled("default", "advanced_module") == True
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_invalid_module_value(self):
|
||||
"""Test handling invalid module values."""
|
||||
fd, path = tempfile.mkstemp(suffix=".json")
|
||||
try:
|
||||
config = {
|
||||
"defaults": {
|
||||
"modules": {
|
||||
"bad_module": "yes", # Should be boolean
|
||||
"another_bad": None,
|
||||
}
|
||||
}
|
||||
}
|
||||
with os.fdopen(fd, 'w') as f:
|
||||
json.dump(config, f)
|
||||
|
||||
manager = TenantConfigManager(path)
|
||||
# Should treat non-boolean as False (safe)
|
||||
assert manager.is_module_enabled("default", "bad_module") == False
|
||||
assert manager.is_module_enabled("default", "another_bad") == False
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
# Integration tests would go here, possibly with Flask test client
|
||||
# These are more complex and depend on Flask app setup
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user