apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-05-11 16:52:38 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

test(e2e): serve binary PDF bytes from storage connector fakes

Browse source
This commit is contained in:
Anish Sarkar 2026-05-09 05:03:07 +05:30
parent ad226853e5
commit 5a2357b981
4 changed files with 36 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

19
surfsense_backend/tests/e2e/fakes/composio_module.py
View file

@ -22,6 +22,8 @@ import logging
from pathlib import Path
from typing import Any
from .binary_loader import _resolve_file_bytes
logger = logging.getLogger(__name__)
@ -29,9 +31,10 @@ logger = logging.getLogger(__name__)
# Fixture loading
# ---------------------------------------------------------------------------
_DRIVE_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "drive_files.json"
_GMAIL_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "gmail_messages.json"
_CALENDAR_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "calendar_events.json"
_FIXTURES_DIR = Path(__file__).parent / "fixtures"
_DRIVE_FIXTURE_PATH = _FIXTURES_DIR / "drive_files.json"
_GMAIL_FIXTURE_PATH = _FIXTURES_DIR / "gmail_messages.json"
_CALENDAR_FIXTURE_PATH = _FIXTURES_DIR / "calendar_events.json"
_DRIVE_DOWNLOAD_DIR = Path("/tmp/surfsense-e2e-composio-downloads")
@ -408,7 +411,7 @@ def _drive_download_file(args: dict[str, Any]) -> dict[str, Any]:
directory and returning the path.
"""
file_id = args.get("file_id", "")
contents = _DRIVE_FIXTURE.get("_file_contents", {}).get(file_id)
contents = _resolve_file_bytes(_DRIVE_FIXTURE, file_id, _FIXTURES_DIR)
if contents is None:
# Unknown file id is a test bug, fail loudly.
raise NotImplementedError(
@ -418,12 +421,14 @@ def _drive_download_file(args: dict[str, Any]) -> dict[str, Any]:
)
_DRIVE_DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
out_path = _DRIVE_DOWNLOAD_DIR / f"{file_id}.txt"
out_path.write_text(contents, encoding="utf-8")
metadata = _drive_get_metadata({"file_id": file_id})["data"]
file_name = metadata.get("name") or f"{file_id}.txt"
out_path = _DRIVE_DOWNLOAD_DIR / file_name
out_path.write_bytes(contents)
return {
"data": {
"file_path": str(out_path),
"file_name": f"{file_id}.txt",
"file_name": file_name,
"size": len(contents),
}
}

13
surfsense_backend/tests/e2e/fakes/dropbox_api.py
View file

@ -15,7 +15,10 @@ from unittest.mock import patch
import httpx
_DROPBOX_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "dropbox_files.json"
from .binary_loader import _resolve_file_bytes
_FIXTURES_DIR = Path(__file__).parent / "fixtures"
_DROPBOX_FIXTURE_PATH = _FIXTURES_DIR / "dropbox_files.json"
def _load_dropbox_fixture() -> dict[str, Any]:
@ -72,17 +75,17 @@ class _FakeDropboxClient(_StrictFakeMixin):
return metadata, None
async def download_file(self, path: str) -> tuple[bytes | None, str | None]:
content = _DROPBOX_FIXTURE.get("_file_contents", {}).get(path)
content = _resolve_file_bytes(_DROPBOX_FIXTURE, path, _FIXTURES_DIR)
if content is None:
return None, f"E2E Dropbox fake has no content for path={path!r}."
return content.encode("utf-8"), None
return content, None
async def download_file_to_disk(self, path: str, dest_path: str) -> str | None:
content = _DROPBOX_FIXTURE.get("_file_contents", {}).get(path)
content = _resolve_file_bytes(_DROPBOX_FIXTURE, path, _FIXTURES_DIR)
if content is None:
return f"E2E Dropbox fake has no content for path={path!r}."
with open(dest_path, "wb") as f:
f.write(content.encode("utf-8"))
f.write(content)
return None
async def get_current_account(self) -> tuple[dict[str, Any] | None, str | None]:

13
surfsense_backend/tests/e2e/fakes/native_google.py
View file

@ -17,9 +17,12 @@ from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
from google.oauth2.credentials import Credentials
_DRIVE_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "drive_files.json"
_GMAIL_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "gmail_messages.json"
_CALENDAR_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "calendar_events.json"
from .binary_loader import _resolve_file_bytes
_FIXTURES_DIR = Path(__file__).parent / "fixtures"
_DRIVE_FIXTURE_PATH = _FIXTURES_DIR / "drive_files.json"
_GMAIL_FIXTURE_PATH = _FIXTURES_DIR / "gmail_messages.json"
_CALENDAR_FIXTURE_PATH = _FIXTURES_DIR / "calendar_events.json"
def _load_drive_fixture() -> dict[str, Any]:
@ -165,12 +168,12 @@ class _FakeDriveFiles(_StrictFakeMixin):
def get_media(self, **kwargs: Any) -> _FakeMediaRequest:
file_id = kwargs.get("fileId")
content = _DRIVE_FIXTURE.get("_file_contents", {}).get(file_id)
content = _resolve_file_bytes(_DRIVE_FIXTURE, file_id, _FIXTURES_DIR)
if content is None:
raise NotImplementedError(
f"E2E native Google fake has no content for fileId={file_id!r}."
)
return _FakeMediaRequest(content.encode("utf-8"))
return _FakeMediaRequest(content)
def export(self, **kwargs: Any) -> _FakeRequest:
file_id = kwargs.get("fileId")

13
surfsense_backend/tests/e2e/fakes/onedrive_graph.py
View file

@ -15,7 +15,10 @@ from unittest.mock import patch
import httpx
_ONEDRIVE_FIXTURE_PATH = Path(__file__).parent / "fixtures" / "onedrive_files.json"
from .binary_loader import _resolve_file_bytes
_FIXTURES_DIR = Path(__file__).parent / "fixtures"
_ONEDRIVE_FIXTURE_PATH = _FIXTURES_DIR / "onedrive_files.json"
def _load_onedrive_fixture() -> dict[str, Any]:
@ -61,17 +64,17 @@ class _FakeOneDriveClient(_StrictFakeMixin):
return metadata, None
async def download_file(self, item_id: str) -> tuple[bytes | None, str | None]:
content = _ONEDRIVE_FIXTURE.get("_file_contents", {}).get(item_id)
content = _resolve_file_bytes(_ONEDRIVE_FIXTURE, item_id, _FIXTURES_DIR)
if content is None:
return None, f"E2E OneDrive fake has no content for item_id={item_id!r}."
return content.encode("utf-8"), None
return content, None
async def download_file_to_disk(self, item_id: str, dest_path: str) -> str | None:
content = _ONEDRIVE_FIXTURE.get("_file_contents", {}).get(item_id)
content = _resolve_file_bytes(_ONEDRIVE_FIXTURE, item_id, _FIXTURES_DIR)
if content is None:
return f"E2E OneDrive fake has no content for item_id={item_id!r}."
with open(dest_path, "wb") as f:
f.write(content.encode("utf-8"))
f.write(content)
return None
async def get_delta(