diff --git a/src/aipass/backup/apps/backup.py b/src/aipass/backup/apps/backup.py index 8dc2fe5f..958afe25 100644 --- a/src/aipass/backup/apps/backup.py +++ b/src/aipass/backup/apps/backup.py @@ -73,6 +73,7 @@ def print_help() -> None: console.print(" [green]drive_sync[/green] Sync backups to the remote drive") console.print(" [green]drive_check[/green] Test the remote drive connection") console.print(" [green]drive_stats[/green] Drive usage statistics") + console.print(" [green]share[/green] Upload a single file to Drive + get a shareable link") console.print(" [green]drive_clear[/green] Clear backups from the remote drive") console.print() diff --git a/src/aipass/backup/apps/handlers/drive/share.py b/src/aipass/backup/apps/handlers/drive/share.py new file mode 100644 index 00000000..b1886906 --- /dev/null +++ b/src/aipass/backup/apps/handlers/drive/share.py @@ -0,0 +1,200 @@ +# =================== AIPass ==================== +# Name: share.py +# Description: Drive sharing — upload + permission + shareable link retrieval +# Version: 1.0.0 +# Created: 2026-07-01 +# Modified: 2026-07-01 +# ============================================= + +"""Drive file sharing. + +Uploads a single file to Drive (under ``AIPass Backups/Shared``), +sets a read permission, and retrieves a shareable webViewLink. +Idempotent: reuses an existing file if one is found by name. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from aipass.prax import logger + +from ..json import json_handler +from . import upload as upload_mod + +if TYPE_CHECKING: + from .client import DriveClient + +SHARE_PROJECT_NAME = "Shared" + + +def _get_authenticated_email(client: DriveClient) -> str | None: + """Derive the authenticated user's email from Drive ``about`` API.""" + try: + request = client.drive_service.about().get(fields="user") # type: ignore[union-attr] + result = client._api_call(request) + if result and result.get("user"): + return result["user"].get("emailAddress") + except Exception as exc: + logger.info(f"Failed to get authenticated email: {exc}") + return None + + +def upload_for_share( + client: DriveClient, + local_file: Path, + note: str = "", +) -> str | None: + """Upload a file for sharing. Returns the Drive file ID or ``None``. + + Idempotent: checks the ``Shared`` project folder first and reuses an + existing file (matched by name) instead of re-uploading. Falls back + to :func:`upload_single_file` for the actual upload. + """ + folder_id = client.get_or_create_project_folder(SHARE_PROJECT_NAME) + if not folder_id: + return None + + existing = client._find_existing_file(local_file.name, folder_id) + if existing: + logger.info(f"File already on Drive: {existing['id']}") + return existing["id"] + + client.file_tracker = {} + success = upload_mod.upload_single_file( + client, + local_file, + SHARE_PROJECT_NAME, + local_file.parent, + note=note, + ) + if not success: + return None + + entry = client.file_tracker.get(local_file.name, {}) + return entry.get("drive_id") + + +def set_share_permission( + client: DriveClient, + file_id: str, + *, + public: bool = False, +) -> str | None: + """Set a read permission on *file_id*. + + *public* ``False`` (default): restricted to the authenticated user + (``type=user``, ``role=reader``). ``True``: anyone with the link + (``type=anyone``, ``role=reader``). + + Returns the permission ID, or ``None`` on failure. + """ + if public: + body: dict[str, Any] = {"type": "anyone", "role": "reader"} + else: + email = _get_authenticated_email(client) + if not email: + client.last_error = "Could not determine authenticated email" + return None + body = {"type": "user", "role": "reader", "emailAddress": email} + + try: + request = client.drive_service.permissions().create( # type: ignore[union-attr] + fileId=file_id, + body=body, + fields="id", + ) + result = client._api_call(request) + if result: + return result.get("id") + except Exception as exc: + client.last_error = str(exc) + logger.warning(f"Failed to set permission on {file_id}: {exc}") + + return None + + +def get_share_link(client: DriveClient, file_id: str) -> str | None: + """Retrieve the shareable link for *file_id*. + + Prefers ``webViewLink``; falls back to ``webContentLink``. + """ + try: + request = client.drive_service.files().get( # type: ignore[union-attr] + fileId=file_id, + fields="webViewLink,webContentLink", + ) + result = client._api_call(request) + if result: + return result.get("webViewLink") or result.get("webContentLink") + except Exception as exc: + client.last_error = str(exc) + logger.warning(f"Failed to get share link for {file_id}: {exc}") + + return None + + +def share_file( + client: DriveClient, + local_file: str | Path, + *, + public: bool = False, + note: str = "", +) -> dict[str, Any]: + """Upload, share, and return a link — the complete pipeline. + + Returns a dict with ``success``, ``link``, ``file_id``, and + ``error`` keys. + """ + local_file = Path(local_file).resolve() + + if not local_file.is_file(): + return { + "success": False, + "link": None, + "file_id": None, + "error": f"Not a file: {local_file}", + } + + file_id = upload_for_share(client, local_file, note=note) + if not file_id: + return { + "success": False, + "link": None, + "file_id": None, + "error": f"Upload failed: {client.last_error or 'unknown'}", + } + + perm_id = set_share_permission(client, file_id, public=public) + if not perm_id: + return { + "success": False, + "link": None, + "file_id": file_id, + "error": f"Permission failed: {client.last_error or 'unknown'}", + } + + link = get_share_link(client, file_id) + if not link: + return { + "success": False, + "link": None, + "file_id": file_id, + "error": f"Link retrieval failed: {client.last_error or 'unknown'}", + } + + json_handler.log_operation( + "share_file", + { + "file": str(local_file), + "file_id": file_id, + "public": public, + "link": link, + }, + ) + + return {"success": True, "link": link, "file_id": file_id, "error": None} + + +# ============================================= diff --git a/src/aipass/backup/apps/modules/share.py b/src/aipass/backup/apps/modules/share.py new file mode 100644 index 00000000..9ce15660 --- /dev/null +++ b/src/aipass/backup/apps/modules/share.py @@ -0,0 +1,124 @@ +# =================== AIPass ==================== +# Name: share.py +# Description: Share module — single-file Drive upload with shareable link +# Version: 1.0.0 +# Created: 2026-07-01 +# Modified: 2026-07-01 +# ============================================= + +"""Share Module — upload a single file to Drive and return a shareable link.""" + +import sys + +from aipass.prax import logger +from aipass.cli.apps.modules import console + +from aipass.backup.apps.handlers.json import json_handler + +MODULE_NAME = "share" +PRIMARY_COMMAND = "share" + + +def print_introspection(): + """Display module info and connected handlers.""" + console.print(f"[bold cyan]{MODULE_NAME} Module[/bold cyan]") + console.print(f" Primary command: [yellow]{PRIMARY_COMMAND}[/yellow]") + console.print(" Status: Live — single-file Drive upload + share link") + console.print(" Handlers: drive/client, drive/upload, drive/share") + + +def print_help(): + """Display help for this module.""" + print_introspection() + console.print() + console.print("[bold cyan]USAGE:[/bold cyan]") + console.print() + console.print(" [dim]drone @backup share [--public][/dim]") + console.print() + console.print("[bold cyan]OPTIONS:[/bold cyan]") + console.print() + console.print(" [green]--public[/green] Share with anyone (default: restricted to owner)") + console.print() + console.print("[bold cyan]OUTPUT:[/bold cyan]") + console.print() + console.print(" Final stdout line is the shareable Google Drive link.") + console.print() + + +def run_share(file_path: str, *, public: bool = False) -> dict: + """Upload a single file to Drive and return a shareable link. + + Authenticates via @api, uploads (or reuses existing), sets a + share permission, and prints the webViewLink as the last line. + + Args: + file_path: Path to the file to share. + public: If True, anyone-with-link; else restricted to owner. + + Returns: + Result dict with success, link, file_id, error. + """ + from aipass.backup.apps.handlers.drive.client import DriveClient + from aipass.backup.apps.handlers.drive.share import share_file + + client = DriveClient() + if not client.authenticate(): + error = f"Drive authentication failed: {client.last_error}" + console.print(f"[red]{error}[/red]") + logger.warning(f"[backup] {error}") + return {"success": False, "link": None, "file_id": None, "error": error} + + console.print(f"[dim]Uploading {file_path}...[/dim]") + result = share_file(client, file_path, public=public) + + if result["success"]: + mode = "public" if public else "restricted" + console.print(f"[green]Shared ({mode}):[/green] {result['link']}") + logger.info(f"[backup] Shared {file_path} ({mode})") + else: + console.print(f"[red]Share failed:[/red] {result['error']}") + logger.warning(f"[backup] Share failed: {result['error']}") + + if result.get("link"): + console.print(result["link"], highlight=False) + + json_handler.log_operation( + "share_command", + { + "file": file_path, + "public": public, + "success": result["success"], + }, + ) + + return result + + +def handle_command(command: str, args: list) -> bool: + """Handle the share command. Returns True if handled.""" + if command != PRIMARY_COMMAND: + return False + + if not args: + print_introspection() + return True + + if args[0] in ("--help", "-h", "help"): + print_help() + return True + + file_path = args[0] + public = "--public" in args + + run_share(file_path, public=public) + return True + + +# ============================================= + +if __name__ == "__main__": + if len(sys.argv) == 1: + print_introspection() + sys.exit(0) + handle_command(PRIMARY_COMMAND, sys.argv[1:]) + sys.exit(0) diff --git a/src/aipass/backup/tests/test_share.py b/src/aipass/backup/tests/test_share.py new file mode 100644 index 00000000..4cf6e46a --- /dev/null +++ b/src/aipass/backup/tests/test_share.py @@ -0,0 +1,408 @@ +# =================== AIPass ==================== +# Name: test_share.py +# Description: Tests for share module + handler (mocked Drive API) +# Version: 1.0.0 +# Created: 2026-07-01 +# Modified: 2026-07-01 +# ============================================= + +"""Tests for share — module routing, handler logic, permission paths.""" + +import importlib +import sys +import types +from unittest.mock import MagicMock, patch + + +def _build_mocks(): + """Build the standard mock dict for importing share modules.""" + mocks: dict[str, object] = {} + + prax = types.ModuleType("aipass.prax") + setattr(prax, "logger", MagicMock()) + mocks["aipass.prax"] = prax + + cli = types.ModuleType("aipass.cli") + cli_apps = types.ModuleType("aipass.cli.apps") + cli_modules = types.ModuleType("aipass.cli.apps.modules") + setattr(cli_modules, "console", MagicMock()) + setattr(cli_modules, "header", MagicMock()) + setattr(cli_modules, "success", MagicMock()) + setattr(cli_modules, "warning", MagicMock()) + setattr(cli_modules, "error", MagicMock()) + mocks["aipass.cli"] = cli + mocks["aipass.cli.apps"] = cli_apps + mocks["aipass.cli.apps.modules"] = cli_modules + + json_pkg = types.ModuleType("aipass.backup.apps.handlers.json") + json_handler = types.ModuleType( + "aipass.backup.apps.handlers.json.json_handler", + ) + setattr(json_handler, "log_operation", MagicMock()) + setattr(json_handler, "load_json", MagicMock(return_value={})) + setattr(json_handler, "save_json", MagicMock()) + mocks["aipass.backup.apps.handlers.json"] = json_pkg + mocks["aipass.backup.apps.handlers.json.json_handler"] = json_handler + + google_client = types.ModuleType("aipass.api.apps.modules.google_client") + setattr(google_client, "get_drive_service", MagicMock()) + setattr(google_client, "api_call_with_retry", MagicMock()) + mocks["aipass.api.apps.modules.google_client"] = google_client + mocks["aipass.api"] = types.ModuleType("aipass.api") + mocks["aipass.api.apps"] = types.ModuleType("aipass.api.apps") + mocks["aipass.api.apps.modules"] = types.ModuleType("aipass.api.apps.modules") + + ghttp = types.ModuleType("googleapiclient.http") + setattr(ghttp, "MediaFileUpload", MagicMock()) + mocks["googleapiclient"] = types.ModuleType("googleapiclient") + mocks["googleapiclient.http"] = ghttp + + return mocks + + +def _fresh_import(module_path: str, mocks: dict): + """Import a module with mocked dependencies, clearing stale entries.""" + stale = [k for k in sys.modules if k.startswith(module_path.rsplit(".", 1)[0])] + with patch.dict(sys.modules, mocks): + for k in stale: + sys.modules.pop(k, None) + return importlib.import_module(module_path) + + +# ── Module routing tests ───────────────────────────────────────────── + + +class TestShareModuleRouting: + """Verify handle_command routing for the share module.""" + + def test_handle_command_returns_true_for_primary(self) -> None: + """handle_command returns True for the primary command.""" + mocks = _build_mocks() + mod = _fresh_import("aipass.backup.apps.modules.share", mocks) + assert mod.handle_command(mod.PRIMARY_COMMAND, []) is True + + def test_handle_command_returns_false_for_unknown(self) -> None: + """handle_command returns False for unknown commands.""" + mocks = _build_mocks() + mod = _fresh_import("aipass.backup.apps.modules.share", mocks) + assert mod.handle_command("nonexistent", []) is False + + def test_handle_command_help(self) -> None: + """--help flag returns True early.""" + mocks = _build_mocks() + mod = _fresh_import("aipass.backup.apps.modules.share", mocks) + assert mod.handle_command(mod.PRIMARY_COMMAND, ["--help"]) is True + + def test_handle_command_help_short(self) -> None: + """-h flag returns True early.""" + mocks = _build_mocks() + mod = _fresh_import("aipass.backup.apps.modules.share", mocks) + assert mod.handle_command(mod.PRIMARY_COMMAND, ["-h"]) is True + + def test_handle_command_help_word(self) -> None: + """help word returns True early.""" + mocks = _build_mocks() + mod = _fresh_import("aipass.backup.apps.modules.share", mocks) + assert mod.handle_command(mod.PRIMARY_COMMAND, ["help"]) is True + + def test_module_constants(self) -> None: + """MODULE_NAME and PRIMARY_COMMAND are correct.""" + mocks = _build_mocks() + mod = _fresh_import("aipass.backup.apps.modules.share", mocks) + assert mod.MODULE_NAME == "share" + assert mod.PRIMARY_COMMAND == "share" + + +# ── Handler tests ──────────────────────────────────────────────────── + + +def _make_mock_client(): + """Build a mock DriveClient with chainable Drive API methods.""" + client = MagicMock() + client.last_error = None + client.file_tracker = {} + client.backup_folder_id = None + client.project_folder_cache = {} + + client.get_or_create_project_folder.return_value = "folder-shared-123" + client._find_existing_file.return_value = None + + service = MagicMock() + client.drive_service = service + + service.permissions.return_value.create.return_value = MagicMock() + service.files.return_value.get.return_value = MagicMock() + service.files.return_value.create.return_value = MagicMock() + service.files.return_value.update.return_value = MagicMock() + service.files.return_value.list.return_value = MagicMock() + service.about.return_value.get.return_value = MagicMock() + + return client + + +class TestShareHandler: + """Test the share handler functions with mocked Drive API.""" + + def test_share_file_success_public(self, tmp_path) -> None: + """Public share uploads, sets permission, returns webViewLink.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + test_file = tmp_path / "report.pdf" + test_file.write_bytes(b"PDF content") + + client._api_call.side_effect = [ + {"id": "file-abc-123"}, + {"id": "perm-xyz-789"}, + {"webViewLink": "https://drive.google.com/file/d/file-abc-123/view"}, + ] + + result = handler.share_file(client, str(test_file), public=True) + + assert result["success"] is True + assert result["link"] == "https://drive.google.com/file/d/file-abc-123/view" + assert result["file_id"] is not None + assert result["error"] is None + + def test_share_file_success_restricted(self, tmp_path) -> None: + """Restricted share uses authenticated email for permission.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + test_file = tmp_path / "data.csv" + test_file.write_text("a,b,c") + + client._api_call.side_effect = [ + {"id": "file-def-456"}, + {"user": {"emailAddress": "test@gmail.com"}}, + {"id": "perm-abc-123"}, + {"webViewLink": "https://drive.google.com/file/d/file-def-456/view"}, + ] + + result = handler.share_file(client, str(test_file), public=False) + + assert result["success"] is True + assert result["link"] is not None + assert result["error"] is None + + def test_share_file_not_a_file(self, tmp_path) -> None: + """Nonexistent path returns error.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + result = handler.share_file(client, str(tmp_path / "nonexistent.txt")) + + assert result["success"] is False + assert "Not a file" in result["error"] + + def test_share_file_directory_rejected(self, tmp_path) -> None: + """Directory path returns error.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + result = handler.share_file(client, str(tmp_path)) + + assert result["success"] is False + assert "Not a file" in result["error"] + + def test_share_file_upload_failure(self, tmp_path) -> None: + """Upload failure surfaces error.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client.get_or_create_project_folder.return_value = None + client.last_error = "Folder creation failed" + + test_file = tmp_path / "fail.txt" + test_file.write_text("content") + + result = handler.share_file(client, str(test_file)) + + assert result["success"] is False + assert "Upload failed" in result["error"] + + def test_share_file_permission_failure(self, tmp_path) -> None: + """Permission failure after upload surfaces error.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._find_existing_file.return_value = {"id": "existing-file-id"} + + client._api_call.side_effect = [ + {"user": {"emailAddress": "test@gmail.com"}}, + None, + ] + client.last_error = "Permission denied" + + test_file = tmp_path / "secret.txt" + test_file.write_text("restricted") + + result = handler.share_file(client, str(test_file)) + + assert result["success"] is False + assert "Permission failed" in result["error"] + + def test_share_file_link_retrieval_failure(self, tmp_path) -> None: + """Link retrieval failure after permission surfaces error.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._find_existing_file.return_value = {"id": "file-id"} + + client._api_call.side_effect = [ + {"id": "perm-id"}, + None, + ] + client.last_error = "API error" + + test_file = tmp_path / "doc.txt" + test_file.write_text("document") + + result = handler.share_file(client, str(test_file), public=True) + + assert result["success"] is False + assert "Link retrieval failed" in result["error"] + + def test_idempotent_reuses_existing(self, tmp_path) -> None: + """Existing file on Drive is reused, not re-uploaded.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._find_existing_file.return_value = {"id": "already-on-drive"} + + client._api_call.side_effect = [ + {"id": "perm-id"}, + {"webViewLink": "https://drive.google.com/file/d/already-on-drive/view"}, + ] + + test_file = tmp_path / "existing.txt" + test_file.write_text("already uploaded") + + result = handler.share_file(client, str(test_file), public=True) + + assert result["success"] is True + assert result["file_id"] == "already-on-drive" + client.get_or_create_project_folder.assert_called_once_with("Shared") + + def test_webcontentlink_fallback(self, tmp_path) -> None: + """Falls back to webContentLink when webViewLink is absent.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._find_existing_file.return_value = {"id": "file-id"} + + client._api_call.side_effect = [ + {"id": "perm-id"}, + {"webContentLink": "https://drive.google.com/uc?id=file-id"}, + ] + + test_file = tmp_path / "download.bin" + test_file.write_bytes(b"\x00\x01") + + result = handler.share_file(client, str(test_file), public=True) + + assert result["success"] is True + assert "uc?id=file-id" in result["link"] + + +class TestSetSharePermission: + """Test permission-setting specifically.""" + + def test_public_permission_body(self, tmp_path) -> None: + """Public permission uses type=anyone, role=reader.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._api_call.return_value = {"id": "perm-id"} + + handler.set_share_permission(client, "file-123", public=True) + + call_args = client.drive_service.permissions().create.call_args + assert call_args.kwargs["body"]["type"] == "anyone" + assert call_args.kwargs["body"]["role"] == "reader" + + def test_restricted_permission_body(self, tmp_path) -> None: + """Restricted permission uses type=user with authenticated email.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._api_call.side_effect = [ + {"user": {"emailAddress": "user@example.com"}}, + {"id": "perm-id"}, + ] + + handler.set_share_permission(client, "file-123", public=False) + + call_args = client.drive_service.permissions().create.call_args + assert call_args.kwargs["body"]["type"] == "user" + assert call_args.kwargs["body"]["emailAddress"] == "user@example.com" + + def test_restricted_fails_without_email(self) -> None: + """Restricted permission fails when email lookup returns None.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._api_call.return_value = None + + result = handler.set_share_permission(client, "file-123", public=False) + + assert result is None + assert "email" in client.last_error.lower() + + +class TestGetShareLink: + """Test link retrieval.""" + + def test_prefers_webviewlink(self) -> None: + """webViewLink is preferred over webContentLink.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._api_call.return_value = { + "webViewLink": "https://view-link", + "webContentLink": "https://content-link", + } + + link = handler.get_share_link(client, "file-id") + assert link == "https://view-link" + + def test_falls_back_to_webcontentlink(self) -> None: + """Falls back to webContentLink when webViewLink is None.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._api_call.return_value = { + "webViewLink": None, + "webContentLink": "https://content-link", + } + + link = handler.get_share_link(client, "file-id") + assert link == "https://content-link" + + def test_returns_none_on_api_failure(self) -> None: + """Returns None when API call fails.""" + mocks = _build_mocks() + handler = _fresh_import("aipass.backup.apps.handlers.drive.share", mocks) + + client = _make_mock_client() + client._api_call.return_value = None + + link = handler.get_share_link(client, "file-id") + assert link is None + + +# =============================================