f103d5f44f
Build the proton-drive Hermes skill following the Phase 4 spec
from ARCHITECTURE.md (§5). Primary path: rclone protondrive backend
with Drive SDK as a fallback option.
Skill components:
- skills/proton-drive/SKILL.md — YAML frontmatter + full docs for
all 9 tools (list, read, download, upload, search, mkdir,
delete, stat, sync) with usage, error handling, security notes
- skills/proton-drive/__init__.py — package init with exports
- skills/proton-drive/tools.py — Python subprocess wrappers for
each tool, plus rclone availability/remote checks
- tests/test_drive.py — 25 unit tests (all pass) with mocked
subprocess.run
All 9 Proton Drive tools implemented:
proton_drive_list, proton_drive_read, proton_drive_download,
proton_drive_upload, proton_drive_search, proton_drive_mkdir,
proton_drive_delete, proton_drive_stat, proton_drive_sync
Signed-off-by: Bee <bee@trentuna.com>
277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""Tests for the Proton Drive skill — all 9 rclone-backed tool handlers.
|
|
|
|
Uses unittest with monkeypatched subprocess.run to avoid needing rclone.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
from types import SimpleNamespace
|
|
|
|
# Load the tools module directly (hyphenated dir can't be imported as a package)
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import sys
|
|
TOOLS_PATH = os.path.join(
|
|
os.path.dirname(__file__), "..", "skills", "proton-drive", "tools.py"
|
|
)
|
|
loader = importlib.machinery.SourceFileLoader("proton_drive_tools", os.path.abspath(TOOLS_PATH))
|
|
spec = importlib.util.spec_from_loader("proton_drive_tools", loader, origin=os.path.abspath(TOOLS_PATH))
|
|
DRIVE = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(DRIVE)
|
|
|
|
# Register in sys.modules so unittest.mock.patch can resolve dotted paths
|
|
sys.modules["proton_drive_tools"] = DRIVE
|
|
|
|
# The mock target prefix matches the module name used during load
|
|
MOD = "proton_drive_tools"
|
|
|
|
|
|
def _fake_run(returncode=0, stdout="", stderr=""):
|
|
"""Create a mock subprocess.run replacement."""
|
|
def fake_run(*args, **kwargs):
|
|
return SimpleNamespace(
|
|
returncode=returncode,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
args=args,
|
|
)
|
|
return fake_run
|
|
|
|
|
|
class TestProtonDriveList(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_list_root(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(
|
|
returncode=0,
|
|
stdout=json.dumps({"Name": "Documents", "IsDir": True}) + "\n"
|
|
+ json.dumps({"Name": "notes.txt", "Size": 1024, "IsDir": False}),
|
|
stderr="",
|
|
)
|
|
result = DRIVE.proton_drive_list()
|
|
self.assertIn("items", result)
|
|
self.assertEqual(result["count"], 2)
|
|
self.assertEqual(result["items"][0]["Name"], "Documents")
|
|
self.assertEqual(result["items"][1]["Name"], "notes.txt")
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_list_recursive(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(
|
|
returncode=0,
|
|
stdout=json.dumps({"Name": "sub/file.pdf", "Size": 2048, "IsDir": False}) + "\n",
|
|
stderr="",
|
|
)
|
|
result = DRIVE.proton_drive_list(path="Documents", recursive=True)
|
|
self.assertEqual(result["items"][0]["Name"], "sub/file.pdf")
|
|
|
|
def test_default_remote(self):
|
|
self.assertEqual(DRIVE._get_remote(), "protondrive")
|
|
|
|
def test_remote_path(self):
|
|
path = DRIVE._remote_path("Docs/File.txt")
|
|
self.assertEqual(path, "protondrive:Docs/File.txt")
|
|
|
|
|
|
class TestProtonDriveRead(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_read_file(self, mock_run):
|
|
def side_effect(*args, **kwargs):
|
|
cmd = args[0]
|
|
if "lsl" in cmd:
|
|
return SimpleNamespace(returncode=0, stdout="5000 2026-01-01 12:00:00 notes.txt", stderr="")
|
|
elif "cat" in cmd:
|
|
return SimpleNamespace(returncode=0, stdout="Hello, Proton Drive!\nLine 2\nLine 3", stderr="")
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
mock_run.side_effect = side_effect
|
|
|
|
result = DRIVE.proton_drive_read("notes.txt")
|
|
self.assertIn("content", result)
|
|
self.assertIn("Proton Drive", result["content"])
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_read_large_file_redirects(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(
|
|
returncode=0,
|
|
stdout="15000000 2026-01-01 12:00:00 big_file.mp4",
|
|
stderr="",
|
|
)
|
|
result = DRIVE.proton_drive_read("big_file.mp4")
|
|
self.assertIn("error", result)
|
|
self.assertIn("proton_drive_download", result["error"])
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_read_head(self, mock_run):
|
|
def side_effect(*args, **kwargs):
|
|
cmd = args[0]
|
|
if "lsl" in cmd:
|
|
return SimpleNamespace(returncode=0, stdout="200 2026-01-01 12:00:00 file.txt", stderr="")
|
|
elif "cat" in cmd:
|
|
return SimpleNamespace(returncode=0, stdout="a\nb\nc\nd\ne", stderr="")
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
mock_run.side_effect = side_effect
|
|
|
|
result = DRIVE.proton_drive_read("file.txt", head=2)
|
|
lines = result["content"].split("\n")
|
|
self.assertEqual(len(lines), 2)
|
|
|
|
|
|
class TestProtonDriveDownload(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.os.path.isfile")
|
|
@patch(f"{MOD}.os.path.getsize")
|
|
def test_download(self, mock_size, mock_isfile, mock_run):
|
|
mock_run.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
mock_isfile.return_value = True
|
|
mock_size.return_value = 1024
|
|
|
|
result = DRIVE.proton_drive_download("remote/file.pdf", "/tmp/out.pdf")
|
|
self.assertEqual(result["status"], "downloaded")
|
|
self.assertEqual(result["size_bytes"], 1024)
|
|
|
|
|
|
class TestProtonDriveUpload(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
@patch(f"{MOD}.os.path.exists")
|
|
@patch(f"{MOD}.os.path.isfile")
|
|
@patch(f"{MOD}.os.path.getsize")
|
|
def test_upload(self, mock_size, mock_isfile, mock_exists, mock_run):
|
|
mock_exists.return_value = True
|
|
mock_isfile.return_value = True
|
|
mock_size.return_value = 8192
|
|
mock_run.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
result = DRIVE.proton_drive_upload("/tmp/mydoc.pdf", "Documents/mydoc.pdf")
|
|
self.assertEqual(result["status"], "uploaded")
|
|
|
|
def test_upload_missing_file(self):
|
|
result = DRIVE.proton_drive_upload("/nonexistent/file.txt", "remote/")
|
|
self.assertIn("error", result)
|
|
|
|
|
|
class TestProtonDriveSearch(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_search(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(
|
|
returncode=0,
|
|
stdout=json.dumps({"Name": "report_q1.pdf", "Size": 5000}) + "\n"
|
|
+ json.dumps({"Name": "report_q2.pdf", "Size": 6000}) + "\n"
|
|
+ json.dumps({"Name": "notes.txt", "Size": 200}),
|
|
stderr="",
|
|
)
|
|
result = DRIVE.proton_drive_search("report")
|
|
self.assertEqual(result["count"], 2)
|
|
self.assertEqual(result["items"][0]["Name"], "report_q1.pdf")
|
|
|
|
|
|
class TestProtonDriveMkdir(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_mkdir(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
result = DRIVE.proton_drive_mkdir("Documents/NewFolder")
|
|
self.assertEqual(result["status"], "created")
|
|
|
|
|
|
class TestProtonDriveDelete(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_delete_file(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
result = DRIVE.proton_drive_delete("old_file.txt")
|
|
self.assertEqual(result["status"], "deleted")
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_delete_recursive(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
result = DRIVE.proton_drive_delete("OldFolder", recursive=True)
|
|
self.assertTrue(result["recursive"])
|
|
|
|
|
|
class TestProtonDriveStat(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_stat_file(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(
|
|
returncode=0,
|
|
stdout="1500 2026-06-01 10:30:00 file.txt",
|
|
stderr="",
|
|
)
|
|
result = DRIVE.proton_drive_stat("file.txt")
|
|
self.assertEqual(result["Size"], "1500")
|
|
|
|
|
|
class TestProtonDriveSync(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_sync_dry_run(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(
|
|
returncode=0,
|
|
stdout="file1.txt\nfile2.txt",
|
|
stderr="",
|
|
)
|
|
result = DRIVE.proton_drive_sync(
|
|
source="/tmp/mydir",
|
|
dest="protondrive:backup/",
|
|
dry_run=True,
|
|
)
|
|
self.assertEqual(result["status"], "dry_run")
|
|
self.assertTrue(result["dry_run"])
|
|
|
|
|
|
class TestRcloneHelpers(unittest.TestCase):
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_check_available(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(returncode=0, stdout="rclone v1.69", stderr="")
|
|
result = DRIVE.check_rclone_availability()
|
|
self.assertIn("available", result)
|
|
|
|
@patch(f"{MOD}.subprocess.run")
|
|
def test_check_remote_missing(self, mock_run):
|
|
mock_run.return_value = SimpleNamespace(returncode=0, stdout="otherremote:\n", stderr="")
|
|
result = DRIVE.check_rclone_remote()
|
|
self.assertFalse(result["configured"])
|
|
|
|
@patch.dict(os.environ, {"PROTON_RCLONE_REMOTE": "customremote"}, clear=False)
|
|
def test_custom_remote(self):
|
|
self.assertEqual(DRIVE._get_remote(), "customremote")
|
|
|
|
@patch.dict(os.environ, {"PROTON_RCLONE_PATH": "/usr/local/bin/rclone"}, clear=False)
|
|
def test_custom_rclone_path(self):
|
|
self.assertEqual(DRIVE._get_rclone_path(), "/usr/local/bin/rclone")
|
|
|
|
def test_get_rclone_remote(self):
|
|
self.assertEqual(DRIVE.get_rclone_remote(), "protondrive")
|
|
|
|
|
|
class TestParseHelpers(unittest.TestCase):
|
|
|
|
def test_parse_lsf_json(self):
|
|
output = json.dumps({"Name": "a.txt", "Size": 10}) + "\n" + json.dumps({"Name": "b.txt", "Size": 20})
|
|
items = DRIVE._parse_lsf_json(output)
|
|
self.assertEqual(len(items), 2)
|
|
|
|
def test_parse_lsf_json_empty(self):
|
|
items = DRIVE._parse_lsf_json("")
|
|
self.assertEqual(items, [])
|
|
|
|
def test_parse_lsl_output(self):
|
|
output = "1024 2026-01-01 12:00:00 file.txt\n2048 2026-06-01 15:00:00 other.pdf"
|
|
items = DRIVE._parse_lsl_output(output)
|
|
self.assertEqual(len(items), 2)
|
|
self.assertEqual(items[0]["Size"], "1024")
|
|
|
|
def test_parse_sync_output(self):
|
|
result = DRIVE._parse_sync_output("file1.txt\nfile2.txt", "ERROR on file3.txt")
|
|
self.assertEqual(result["change_count"], 2)
|
|
self.assertEqual(result["error_count"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|