Gan, Jimmy
b981c06d59
- Fix unit test imports: add env setup in conftest.py before module imports - Add 24 new auth router tests (RBAC, preferences, password validation) - Add 16 new tests for litellm and chat_summary routers - Apply black formatting and ruff linting across codebase - Add pre-commit hooks configuration (black, ruff, file checks) - Increase CI coverage threshold from 40% to 50% Test Results: - 206 tests passing (91 unit + 115 integration) - Coverage: 58.79% on core modules - auth.py: 57% → 85%, litellm.py: 23% → 87%, chat_summary.py: 41% → 100% - auth_service: 96.51%, config: 100%, rbac: 93.48%
142 lines
5.2 KiB
Python
142 lines
5.2 KiB
Python
"""
|
|
Integration tests for file operations.
|
|
"""
|
|
|
|
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 "entries" 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, params={"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, params={"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 datetime import timedelta
|
|
|
|
from auth_service import create_access_token
|
|
|
|
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, params={"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 datetime import timedelta
|
|
|
|
from auth_service import create_access_token
|
|
|
|
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
|