From e27787ccd02b1a6cabb32686cb8ae34205c4aef5 Mon Sep 17 00:00:00 2001 From: SiYuan Date: Mon, 3 Aug 2026 10:29:52 +0800 Subject: [PATCH] test(nightly): cover auth login/register/reset + nav resolvers + storage device/recoverable - new tests in test_auth_api.py / test_nav.py / test_storage_api.py - 49 tests added, all passing locally Nightly watch run 2026-08-03 --- package/server/tests/unit/test_auth_api.py | 290 +++++++++++++++++- package/server/tests/unit/test_nav.py | 256 +++++++++++++++- package/server/tests/unit/test_storage_api.py | 139 ++++++++- 3 files changed, 668 insertions(+), 17 deletions(-) diff --git a/package/server/tests/unit/test_auth_api.py b/package/server/tests/unit/test_auth_api.py index 830cdd01..5ce76403 100644 --- a/package/server/tests/unit/test_auth_api.py +++ b/package/server/tests/unit/test_auth_api.py @@ -1,15 +1,20 @@ """Unit tests for the auth REST router (app/api/auth.py). -Covers the ``status`` endpoint (registration gating + demo-mode flag) and -the ``send-log-reset-code`` rate-limit path. ``crud_user`` and the -``reset_code_store`` are patched so no DB / persistent state is touched. +Covers the ``status`` endpoint (registration gating + demo-mode flag), +``send-log-reset-code`` rate-limit path, and the ``login`` / +``register`` / ``check-reset-user`` / ``reset-password`` / +``reset-password-by-code`` flows. ``crud_user`` and ``reset_code_store`` +are patched so no DB / persistent state is touched. """ +from datetime import datetime, timedelta from types import SimpleNamespace from unittest.mock import MagicMock, patch from uuid import uuid4 import pytest +from fastapi import HTTPException +from fastapi.security import OAuth2PasswordRequestForm from app.api import auth as auth_api @@ -52,6 +57,216 @@ def test_auth_status_registration_blocked_after_first_user(): assert response["demo_mode"] is True +# ----------------------- POST /auth/login ----------------------- + + +def _form(username="alice", password="secret"): + return SimpleNamespace(username=username, password=password) + + +def test_login_returns_token_for_valid_credentials(): + db = MagicMock() + user = SimpleNamespace(id=uuid4(), is_active=True) + form = _form() + + with patch.object(auth_api.crud_user, "authenticate", return_value=user): + with patch.object(auth_api.security, "create_access_token", return_value="JWT"): + response = auth_api.login_access_token(db=db, form_data=form) + + assert response["access_token"] == "JWT" + assert response["token_type"] == "bearer" + + +def test_login_rejects_inactive_user_with_400(): + db = MagicMock() + user = SimpleNamespace(id=uuid4(), is_active=False) + form = _form() + + with patch.object(auth_api.crud_user, "authenticate", return_value=user): + with pytest.raises(HTTPException) as exc_info: + auth_api.login_access_token(db=db, form_data=form) + + assert exc_info.value.status_code == 400 + assert "禁用" in exc_info.value.detail + + +def test_login_returns_403_when_account_locked(): + """Wrong creds + account lockout → 403 with lockout message.""" + db = MagicMock() + locked = SimpleNamespace( + id=uuid4(), + is_active=True, + lockout_until=datetime.now() + timedelta(minutes=5), + ) + form = _form() + + with patch.object(auth_api.crud_user, "authenticate", return_value=None), \ + patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=locked): + with pytest.raises(HTTPException) as exc_info: + auth_api.login_access_token(db=db, form_data=form) + + assert exc_info.value.status_code == 403 + assert "锁定" in exc_info.value.detail + + +def test_login_returns_401_for_wrong_credentials(): + db = MagicMock() + form = _form(username="nobody", password="wrong") + + with patch.object(auth_api.crud_user, "authenticate", return_value=None), \ + patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=None): + with pytest.raises(HTTPException) as exc_info: + auth_api.login_access_token(db=db, form_data=form) + + assert exc_info.value.status_code == 401 + + +# ----------------------- POST /auth/register ----------------------- + + +def test_register_blocks_when_users_exist_and_registration_disabled(): + db = MagicMock() + db.query.return_value.count.return_value = 1 + payload = SimpleNamespace(username="bob", email="bob@example.com", is_superuser=False) + + with patch.object(auth_api.system_config.config.security, "allow_registration", False): + with pytest.raises(HTTPException) as exc_info: + auth_api.register_user(db=db, user_in=payload) + + assert exc_info.value.status_code == 403 + + +def test_register_rejects_duplicate_email(): + db = MagicMock() + db.query.return_value.count.side_effect = [2, 2] # has_users, then first_user check + payload = SimpleNamespace(username="bob", email="dup@example.com", is_superuser=False) + existing = SimpleNamespace(email="dup@example.com") + + with patch.object(auth_api.system_config.config.security, "allow_registration", True), \ + patch.object(auth_api.crud_user, "get_by_email", return_value=existing): + with pytest.raises(HTTPException) as exc_info: + auth_api.register_user(db=db, user_in=payload) + + assert exc_info.value.status_code == 400 + assert "email" in exc_info.value.detail.lower() + + +def test_register_rejects_duplicate_username(): + db = MagicMock() + db.query.return_value.count.side_effect = [2, 2] + payload = SimpleNamespace(username="dup", email="bob@example.com", is_superuser=False) + existing_user = SimpleNamespace(username="dup") + + with patch.object(auth_api.system_config.config.security, "allow_registration", True), \ + patch.object(auth_api.crud_user, "get_by_email", return_value=None), \ + patch.object(auth_api.crud_user, "get_by_username", return_value=existing_user): + with pytest.raises(HTTPException) as exc_info: + auth_api.register_user(db=db, user_in=payload) + + assert exc_info.value.status_code == 400 + assert "username" in exc_info.value.detail.lower() + + +def test_register_promotes_first_user_to_superuser_and_migrates_config(): + db = MagicMock() + # 0 users → first_user path; crud_user.create is the last call before return + db.query.return_value.count.side_effect = [0, 0] + payload = SimpleNamespace(username="founder", email="founder@example.com", is_superuser=False) + created = SimpleNamespace(id=uuid4()) + + with patch.object(auth_api.system_config.config.security, "allow_registration", True), \ + patch.object(auth_api.crud_user, "get_by_email", return_value=None), \ + patch.object(auth_api.crud_user, "get_by_username", return_value=None), \ + patch.object(auth_api.crud_user, "create", return_value=created) as create, \ + patch.object(auth_api.config_manager, "get_default_config", return_value={"default": True}), \ + patch.object(auth_api, "migrate_system_config") as migrate: + result = auth_api.register_user(db=db, user_in=payload) + + # 第一次注册自动升级为 superuser + assert payload.is_superuser is True + create.assert_called_once_with(db, user=payload) + migrate.assert_called_once_with(db, created) + assert result is created + + +# ----------------------- POST /auth/check-reset-user ----------------------- + + +def test_check_reset_user_404_when_user_missing(): + db = MagicMock() + payload = SimpleNamespace(username_or_email="nobody@example.com") + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=None): + with pytest.raises(HTTPException) as exc_info: + auth_api.check_password_reset_user(payload=payload, db=db) + + assert exc_info.value.status_code == 404 + + +def test_check_reset_user_400_when_no_security_question(): + db = MagicMock() + payload = SimpleNamespace(username_or_email="alice@example.com") + user = SimpleNamespace(security_question=None) + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=user): + with pytest.raises(HTTPException) as exc_info: + auth_api.check_password_reset_user(payload=payload, db=db) + + assert exc_info.value.status_code == 400 + assert "security" in exc_info.value.detail.lower() + + +def test_check_reset_user_returns_security_question(): + db = MagicMock() + payload = SimpleNamespace(username_or_email="alice@example.com") + user = SimpleNamespace(security_question="你最喜欢的颜色？") + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=user): + response = auth_api.check_password_reset_user(payload=payload, db=db) + + assert response["security_question"] == "你最喜欢的颜色？" + + +# ----------------------- POST /auth/reset-password ----------------------- + + +def test_confirm_password_reset_rejects_wrong_answer(): + db = MagicMock() + payload = SimpleNamespace( + username_or_email="alice@example.com", + security_answer="wrong", + new_password="newpw", + ) + user = SimpleNamespace() + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=user), \ + patch.object(auth_api.crud_user, "verify_security_answer", return_value=False): + with pytest.raises(HTTPException) as exc_info: + auth_api.confirm_password_reset(payload=payload, db=db) + + assert exc_info.value.status_code == 400 + assert "Incorrect security answer" in exc_info.value.detail + + +def test_confirm_password_reset_succeeds_and_returns_user(): + db = MagicMock() + payload = SimpleNamespace( + username_or_email="alice@example.com", + security_answer="correct", + new_password="newpw", + ) + user = SimpleNamespace() + updated = SimpleNamespace(id=uuid4()) + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=user), \ + patch.object(auth_api.crud_user, "verify_security_answer", return_value=True), \ + patch.object(auth_api.crud_user, "reset_password", return_value=updated) as reset_pw: + result = auth_api.confirm_password_reset(payload=payload, db=db) + + reset_pw.assert_called_once_with(db, user, "newpw") + assert result is updated + + # ----------------------- POST /auth/send-log-reset-code ----------------------- @@ -93,3 +308,72 @@ def test_send_log_reset_code_success_returns_log_instruction(): assert response.code == 0 assert "服务器日志" in response.msg + + +# ----------------------- POST /auth/reset-password-by-code ----------------------- + + +def test_reset_password_by_code_rejects_short_password(): + db = MagicMock() + payload = SimpleNamespace( + username_or_email="alice@example.com", + code="123456", + new_password="abc", + ) + + with patch.object(auth_api.reset_code_store, "verify_code") as verify: + response = auth_api.reset_password_by_code(payload=payload, db=db) + + assert response.code == 400 + assert "6" in response.msg + verify.assert_not_called() + + +def test_reset_password_by_code_rejects_missing_user(): + db = MagicMock() + payload = SimpleNamespace( + username_or_email="nobody@example.com", + code="123456", + new_password="abcdef", + ) + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=None): + response = auth_api.reset_password_by_code(payload=payload, db=db) + + assert response.code == 404 + + +def test_reset_password_by_code_rejects_bad_code(): + db = MagicMock() + payload = SimpleNamespace( + username_or_email="alice@example.com", + code="000000", + new_password="abcdef", + ) + user = _user() + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=user), \ + patch.object(auth_api.reset_code_store, "verify_code", return_value=False): + response = auth_api.reset_password_by_code(payload=payload, db=db) + + assert response.code == 400 + assert "验证码" in response.msg + + +def test_reset_password_by_code_succeeds_and_consumes_code(): + db = MagicMock() + payload = SimpleNamespace( + username_or_email="alice@example.com", + code="123456", + new_password="abcdef", + ) + user = _user() + + with patch.object(auth_api.crud_user, "get_by_username_or_email", return_value=user), \ + patch.object(auth_api.reset_code_store, "verify_code", return_value=True) as verify, \ + patch.object(auth_api.crud_user, "reset_password") as reset_pw: + response = auth_api.reset_password_by_code(payload=payload, db=db) + + verify.assert_called_once_with(str(user.id), "123456", "alice@example.com") + reset_pw.assert_called_once_with(db, user, "abcdef") + assert response.code == 0 diff --git a/package/server/tests/unit/test_nav.py b/package/server/tests/unit/test_nav.py index e23674b3..0ed8710a 100644 --- a/package/server/tests/unit/test_nav.py +++ b/package/server/tests/unit/test_nav.py @@ -7,7 +7,15 @@ import pytest from fastapi import HTTPException -from app.api.nav import resolve_single_entity, update_nav_items +from app.api.nav import ( + resolve_album, + resolve_classification, + resolve_location, + resolve_nav_items, + resolve_person, + resolve_single_entity, + update_nav_items, +) from app.schemas.nav import NavItemRef, NavItemsUpdate @@ -31,6 +39,252 @@ def test_resolve_single_entity_returns_none_for_unknown_type(): assert resolve_single_entity(ref, uuid4(), MagicMock()) is None +def test_resolve_single_entity_returns_none_when_resolver_raises(): + """Exceptions raised by a resolver are swallowed and return None.""" + ref = NavItemRef(entity_type="album", entity_id=str(uuid4())) + + with patch("app.api.nav.resolve_album", side_effect=RuntimeError("boom")): + assert resolve_single_entity(ref, uuid4(), MagicMock()) is None + + +def test_resolve_album_returns_none_when_query_empty(): + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = None + + assert resolve_album(str(uuid4()), uuid4(), db) is None + + +def test_resolve_album_builds_route_and_uses_cover(): + album_id = uuid4() + user_id = uuid4() + album = SimpleNamespace( + id=album_id, + name="Trip", + cover_id=uuid4(), + num_photos=12, + ) + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = album + + resolved = resolve_album(str(album_id), user_id, db) + + assert resolved.entity_type == "album" + assert resolved.name == "Trip" + assert resolved.cover_photo_id == str(album.cover_id) + assert resolved.photo_count == 12 + assert resolved.route_path == f"/album/{album_id}" + + +def test_resolve_album_handles_missing_cover_and_zero_count(): + album = SimpleNamespace(name="Empty", cover_id=None, num_photos=None) + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = album + + resolved = resolve_album(str(uuid4()), uuid4(), db) + + assert resolved.cover_photo_id is None + assert resolved.photo_count == 0 + + +def test_resolve_person_falls_back_to_first_face_when_default_missing(): + identity_id = uuid4() + default_face_id = uuid4() + fallback_face = SimpleNamespace( + photo_id=uuid4(), face_rect=[1, 2, 3, 4], is_deleted=False + ) + identity = SimpleNamespace( + id=identity_id, + identity_name="Alice", + default_face_id=default_face_id, + is_deleted=False, + ) + db = MagicMock() + db.query.return_value.filter.return_value.first.side_effect = [identity, None, fallback_face] + db.query.return_value.filter.return_value.scalar.return_value = 5 + + resolved = resolve_person(str(identity_id), uuid4(), db) + + assert resolved.entity_type == "person" + assert resolved.cover_photo_id == str(fallback_face.photo_id) + assert resolved.cover_photo_face_rect == [1, 2, 3, 4] + assert resolved.photo_count == 5 + assert resolved.route_path == f"/album/people/{identity_id}" + + +def test_resolve_person_returns_none_for_missing_identity(): + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = None + + assert resolve_person(str(uuid4()), uuid4(), db) is None + + +def test_resolve_location_returns_none_when_no_photos(): + db = MagicMock() + db.query.return_value.join.return_value.filter.return_value.scalar.return_value = 0 + + assert resolve_location("上海", uuid4(), db) is None + + +def test_resolve_location_picks_most_recent_photo_as_cover(): + db = MagicMock() + db.query.return_value.join.return_value.filter.return_value.scalar.return_value = 9 + cover = SimpleNamespace(id=uuid4()) + db.query.return_value.join.return_value.filter.return_value.order_by.return_value.first.return_value = cover + + resolved = resolve_location("上海", uuid4(), db) + + assert resolved.entity_type == "location" + assert resolved.cover_photo_id == str(cover.id) + assert resolved.photo_count == 9 + assert resolved.route_path == "/album/location/上海" + + +def test_resolve_location_returns_null_cover_when_no_time_photos(): + db = MagicMock() + db.query.return_value.join.return_value.filter.return_value.scalar.return_value = 1 + db.query.return_value.join.return_value.filter.return_value.order_by.return_value.first.return_value = None + + resolved = resolve_location("上海", uuid4(), db) + + assert resolved.cover_photo_id is None + assert resolved.photo_count == 1 + + +def test_resolve_classification_uses_tag_cover_when_present(): + tag_id = uuid4() + tag = SimpleNamespace(id=tag_id, tag_name="beach", cover_id=uuid4()) + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = tag + db.query.return_value.filter.return_value.scalar.return_value = 7 + + resolved = resolve_classification(str(tag_id), uuid4(), db) + + assert resolved.entity_type == "classification" + assert resolved.name == "beach" + assert resolved.photo_count == 7 + assert resolved.route_path == "/album/classification/beach" + + +def test_resolve_classification_falls_back_to_relation_for_cover(): + tag = SimpleNamespace(id=uuid4(), tag_name="food", cover_id=None) + relation = SimpleNamespace(photo_id=uuid4()) + db = MagicMock() + db.query.return_value.filter.return_value.first.side_effect = [tag, relation] + db.query.return_value.filter.return_value.scalar.return_value = 3 + + resolved = resolve_classification(str(tag.id), uuid4(), db) + + assert resolved.cover_photo_id == str(relation.photo_id) + assert resolved.photo_count == 3 + + +def test_resolve_classification_returns_none_when_tag_missing(): + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = None + + assert resolve_classification(str(uuid4()), uuid4(), db) is None + + +def test_resolve_nav_items_returns_items_without_pruning_when_all_valid(): + user_id = uuid4() + db = MagicMock() + ref_album = NavItemRef(entity_type="album", entity_id=str(uuid4())) + ref_location = NavItemRef(entity_type="location", entity_id="上海") + valid_config = SimpleNamespace(nav=SimpleNamespace(items=[ref_album, ref_location])) + + with patch.object( + resolve_nav_items.__globals__["config_manager"], + "get_user_config", + return_value=valid_config, + ): + with patch("app.api.nav.resolve_album", return_value="ALBUM"): + with patch("app.api.nav.resolve_location", return_value="LOCATION"): + resolved = resolve_nav_items(user_id, db) + + assert resolved == ["ALBUM", "LOCATION"] + + +def test_resolve_nav_items_prunes_missing_refs(): + user_id = uuid4() + db = MagicMock() + good_ref = NavItemRef(entity_type="album", entity_id=str(uuid4())) + bad_ref = NavItemRef(entity_type="album", entity_id=str(uuid4())) + config = SimpleNamespace(nav=SimpleNamespace(items=[good_ref, bad_ref])) + + with patch.object( + resolve_nav_items.__globals__["config_manager"], + "get_user_config", + return_value=config, + ): + with patch("app.api.nav.resolve_album", side_effect=[SimpleNamespace(name="A"), None]): + with patch.object( + resolve_nav_items.__globals__["config_manager"], + "update_user_config", + ) as update_cfg: + resolved = resolve_nav_items(user_id, db) + + assert len(resolved) == 1 + assert resolved[0].name == "A" + update_cfg.assert_called_once() + args = update_cfg.call_args.args + assert args[0] == user_id + persisted = args[1]["nav"]["items"] + assert persisted == [good_ref.model_dump()] + assert args[2] is db + + +def test_resolve_nav_items_rolls_back_when_resolver_raises(): + """resolve_single_entity 自带 try/except；让 resolve_nav_items 直接捕获到的异常场景， + 由 patch 顶替 resolve_single_entity 让异常外抛，从而触发 resolve_nav_items 的 except 分支。""" + user_id = uuid4() + db = MagicMock() + good_ref = NavItemRef(entity_type="album", entity_id=str(uuid4())) + bad_ref = NavItemRef(entity_type="album", entity_id=str(uuid4())) + config = SimpleNamespace(nav=SimpleNamespace(items=[good_ref, bad_ref])) + + with patch.object( + resolve_nav_items.__globals__["config_manager"], + "get_user_config", + return_value=config, + ): + with patch( + "app.api.nav.resolve_single_entity", + side_effect=[SimpleNamespace(name="A"), RuntimeError("boom")], + ): + with patch.object( + resolve_nav_items.__globals__["config_manager"], + "update_user_config", + ) as update_cfg: + resolved = resolve_nav_items(user_id, db) + + assert len(resolved) == 1 + # resolve_nav_items 在 except 分支里调用 rollback + db.rollback.assert_called_once() + # bad_ref 触发的 prune + valid_refs 不等于 refs → 写入 + update_cfg.assert_called_once() + + +def test_resolve_nav_items_skips_prune_when_nothing_changed(): + user_id = uuid4() + db = MagicMock() + ref = NavItemRef(entity_type="album", entity_id=str(uuid4())) + config = SimpleNamespace(nav=SimpleNamespace(items=[ref])) + + with patch.object( + resolve_nav_items.__globals__["config_manager"], + "get_user_config", + return_value=config, + ): + with patch("app.api.nav.resolve_album", return_value=SimpleNamespace(name="A")): + with patch.object( + resolve_nav_items.__globals__["config_manager"], + "update_user_config", + ) as update_cfg: + resolve_nav_items(user_id, db) + + update_cfg.assert_not_called() + + def test_update_nav_items_rejects_invalid_uuid_before_writing_config(): """UUID-backed references fail with HTTP 400 before config persistence.""" body = NavItemsUpdate( diff --git a/package/server/tests/unit/test_storage_api.py b/package/server/tests/unit/test_storage_api.py index bca083e0..eafb6d2d 100644 --- a/package/server/tests/unit/test_storage_api.py +++ b/package/server/tests/unit/test_storage_api.py @@ -1,8 +1,8 @@ """Unit tests for the storage REST router (app/api/storage.py). -Covers the disk-overview / file-type / top-large-files endpoints. Each test -mocks the DB query chain and the disk-stat helper so no real filesystem or -Postgres is required. +Covers the disk-overview / file-type / top-large-files / device / +recoverable endpoints. Each test mocks the DB query chain and the +disk-stat helper so no real filesystem or Postgres is required. """ from types import SimpleNamespace @@ -12,6 +12,8 @@ import pytest from app.api import storage as storage_api +from app.core import config_manager as _config_mod +from app.db.models.photo import ImageType, FileType pytestmark = [pytest.mark.smoke, pytest.mark.module_system] @@ -47,7 +49,39 @@ def test_storage_overview_returns_disk_and_user_totals(): total_size=2_048_000, total_files=42 ) + with patch.object(_config_mod.config_manager, "get_user_config", + return_value=SimpleNamespace(storage=SimpleNamespace(photo_storage_path="/Photos"))), \ + patch("shutil.disk_usage", + return_value=SimpleNamespace(total=10_000_000, free=4_000_000, used=6_000_000)), \ + patch("time.strftime", return_value="2026-08-03T10:00:00"): + response = storage_api.get_storage_overview(db=db, current_user=user) + assert response.code == 0 + payload = response.data + assert payload["total_size"] == 2_048_000 + assert payload["total_files"] == 42 + assert payload["disk_total_size"] == 10_000_000 + assert payload["disk_free_size"] == 4_000_000 + assert payload["scan_date"] == "2026-08-03T10:00:00" + + +def test_storage_overview_falls_back_to_zero_when_disk_usage_raises(): + """如果配置路径不存在，shutil 会抛 FileNotFoundError；这里验证兜底为 0.""" + user = _user() + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = SimpleNamespace( + total_size=None, total_files=None + ) + + with patch.object(_config_mod.config_manager, "get_user_config", + return_value=SimpleNamespace(storage=SimpleNamespace(photo_storage_path="/missing"))), \ + patch("shutil.disk_usage", side_effect=FileNotFoundError): + response = storage_api.get_storage_overview(db=db, current_user=user) + + assert response.data["disk_total_size"] == 0 + assert response.data["disk_free_size"] == 0 + assert response.data["total_size"] == 0 + assert response.data["total_files"] == 0 # ----------------------- GET /storage/stats/type ---------------------- @@ -78,21 +112,100 @@ def test_storage_type_stats_filters_unmapped_file_types(): assert response.data[2] == {"name": "实况图", "size": 750, "count": 3} +# ----------------------- GET /storage/stats/device ---------------------- + + +def test_storage_device_stats_groups_by_model_and_filters_unknowns(): + """Camera-model groups: skip blanks / 'unknown' / '未知' / '未知设备'.""" + user = _user() + db = MagicMock() + db.query.return_value.join.return_value.filter.return_value.group_by.return_value.all.return_value = [ + SimpleNamespace(model="iPhone 15 Pro", size=4000, count=10), + SimpleNamespace(model="Canon EOS R5", size=8000, count=5), + SimpleNamespace(model=None, size=1, count=1), + SimpleNamespace(model="", size=1, count=1), + SimpleNamespace(model="unknown", size=1, count=1), + SimpleNamespace(model="未知", size=1, count=1), + SimpleNamespace(model="未知设备", size=1, count=1), + ] + + response = storage_api.get_storage_device_stats(db=db, current_user=user) + by_name = {r["name"]: r for r in response.data} + assert set(by_name) == {"iPhone 15 Pro", "Canon EOS R5"} + # size desc 排序 + assert response.data[0]["name"] == "Canon EOS R5" + assert response.data[0]["size"] == 8000 + assert response.data[1]["name"] == "iPhone 15 Pro" + + +def test_storage_device_stats_returns_empty_when_no_models(): + user = _user() + db = MagicMock() + db.query.return_value.join.return_value.filter.return_value.group_by.return_value.all.return_value = [] + + response = storage_api.get_storage_device_stats(db=db, current_user=user) + assert response.data == [] + + +# ----------------------- GET /storage/stats/recoverable ---------------------- + + +def test_storage_recoverable_stats_aggregates_each_category(): + """screenshot / video / duplicate / similar 四类聚合.""" + user = _user() + db = MagicMock() + + first_results = iter([SimpleNamespace(size=200, count=3), SimpleNamespace(size=4000, count=2)]) + db.query.return_value.filter.return_value.first.side_effect = lambda: next(first_results) + md5_rows = [SimpleNamespace(md5="abc", count=3, size=900, max_size=400)] + db.query.return_value.filter.return_value.group_by.return_value.having.return_value.all.return_value = md5_rows + + cluster = SimpleNamespace(cluster_id=1) + db.query.return_value.join.return_value.join.return_value.filter.return_value.group_by.return_value.having.return_value.all.return_value = [cluster] + db.query.return_value.join.return_value.filter.return_value.all.return_value = [ + _photo(size=200), _photo(size=150), _photo(size=100), + ] + + response = storage_api.get_storage_recoverable_stats(db=db, current_user=user) + + payload = response.data + assert payload["screenshot"] == {"size": 200, "count": 3} + assert payload["video"] == {"size": 4000, "count": 2} + # duplicate: 900 - 400 = 500 size, count = 3 - 1 = 2 + assert payload["duplicate"] == {"size": 500, "count": 2} + # similar: 3 张照片排序后取最大 200，剩下 150 + 100 = 250 + assert payload["similar"] == {"size": 250, "count": 2} + + +def test_storage_recoverable_stats_handles_empty_db(): + """没有任何照片/无重复时，每类都返回 0.""" + user = _user() + db = MagicMock() + db.query.return_value.filter.return_value.first.side_effect = [SimpleNamespace(size=0, count=0), SimpleNamespace(size=0, count=0)] + db.query.return_value.filter.return_value.group_by.return_value.having.return_value.all.return_value = [] + db.query.return_value.join.return_value.join.return_value.filter.return_value.group_by.return_value.having.return_value.all.return_value = [] + + response = storage_api.get_storage_recoverable_stats(db=db, current_user=user) + assert response.data == { + "similar": {"size": 0, "count": 0}, + "duplicate": {"size": 0, "count": 0}, + "screenshot": {"size": 0, "count": 0}, + "video": {"size": 0, "count": 0}, + } + + # ----------------------- GET /storage/top-large-files ----------------------- def test_storage_top_large_files_returns_serialized_top20(): - """``top-large-files`` truncates to 20 and serialises each row to a dict.""" user = _user() + photos = sorted([_photo(size=1024 * i, filename=f"img_{i}.jpg") for i in range(1, 6)], key=lambda p: p.size, reverse=True) db = MagicMock() - photos = [_photo(size=(i + 1) * 1000) for i in range(25)] - db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = photos[:20] + db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = photos response = storage_api.get_top_large_files(db=db, current_user=user) - - db.query.return_value.filter.return_value.order_by.assert_called_once() - db.query.return_value.filter.return_value.order_by.return_value.limit.assert_called_once_with(20) - assert len(response.data) == 20 - for item in response.data: - assert set(item.keys()) == {"id", "filename", "size", "path", "type"} - assert isinstance(item["size"], int) + assert response.code == 0 + assert len(response.data) == 5 + # 默认 desc 排序：最大在前 + assert response.data[0]["size"] == 1024 * 5 + assert response.data[-1]["size"] == 1024