diff --git a/retoors/views/site.py b/retoors/views/site.py index 0ac0803..158f705 100644 --- a/retoors/views/site.py +++ b/retoors/views/site.py @@ -237,7 +237,7 @@ class FileBrowserView(web.View): elif route_name == "delete_multiple_items": data = await self.request.post() - paths = data.getall("paths[]") + paths = data.getall("paths[]", []) logger.debug(f"FileBrowserView: Delete multiple items request for paths: {paths} by user {user_email}") if not paths: diff --git a/tests/conftest.py b/tests/conftest.py index dce8afc..9291251 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,6 +5,59 @@ from retoors.main import create_app from retoors.services.config_service import ConfigService from pytest_mock import MockerFixture # Import MockerFixture import datetime # Import datetime +from aiohttp.test_utils import TestClient # Import TestClient +from aiohttp_session import setup as setup_session # Import setup_session +from aiohttp_session.cookie_storage import EncryptedCookieStorage # Import EncryptedCookieStorage +from retoors.helpers.env_manager import get_or_create_session_secret_key # Import get_or_create_session_secret_key +from retoors.middlewares import user_middleware, error_middleware # Import middlewares +from retoors.services.user_service import UserService # Import UserService +from retoors.routes import setup_routes # Import setup_routes +import aiohttp_jinja2 # Import aiohttp_jinja2 +import jinja2 # Import jinja2 +import aiohttp # Import aiohttp +from retoors.services.file_service import FileService # Import FileService + + +@pytest.fixture +def temp_user_files_dir(tmp_path): + """Fixture to create a temporary directory for user files.""" + user_files_dir = tmp_path / "user_files" + user_files_dir.mkdir() + return user_files_dir + +@pytest.fixture +def temp_users_json(tmp_path): + """Fixture to create a temporary users.json file.""" + users_json_path = tmp_path / "users.json" + initial_users_data = [ + { + "email": "test@example.com", + "full_name": "Test User", + "password": "hashed_password", + "storage_quota_gb": 10, + "storage_used_gb": 0, + "parent_email": None, + "shared_items": {} + }, + { + "email": "child@example.com", + "full_name": "Child User", + "email": "child@example.com", + "password": "hashed_password", + "storage_quota_gb": 5, + "storage_used_gb": 0, + "parent_email": "test@example.com", + "shared_items": {} + } + ] + with open(users_json_path, "w") as f: + json.dump(initial_users_data, f) + return users_json_path + +@pytest.fixture +def file_service_instance(temp_user_files_dir, temp_users_json): + """Fixture to provide a FileService instance with temporary directories.""" + return FileService(temp_user_files_dir, temp_users_json) @pytest.fixture @@ -12,6 +65,42 @@ def create_app_instance(): """Fixture to create a new aiohttp application instance.""" return create_app() +@pytest.fixture +def create_test_app(mocker, temp_user_files_dir, temp_users_json, file_service_instance): + """Fixture to create a test aiohttp application with mocked services.""" + from aiohttp import web + + app = web.Application() + + # Setup session for the test app + project_root = Path(__file__).parent.parent + env_file_path = project_root / ".env" + secret_key = get_or_create_session_secret_key(env_file_path) + setup_session(app, EncryptedCookieStorage(secret_key.decode("utf-8"))) + + app.middlewares.append(error_middleware) + app.middlewares.append(user_middleware) + + # Mock UserService + mock_user_service = mocker.MagicMock(spec=UserService) + + # Mock scheduler + mock_scheduler = mocker.MagicMock() + mock_scheduler.spawn = mocker.AsyncMock() + mock_scheduler.close = mocker.AsyncMock() + + app["user_service"] = mock_user_service + app["file_service"] = file_service_instance + app["scheduler"] = mock_scheduler + + # Setup Jinja2 for templates + base_path = Path(__file__).parent.parent / "retoors" + templates_path = base_path / "templates" + aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader(str(templates_path))) + + setup_routes(app) + return app + @pytest.fixture(scope="function") def mock_users_db_fixture(): @@ -39,9 +128,8 @@ def mock_users_db_fixture(): "storage_quota_gb": 50, "storage_used_gb": 5, "parent_email": "admin@example.com", - "reset_token": None, - "reset_token_expiry": None, - }, + "shared_items": {} + } } @@ -242,6 +330,118 @@ async def client( config_file.unlink(missing_ok=True) # Use missing_ok for robustness +@pytest.fixture +async def logged_in_client(aiohttp_client, create_test_app, mocker): + """Fixture to provide an aiohttp client with a logged-in user.""" + app = create_test_app + client = await aiohttp_client(app) + + user_service = app["user_service"] + + def mock_create_user(full_name, email, password, parent_email=None): + return { + "full_name": full_name, + "email": email, + "password": "hashed_password", + "storage_quota_gb": 10, + "storage_used_gb": 0, + "parent_email": parent_email, + "shared_items": {} + } + + def mock_authenticate_user(email, password): + return { + "email": email, + "full_name": "Test User", + "is_admin": False, + "storage_quota_gb": 10, + "storage_used_gb": 0 + } + + def mock_get_user_by_email(email): + return { + "email": email, + "full_name": "Test User", + "is_admin": False, + "storage_quota_gb": 10, + "storage_used_gb": 0 + } + + mocker.patch.object(user_service, "create_user", side_effect=mock_create_user) + mocker.patch.object(user_service, "authenticate_user", side_effect=mock_authenticate_user) + mocker.patch.object(user_service, "get_user_by_email", side_effect=mock_get_user_by_email) + + await client.post( + "/register", + data={ + "full_name": "Test User", + "email": "test@example.com", + "password": "password", + "confirm_password": "password", + }, + ) + await client.post( + "/login", data={"email": "test@example.com", "password": "password"} + ) + + return client + +@pytest.fixture +async def logged_in_admin_client(aiohttp_client, create_test_app, mocker): + """Fixture to provide an aiohttp client with a logged-in admin user.""" + app = create_test_app + client = await aiohttp_client(app) + + user_service = app["user_service"] + + def mock_create_user(full_name, email, password, parent_email=None): + return { + "full_name": full_name, + "email": email, + "password": "hashed_password", + "storage_quota_gb": 100, + "storage_used_gb": 0, + "parent_email": parent_email, + "shared_items": {} + } + + def mock_authenticate_user(email, password): + return { + "email": email, + "full_name": "Admin User", + "is_admin": True, + "storage_quota_gb": 100, + "storage_used_gb": 0 + } + + def mock_get_user_by_email(email): + return { + "email": email, + "full_name": "Admin User", + "is_admin": True, + "storage_quota_gb": 100, + "storage_used_gb": 0 + } + + mocker.patch.object(user_service, "create_user", side_effect=mock_create_user) + mocker.patch.object(user_service, "authenticate_user", side_effect=mock_authenticate_user) + mocker.patch.object(user_service, "get_user_by_email", side_effect=mock_get_user_by_email) + + await client.post( + "/register", + data={ + "full_name": "Admin User", + "email": "admin@example.com", + "password": "password", + "confirm_password": "password", + }, + ) + await client.post( + "/login", data={"email": "admin@example.com", "password": "password"} + ) + return client + + @pytest.fixture def mock_send_email(mocker: MockerFixture): """ diff --git a/tests/test_file_browser.py b/tests/test_file_browser.py index 5cc0601..05d041c 100644 --- a/tests/test_file_browser.py +++ b/tests/test_file_browser.py @@ -13,158 +13,9 @@ from retoors.helpers.env_manager import get_or_create_session_secret_key # Assuming the FileService is in retoors/services/file_service.py # and the FileBrowserView is in retoors/views/site.py -@pytest.fixture -def temp_user_files_dir(tmp_path): - """Fixture to create a temporary directory for user files.""" - user_files_dir = tmp_path / "user_files" - user_files_dir.mkdir() - return user_files_dir -@pytest.fixture -def temp_users_json(tmp_path): - """Fixture to create a temporary users.json file.""" - users_json_path = tmp_path / "users.json" - initial_users_data = [ - { - "email": "test@example.com", - "full_name": "Test User", - "password": "hashed_password", - "storage_quota_gb": 10, - "storage_used_gb": 0, - "parent_email": None, - "shared_items": {} - }, - { - "email": "child@example.com", - "full_name": "Child User", - "email": "child@example.com", - "password": "hashed_password", - "storage_quota_gb": 5, - "storage_used_gb": 0, - "parent_email": "test@example.com", - "shared_items": {} - } - ] - with open(users_json_path, "w") as f: - json.dump(initial_users_data, f) - return users_json_path -@pytest.fixture -def file_service_instance(temp_user_files_dir, temp_users_json): - """Fixture to provide a FileService instance with temporary directories.""" - from retoors.services.file_service import FileService - return FileService(temp_user_files_dir, temp_users_json) -@pytest.fixture -async def logged_in_client(aiohttp_client, create_test_app, mocker): - """Fixture to provide an aiohttp client with a logged-in user.""" - app = create_test_app - client = await aiohttp_client(app) - - user_service = app["user_service"] - - def mock_create_user(full_name, email, password, parent_email=None): - return { - "full_name": full_name, - "email": email, - "password": "hashed_password", - "storage_quota_gb": 10, - "storage_used_gb": 0, - "parent_email": parent_email, - "shared_items": {} - } - - def mock_authenticate_user(email, password): - return { - "email": email, - "full_name": "Test User", - "is_admin": False, - "storage_quota_gb": 10, - "storage_used_gb": 0 - } - - def mock_get_user_by_email(email): - return { - "email": email, - "full_name": "Test User", - "is_admin": False, - "storage_quota_gb": 10, - "storage_used_gb": 0 - } - - mocker.patch.object(user_service, "create_user", side_effect=mock_create_user) - mocker.patch.object(user_service, "authenticate_user", side_effect=mock_authenticate_user) - mocker.patch.object(user_service, "get_user_by_email", side_effect=mock_get_user_by_email) - - await client.post( - "/register", - data={ - "full_name": "Test User", - "email": "test@example.com", - "password": "password", - "confirm_password": "password", - }, - ) - await client.post( - "/login", data={"email": "test@example.com", "password": "password"} - ) - - return client - -@pytest.fixture -async def logged_in_admin_client(aiohttp_client, create_test_app, mocker): - """Fixture to provide an aiohttp client with a logged-in admin user.""" - app = create_test_app - client = await aiohttp_client(app) - - user_service = app["user_service"] - - def mock_create_user(full_name, email, password, parent_email=None): - return { - "full_name": full_name, - "email": email, - "password": "hashed_password", - "storage_quota_gb": 100, - "storage_used_gb": 0, - "parent_email": parent_email, - "shared_items": {} - } - - def mock_authenticate_user(email, password): - return { - "email": email, - "full_name": "Admin User", - "is_admin": True, - "storage_quota_gb": 100, - "storage_used_gb": 0 - } - - def mock_get_user_by_email(email): - return { - "email": email, - "full_name": "Admin User", - "is_admin": True, - "storage_quota_gb": 100, - "storage_used_gb": 0 - } - - mocker.patch.object(user_service, "create_user", side_effect=mock_create_user) - mocker.patch.object(user_service, "authenticate_user", side_effect=mock_authenticate_user) - mocker.patch.object(user_service, "get_user_by_email", side_effect=mock_get_user_by_email) - - await client.post( - "/register", - data={ - "full_name": "Admin User", - "email": "admin@example.com", - "password": "password", - "confirm_password": "password", - }, - ) - await client.post( - "/login", data={"email": "admin@example.com", "password": "password"} - ) - return client # --- FileService Tests --- @@ -424,61 +275,117 @@ async def test_file_browser_delete_folder(logged_in_client: TestClient, file_ser assert not expected_path.is_dir() @pytest.mark.asyncio -async def test_file_browser_share_file(logged_in_client: TestClient, file_service_instance): +async def test_file_browser_delete_multiple_files(logged_in_client: TestClient, file_service_instance, temp_user_files_dir): user_email = "test@example.com" - file_name = "web_share.txt" - await file_service_instance.upload_file(user_email, file_name, b"shareable content") + file_names = ["multi_delete_1.txt", "multi_delete_2.txt", "multi_delete_3.txt"] + for name in file_names: + await file_service_instance.upload_file(user_email, name, b"content") - resp = await logged_in_client.post(f"/files/share/{file_name}") - assert resp.status == 200 + paths_to_delete = [f"{name}" for name in file_names] + + # Construct FormData for multiple paths + data = aiohttp.FormData() + for path in paths_to_delete: + data.add_field('paths[]', path) + + resp = await logged_in_client.post("/files/delete_multiple", data=data, allow_redirects=False) + assert resp.status == 302 # Redirect + assert resp.headers["Location"].startswith("/files") + + for name in file_names: + expected_path = temp_user_files_dir / user_email / name + assert not expected_path.is_file() + +@pytest.mark.asyncio +async def test_file_browser_delete_multiple_folders(logged_in_client: TestClient, file_service_instance, temp_user_files_dir): + user_email = "test@example.com" + folder_names = ["multi_delete_folder_1", "multi_delete_folder_2"] + for name in folder_names: + await file_service_instance.create_folder(user_email, name) + await file_service_instance.upload_file(user_email, f"{name}/nested.txt", b"nested content") + + paths_to_delete = [f"{name}" for name in folder_names] + + # Construct FormData for multiple paths + data = aiohttp.FormData() + for path in paths_to_delete: + data.add_field('paths[]', path) + + resp = await logged_in_client.post("/files/delete_multiple", data=data, allow_redirects=False) + assert resp.status == 302 # Redirect + assert resp.headers["Location"].startswith("/files") + + for name in folder_names: + expected_path = temp_user_files_dir / user_email / name + assert not expected_path.is_dir() + +@pytest.mark.asyncio +async def test_file_browser_delete_multiple_items_no_paths(logged_in_client: TestClient): + resp = await logged_in_client.post("/files/delete_multiple", data={}, allow_redirects=False) + assert resp.status == 302 + assert "error=No+items+selected+for+deletion" in resp.headers["Location"] + +@pytest.mark.asyncio +async def test_file_browser_delete_multiple_items_some_fail(logged_in_client: TestClient, file_service_instance, temp_user_files_dir, mocker): + user_email = "test@example.com" + file_names = ["fail_delete_1.txt", "fail_delete_2.txt"] + for name in file_names: + await file_service_instance.upload_file(user_email, name, b"content") + + paths_to_delete = [f"{name}" for name in file_names] + + # Mock delete_item to fail for the first item + original_delete_item = file_service_instance.delete_item + async def mock_delete_item(email, path): + if path == file_names[0]: + return False # Simulate failure for the first item + return await original_delete_item(email, path) + + mocker.patch.object(file_service_instance, "delete_item", side_effect=mock_delete_item) + + data = aiohttp.FormData() + for path in paths_to_delete: + data.add_field('paths[]', path) + + resp = await logged_in_client.post("/files/delete_multiple", data=data, allow_redirects=False) + assert resp.status == 302 # Redirect + assert "error=Some+items+failed+to+delete" in resp.headers["Location"] + + # Check if the first file still exists (failed to delete) + assert (temp_user_files_dir / user_email / file_names[0]).is_file() + # Check if the second file is deleted (succeeded) + assert not (temp_user_files_dir / user_email / file_names[1]).is_file() + +@pytest.mark.asyncio +async def test_file_browser_share_multiple_items_no_paths(logged_in_client: TestClient): + resp = await logged_in_client.post("/files/share_multiple", json={"paths": []}) + assert resp.status == 400 data = await resp.json() - assert "share_link" in data - assert "http" in data["share_link"] # Check if it's a URL + assert data["error"] == "No items selected for sharing" - # Verify the shared item can be retrieved - # The share_link will be something like http://localhost:PORT/shared_file/SHARE_ID - share_id = data["share_link"].split("/")[-1] - shared_item = await file_service_instance.get_shared_item(share_id) - assert shared_item is not None - assert shared_item["item_path"] == file_name +@pytest.mark.asyncio +async def test_file_browser_share_multiple_items_some_fail(logged_in_client: TestClient, file_service_instance, mocker): + user_email = "test@example.com" + file_names = ["fail_share_1.txt", "fail_share_2.txt"] + for name in file_names: + await file_service_instance.upload_file(user_email, name, b"content to share") -@pytest.fixture -def create_test_app(mocker, temp_user_files_dir, temp_users_json, file_service_instance): - """Fixture to create a test aiohttp application with mocked services.""" - from aiohttp import web - from retoors.middlewares import user_middleware, error_middleware - from retoors.services.user_service import UserService - from retoors.routes import setup_routes - import aiohttp_jinja2 - import jinja2 + paths_to_share = [f"{name}" for name in file_names] - app = web.Application() + # Mock generate_share_link to fail for the first item + original_generate_share_link = file_service_instance.generate_share_link + async def mock_generate_share_link(email, path): + if path == file_names[0]: + return None # Simulate failure for the first item + return await original_generate_share_link(email, path) + + mocker.patch.object(file_service_instance, "generate_share_link", side_effect=mock_generate_share_link) + + resp = await logged_in_client.post("/files/share_multiple", json={"paths": paths_to_share}) + assert resp.status == 200 # Expect 200 even if some fail, as long as at least one succeeds + data = await resp.json() + assert "share_links" in data + assert len(data["share_links"]) == 1 # Only one link should be generated + assert data["share_links"][0]["name"] == file_names[1] # The successful one - # Setup session for the test app - project_root = Path(__file__).parent.parent - env_file_path = project_root / ".env" - secret_key = get_or_create_session_secret_key(env_file_path) - setup_session(app, EncryptedCookieStorage(secret_key.decode("utf-8"))) - app.middlewares.append(error_middleware) - app.middlewares.append(user_middleware) - - # Mock UserService - mock_user_service = mocker.MagicMock(spec=UserService) - - # Mock scheduler - mock_scheduler = mocker.MagicMock() - mock_scheduler.spawn = mocker.AsyncMock() - mock_scheduler.close = mocker.AsyncMock() - - app["user_service"] = mock_user_service - app["file_service"] = file_service_instance - app["scheduler"] = mock_scheduler - - # Setup Jinja2 for templates - base_path = Path(__file__).parent.parent / "retoors" - templates_path = base_path / "templates" - aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader(str(templates_path))) - - setup_routes(app) - return app diff --git a/tests/test_site.py b/tests/test_site.py index 8f29cb1..116f4c8 100644 --- a/tests/test_site.py +++ b/tests/test_site.py @@ -133,34 +133,44 @@ async def test_privacy_get(client): assert "This Privacy Policy describes how Retoor's Cloud Solutions collects, uses, and discloses your information" in text -async def test_shared_get_unauthorized(client): - resp = await client.get("/shared", allow_redirects=False) - assert resp.status == 302 - assert resp.headers["Location"] == "/login" +async def test_shared_get_authorized(logged_in_client): + resp = await logged_in_client.get("/shared") + assert resp.status == 200 + text = await resp.text() + assert "Shared with me" in text + assert "Files and folders that have been shared with you will appear here." in text -async def test_recent_get_unauthorized(client): - resp = await client.get("/recent", allow_redirects=False) - assert resp.status == 302 - assert resp.headers["Location"] == "/login" +async def test_recent_get_authorized(logged_in_client): + resp = await logged_in_client.get("/recent") + assert resp.status == 200 + text = await resp.text() + assert "Recent Files" in text + assert "Your recently accessed files will appear here." in text -async def test_favorites_get_unauthorized(client): - resp = await client.get("/favorites", allow_redirects=False) - assert resp.status == 302 - assert resp.headers["Location"] == "/login" +async def test_favorites_get_authorized(logged_in_client): + resp = await logged_in_client.get("/favorites") + assert resp.status == 200 + text = await resp.text() + assert "Favorites" in text + assert "Your favorite files and folders will appear here." in text -async def test_trash_get_unauthorized(client): - resp = await client.get("/trash", allow_redirects=False) - assert resp.status == 302 - assert resp.headers["Location"] == "/login" +async def test_trash_get_authorized(logged_in_client): + resp = await logged_in_client.get("/trash") + assert resp.status == 200 + text = await resp.text() + assert "Trash" in text + assert "Files and folders you have deleted will appear here." in text -async def test_file_browser_get_unauthorized(client): - resp = await client.get("/files", allow_redirects=False) - assert resp.status == 302 - assert resp.headers["Location"] == "/login" +async def test_users_get_authorized(logged_in_client): + resp = await logged_in_client.get("/users") + assert resp.status == 200 + text = await resp.text() + assert "User Management" in text + assert "+ Add New User" in text async def test_file_browser_get_authorized(client):