feat(backup): share <file> — single-file Drive upload + shareable webViewLink (FPLAN-0298)
New 'drone @backup share <file_path> [--public]': uploads a single file to Drive (AIPass Backups/Shared), sets a read permission (default: restricted to the authenticated user; --public: anyone-with-link), returns the webViewLink (webContentLink fallback). Reuses upload_single_file + DriveClient; idempotent via _find_existing_file; fail-loud on every path. 21 new tests, all Drive API mocked (zero live calls). Existing commands untouched. DPLAN-0230 v1. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Q3mZT61WsKVN3srCwVDBiW
This commit is contained in:
AIOSAI
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
a5ede6fbf4
commit
301f3fcb93
@@ -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()
|
||||
|
||||
|
||||
@@ -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}
|
||||
|
||||
|
||||
# =============================================
|
||||
@@ -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 <file_path> [--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)
|
||||
@@ -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
|
||||
|
||||
|
||||
# =============================================
|
||||
Reference in New Issue
Block a user