feat: add comprehensive unit and integration testing infrastructure
- Backend: pytest with unit tests (auth, rbac, config) and integration tests (auth flow, docker, files) - Frontend: vitest with unit tests (api client) and component tests (login) - CI: Gitea Actions workflow for automated testing with coverage reports - Documentation: TESTING.md guide with setup, usage, and best practices - Coverage goals: 80%+ backend, 70%+ frontend
This commit is contained in:
Gan, Jimmy
@@ -0,0 +1 @@
|
||||
# Backend tests
|
||||
@@ -0,0 +1,192 @@
|
||||
"""
|
||||
Shared pytest fixtures for backend tests.
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
import pytest
|
||||
from datetime import timedelta
|
||||
from fastapi.testclient import TestClient
|
||||
from unittest.mock import Mock, AsyncMock, patch
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_volume_root(tmp_path):
|
||||
"""Create temporary volume root for testing."""
|
||||
volume_root = tmp_path / "volume1"
|
||||
volume_root.mkdir()
|
||||
(volume_root / "docker" / "nas-dashboard").mkdir(parents=True)
|
||||
return str(volume_root)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config(temp_volume_root, monkeypatch):
|
||||
"""Mock config module with test values."""
|
||||
monkeypatch.setenv("SECRET_KEY", "test-secret-key-at-least-32-chars-long")
|
||||
monkeypatch.setenv("ADMIN_USER", "testadmin")
|
||||
monkeypatch.setenv("ADMIN_PASSWORD_HASH", "$pbkdf2-sha256$29000$N2YMIQQgBAAgxBgjhPD.vw$1234567890abcdef")
|
||||
monkeypatch.setenv("VOLUME_ROOT", temp_volume_root)
|
||||
monkeypatch.setenv("DOCKER_HOST", "tcp://mock-docker:2375")
|
||||
monkeypatch.setenv("GITEA_URL", "http://mock-gitea:3000")
|
||||
monkeypatch.setenv("GITEA_TOKEN", "mock-token")
|
||||
|
||||
# Reload config module to pick up new env vars
|
||||
import config
|
||||
import importlib
|
||||
importlib.reload(config)
|
||||
|
||||
return config
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_auth_file(temp_volume_root):
|
||||
"""Create temporary auth.json file."""
|
||||
auth_file = os.path.join(temp_volume_root, "docker/nas-dashboard/auth.json")
|
||||
data = {
|
||||
"totp_secret": "",
|
||||
"passkey_credentials": [],
|
||||
"token_version": 0,
|
||||
"last_totp_ts": 0
|
||||
}
|
||||
with open(auth_file, "w") as f:
|
||||
json.dump(data, f)
|
||||
return auth_file
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_rbac_file(temp_volume_root):
|
||||
"""Create temporary rbac.json file."""
|
||||
rbac_file = os.path.join(temp_volume_root, "docker/nas-dashboard/rbac.json")
|
||||
data = {
|
||||
"role_defaults": {
|
||||
"admin": {"pages": "*", "sidebar_links": "*"},
|
||||
"member": {"pages": ["dashboard", "docker"], "sidebar_links": "*"},
|
||||
"viewer": {"pages": ["dashboard"], "sidebar_links": ["dashboard"]}
|
||||
},
|
||||
"user_overrides": {}
|
||||
}
|
||||
with open(rbac_file, "w") as f:
|
||||
json.dump(data, f)
|
||||
return rbac_file
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_access_token(mock_config):
|
||||
"""Generate a valid access token for testing."""
|
||||
from auth import create_access_token
|
||||
return create_access_token(
|
||||
data={"sub": "testuser", "role": "admin"},
|
||||
expires_delta=timedelta(minutes=30)
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_refresh_token(mock_config, temp_auth_file):
|
||||
"""Generate a valid refresh token for testing."""
|
||||
from auth import create_refresh_token
|
||||
return create_refresh_token(data={"sub": "testuser", "role": "admin"})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def expired_access_token(mock_config):
|
||||
"""Generate an expired access token for testing."""
|
||||
from auth import create_access_token
|
||||
return create_access_token(
|
||||
data={"sub": "testuser", "role": "admin"},
|
||||
expires_delta=timedelta(seconds=-1)
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_docker_client():
|
||||
"""Mock Docker client for testing."""
|
||||
client = Mock()
|
||||
|
||||
# Mock container
|
||||
container = Mock()
|
||||
container.id = "abc123"
|
||||
container.name = "test-container"
|
||||
container.status = "running"
|
||||
container.attrs = {
|
||||
"State": {"Health": {"Status": "healthy"}},
|
||||
"Config": {"Image": "test-image:latest"}
|
||||
}
|
||||
container.logs.return_value = b"test log line 1\ntest log line 2\n"
|
||||
|
||||
client.containers.list.return_value = [container]
|
||||
client.containers.get.return_value = container
|
||||
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ssh_connection():
|
||||
"""Mock asyncssh connection for testing."""
|
||||
conn = AsyncMock()
|
||||
process = AsyncMock()
|
||||
process.stdout = AsyncMock()
|
||||
process.stdin = AsyncMock()
|
||||
process.stderr = AsyncMock()
|
||||
conn.create_process.return_value = process
|
||||
return conn
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_httpx_client():
|
||||
"""Mock httpx AsyncClient for testing."""
|
||||
client = AsyncMock()
|
||||
response = AsyncMock()
|
||||
response.status_code = 200
|
||||
response.json.return_value = {"status": "ok"}
|
||||
client.get.return_value = response
|
||||
client.post.return_value = response
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_app(mock_config, temp_auth_file, temp_rbac_file, mock_docker_client):
|
||||
"""Create FastAPI test client with mocked dependencies."""
|
||||
# Patch Docker client before importing main
|
||||
with patch("docker.DockerClient", return_value=mock_docker_client):
|
||||
from main import app
|
||||
client = TestClient(app)
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_headers(valid_access_token):
|
||||
"""Headers with admin access token."""
|
||||
return {"Authorization": f"Bearer {valid_access_token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_rbac_data():
|
||||
"""Sample RBAC configuration for testing."""
|
||||
return {
|
||||
"role_defaults": {
|
||||
"admin": {"pages": "*", "sidebar_links": "*"},
|
||||
"member": {"pages": ["dashboard", "docker", "files"], "sidebar_links": "*"},
|
||||
"viewer": {"pages": ["dashboard"], "sidebar_links": ["dashboard"]}
|
||||
},
|
||||
"user_overrides": {
|
||||
"customuser": {
|
||||
"pages": ["dashboard", "docker"],
|
||||
"sidebar_links": ["dashboard", "docker", "files"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_totp_secret():
|
||||
"""Sample TOTP secret for testing."""
|
||||
return "JBSWY3DPEHPK3PXP" # Base32 encoded secret
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_request():
|
||||
"""Mock FastAPI Request object."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "192.168.31.100"
|
||||
request.headers = {}
|
||||
return request
|
||||
@@ -0,0 +1 @@
|
||||
# Integration tests
|
||||
@@ -0,0 +1,318 @@
|
||||
"""
|
||||
Integration tests for authentication flow.
|
||||
"""
|
||||
import pytest
|
||||
import pyotp
|
||||
from fastapi import status
|
||||
from auth import get_password_hash, save_totp_secret, save_password_hash
|
||||
|
||||
|
||||
class TestLoginFlow:
|
||||
"""Test login endpoint."""
|
||||
|
||||
def test_login_success_no_2fa(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test successful login without 2FA."""
|
||||
# Set up password
|
||||
password = "test_password"
|
||||
hashed = get_password_hash(password)
|
||||
save_password_hash(hashed)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "testadmin", "password": password, "totp_code": ""}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["user"] == "testadmin"
|
||||
assert data["role"] == "admin"
|
||||
assert data["has_2fa"] is False
|
||||
assert "access_token" in data
|
||||
assert "refresh_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
|
||||
def test_login_success_with_2fa(self, test_app, mock_config, temp_auth_file, sample_totp_secret):
|
||||
"""Test successful login with 2FA."""
|
||||
password = "test_password"
|
||||
hashed = get_password_hash(password)
|
||||
save_password_hash(hashed)
|
||||
save_totp_secret(sample_totp_secret)
|
||||
|
||||
totp = pyotp.TOTP(sample_totp_secret)
|
||||
valid_code = totp.now()
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "testadmin", "password": password, "totp_code": valid_code}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["user"] == "testadmin"
|
||||
assert data["has_2fa"] is True
|
||||
|
||||
def test_login_wrong_password(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test login with wrong password."""
|
||||
password = "correct_password"
|
||||
hashed = get_password_hash(password)
|
||||
save_password_hash(hashed)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "testadmin", "password": "wrong_password", "totp_code": ""}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "Invalid credentials" in response.json()["detail"]
|
||||
|
||||
def test_login_wrong_username(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test login with wrong username."""
|
||||
password = "test_password"
|
||||
hashed = get_password_hash(password)
|
||||
save_password_hash(hashed)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "wronguser", "password": password, "totp_code": ""}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_login_missing_2fa_code(self, test_app, mock_config, temp_auth_file, sample_totp_secret):
|
||||
"""Test login with 2FA enabled but no code provided."""
|
||||
password = "test_password"
|
||||
hashed = get_password_hash(password)
|
||||
save_password_hash(hashed)
|
||||
save_totp_secret(sample_totp_secret)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "testadmin", "password": password, "totp_code": ""}
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "2FA code required" in response.json()["detail"]
|
||||
|
||||
def test_login_invalid_2fa_code(self, test_app, mock_config, temp_auth_file, sample_totp_secret):
|
||||
"""Test login with invalid 2FA code."""
|
||||
password = "test_password"
|
||||
hashed = get_password_hash(password)
|
||||
save_password_hash(hashed)
|
||||
save_totp_secret(sample_totp_secret)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "testadmin", "password": password, "totp_code": "000000"}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_login_sets_cookies(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test that login sets auth cookies."""
|
||||
password = "test_password"
|
||||
hashed = get_password_hash(password)
|
||||
save_password_hash(hashed)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "testadmin", "password": password, "totp_code": ""}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Check that cookies are set
|
||||
assert "nas_access_token" in response.cookies
|
||||
assert "nas_refresh_token" in response.cookies
|
||||
|
||||
|
||||
class TestRefreshFlow:
|
||||
"""Test token refresh endpoint."""
|
||||
|
||||
def test_refresh_with_valid_token(self, test_app, mock_config, temp_auth_file, valid_refresh_token):
|
||||
"""Test refreshing with valid refresh token."""
|
||||
response = test_app.post(
|
||||
"/api/auth/refresh",
|
||||
json={"refresh_token": valid_refresh_token}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "access_token" in data
|
||||
assert "refresh_token" in data
|
||||
assert data["access_token"] != valid_refresh_token
|
||||
|
||||
def test_refresh_with_cookie(self, test_app, mock_config, temp_auth_file, valid_refresh_token):
|
||||
"""Test refreshing with refresh token in cookie."""
|
||||
test_app.cookies.set("nas_refresh_token", valid_refresh_token)
|
||||
|
||||
response = test_app.post("/api/auth/refresh", json={})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "access_token" in data
|
||||
|
||||
def test_refresh_with_invalid_token(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test refreshing with invalid token."""
|
||||
response = test_app.post(
|
||||
"/api/auth/refresh",
|
||||
json={"refresh_token": "invalid.token.here"}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_refresh_with_expired_token(self, test_app, mock_config, temp_auth_file, expired_access_token):
|
||||
"""Test refreshing with expired token."""
|
||||
response = test_app.post(
|
||||
"/api/auth/refresh",
|
||||
json={"refresh_token": expired_access_token}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_refresh_missing_token(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test refresh without providing token."""
|
||||
response = test_app.post("/api/auth/refresh", json={})
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "Missing refresh token" in response.json()["detail"]
|
||||
|
||||
def test_refresh_with_access_token_fails(self, test_app, mock_config, temp_auth_file, valid_access_token):
|
||||
"""Test that access token cannot be used for refresh."""
|
||||
response = test_app.post(
|
||||
"/api/auth/refresh",
|
||||
json={"refresh_token": valid_access_token}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "Invalid token type" in response.json()["detail"]
|
||||
|
||||
|
||||
class TestLogoutFlow:
|
||||
"""Test logout endpoint."""
|
||||
|
||||
def test_logout_success(self, test_app, mock_config, temp_auth_file, admin_headers):
|
||||
"""Test successful logout."""
|
||||
response = test_app.post("/api/auth/logout", headers=admin_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["message"] == "Logged out successfully"
|
||||
|
||||
def test_logout_clears_cookies(self, test_app, mock_config, temp_auth_file, admin_headers):
|
||||
"""Test that logout clears auth cookies."""
|
||||
response = test_app.post("/api/auth/logout", headers=admin_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Cookies should be deleted (max-age=0 or expires in past)
|
||||
# TestClient doesn't fully simulate cookie deletion, but we can check response
|
||||
|
||||
def test_logout_without_auth(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test logout without authentication."""
|
||||
response = test_app.post("/api/auth/logout")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestUserInfo:
|
||||
"""Test user info endpoint."""
|
||||
|
||||
def test_get_user_info(self, test_app, mock_config, temp_auth_file, temp_rbac_file, admin_headers):
|
||||
"""Test getting current user info."""
|
||||
response = test_app.get("/api/auth/me", headers=admin_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["username"] == "testuser"
|
||||
assert data["role"] == "admin"
|
||||
assert "pages" in data
|
||||
assert "sidebar_links" in data
|
||||
|
||||
def test_get_user_info_without_auth(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test getting user info without authentication."""
|
||||
response = test_app.get("/api/auth/me")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestProxyAuth:
|
||||
"""Test proxy authentication endpoint."""
|
||||
|
||||
def test_proxy_auth_from_trusted_source(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
||||
"""Test proxy auth from trusted proxy."""
|
||||
response = test_app.get(
|
||||
"/api/auth/proxy",
|
||||
headers={
|
||||
"Remote-User": "proxyuser",
|
||||
"Remote-Groups": "dashboard_admin"
|
||||
}
|
||||
)
|
||||
|
||||
# Note: This test may need adjustment based on actual proxy auth implementation
|
||||
# The test client's default host may not be in trusted proxies
|
||||
assert response.status_code in [200, 403]
|
||||
|
||||
def test_proxy_auth_from_untrusted_source(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test proxy auth from untrusted source."""
|
||||
# Mock request from untrusted IP
|
||||
response = test_app.get(
|
||||
"/api/auth/proxy",
|
||||
headers={
|
||||
"Remote-User": "proxyuser",
|
||||
"Remote-Groups": "dashboard_admin"
|
||||
}
|
||||
)
|
||||
|
||||
# Should reject untrusted proxy
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
class TestPasswordChange:
|
||||
"""Test password change endpoint."""
|
||||
|
||||
def test_change_password_success(self, test_app, mock_config, temp_auth_file, admin_headers):
|
||||
"""Test successful password change."""
|
||||
old_password = "old_password"
|
||||
new_password = "new_password_123"
|
||||
|
||||
# Set initial password
|
||||
hashed = get_password_hash(old_password)
|
||||
save_password_hash(hashed)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/change-password",
|
||||
headers=admin_headers,
|
||||
json={
|
||||
"old_password": old_password,
|
||||
"new_password": new_password
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_change_password_wrong_old_password(self, test_app, mock_config, temp_auth_file, admin_headers):
|
||||
"""Test password change with wrong old password."""
|
||||
old_password = "old_password"
|
||||
hashed = get_password_hash(old_password)
|
||||
save_password_hash(hashed)
|
||||
|
||||
response = test_app.post(
|
||||
"/api/auth/change-password",
|
||||
headers=admin_headers,
|
||||
json={
|
||||
"old_password": "wrong_password",
|
||||
"new_password": "new_password_123"
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_change_password_without_auth(self, test_app, mock_config, temp_auth_file):
|
||||
"""Test password change without authentication."""
|
||||
response = test_app.post(
|
||||
"/api/auth/change-password",
|
||||
json={
|
||||
"old_password": "old",
|
||||
"new_password": "new"
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
@@ -0,0 +1,181 @@
|
||||
"""
|
||||
Integration tests for Docker operations.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
|
||||
class TestDockerContainerList:
|
||||
"""Test listing Docker containers."""
|
||||
|
||||
def test_list_containers_success(self, test_app, admin_headers):
|
||||
"""Test listing containers successfully."""
|
||||
response = test_app.get("/api/docker/containers", headers=admin_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
if len(data) > 0:
|
||||
container = data[0]
|
||||
assert "id" in container
|
||||
assert "name" in container
|
||||
assert "status" in container
|
||||
|
||||
def test_list_containers_without_auth(self, test_app):
|
||||
"""Test listing containers without authentication."""
|
||||
response = test_app.get("/api/docker/containers")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestDockerContainerActions:
|
||||
"""Test Docker container start/stop/restart actions."""
|
||||
|
||||
def test_start_container_success(self, test_app, admin_headers, mock_docker_client):
|
||||
"""Test starting a container."""
|
||||
container_id = "abc123"
|
||||
|
||||
response = test_app.post(
|
||||
f"/api/docker/containers/{container_id}/start",
|
||||
headers=admin_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "message" in data or "status" in data
|
||||
|
||||
def test_stop_container_success(self, test_app, admin_headers, mock_docker_client):
|
||||
"""Test stopping a container."""
|
||||
container_id = "abc123"
|
||||
|
||||
response = test_app.post(
|
||||
f"/api/docker/containers/{container_id}/stop",
|
||||
headers=admin_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_restart_container_success(self, test_app, admin_headers, mock_docker_client):
|
||||
"""Test restarting a container."""
|
||||
container_id = "abc123"
|
||||
|
||||
response = test_app.post(
|
||||
f"/api/docker/containers/{container_id}/restart",
|
||||
headers=admin_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_container_action_without_auth(self, test_app):
|
||||
"""Test container action without authentication."""
|
||||
response = test_app.post("/api/docker/containers/abc123/start")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_container_action_invalid_action(self, test_app, admin_headers):
|
||||
"""Test container action with invalid action."""
|
||||
response = test_app.post(
|
||||
"/api/docker/containers/abc123/invalid_action",
|
||||
headers=admin_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
class TestDockerContainerLogs:
|
||||
"""Test Docker container logs retrieval."""
|
||||
|
||||
def test_get_container_logs_success(self, test_app, admin_headers, mock_docker_client):
|
||||
"""Test getting container logs."""
|
||||
container_id = "abc123"
|
||||
|
||||
response = test_app.get(
|
||||
f"/api/docker/containers/{container_id}/logs",
|
||||
headers=admin_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "logs" in data or isinstance(data, str)
|
||||
|
||||
def test_get_container_logs_with_tail(self, test_app, admin_headers, mock_docker_client):
|
||||
"""Test getting container logs with tail limit."""
|
||||
container_id = "abc123"
|
||||
|
||||
response = test_app.get(
|
||||
f"/api/docker/containers/{container_id}/logs?tail=100",
|
||||
headers=admin_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_get_container_logs_without_auth(self, test_app):
|
||||
"""Test getting logs without authentication."""
|
||||
response = test_app.get("/api/docker/containers/abc123/logs")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_get_container_logs_nonexistent(self, test_app, admin_headers, mock_docker_client):
|
||||
"""Test getting logs for nonexistent container."""
|
||||
# Mock container not found
|
||||
mock_docker_client.containers.get.side_effect = Exception("Container not found")
|
||||
|
||||
response = test_app.get(
|
||||
"/api/docker/containers/nonexistent/logs",
|
||||
headers=admin_headers
|
||||
)
|
||||
|
||||
# Should handle error gracefully
|
||||
assert response.status_code in [404, 500]
|
||||
|
||||
|
||||
class TestDockerPermissions:
|
||||
"""Test Docker operation permissions."""
|
||||
|
||||
@pytest.fixture
|
||||
def member_token(self, mock_config, temp_auth_file, temp_rbac_file):
|
||||
"""Create token for member role."""
|
||||
from auth import create_access_token
|
||||
from datetime import timedelta
|
||||
return create_access_token(
|
||||
data={"sub": "memberuser", "role": "member"},
|
||||
expires_delta=timedelta(minutes=30)
|
||||
)
|
||||
|
||||
@pytest.fixture
|
||||
def member_headers(self, member_token):
|
||||
"""Headers with member access token."""
|
||||
return {"Authorization": f"Bearer {member_token}"}
|
||||
|
||||
def test_member_can_list_containers(self, test_app, member_headers):
|
||||
"""Test that member can list containers."""
|
||||
response = test_app.get("/api/docker/containers", headers=member_headers)
|
||||
|
||||
# Members should be able to view containers
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_member_cannot_start_container(self, test_app, member_headers):
|
||||
"""Test that member cannot start containers (admin only)."""
|
||||
response = test_app.post(
|
||||
"/api/docker/containers/abc123/start",
|
||||
headers=member_headers
|
||||
)
|
||||
|
||||
# Should be forbidden for non-admin
|
||||
assert response.status_code == 403
|
||||
|
||||
def test_viewer_can_list_containers(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
||||
"""Test that viewer can list containers if they have docker page access."""
|
||||
from auth import create_access_token
|
||||
from datetime import timedelta
|
||||
|
||||
viewer_token = create_access_token(
|
||||
data={"sub": "vieweruser", "role": "viewer"},
|
||||
expires_delta=timedelta(minutes=30)
|
||||
)
|
||||
headers = {"Authorization": f"Bearer {viewer_token}"}
|
||||
|
||||
response = test_app.get("/api/docker/containers", headers=headers)
|
||||
|
||||
# Viewer may not have docker page access by default
|
||||
assert response.status_code in [200, 403]
|
||||
@@ -0,0 +1,189 @@
|
||||
"""
|
||||
Integration tests for file operations.
|
||||
"""
|
||||
import pytest
|
||||
import os
|
||||
|
||||
|
||||
class TestFileBrowse:
|
||||
"""Test file browsing."""
|
||||
|
||||
def test_browse_root(self, test_app, admin_headers, temp_volume_root):
|
||||
"""Test browsing root directory."""
|
||||
response = test_app.get(
|
||||
"/api/files/browse",
|
||||
headers=admin_headers,
|
||||
params={"path": "/"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "files" in data or isinstance(data, list)
|
||||
|
||||
def test_browse_without_auth(self, test_app):
|
||||
"""Test browsing without authentication."""
|
||||
response = test_app.get("/api/files/browse", params={"path": "/"})
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_browse_blocked_directory(self, test_app, admin_headers):
|
||||
"""Test browsing blocked directory."""
|
||||
response = test_app.get(
|
||||
"/api/files/browse",
|
||||
headers=admin_headers,
|
||||
params={"path": "/@appstore"}
|
||||
)
|
||||
|
||||
# Should be forbidden
|
||||
assert response.status_code == 403
|
||||
|
||||
def test_browse_path_traversal_attempt(self, test_app, admin_headers):
|
||||
"""Test path traversal protection."""
|
||||
response = test_app.get(
|
||||
"/api/files/browse",
|
||||
headers=admin_headers,
|
||||
params={"path": "/../../../etc"}
|
||||
)
|
||||
|
||||
# Should be forbidden
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
class TestFileUpload:
|
||||
"""Test file upload."""
|
||||
|
||||
def test_upload_file_success(self, test_app, admin_headers, temp_volume_root):
|
||||
"""Test uploading a file."""
|
||||
test_content = b"test file content"
|
||||
files = {"file": ("test.txt", test_content, "text/plain")}
|
||||
|
||||
response = test_app.post(
|
||||
"/api/files/upload",
|
||||
headers=admin_headers,
|
||||
files=files,
|
||||
data={"path": "/"}
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 201]
|
||||
|
||||
def test_upload_without_auth(self, test_app):
|
||||
"""Test upload without authentication."""
|
||||
files = {"file": ("test.txt", b"content", "text/plain")}
|
||||
|
||||
response = test_app.post(
|
||||
"/api/files/upload",
|
||||
files=files,
|
||||
data={"path": "/"}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_upload_as_member_forbidden(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
||||
"""Test that non-admin cannot upload."""
|
||||
from auth import create_access_token
|
||||
from datetime import timedelta
|
||||
|
||||
member_token = create_access_token(
|
||||
data={"sub": "memberuser", "role": "member"},
|
||||
expires_delta=timedelta(minutes=30)
|
||||
)
|
||||
headers = {"Authorization": f"Bearer {member_token}"}
|
||||
files = {"file": ("test.txt", b"content", "text/plain")}
|
||||
|
||||
response = test_app.post(
|
||||
"/api/files/upload",
|
||||
headers=headers,
|
||||
files=files,
|
||||
data={"path": "/"}
|
||||
)
|
||||
|
||||
# Upload should be admin-only
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
class TestFileDelete:
|
||||
"""Test file deletion."""
|
||||
|
||||
def test_delete_file_success(self, test_app, admin_headers, temp_volume_root):
|
||||
"""Test deleting a file."""
|
||||
# Create a test file first
|
||||
test_file = os.path.join(temp_volume_root, "test_delete.txt")
|
||||
with open(test_file, "w") as f:
|
||||
f.write("test content")
|
||||
|
||||
response = test_app.delete(
|
||||
"/api/files/delete",
|
||||
headers=admin_headers,
|
||||
params={"path": "/test_delete.txt"}
|
||||
)
|
||||
|
||||
assert response.status_code in [200, 204]
|
||||
|
||||
def test_delete_without_auth(self, test_app):
|
||||
"""Test delete without authentication."""
|
||||
response = test_app.delete(
|
||||
"/api/files/delete",
|
||||
params={"path": "/test.txt"}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_delete_as_member_forbidden(self, test_app, mock_config, temp_auth_file, temp_rbac_file):
|
||||
"""Test that non-admin cannot delete."""
|
||||
from auth import create_access_token
|
||||
from datetime import timedelta
|
||||
|
||||
member_token = create_access_token(
|
||||
data={"sub": "memberuser", "role": "member"},
|
||||
expires_delta=timedelta(minutes=30)
|
||||
)
|
||||
headers = {"Authorization": f"Bearer {member_token}"}
|
||||
|
||||
response = test_app.delete(
|
||||
"/api/files/delete",
|
||||
headers=headers,
|
||||
params={"path": "/test.txt"}
|
||||
)
|
||||
|
||||
# Delete should be admin-only
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
class TestFileDownload:
|
||||
"""Test file download."""
|
||||
|
||||
def test_download_file_success(self, test_app, admin_headers, temp_volume_root):
|
||||
"""Test downloading a file."""
|
||||
# Create a test file
|
||||
test_file = os.path.join(temp_volume_root, "test_download.txt")
|
||||
test_content = "test download content"
|
||||
with open(test_file, "w") as f:
|
||||
f.write(test_content)
|
||||
|
||||
response = test_app.get(
|
||||
"/api/files/download",
|
||||
headers=admin_headers,
|
||||
params={"path": "/test_download.txt"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert test_content in response.text or response.content == test_content.encode()
|
||||
|
||||
def test_download_without_auth(self, test_app):
|
||||
"""Test download without authentication."""
|
||||
response = test_app.get(
|
||||
"/api/files/download",
|
||||
params={"path": "/test.txt"}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_download_nonexistent_file(self, test_app, admin_headers):
|
||||
"""Test downloading nonexistent file."""
|
||||
response = test_app.get(
|
||||
"/api/files/download",
|
||||
headers=admin_headers,
|
||||
params={"path": "/nonexistent.txt"}
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
@@ -0,0 +1 @@
|
||||
# Unit tests
|
||||
@@ -0,0 +1,322 @@
|
||||
"""
|
||||
Unit tests for auth.py - JWT, TOTP, password hashing, encryption.
|
||||
"""
|
||||
import pytest
|
||||
import jwt
|
||||
import pyotp
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from auth import (
|
||||
verify_password,
|
||||
get_password_hash,
|
||||
create_access_token,
|
||||
create_refresh_token,
|
||||
user_from_access_token,
|
||||
verify_totp,
|
||||
load_totp_secret,
|
||||
save_totp_secret,
|
||||
load_password_hash,
|
||||
save_password_hash,
|
||||
increment_token_version,
|
||||
verify_token_version,
|
||||
_encrypt,
|
||||
_decrypt,
|
||||
load_passkey_credentials,
|
||||
save_passkey_credential,
|
||||
clear_passkey_credentials,
|
||||
)
|
||||
from fastapi import HTTPException
|
||||
|
||||
|
||||
class TestPasswordHashing:
|
||||
"""Test password hashing and verification."""
|
||||
|
||||
def test_hash_and_verify_password(self):
|
||||
"""Test that password hashing and verification works."""
|
||||
password = "test_password_123"
|
||||
hashed = get_password_hash(password)
|
||||
|
||||
assert hashed != password
|
||||
assert verify_password(password, hashed)
|
||||
|
||||
def test_verify_wrong_password(self):
|
||||
"""Test that wrong password fails verification."""
|
||||
password = "correct_password"
|
||||
wrong_password = "wrong_password"
|
||||
hashed = get_password_hash(password)
|
||||
|
||||
assert not verify_password(wrong_password, hashed)
|
||||
|
||||
def test_different_hashes_for_same_password(self):
|
||||
"""Test that same password produces different hashes (salt)."""
|
||||
password = "test_password"
|
||||
hash1 = get_password_hash(password)
|
||||
hash2 = get_password_hash(password)
|
||||
|
||||
assert hash1 != hash2
|
||||
assert verify_password(password, hash1)
|
||||
assert verify_password(password, hash2)
|
||||
|
||||
|
||||
class TestJWTTokens:
|
||||
"""Test JWT token creation and validation."""
|
||||
|
||||
def test_create_access_token(self, mock_config):
|
||||
"""Test access token creation."""
|
||||
data = {"sub": "testuser", "role": "admin"}
|
||||
token = create_access_token(data)
|
||||
|
||||
assert token is not None
|
||||
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
||||
assert payload["sub"] == "testuser"
|
||||
assert payload["role"] == "admin"
|
||||
assert payload["type"] == "access"
|
||||
assert "exp" in payload
|
||||
|
||||
def test_create_access_token_with_custom_expiry(self, mock_config):
|
||||
"""Test access token with custom expiration."""
|
||||
data = {"sub": "testuser"}
|
||||
expires_delta = timedelta(minutes=60)
|
||||
token = create_access_token(data, expires_delta)
|
||||
|
||||
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
||||
exp_time = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# Should expire in approximately 60 minutes
|
||||
time_diff = (exp_time - now).total_seconds()
|
||||
assert 3500 < time_diff < 3700 # ~60 minutes with some tolerance
|
||||
|
||||
def test_create_refresh_token(self, mock_config, temp_auth_file):
|
||||
"""Test refresh token creation."""
|
||||
data = {"sub": "testuser", "role": "member"}
|
||||
token = create_refresh_token(data)
|
||||
|
||||
assert token is not None
|
||||
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
||||
assert payload["sub"] == "testuser"
|
||||
assert payload["role"] == "member"
|
||||
assert payload["type"] == "refresh"
|
||||
assert "tv" in payload # token version
|
||||
assert "exp" in payload
|
||||
|
||||
def test_user_from_valid_access_token(self, mock_config, temp_rbac_file):
|
||||
"""Test extracting user from valid access token."""
|
||||
token = create_access_token({"sub": "testuser", "role": "admin"})
|
||||
user = user_from_access_token(token)
|
||||
|
||||
assert user.username == "testuser"
|
||||
assert user.role == "admin"
|
||||
assert user.pages == "*"
|
||||
assert user.sidebar_links == "*"
|
||||
|
||||
def test_user_from_expired_token(self, mock_config, temp_rbac_file):
|
||||
"""Test that expired token raises HTTPException."""
|
||||
token = create_access_token(
|
||||
{"sub": "testuser", "role": "admin"},
|
||||
expires_delta=timedelta(seconds=-1)
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
user_from_access_token(token)
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
def test_user_from_invalid_token(self, mock_config, temp_rbac_file):
|
||||
"""Test that invalid token raises HTTPException."""
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
user_from_access_token("invalid.token.here")
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
def test_user_from_refresh_token_fails(self, mock_config, temp_rbac_file, temp_auth_file):
|
||||
"""Test that refresh token cannot be used as access token."""
|
||||
token = create_refresh_token({"sub": "testuser", "role": "admin"})
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
user_from_access_token(token)
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
def test_user_from_token_without_username(self, mock_config, temp_rbac_file):
|
||||
"""Test that token without username fails."""
|
||||
# Create token without 'sub' field
|
||||
token = jwt.encode(
|
||||
{"role": "admin", "type": "access", "exp": datetime.now(timezone.utc) + timedelta(minutes=30)},
|
||||
mock_config.SECRET_KEY,
|
||||
algorithm=mock_config.ALGORITHM
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
user_from_access_token(token)
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
|
||||
class TestEncryption:
|
||||
"""Test encryption and decryption functions."""
|
||||
|
||||
def test_encrypt_decrypt(self, mock_config):
|
||||
"""Test basic encryption and decryption."""
|
||||
plaintext = "my-secret-totp-key"
|
||||
encrypted = _encrypt(plaintext)
|
||||
|
||||
assert encrypted != plaintext
|
||||
assert encrypted != ""
|
||||
|
||||
decrypted = _decrypt(encrypted)
|
||||
assert decrypted == plaintext
|
||||
|
||||
def test_encrypt_empty_string(self, mock_config):
|
||||
"""Test encrypting empty string."""
|
||||
encrypted = _encrypt("")
|
||||
assert encrypted == ""
|
||||
|
||||
decrypted = _decrypt("")
|
||||
assert decrypted == ""
|
||||
|
||||
def test_decrypt_invalid_ciphertext(self, mock_config):
|
||||
"""Test decrypting invalid ciphertext returns original."""
|
||||
invalid = "not-a-valid-encrypted-string"
|
||||
decrypted = _decrypt(invalid)
|
||||
|
||||
# Should fallback to returning the original string
|
||||
assert decrypted == invalid
|
||||
|
||||
def test_encrypt_decrypt_unicode(self, mock_config):
|
||||
"""Test encryption with unicode characters."""
|
||||
plaintext = "测试密钥🔐"
|
||||
encrypted = _encrypt(plaintext)
|
||||
decrypted = _decrypt(encrypted)
|
||||
|
||||
assert decrypted == plaintext
|
||||
|
||||
|
||||
class TestTOTP:
|
||||
"""Test TOTP functionality."""
|
||||
|
||||
def test_save_and_load_totp_secret(self, mock_config, temp_auth_file):
|
||||
"""Test saving and loading TOTP secret."""
|
||||
secret = "JBSWY3DPEHPK3PXP"
|
||||
save_totp_secret(secret)
|
||||
|
||||
loaded = load_totp_secret()
|
||||
assert loaded == secret
|
||||
|
||||
def test_load_totp_secret_from_env(self, mock_config, temp_auth_file, monkeypatch):
|
||||
"""Test loading TOTP secret from environment variable."""
|
||||
monkeypatch.setenv("TOTP_SECRET", "ENV_SECRET_KEY")
|
||||
import importlib
|
||||
import config
|
||||
importlib.reload(config)
|
||||
|
||||
# When no secret in file, should return env var
|
||||
loaded = load_totp_secret()
|
||||
assert loaded == "ENV_SECRET_KEY"
|
||||
|
||||
def test_verify_totp_valid_code(self, mock_config, temp_auth_file, sample_totp_secret):
|
||||
"""Test verifying valid TOTP code."""
|
||||
save_totp_secret(sample_totp_secret)
|
||||
totp = pyotp.TOTP(sample_totp_secret)
|
||||
valid_code = totp.now()
|
||||
|
||||
assert verify_totp(valid_code)
|
||||
|
||||
def test_verify_totp_invalid_code(self, mock_config, temp_auth_file, sample_totp_secret):
|
||||
"""Test verifying invalid TOTP code."""
|
||||
save_totp_secret(sample_totp_secret)
|
||||
|
||||
assert not verify_totp("000000")
|
||||
|
||||
def test_verify_totp_no_secret_enabled(self, mock_config, temp_auth_file):
|
||||
"""Test that TOTP verification passes when 2FA not enabled."""
|
||||
# No secret saved, should return True (2FA disabled)
|
||||
assert verify_totp("any_code")
|
||||
|
||||
def test_verify_totp_replay_protection(self, mock_config, temp_auth_file, sample_totp_secret):
|
||||
"""Test that same TOTP code cannot be used twice."""
|
||||
save_totp_secret(sample_totp_secret)
|
||||
totp = pyotp.TOTP(sample_totp_secret)
|
||||
valid_code = totp.now()
|
||||
|
||||
# First use should succeed
|
||||
assert verify_totp(valid_code)
|
||||
|
||||
# Second use of same code should fail (replay protection)
|
||||
assert not verify_totp(valid_code)
|
||||
|
||||
|
||||
class TestPasswordPersistence:
|
||||
"""Test password hash persistence."""
|
||||
|
||||
def test_save_and_load_password_hash(self, mock_config, temp_auth_file):
|
||||
"""Test saving and loading password hash."""
|
||||
hashed = get_password_hash("new_password")
|
||||
save_password_hash(hashed)
|
||||
|
||||
loaded = load_password_hash()
|
||||
assert loaded == hashed
|
||||
|
||||
def test_load_password_hash_from_env(self, mock_config, temp_auth_file):
|
||||
"""Test loading password hash from environment variable."""
|
||||
# When no hash in file, should return env var
|
||||
loaded = load_password_hash()
|
||||
assert loaded == mock_config.ADMIN_PASSWORD_HASH
|
||||
|
||||
|
||||
class TestPasskeyCredentials:
|
||||
"""Test passkey credential management."""
|
||||
|
||||
def test_save_and_load_passkey_credentials(self, mock_config, temp_auth_file):
|
||||
"""Test saving and loading passkey credentials."""
|
||||
cred1 = {"id": "cred1", "public_key": "key1"}
|
||||
cred2 = {"id": "cred2", "public_key": "key2"}
|
||||
|
||||
save_passkey_credential(cred1)
|
||||
save_passkey_credential(cred2)
|
||||
|
||||
creds = load_passkey_credentials()
|
||||
assert len(creds) == 2
|
||||
assert cred1 in creds
|
||||
assert cred2 in creds
|
||||
|
||||
def test_clear_passkey_credentials(self, mock_config, temp_auth_file):
|
||||
"""Test clearing all passkey credentials."""
|
||||
save_passkey_credential({"id": "cred1"})
|
||||
save_passkey_credential({"id": "cred2"})
|
||||
|
||||
clear_passkey_credentials()
|
||||
|
||||
creds = load_passkey_credentials()
|
||||
assert len(creds) == 0
|
||||
|
||||
|
||||
class TestTokenVersion:
|
||||
"""Test token version management for refresh token rotation."""
|
||||
|
||||
def test_increment_token_version(self, mock_config, temp_auth_file):
|
||||
"""Test incrementing token version."""
|
||||
v1 = increment_token_version()
|
||||
assert v1 == 1
|
||||
|
||||
v2 = increment_token_version()
|
||||
assert v2 == 2
|
||||
|
||||
v3 = increment_token_version()
|
||||
assert v3 == 3
|
||||
|
||||
def test_verify_token_version(self, mock_config, temp_auth_file):
|
||||
"""Test verifying token version."""
|
||||
v1 = increment_token_version()
|
||||
assert verify_token_version(v1)
|
||||
|
||||
v2 = increment_token_version()
|
||||
assert verify_token_version(v2)
|
||||
assert not verify_token_version(v1) # Old version should fail
|
||||
|
||||
def test_token_version_in_refresh_token(self, mock_config, temp_auth_file):
|
||||
"""Test that refresh token includes current token version."""
|
||||
current_version = increment_token_version()
|
||||
token = create_refresh_token({"sub": "testuser"})
|
||||
|
||||
payload = jwt.decode(token, mock_config.SECRET_KEY, algorithms=[mock_config.ALGORITHM])
|
||||
assert payload["tv"] == current_version
|
||||
@@ -0,0 +1,297 @@
|
||||
"""
|
||||
Unit tests for config.py - IP parsing, environment validation.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import Mock
|
||||
from config import (
|
||||
_parse_ip,
|
||||
_ip_matches_ranges,
|
||||
is_lan_ip,
|
||||
is_tailscale_ip,
|
||||
is_trusted_proxy,
|
||||
get_forwarded_client_ip,
|
||||
get_client_ip,
|
||||
)
|
||||
|
||||
|
||||
class TestParseIP:
|
||||
"""Test IP address parsing."""
|
||||
|
||||
def test_parse_valid_ipv4(self):
|
||||
"""Test parsing valid IPv4 address."""
|
||||
ip = _parse_ip("192.168.1.1")
|
||||
assert ip is not None
|
||||
assert str(ip) == "192.168.1.1"
|
||||
|
||||
def test_parse_valid_ipv6(self):
|
||||
"""Test parsing valid IPv6 address."""
|
||||
ip = _parse_ip("2001:db8::1")
|
||||
assert ip is not None
|
||||
assert str(ip) == "2001:db8::1"
|
||||
|
||||
def test_parse_localhost_ipv4(self):
|
||||
"""Test parsing localhost IPv4."""
|
||||
ip = _parse_ip("127.0.0.1")
|
||||
assert ip is not None
|
||||
assert str(ip) == "127.0.0.1"
|
||||
|
||||
def test_parse_localhost_ipv6(self):
|
||||
"""Test parsing localhost IPv6."""
|
||||
ip = _parse_ip("::1")
|
||||
assert ip is not None
|
||||
assert str(ip) == "::1"
|
||||
|
||||
def test_parse_invalid_ip(self):
|
||||
"""Test parsing invalid IP returns None."""
|
||||
assert _parse_ip("not.an.ip.address") is None
|
||||
assert _parse_ip("999.999.999.999") is None
|
||||
assert _parse_ip("") is None
|
||||
assert _parse_ip(" ") is None
|
||||
|
||||
def test_parse_ip_with_whitespace(self):
|
||||
"""Test parsing IP with surrounding whitespace."""
|
||||
ip = _parse_ip(" 192.168.1.1 ")
|
||||
assert ip is not None
|
||||
assert str(ip) == "192.168.1.1"
|
||||
|
||||
|
||||
class TestIPMatchesRanges:
|
||||
"""Test IP range matching."""
|
||||
|
||||
def test_match_single_ip(self):
|
||||
"""Test matching against single IP."""
|
||||
assert _ip_matches_ranges("192.168.1.1", ["192.168.1.1"])
|
||||
assert not _ip_matches_ranges("192.168.1.2", ["192.168.1.1"])
|
||||
|
||||
def test_match_cidr_range(self):
|
||||
"""Test matching against CIDR range."""
|
||||
ranges = ["192.168.1.0/24"]
|
||||
assert _ip_matches_ranges("192.168.1.1", ranges)
|
||||
assert _ip_matches_ranges("192.168.1.100", ranges)
|
||||
assert _ip_matches_ranges("192.168.1.254", ranges)
|
||||
assert not _ip_matches_ranges("192.168.2.1", ranges)
|
||||
|
||||
def test_match_multiple_ranges(self):
|
||||
"""Test matching against multiple ranges."""
|
||||
ranges = ["192.168.1.0/24", "10.0.0.0/8", "172.16.0.1"]
|
||||
assert _ip_matches_ranges("192.168.1.50", ranges)
|
||||
assert _ip_matches_ranges("10.5.10.20", ranges)
|
||||
assert _ip_matches_ranges("172.16.0.1", ranges)
|
||||
assert not _ip_matches_ranges("8.8.8.8", ranges)
|
||||
|
||||
def test_match_ipv6_range(self):
|
||||
"""Test matching IPv6 against range."""
|
||||
ranges = ["2001:db8::/32"]
|
||||
assert _ip_matches_ranges("2001:db8::1", ranges)
|
||||
assert _ip_matches_ranges("2001:db8:1234::5678", ranges)
|
||||
assert not _ip_matches_ranges("2001:db9::1", ranges)
|
||||
|
||||
def test_match_invalid_ip(self):
|
||||
"""Test that invalid IP returns False."""
|
||||
assert not _ip_matches_ranges("invalid", ["192.168.1.0/24"])
|
||||
assert not _ip_matches_ranges("", ["192.168.1.0/24"])
|
||||
|
||||
def test_match_empty_ranges(self):
|
||||
"""Test matching against empty ranges."""
|
||||
assert not _ip_matches_ranges("192.168.1.1", [])
|
||||
|
||||
def test_match_with_whitespace_in_ranges(self):
|
||||
"""Test matching with whitespace in range strings."""
|
||||
ranges = [" 192.168.1.0/24 ", " 10.0.0.1 "]
|
||||
assert _ip_matches_ranges("192.168.1.50", ranges)
|
||||
assert _ip_matches_ranges("10.0.0.1", ranges)
|
||||
|
||||
def test_match_with_empty_string_in_ranges(self):
|
||||
"""Test that empty strings in ranges are skipped."""
|
||||
ranges = ["192.168.1.0/24", "", " ", "10.0.0.1"]
|
||||
assert _ip_matches_ranges("192.168.1.50", ranges)
|
||||
assert _ip_matches_ranges("10.0.0.1", ranges)
|
||||
|
||||
def test_match_with_invalid_range(self):
|
||||
"""Test that invalid ranges are skipped."""
|
||||
ranges = ["192.168.1.0/24", "invalid-range", "10.0.0.1"]
|
||||
assert _ip_matches_ranges("192.168.1.50", ranges)
|
||||
assert _ip_matches_ranges("10.0.0.1", ranges)
|
||||
assert not _ip_matches_ranges("8.8.8.8", ranges)
|
||||
|
||||
|
||||
class TestIsLanIP:
|
||||
"""Test LAN IP detection."""
|
||||
|
||||
def test_is_lan_ip_default_range(self, mock_config):
|
||||
"""Test LAN IP detection with default range."""
|
||||
# Default is 192.168.31.0/24
|
||||
assert is_lan_ip("192.168.31.1")
|
||||
assert is_lan_ip("192.168.31.100")
|
||||
assert is_lan_ip("192.168.31.254")
|
||||
assert not is_lan_ip("192.168.30.1")
|
||||
assert not is_lan_ip("192.168.32.1")
|
||||
|
||||
def test_is_lan_ip_not_tailscale(self, mock_config):
|
||||
"""Test that Tailscale IPs are not considered LAN."""
|
||||
assert not is_lan_ip("100.64.0.1")
|
||||
assert not is_lan_ip("100.78.131.124")
|
||||
|
||||
|
||||
class TestIsTailscaleIP:
|
||||
"""Test Tailscale IP detection."""
|
||||
|
||||
def test_is_tailscale_ip_valid(self):
|
||||
"""Test Tailscale IP detection."""
|
||||
assert is_tailscale_ip("100.64.0.1")
|
||||
assert is_tailscale_ip("100.78.131.124")
|
||||
assert is_tailscale_ip("100.100.100.100")
|
||||
assert is_tailscale_ip("100.127.255.254")
|
||||
|
||||
def test_is_tailscale_ip_invalid(self):
|
||||
"""Test non-Tailscale IPs."""
|
||||
assert not is_tailscale_ip("100.63.255.255") # Just below range
|
||||
assert not is_tailscale_ip("100.128.0.0") # Just above range
|
||||
assert not is_tailscale_ip("192.168.1.1")
|
||||
assert not is_tailscale_ip("10.0.0.1")
|
||||
|
||||
def test_is_tailscale_ip_boundary(self):
|
||||
"""Test Tailscale IP range boundaries."""
|
||||
# 100.64.0.0/10 means 100.64.0.0 to 100.127.255.255
|
||||
assert is_tailscale_ip("100.64.0.0") # Start of range
|
||||
assert is_tailscale_ip("100.127.255.255") # End of range
|
||||
|
||||
|
||||
class TestIsTrustedProxy:
|
||||
"""Test trusted proxy detection."""
|
||||
|
||||
def test_is_trusted_proxy_default(self, mock_config):
|
||||
"""Test trusted proxy detection with default config."""
|
||||
# Default: 127.0.0.1,::1,172.21.0.1
|
||||
assert is_trusted_proxy("127.0.0.1")
|
||||
assert is_trusted_proxy("::1")
|
||||
assert is_trusted_proxy("172.21.0.1")
|
||||
assert not is_trusted_proxy("192.168.1.1")
|
||||
|
||||
|
||||
class TestGetForwardedClientIP:
|
||||
"""Test forwarded client IP extraction."""
|
||||
|
||||
def test_get_forwarded_ip_from_trusted_proxy(self, mock_config):
|
||||
"""Test extracting forwarded IP from trusted proxy."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "127.0.0.1" # Trusted proxy
|
||||
request.headers = {"x-forwarded-for": "203.0.113.1"}
|
||||
|
||||
ip = get_forwarded_client_ip(request)
|
||||
assert ip == "203.0.113.1"
|
||||
|
||||
def test_get_forwarded_ip_from_untrusted_proxy(self, mock_config):
|
||||
"""Test that untrusted proxy returns None."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "8.8.8.8" # Not trusted
|
||||
request.headers = {"x-forwarded-for": "203.0.113.1"}
|
||||
|
||||
ip = get_forwarded_client_ip(request)
|
||||
assert ip is None
|
||||
|
||||
def test_get_forwarded_ip_multiple_hops(self, mock_config):
|
||||
"""Test extracting first IP from multiple forwarded IPs."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "127.0.0.1"
|
||||
request.headers = {"x-forwarded-for": "203.0.113.1, 198.51.100.1, 192.0.2.1"}
|
||||
|
||||
ip = get_forwarded_client_ip(request)
|
||||
assert ip == "203.0.113.1"
|
||||
|
||||
def test_get_forwarded_ip_no_header(self, mock_config):
|
||||
"""Test when x-forwarded-for header is missing."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "127.0.0.1"
|
||||
request.headers = {}
|
||||
|
||||
ip = get_forwarded_client_ip(request)
|
||||
assert ip is None
|
||||
|
||||
def test_get_forwarded_ip_empty_header(self, mock_config):
|
||||
"""Test when x-forwarded-for header is empty."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "127.0.0.1"
|
||||
request.headers = {"x-forwarded-for": ""}
|
||||
|
||||
ip = get_forwarded_client_ip(request)
|
||||
assert ip is None
|
||||
|
||||
def test_get_forwarded_ip_no_client(self, mock_config):
|
||||
"""Test when request has no client."""
|
||||
request = Mock()
|
||||
request.client = None
|
||||
request.headers = {"x-forwarded-for": "203.0.113.1"}
|
||||
|
||||
ip = get_forwarded_client_ip(request)
|
||||
assert ip is None
|
||||
|
||||
|
||||
class TestGetClientIP:
|
||||
"""Test client IP extraction."""
|
||||
|
||||
def test_get_client_ip_direct(self, mock_config):
|
||||
"""Test getting client IP from direct connection."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "192.168.1.100"
|
||||
request.headers = {}
|
||||
|
||||
ip = get_client_ip(request)
|
||||
assert ip == "192.168.1.100"
|
||||
|
||||
def test_get_client_ip_via_trusted_proxy(self, mock_config):
|
||||
"""Test getting client IP via trusted proxy."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "127.0.0.1"
|
||||
request.headers = {"x-forwarded-for": "203.0.113.1"}
|
||||
|
||||
ip = get_client_ip(request)
|
||||
assert ip == "203.0.113.1"
|
||||
|
||||
def test_get_client_ip_via_untrusted_proxy(self, mock_config):
|
||||
"""Test getting client IP via untrusted proxy falls back to direct."""
|
||||
request = Mock()
|
||||
request.client = Mock()
|
||||
request.client.host = "8.8.8.8"
|
||||
request.headers = {"x-forwarded-for": "203.0.113.1"}
|
||||
|
||||
ip = get_client_ip(request)
|
||||
assert ip == "8.8.8.8"
|
||||
|
||||
def test_get_client_ip_no_client(self, mock_config):
|
||||
"""Test getting client IP when request has no client."""
|
||||
request = Mock()
|
||||
request.client = None
|
||||
request.headers = {}
|
||||
|
||||
ip = get_client_ip(request)
|
||||
assert ip == ""
|
||||
|
||||
|
||||
class TestConfigValidation:
|
||||
"""Test configuration validation."""
|
||||
|
||||
def test_secret_key_validation(self, monkeypatch):
|
||||
"""Test that short SECRET_KEY raises error."""
|
||||
monkeypatch.setenv("SECRET_KEY", "short")
|
||||
|
||||
with pytest.raises(RuntimeError, match="SECRET_KEY must be set and at least 32 characters"):
|
||||
import config
|
||||
import importlib
|
||||
importlib.reload(config)
|
||||
|
||||
def test_secret_key_empty(self, monkeypatch):
|
||||
"""Test that empty SECRET_KEY raises error."""
|
||||
monkeypatch.delenv("SECRET_KEY", raising=False)
|
||||
|
||||
with pytest.raises(RuntimeError, match="SECRET_KEY must be set and at least 32 characters"):
|
||||
import config
|
||||
import importlib
|
||||
importlib.reload(config)
|
||||
@@ -0,0 +1,316 @@
|
||||
"""
|
||||
Unit tests for rbac.py - Role-based access control.
|
||||
"""
|
||||
import pytest
|
||||
import json
|
||||
import os
|
||||
from rbac import (
|
||||
load_rbac,
|
||||
save_rbac,
|
||||
update_rbac,
|
||||
resolve_role,
|
||||
get_pages,
|
||||
get_sidebar_links,
|
||||
get_sidebar_order,
|
||||
DEFAULT_RBAC,
|
||||
ROLE_GROUP_MAP,
|
||||
)
|
||||
|
||||
|
||||
class TestLoadRBAC:
|
||||
"""Test loading RBAC configuration."""
|
||||
|
||||
def test_load_rbac_from_file(self, mock_config, temp_rbac_file):
|
||||
"""Test loading RBAC from existing file."""
|
||||
rbac = load_rbac()
|
||||
|
||||
assert "role_defaults" in rbac
|
||||
assert "user_overrides" in rbac
|
||||
assert "admin" in rbac["role_defaults"]
|
||||
assert "member" in rbac["role_defaults"]
|
||||
assert "viewer" in rbac["role_defaults"]
|
||||
|
||||
def test_load_rbac_missing_file(self, mock_config, temp_volume_root):
|
||||
"""Test loading RBAC when file doesn't exist returns defaults."""
|
||||
# Don't create rbac file
|
||||
rbac = load_rbac()
|
||||
|
||||
assert rbac == DEFAULT_RBAC
|
||||
|
||||
def test_load_rbac_corrupted_file(self, mock_config, temp_volume_root):
|
||||
"""Test loading RBAC with corrupted file returns defaults."""
|
||||
rbac_file = os.path.join(temp_volume_root, "docker/nas-dashboard/rbac.json")
|
||||
os.makedirs(os.path.dirname(rbac_file), exist_ok=True)
|
||||
|
||||
# Write invalid JSON
|
||||
with open(rbac_file, "w") as f:
|
||||
f.write("invalid json {{{")
|
||||
|
||||
rbac = load_rbac()
|
||||
assert rbac == DEFAULT_RBAC
|
||||
|
||||
|
||||
class TestSaveRBAC:
|
||||
"""Test saving RBAC configuration."""
|
||||
|
||||
def test_save_rbac(self, mock_config, temp_volume_root):
|
||||
"""Test saving RBAC configuration."""
|
||||
rbac_file = os.path.join(temp_volume_root, "docker/nas-dashboard/rbac.json")
|
||||
test_data = {
|
||||
"role_defaults": {"admin": {"pages": "*"}},
|
||||
"user_overrides": {"testuser": {"pages": ["dashboard"]}}
|
||||
}
|
||||
|
||||
save_rbac(test_data)
|
||||
|
||||
assert os.path.exists(rbac_file)
|
||||
with open(rbac_file) as f:
|
||||
loaded = json.load(f)
|
||||
assert loaded == test_data
|
||||
|
||||
def test_save_rbac_creates_directory(self, mock_config, temp_volume_root):
|
||||
"""Test that save_rbac creates directory if it doesn't exist."""
|
||||
rbac_file = os.path.join(temp_volume_root, "docker/nas-dashboard/rbac.json")
|
||||
|
||||
# Remove directory
|
||||
import shutil
|
||||
if os.path.exists(os.path.dirname(rbac_file)):
|
||||
shutil.rmtree(os.path.dirname(rbac_file))
|
||||
|
||||
test_data = {"role_defaults": {}, "user_overrides": {}}
|
||||
save_rbac(test_data)
|
||||
|
||||
assert os.path.exists(rbac_file)
|
||||
|
||||
|
||||
class TestUpdateRBAC:
|
||||
"""Test atomic RBAC updates."""
|
||||
|
||||
def test_update_rbac_with_mutator(self, mock_config, temp_rbac_file):
|
||||
"""Test updating RBAC with mutator function."""
|
||||
def add_user_override(data):
|
||||
data["user_overrides"]["newuser"] = {"pages": ["dashboard", "docker"]}
|
||||
return "success"
|
||||
|
||||
result = update_rbac(add_user_override)
|
||||
|
||||
assert result == "success"
|
||||
|
||||
# Verify changes persisted
|
||||
rbac = load_rbac()
|
||||
assert "newuser" in rbac["user_overrides"]
|
||||
assert rbac["user_overrides"]["newuser"]["pages"] == ["dashboard", "docker"]
|
||||
|
||||
def test_update_rbac_thread_safe(self, mock_config, temp_rbac_file):
|
||||
"""Test that update_rbac is thread-safe."""
|
||||
import threading
|
||||
|
||||
results = []
|
||||
|
||||
def increment_counter(data):
|
||||
current = data.get("counter", 0)
|
||||
data["counter"] = current + 1
|
||||
return data["counter"]
|
||||
|
||||
# Run multiple updates concurrently
|
||||
threads = []
|
||||
for _ in range(10):
|
||||
t = threading.Thread(target=lambda: results.append(update_rbac(increment_counter)))
|
||||
threads.append(t)
|
||||
t.start()
|
||||
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
# Final counter should be 10
|
||||
rbac = load_rbac()
|
||||
assert rbac.get("counter") == 10
|
||||
|
||||
|
||||
class TestResolveRole:
|
||||
"""Test role resolution from groups."""
|
||||
|
||||
def test_resolve_role_admin(self):
|
||||
"""Test resolving admin role."""
|
||||
groups = ["dashboard_admin", "other_group"]
|
||||
role = resolve_role(groups)
|
||||
assert role == "admin"
|
||||
|
||||
def test_resolve_role_member(self):
|
||||
"""Test resolving member role."""
|
||||
groups = ["dashboard_member"]
|
||||
role = resolve_role(groups)
|
||||
assert role == "member"
|
||||
|
||||
def test_resolve_role_viewer(self):
|
||||
"""Test resolving viewer role."""
|
||||
groups = ["dashboard_viewer", "unrelated_group"]
|
||||
role = resolve_role(groups)
|
||||
assert role == "viewer"
|
||||
|
||||
def test_resolve_role_no_match(self):
|
||||
"""Test resolving role with no matching groups."""
|
||||
groups = ["other_group", "another_group"]
|
||||
role = resolve_role(groups)
|
||||
assert role is None
|
||||
|
||||
def test_resolve_role_empty_groups(self):
|
||||
"""Test resolving role with empty groups list."""
|
||||
role = resolve_role([])
|
||||
assert role is None
|
||||
|
||||
def test_resolve_role_priority(self):
|
||||
"""Test that first matching group is used."""
|
||||
groups = ["dashboard_admin", "dashboard_member"]
|
||||
role = resolve_role(groups)
|
||||
assert role == "admin"
|
||||
|
||||
|
||||
class TestGetPages:
|
||||
"""Test page access resolution."""
|
||||
|
||||
def test_get_pages_admin_default(self, mock_config, temp_rbac_file):
|
||||
"""Test getting pages for admin with default settings."""
|
||||
pages = get_pages("testuser", "admin")
|
||||
assert pages == "*"
|
||||
|
||||
def test_get_pages_member_default(self, mock_config, temp_rbac_file):
|
||||
"""Test getting pages for member with default settings."""
|
||||
pages = get_pages("testuser", "member")
|
||||
assert isinstance(pages, list)
|
||||
assert "dashboard" in pages
|
||||
assert "docker" in pages
|
||||
|
||||
def test_get_pages_viewer_default(self, mock_config, temp_rbac_file):
|
||||
"""Test getting pages for viewer with default settings."""
|
||||
pages = get_pages("testuser", "viewer")
|
||||
assert pages == ["dashboard"]
|
||||
|
||||
def test_get_pages_with_user_override(self, mock_config, temp_rbac_file):
|
||||
"""Test getting pages with user-specific override."""
|
||||
# Add user override
|
||||
def add_override(data):
|
||||
data["user_overrides"]["customuser"] = {"pages": ["dashboard", "files"]}
|
||||
|
||||
update_rbac(add_override)
|
||||
|
||||
pages = get_pages("customuser", "admin")
|
||||
assert pages == ["dashboard", "files"]
|
||||
|
||||
def test_get_pages_unknown_role(self, mock_config, temp_rbac_file):
|
||||
"""Test getting pages for unknown role returns empty list."""
|
||||
pages = get_pages("testuser", "unknown_role")
|
||||
assert pages == []
|
||||
|
||||
|
||||
class TestGetSidebarLinks:
|
||||
"""Test sidebar link access resolution."""
|
||||
|
||||
def test_get_sidebar_links_admin_default(self, mock_config, temp_rbac_file):
|
||||
"""Test getting sidebar links for admin."""
|
||||
links = get_sidebar_links("testuser", "admin")
|
||||
assert links == "*"
|
||||
|
||||
def test_get_sidebar_links_member_default(self, mock_config, temp_rbac_file):
|
||||
"""Test getting sidebar links for member."""
|
||||
links = get_sidebar_links("testuser", "member")
|
||||
assert links == "*"
|
||||
|
||||
def test_get_sidebar_links_viewer_default(self, mock_config, temp_rbac_file):
|
||||
"""Test getting sidebar links for viewer."""
|
||||
links = get_sidebar_links("testuser", "viewer")
|
||||
assert links == ["dashboard"]
|
||||
|
||||
def test_get_sidebar_links_with_user_override(self, mock_config, temp_rbac_file):
|
||||
"""Test getting sidebar links with user override."""
|
||||
def add_override(data):
|
||||
data["user_overrides"]["customuser"] = {
|
||||
"pages": "*",
|
||||
"sidebar_links": ["dashboard", "docker", "files"]
|
||||
}
|
||||
|
||||
update_rbac(add_override)
|
||||
|
||||
links = get_sidebar_links("customuser", "admin")
|
||||
assert links == ["dashboard", "docker", "files"]
|
||||
|
||||
def test_get_sidebar_links_no_override(self, mock_config, temp_rbac_file):
|
||||
"""Test that user without sidebar_links override gets role default."""
|
||||
def add_override(data):
|
||||
data["user_overrides"]["customuser"] = {"pages": ["dashboard"]}
|
||||
# No sidebar_links specified
|
||||
|
||||
update_rbac(add_override)
|
||||
|
||||
links = get_sidebar_links("customuser", "admin")
|
||||
assert links == "*" # Should fall back to admin default
|
||||
|
||||
|
||||
class TestGetSidebarOrder:
|
||||
"""Test sidebar order retrieval."""
|
||||
|
||||
def test_get_sidebar_order_no_override(self, mock_config, temp_rbac_file):
|
||||
"""Test getting sidebar order when user has no override."""
|
||||
order = get_sidebar_order("testuser")
|
||||
assert order is None
|
||||
|
||||
def test_get_sidebar_order_with_override(self, mock_config, temp_rbac_file):
|
||||
"""Test getting sidebar order with user override."""
|
||||
custom_order = {
|
||||
"Main": ["dashboard", "docker"],
|
||||
"Media": ["navidrome", "immich"]
|
||||
}
|
||||
|
||||
def add_override(data):
|
||||
data["user_overrides"]["customuser"] = {"sidebar_order": custom_order}
|
||||
|
||||
update_rbac(add_override)
|
||||
|
||||
order = get_sidebar_order("customuser")
|
||||
assert order == custom_order
|
||||
|
||||
|
||||
class TestDefaultRBAC:
|
||||
"""Test default RBAC configuration."""
|
||||
|
||||
def test_default_rbac_structure(self):
|
||||
"""Test that DEFAULT_RBAC has correct structure."""
|
||||
assert "role_defaults" in DEFAULT_RBAC
|
||||
assert "user_overrides" in DEFAULT_RBAC
|
||||
|
||||
assert "admin" in DEFAULT_RBAC["role_defaults"]
|
||||
assert "member" in DEFAULT_RBAC["role_defaults"]
|
||||
assert "viewer" in DEFAULT_RBAC["role_defaults"]
|
||||
|
||||
def test_default_rbac_admin_permissions(self):
|
||||
"""Test admin has full permissions by default."""
|
||||
admin = DEFAULT_RBAC["role_defaults"]["admin"]
|
||||
assert admin["pages"] == "*"
|
||||
assert admin["sidebar_links"] == "*"
|
||||
|
||||
def test_default_rbac_member_permissions(self):
|
||||
"""Test member has limited permissions."""
|
||||
member = DEFAULT_RBAC["role_defaults"]["member"]
|
||||
assert isinstance(member["pages"], list)
|
||||
assert "dashboard" in member["pages"]
|
||||
assert member["sidebar_links"] == "*"
|
||||
|
||||
def test_default_rbac_viewer_permissions(self):
|
||||
"""Test viewer has minimal permissions."""
|
||||
viewer = DEFAULT_RBAC["role_defaults"]["viewer"]
|
||||
assert viewer["pages"] == ["dashboard"]
|
||||
assert viewer["sidebar_links"] == ["dashboard"]
|
||||
|
||||
|
||||
class TestRoleGroupMap:
|
||||
"""Test role group mapping."""
|
||||
|
||||
def test_role_group_map_completeness(self):
|
||||
"""Test that ROLE_GROUP_MAP has all expected mappings."""
|
||||
assert "dashboard_admin" in ROLE_GROUP_MAP
|
||||
assert "dashboard_member" in ROLE_GROUP_MAP
|
||||
assert "dashboard_viewer" in ROLE_GROUP_MAP
|
||||
|
||||
assert ROLE_GROUP_MAP["dashboard_admin"] == "admin"
|
||||
assert ROLE_GROUP_MAP["dashboard_member"] == "member"
|
||||
assert ROLE_GROUP_MAP["dashboard_viewer"] == "viewer"
|
||||
Reference in New Issue
Block a user