PlanOpticon

planopticon / tests / test_cloud_sources.py
Source Blame History 213 lines
a6b6869… leo 1 """Tests for cloud source integrations."""
a6b6869… leo 2
a6b6869… leo 3 import json
a6b6869… leo 4 from pathlib import Path
a6b6869… leo 5 from unittest.mock import MagicMock, patch
a6b6869… leo 6
a6b6869… leo 7 import pytest
a6b6869… leo 8
a6b6869… leo 9 from video_processor.sources.base import BaseSource, SourceFile
a6b6869… leo 10
a6b6869… leo 11
a6b6869… leo 12 class TestSourceFile:
a6b6869… leo 13 def test_basic(self):
a6b6869… leo 14 f = SourceFile(name="video.mp4", id="abc123")
a6b6869… leo 15 assert f.name == "video.mp4"
a6b6869… leo 16 assert f.id == "abc123"
a6b6869… leo 17
a6b6869… leo 18 def test_full(self):
a6b6869… leo 19 f = SourceFile(
a6b6869… leo 20 name="meeting.mp4",
a6b6869… leo 21 id="xyz",
a6b6869… leo 22 size_bytes=1024000,
a6b6869… leo 23 mime_type="video/mp4",
a6b6869… leo 24 modified_at="2025-01-01T00:00:00Z",
a6b6869… leo 25 path="/recordings/meeting.mp4",
a6b6869… leo 26 )
a6b6869… leo 27 assert f.size_bytes == 1024000
a6b6869… leo 28 assert f.path == "/recordings/meeting.mp4"
a6b6869… leo 29
a6b6869… leo 30 def test_round_trip(self):
a6b6869… leo 31 f = SourceFile(name="test.mp4", id="1")
a6b6869… leo 32 data = f.model_dump_json()
a6b6869… leo 33 restored = SourceFile.model_validate_json(data)
a6b6869… leo 34 assert restored.name == f.name
a6b6869… leo 35
a6b6869… leo 36
a6b6869… leo 37 class TestBaseSource:
a6b6869… leo 38 def test_download_all(self, tmp_path):
a6b6869… leo 39 class FakeSource(BaseSource):
a6b6869… leo 40 def authenticate(self):
a6b6869… leo 41 return True
a6b6869… leo 42
a6b6869… leo 43 def list_videos(self, **kwargs):
a6b6869… leo 44 return []
a6b6869… leo 45
a6b6869… leo 46 def download(self, file, destination):
a6b6869… leo 47 destination.write_text("fake video data")
a6b6869… leo 48 return destination
a6b6869… leo 49
a6b6869… leo 50 source = FakeSource()
a6b6869… leo 51 files = [
a6b6869… leo 52 SourceFile(name="a.mp4", id="1"),
a6b6869… leo 53 SourceFile(name="b.mp4", id="2"),
a6b6869… leo 54 ]
a6b6869… leo 55 paths = source.download_all(files, tmp_path / "downloads")
a6b6869… leo 56 assert len(paths) == 2
a6b6869… leo 57 assert (tmp_path / "downloads" / "a.mp4").exists()
a6b6869… leo 58 assert (tmp_path / "downloads" / "b.mp4").exists()
a6b6869… leo 59
a6b6869… leo 60 def test_download_all_handles_errors(self, tmp_path):
a6b6869… leo 61 class FailingSource(BaseSource):
a6b6869… leo 62 def authenticate(self):
a6b6869… leo 63 return True
a6b6869… leo 64
a6b6869… leo 65 def list_videos(self, **kwargs):
a6b6869… leo 66 return []
a6b6869… leo 67
a6b6869… leo 68 def download(self, file, destination):
a6b6869… leo 69 raise RuntimeError("Download failed")
a6b6869… leo 70
a6b6869… leo 71 source = FailingSource()
a6b6869… leo 72 files = [SourceFile(name="fail.mp4", id="1")]
a6b6869… leo 73 paths = source.download_all(files, tmp_path / "downloads")
a6b6869… leo 74 assert len(paths) == 0
a6b6869… leo 75
a6b6869… leo 76
a6b6869… leo 77 class TestGoogleDriveSource:
a6b6869… leo 78 def test_init_defaults(self):
a6b6869… leo 79 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 80
a6b6869… leo 81 source = GoogleDriveSource()
a6b6869… leo 82 assert source.service is None
a6b6869… leo 83 assert source.use_service_account is None
a6b6869… leo 84
a6b6869… leo 85 def test_init_with_credentials(self):
a6b6869… leo 86 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 87
a6b6869… leo 88 source = GoogleDriveSource(
a6b6869… leo 89 credentials_path="/path/to/creds.json",
a6b6869… leo 90 use_service_account=True,
a6b6869… leo 91 )
a6b6869… leo 92 assert source.credentials_path == "/path/to/creds.json"
a6b6869… leo 93 assert source.use_service_account is True
a6b6869… leo 94
a6b6869… leo 95 def test_is_service_account_true(self, tmp_path):
a6b6869… leo 96 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 97
a6b6869… leo 98 creds_file = tmp_path / "sa.json"
a6b6869… leo 99 creds_file.write_text(json.dumps({"type": "service_account"}))
a6b6869… leo 100 source = GoogleDriveSource(credentials_path=str(creds_file))
a6b6869… leo 101 assert source._is_service_account() is True
a6b6869… leo 102
a6b6869… leo 103 def test_is_service_account_false(self, tmp_path):
a6b6869… leo 104 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 105
a6b6869… leo 106 creds_file = tmp_path / "oauth.json"
a6b6869… leo 107 creds_file.write_text(json.dumps({"installed": {}}))
a6b6869… leo 108 source = GoogleDriveSource(credentials_path=str(creds_file))
a6b6869… leo 109 assert source._is_service_account() is False
a6b6869… leo 110
a6b6869… leo 111 @patch.dict("os.environ", {}, clear=True)
a6b6869… leo 112 def test_is_service_account_no_path(self):
a6b6869… leo 113 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 114
a6b6869… leo 115 source = GoogleDriveSource(credentials_path=None)
a6b6869… leo 116 source.credentials_path = None # Override any env var fallback
a6b6869… leo 117 assert source._is_service_account() is False
a6b6869… leo 118
a6b6869… leo 119 def test_list_videos_not_authenticated(self):
a6b6869… leo 120 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 121
a6b6869… leo 122 source = GoogleDriveSource()
a6b6869… leo 123 with pytest.raises(RuntimeError, match="Not authenticated"):
a6b6869… leo 124 source.list_videos(folder_id="abc")
a6b6869… leo 125
a6b6869… leo 126 def test_download_not_authenticated(self):
a6b6869… leo 127 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 128
a6b6869… leo 129 source = GoogleDriveSource()
a6b6869… leo 130 f = SourceFile(name="test.mp4", id="1")
a6b6869… leo 131 with pytest.raises(RuntimeError, match="Not authenticated"):
a6b6869… leo 132 source.download(f, Path("/tmp/test.mp4"))
a6b6869… leo 133
a6b6869… leo 134 @patch("video_processor.sources.google_drive.GoogleDriveSource._auth_service_account")
a6b6869… leo 135 def test_authenticate_import_error(self, mock_auth):
a6b6869… leo 136 from video_processor.sources.google_drive import GoogleDriveSource
a6b6869… leo 137
a6b6869… leo 138 source = GoogleDriveSource()
829e24a… leo 139 with patch.dict(
829e24a… leo 140 "sys.modules", {"google.oauth2": None, "google.oauth2.service_account": None}
829e24a… leo 141 ):
a6b6869… leo 142 # The import will fail inside authenticate
a6b6869… leo 143 result = source.authenticate()
a6b6869… leo 144 assert result is False
a6b6869… leo 145
a6b6869… leo 146
a6b6869… leo 147 class TestDropboxSource:
a6b6869… leo 148 def test_init_defaults(self):
a6b6869… leo 149 from video_processor.sources.dropbox_source import DropboxSource
a6b6869… leo 150
a6b6869… leo 151 source = DropboxSource()
a6b6869… leo 152 assert source.dbx is None
a6b6869… leo 153
a6b6869… leo 154 def test_init_with_token(self):
a6b6869… leo 155 from video_processor.sources.dropbox_source import DropboxSource
a6b6869… leo 156
a6b6869… leo 157 source = DropboxSource(access_token="test_token")
a6b6869… leo 158 assert source.access_token == "test_token"
a6b6869… leo 159
a6b6869… leo 160 def test_init_with_app_key(self):
a6b6869… leo 161 from video_processor.sources.dropbox_source import DropboxSource
a6b6869… leo 162
a6b6869… leo 163 source = DropboxSource(app_key="key", app_secret="secret")
a6b6869… leo 164 assert source.app_key == "key"
a6b6869… leo 165 assert source.app_secret == "secret"
a6b6869… leo 166
a6b6869… leo 167 def test_list_videos_not_authenticated(self):
a6b6869… leo 168 from video_processor.sources.dropbox_source import DropboxSource
a6b6869… leo 169
a6b6869… leo 170 source = DropboxSource()
a6b6869… leo 171 with pytest.raises(RuntimeError, match="Not authenticated"):
a6b6869… leo 172 source.list_videos(folder_path="/recordings")
a6b6869… leo 173
a6b6869… leo 174 def test_download_not_authenticated(self):
a6b6869… leo 175 from video_processor.sources.dropbox_source import DropboxSource
a6b6869… leo 176
a6b6869… leo 177 source = DropboxSource()
a6b6869… leo 178 f = SourceFile(name="test.mp4", id="1", path="/test.mp4")
a6b6869… leo 179 with pytest.raises(RuntimeError, match="Not authenticated"):
a6b6869… leo 180 source.download(f, Path("/tmp/test.mp4"))
a6b6869… leo 181
a6b6869… leo 182 def test_authenticate_no_sdk(self):
a6b6869… leo 183 from video_processor.sources.dropbox_source import DropboxSource
a6b6869… leo 184
a6b6869… leo 185 source = DropboxSource()
a6b6869… leo 186 with patch.dict("sys.modules", {"dropbox": None}):
a6b6869… leo 187 result = source.authenticate()
a6b6869… leo 188 assert result is False
a6b6869… leo 189
a6b6869… leo 190 def test_auth_saved_token(self, tmp_path):
a6b6869… leo 191 pytest.importorskip("dropbox")
a6b6869… leo 192 from video_processor.sources.dropbox_source import DropboxSource
a6b6869… leo 193
a6b6869… leo 194 token_file = tmp_path / "token.json"
829e24a… leo 195 token_file.write_text(
829e24a… leo 196 json.dumps(
829e24a… leo 197 {
829e24a… leo 198 "refresh_token": "rt_test",
829e24a… leo 199 "app_key": "key",
829e24a… leo 200 "app_secret": "secret",
829e24a… leo 201 }
829e24a… leo 202 )
829e24a… leo 203 )
a6b6869… leo 204
a6b6869… leo 205 source = DropboxSource(token_path=token_file, app_key="key", app_secret="secret")
a6b6869… leo 206
a6b6869… leo 207 mock_dbx = MagicMock()
a6b6869… leo 208 with patch("dropbox.Dropbox", return_value=mock_dbx):
a6b6869… leo 209 import dropbox
829e24a… leo 210
a6b6869… leo 211 result = source._auth_saved_token(dropbox)
a6b6869… leo 212 assert result is True
a6b6869… leo 213 assert source.dbx is mock_dbx

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button