"""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()