Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pydrivedol/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,4 +41,5 @@
# Needing API setup:
GDReader, # read-only Mapping interface to a Google Drive folder
GDStore, # read-write MutableMapping interface to a Google Drive folder
drive_from_service_account, # headless auth: build a GoogleDrive from an SA key
)
44 changes: 43 additions & 1 deletion pydrivedol/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,42 @@ def _init_google_drive(
return GoogleDrive(gauth)


def drive_from_service_account(
key_file: str,
*,
scopes=("https://www.googleapis.com/auth/drive",),
subject: Optional[str] = None,
):
"""Build an authenticated ``GoogleDrive`` from a service-account key file.

For headless / server use — no browser OAuth flow. Share the target Drive folder
with the service account's ``client_email`` (Viewer for read, Editor for write).
Pass the resulting drive to ``GDReader``/``GDStore`` via their ``drive=`` argument.

Args:
key_file: path to the service-account JSON key.
scopes: OAuth scopes; default is full Drive (use ``.../auth/drive.readonly``
to enforce read-only at the token level).
subject: optional user email to impersonate (domain-wide delegation).

>>> drive = drive_from_service_account('sa-key.json') # doctest: +SKIP
>>> reader = GDReader(folder_url, drive=drive) # doctest: +SKIP
"""
_require_pydrive2()
service_config = {"client_json_file_path": key_file}
if subject:
service_config["client_user_email"] = subject
gauth = GoogleAuth(
settings={
"client_config_backend": "service",
"service_config": service_config,
"oauth_scope": list(scopes),
}
)
gauth.ServiceAuth()
return GoogleDrive(gauth)


class GDReader(Mapping):
"""
Read-only Mapping to Google Drive folder.
Expand All @@ -219,6 +255,7 @@ def __init__(
credentials_file: str = "client_secrets.json",
settings_file: str = "settings.yaml",
include_hidden: bool = False,
drive=None,
):
"""
Initialize reader.
Expand All @@ -229,6 +266,9 @@ def __init__(
credentials_file: OAuth2 credentials path
settings_file: Auth settings path
include_hidden: Include files starting with '.'
drive: a pre-authenticated ``GoogleDrive`` (e.g. from
``drive_from_service_account``). When given, the OAuth
``credentials_file``/``settings_file`` flow is skipped.
"""
_require_pydrive2()

Expand All @@ -242,7 +282,9 @@ def __init__(
self._credentials_file = credentials_file
self._settings_file = settings_file

self._drive = _init_google_drive(credentials_file, settings_file)
self._drive = (
drive if drive is not None else _init_google_drive(credentials_file, settings_file)
)
self._file_cache = None

def _list_files(self, folder_id: str, prefix: str = "", level: int = 0):
Expand Down
64 changes: 64 additions & 0 deletions pydrivedol/tests/test_service_account.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
"""Tests for service-account (headless) auth support.

Mockable — no live Drive. Verifies ``drive_from_service_account`` builds the pydrive2
service-account settings, and that ``GDReader``/``GDStore`` use an injected ``drive=``
client instead of the interactive OAuth flow.
"""

import pydrivedol.base as base
from pydrivedol import drive_from_service_account, GDReader


def test_drive_from_service_account_builds_service_settings(monkeypatch):
captured = {}

class FakeAuth:
def __init__(self, settings=None):
captured["settings"] = settings

def ServiceAuth(self):
captured["service_auth"] = True

monkeypatch.setattr(base, "_PYDRIVE2_AVAILABLE", True)
monkeypatch.setattr(base, "GoogleAuth", FakeAuth)
monkeypatch.setattr(base, "GoogleDrive", lambda auth: ("DRIVE", auth))

drive = drive_from_service_account("/key.json", scopes=("scope-a",), subject="u@x.com")

s = captured["settings"]
assert s["client_config_backend"] == "service"
assert s["service_config"]["client_json_file_path"] == "/key.json"
assert s["service_config"]["client_user_email"] == "u@x.com" # impersonation when subject given
assert s["oauth_scope"] == ["scope-a"]
assert captured["service_auth"] is True
assert drive[0] == "DRIVE"


def test_drive_from_service_account_no_subject_omits_impersonation(monkeypatch):
captured = {}

class FakeAuth:
def __init__(self, settings=None):
captured["settings"] = settings

def ServiceAuth(self):
pass

monkeypatch.setattr(base, "_PYDRIVE2_AVAILABLE", True)
monkeypatch.setattr(base, "GoogleAuth", FakeAuth)
monkeypatch.setattr(base, "GoogleDrive", lambda auth: "DRIVE")
drive_from_service_account("/key.json")
assert "client_user_email" not in captured["settings"]["service_config"]


def test_gdreader_uses_injected_drive(monkeypatch):
monkeypatch.setattr(base, "_PYDRIVE2_AVAILABLE", True)

def _no_oauth(*a, **k):
raise AssertionError("must not run the OAuth flow when drive= is provided")

monkeypatch.setattr(base, "_init_google_drive", _no_oauth)
sentinel = object()
reader = GDReader("https://drive.google.com/drive/folders/ABC123", drive=sentinel)
assert reader._drive is sentinel
assert reader.folder_id == "ABC123"
Loading