From a8d44b189c4212144fe5f772f89e2a6f8f05501c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 5 Mar 2026 11:37:54 +0000 Subject: [PATCH] refactor(multi-user): address code review - module imports, explicit false(), string length Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/files.py | 9 +--- app/utils/user_scope.py | 3 +- .../versions/012_add_multi_user_support.py | 2 +- tests/test_multi_user.py | 44 +++++++++---------- 4 files changed, 26 insertions(+), 32 deletions(-) diff --git a/app/api/files.py b/app/api/files.py index 4d7fb078..f2b74b1f 100644 --- a/app/api/files.py +++ b/app/api/files.py @@ -28,6 +28,7 @@ from app.utils.file_queries import apply_status_filter from app.utils.file_status import get_files_processing_status from app.utils.filename_utils import sanitize_filename from app.utils.input_validation import validate_search_query, validate_sort_field, validate_sort_order +from app.utils.user_scope import apply_owner_filter, get_current_owner_id # Set up logging logger = logging.getLogger(__name__) @@ -100,8 +101,6 @@ def list_files_api( search = validate_search_query(search) # Start with base query, scoped to the current user in multi-user mode - from app.utils.user_scope import apply_owner_filter - query = db.query(FileRecord) query = apply_owner_filter(query, request) @@ -245,8 +244,6 @@ def get_file_details(request: Request, file_id: int, db: DbSession): Get detailed information about a specific file including processing history. """ # Find the file record, scoped to the current user in multi-user mode - from app.utils.user_scope import apply_owner_filter - query = db.query(FileRecord).filter(FileRecord.id == file_id) query = apply_owner_filter(query, request) file_record = query.first() @@ -308,8 +305,6 @@ def delete_file_record(request: Request, file_id: int, db: DbSession): try: # Find the file record, scoped to the current user in multi-user mode - from app.utils.user_scope import apply_owner_filter - query = db.query(FileRecord).filter(FileRecord.id == file_id) query = apply_owner_filter(query, request) file_record = query.first() @@ -1298,8 +1293,6 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(... file_ext = os.path.splitext(target_path)[1].lower() # Determine the owner_id for multi-user document isolation - from app.utils.user_scope import get_current_owner_id - upload_owner_id = get_current_owner_id(request) if settings.multi_user_enabled else None # Check if it's a PDF by extension or MIME type diff --git a/app/utils/user_scope.py b/app/utils/user_scope.py index a89ec4ca..d1be44e3 100644 --- a/app/utils/user_scope.py +++ b/app/utils/user_scope.py @@ -11,6 +11,7 @@ import logging from fastapi import Request from sqlalchemy.orm import Query +from sqlalchemy.sql import false from app.config import settings from app.models import FileRecord @@ -66,6 +67,6 @@ def apply_owner_filter(query: Query, request: Request) -> Query: owner_id = get_current_owner_id(request) if owner_id is None: # No authenticated user — return empty result set - return query.filter(FileRecord.id < 0) + return query.filter(false()) return query.filter(FileRecord.owner_id == owner_id) diff --git a/migrations/versions/012_add_multi_user_support.py b/migrations/versions/012_add_multi_user_support.py index 655c413f..6dd6e804 100644 --- a/migrations/versions/012_add_multi_user_support.py +++ b/migrations/versions/012_add_multi_user_support.py @@ -19,7 +19,7 @@ depends_on: Union[str, None] = None def upgrade() -> None: """Add owner_id column to files table for multi-user document isolation.""" - op.add_column("files", sa.Column("owner_id", sa.String(), nullable=True)) + op.add_column("files", sa.Column("owner_id", sa.String(255), nullable=True)) op.create_index("ix_files_owner_id", "files", ["owner_id"]) diff --git a/tests/test_multi_user.py b/tests/test_multi_user.py index 9f3b7b2c..5c9c0f5d 100644 --- a/tests/test_multi_user.py +++ b/tests/test_multi_user.py @@ -47,7 +47,7 @@ def mu_session(mu_engine): session.close() -def _make_file(session, owner_id=None, filename="test.pdf"): +def _create_file_record(session, owner_id=None, filename="test.pdf"): """Helper to insert a minimal FileRecord.""" rec = FileRecord( filehash="abc123", @@ -87,21 +87,21 @@ class TestFileRecordOwnerField: @pytest.mark.unit def test_owner_id_defaults_to_none(self, mu_session): """FileRecord created without owner_id should have None.""" - rec = _make_file(mu_session) + rec = _create_file_record(mu_session) assert rec.owner_id is None @pytest.mark.unit def test_owner_id_stores_value(self, mu_session): """FileRecord created with owner_id should persist it.""" - rec = _make_file(mu_session, owner_id="user@example.com") + rec = _create_file_record(mu_session, owner_id="user@example.com") assert rec.owner_id == "user@example.com" @pytest.mark.unit def test_owner_id_filterable(self, mu_session): """Can query FileRecord by owner_id.""" - _make_file(mu_session, owner_id="alice") - _make_file(mu_session, owner_id="bob") - _make_file(mu_session, owner_id=None) + _create_file_record(mu_session, owner_id="alice") + _create_file_record(mu_session, owner_id="bob") + _create_file_record(mu_session, owner_id=None) alice_files = mu_session.query(FileRecord).filter(FileRecord.owner_id == "alice").all() assert len(alice_files) == 1 @@ -171,9 +171,9 @@ class TestApplyOwnerFilter: """When multi_user_enabled=False, all files are returned.""" from app.utils.user_scope import apply_owner_filter - _make_file(mu_session, owner_id="alice") - _make_file(mu_session, owner_id="bob") - _make_file(mu_session, owner_id=None) + _create_file_record(mu_session, owner_id="alice") + _create_file_record(mu_session, owner_id="bob") + _create_file_record(mu_session, owner_id=None) request = _mock_request(user={"preferred_username": "alice"}) query = mu_session.query(FileRecord) @@ -188,9 +188,9 @@ class TestApplyOwnerFilter: """When multi_user_enabled=True, only user's files are returned.""" from app.utils.user_scope import apply_owner_filter - _make_file(mu_session, owner_id="alice") - _make_file(mu_session, owner_id="bob") - _make_file(mu_session, owner_id=None) + _create_file_record(mu_session, owner_id="alice") + _create_file_record(mu_session, owner_id="bob") + _create_file_record(mu_session, owner_id=None) request = _mock_request(user={"preferred_username": "alice"}) query = mu_session.query(FileRecord) @@ -207,9 +207,9 @@ class TestApplyOwnerFilter: """Admin users bypass the owner filter in multi-user mode.""" from app.utils.user_scope import apply_owner_filter - _make_file(mu_session, owner_id="alice") - _make_file(mu_session, owner_id="bob") - _make_file(mu_session, owner_id=None) + _create_file_record(mu_session, owner_id="alice") + _create_file_record(mu_session, owner_id="bob") + _create_file_record(mu_session, owner_id=None) request = _mock_request(user={"preferred_username": "admin", "is_admin": True}) query = mu_session.query(FileRecord) @@ -224,7 +224,7 @@ class TestApplyOwnerFilter: """When no user is logged in and multi-user is enabled, return empty.""" from app.utils.user_scope import apply_owner_filter - _make_file(mu_session, owner_id="alice") + _create_file_record(mu_session, owner_id="alice") request = _mock_request(user=None) query = mu_session.query(FileRecord) @@ -319,8 +319,8 @@ class TestFilesAPIMultiUser: @pytest.mark.integration def test_list_files_unscoped_single_user(self, client, db_session): """In single-user mode all files are visible.""" - _make_file(db_session, owner_id="alice", filename="a.pdf") - _make_file(db_session, owner_id="bob", filename="b.pdf") + _create_file_record(db_session, owner_id="alice", filename="a.pdf") + _create_file_record(db_session, owner_id="bob", filename="b.pdf") with _patch_multi_user(False): response = client.get("/api/files") @@ -332,8 +332,8 @@ class TestFilesAPIMultiUser: @pytest.mark.integration def test_list_files_scoped_multi_user(self, client, db_session): """In multi-user mode only the user's files should be returned.""" - _make_file(db_session, owner_id="alice", filename="a.pdf") - _make_file(db_session, owner_id="bob", filename="b.pdf") + _create_file_record(db_session, owner_id="alice", filename="a.pdf") + _create_file_record(db_session, owner_id="bob", filename="b.pdf") with _patch_multi_user(True): # Without a real session, the filter will return no results @@ -347,7 +347,7 @@ class TestFilesAPIMultiUser: @pytest.mark.integration def test_get_file_detail_respects_scope(self, client, db_session): """File detail endpoint should return 404 for files owned by other users.""" - rec = _make_file(db_session, owner_id="alice", filename="a.pdf") + rec = _create_file_record(db_session, owner_id="alice", filename="a.pdf") with _patch_multi_user(True): response = client.get(f"/api/files/{rec.id}") @@ -357,7 +357,7 @@ class TestFilesAPIMultiUser: @pytest.mark.integration def test_get_file_detail_single_user_mode(self, client, db_session): """File detail endpoint should work normally in single-user mode.""" - rec = _make_file(db_session, owner_id="alice", filename="a.pdf") + rec = _create_file_record(db_session, owner_id="alice", filename="a.pdf") with _patch_multi_user(False): response = client.get(f"/api/files/{rec.id}")