mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-28 06:51:16 +08:00
fix: update tests for gws migration
- Rewrite test_google_workspace_api.py: test bridge token handling and calendar date range instead of removed get_credentials() - Update test_google_oauth_setup.py: partial scopes now accepted with warning instead of rejected with SystemExit
This commit is contained in:
@@ -211,14 +211,15 @@ class TestExchangeAuthCode:
|
|||||||
assert setup_module.PENDING_AUTH_PATH.exists()
|
assert setup_module.PENDING_AUTH_PATH.exists()
|
||||||
assert not setup_module.TOKEN_PATH.exists()
|
assert not setup_module.TOKEN_PATH.exists()
|
||||||
|
|
||||||
def test_refuses_to_overwrite_existing_token_with_narrower_scopes(self, setup_module, capsys):
|
def test_accepts_narrower_scopes_with_warning(self, setup_module, capsys):
|
||||||
|
"""Partial scopes are accepted with a warning (gws migration: v2.0)."""
|
||||||
setup_module.PENDING_AUTH_PATH.write_text(
|
setup_module.PENDING_AUTH_PATH.write_text(
|
||||||
json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"})
|
json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"})
|
||||||
)
|
)
|
||||||
setup_module.TOKEN_PATH.write_text(json.dumps({"token": "existing-token", "scopes": setup_module.SCOPES}))
|
setup_module.TOKEN_PATH.write_text(json.dumps({"token": "***", "scopes": setup_module.SCOPES}))
|
||||||
FakeFlow.credentials_payload = {
|
FakeFlow.credentials_payload = {
|
||||||
"token": "narrow-token",
|
"token": "***",
|
||||||
"refresh_token": "refresh-token",
|
"refresh_token": "***",
|
||||||
"token_uri": "https://oauth2.googleapis.com/token",
|
"token_uri": "https://oauth2.googleapis.com/token",
|
||||||
"client_id": "client-id",
|
"client_id": "client-id",
|
||||||
"client_secret": "client-secret",
|
"client_secret": "client-secret",
|
||||||
@@ -228,10 +229,12 @@ class TestExchangeAuthCode:
|
|||||||
],
|
],
|
||||||
}
|
}
|
||||||
|
|
||||||
with pytest.raises(SystemExit):
|
setup_module.exchange_auth_code("4/test-auth-code")
|
||||||
setup_module.exchange_auth_code("4/test-auth-code")
|
|
||||||
|
|
||||||
out = capsys.readouterr().out
|
out = capsys.readouterr().out
|
||||||
assert "refusing to save incomplete google workspace token" in out.lower()
|
assert "warning" in out.lower()
|
||||||
assert json.loads(setup_module.TOKEN_PATH.read_text())["token"] == "existing-token"
|
assert "missing" in out.lower()
|
||||||
assert setup_module.PENDING_AUTH_PATH.exists()
|
# Token is saved (partial scopes accepted)
|
||||||
|
assert setup_module.TOKEN_PATH.exists()
|
||||||
|
# Pending auth is cleaned up
|
||||||
|
assert not setup_module.PENDING_AUTH_PATH.exists()
|
||||||
|
|||||||
@@ -1,117 +1,175 @@
|
|||||||
"""Regression tests for Google Workspace API credential validation."""
|
"""Tests for Google Workspace gws bridge and CLI wrapper."""
|
||||||
|
|
||||||
import importlib.util
|
import importlib.util
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
import types
|
import types
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
SCRIPT_PATH = (
|
BRIDGE_PATH = (
|
||||||
|
Path(__file__).resolve().parents[2]
|
||||||
|
/ "skills/productivity/google-workspace/scripts/gws_bridge.py"
|
||||||
|
)
|
||||||
|
API_PATH = (
|
||||||
Path(__file__).resolve().parents[2]
|
Path(__file__).resolve().parents[2]
|
||||||
/ "skills/productivity/google-workspace/scripts/google_api.py"
|
/ "skills/productivity/google-workspace/scripts/google_api.py"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class FakeAuthorizedCredentials:
|
|
||||||
def __init__(self, *, valid=True, expired=False, refresh_token="refresh-token"):
|
|
||||||
self.valid = valid
|
|
||||||
self.expired = expired
|
|
||||||
self.refresh_token = refresh_token
|
|
||||||
self.refresh_calls = 0
|
|
||||||
|
|
||||||
def refresh(self, _request):
|
|
||||||
self.refresh_calls += 1
|
|
||||||
self.valid = True
|
|
||||||
self.expired = False
|
|
||||||
|
|
||||||
def to_json(self):
|
|
||||||
return json.dumps({
|
|
||||||
"token": "refreshed-token",
|
|
||||||
"refresh_token": self.refresh_token,
|
|
||||||
"token_uri": "https://oauth2.googleapis.com/token",
|
|
||||||
"client_id": "client-id",
|
|
||||||
"client_secret": "client-secret",
|
|
||||||
"scopes": [
|
|
||||||
"https://www.googleapis.com/auth/gmail.readonly",
|
|
||||||
"https://www.googleapis.com/auth/gmail.send",
|
|
||||||
"https://www.googleapis.com/auth/gmail.modify",
|
|
||||||
"https://www.googleapis.com/auth/calendar",
|
|
||||||
"https://www.googleapis.com/auth/drive.readonly",
|
|
||||||
"https://www.googleapis.com/auth/contacts.readonly",
|
|
||||||
"https://www.googleapis.com/auth/spreadsheets",
|
|
||||||
"https://www.googleapis.com/auth/documents.readonly",
|
|
||||||
],
|
|
||||||
})
|
|
||||||
|
|
||||||
|
|
||||||
class FakeCredentialsFactory:
|
|
||||||
creds = FakeAuthorizedCredentials()
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_authorized_user_file(cls, _path, _scopes):
|
|
||||||
return cls.creds
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def google_api_module(monkeypatch, tmp_path):
|
def bridge_module(monkeypatch, tmp_path):
|
||||||
google_module = types.ModuleType("google")
|
hermes_home = tmp_path / ".hermes"
|
||||||
oauth2_module = types.ModuleType("google.oauth2")
|
hermes_home.mkdir()
|
||||||
credentials_module = types.ModuleType("google.oauth2.credentials")
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||||
credentials_module.Credentials = FakeCredentialsFactory
|
|
||||||
auth_module = types.ModuleType("google.auth")
|
|
||||||
transport_module = types.ModuleType("google.auth.transport")
|
|
||||||
requests_module = types.ModuleType("google.auth.transport.requests")
|
|
||||||
requests_module.Request = object
|
|
||||||
|
|
||||||
monkeypatch.setitem(sys.modules, "google", google_module)
|
spec = importlib.util.spec_from_file_location("gws_bridge_test", BRIDGE_PATH)
|
||||||
monkeypatch.setitem(sys.modules, "google.oauth2", oauth2_module)
|
|
||||||
monkeypatch.setitem(sys.modules, "google.oauth2.credentials", credentials_module)
|
|
||||||
monkeypatch.setitem(sys.modules, "google.auth", auth_module)
|
|
||||||
monkeypatch.setitem(sys.modules, "google.auth.transport", transport_module)
|
|
||||||
monkeypatch.setitem(sys.modules, "google.auth.transport.requests", requests_module)
|
|
||||||
|
|
||||||
spec = importlib.util.spec_from_file_location("google_workspace_api_test", SCRIPT_PATH)
|
|
||||||
module = importlib.util.module_from_spec(spec)
|
module = importlib.util.module_from_spec(spec)
|
||||||
assert spec.loader is not None
|
assert spec.loader is not None
|
||||||
spec.loader.exec_module(module)
|
spec.loader.exec_module(module)
|
||||||
|
|
||||||
monkeypatch.setattr(module, "TOKEN_PATH", tmp_path / "google_token.json")
|
|
||||||
return module
|
return module
|
||||||
|
|
||||||
|
|
||||||
def _write_token(path: Path, scopes):
|
@pytest.fixture
|
||||||
path.write_text(json.dumps({
|
def api_module(monkeypatch, tmp_path):
|
||||||
"token": "access-token",
|
hermes_home = tmp_path / ".hermes"
|
||||||
"refresh_token": "refresh-token",
|
hermes_home.mkdir()
|
||||||
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||||
|
|
||||||
|
spec = importlib.util.spec_from_file_location("gws_api_test", API_PATH)
|
||||||
|
module = importlib.util.module_from_spec(spec)
|
||||||
|
assert spec.loader is not None
|
||||||
|
spec.loader.exec_module(module)
|
||||||
|
return module
|
||||||
|
|
||||||
|
|
||||||
|
def _write_token(path: Path, *, token="ya29.test", expiry=None, **extra):
|
||||||
|
data = {
|
||||||
|
"token": token,
|
||||||
|
"refresh_token": "1//refresh",
|
||||||
|
"client_id": "123.apps.googleusercontent.com",
|
||||||
|
"client_secret": "secret",
|
||||||
"token_uri": "https://oauth2.googleapis.com/token",
|
"token_uri": "https://oauth2.googleapis.com/token",
|
||||||
"client_id": "client-id",
|
**extra,
|
||||||
"client_secret": "client-secret",
|
}
|
||||||
"scopes": scopes,
|
if expiry is not None:
|
||||||
}))
|
data["expiry"] = expiry
|
||||||
|
path.write_text(json.dumps(data))
|
||||||
|
|
||||||
|
|
||||||
def test_get_credentials_rejects_missing_scopes(google_api_module, capsys):
|
def test_bridge_returns_valid_token(bridge_module, tmp_path):
|
||||||
FakeCredentialsFactory.creds = FakeAuthorizedCredentials(valid=True)
|
"""Non-expired token is returned without refresh."""
|
||||||
_write_token(google_api_module.TOKEN_PATH, [
|
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
||||||
"https://www.googleapis.com/auth/drive.readonly",
|
token_path = bridge_module.get_token_path()
|
||||||
"https://www.googleapis.com/auth/spreadsheets",
|
_write_token(token_path, token="ya29.valid", expiry=future)
|
||||||
])
|
|
||||||
|
|
||||||
|
result = bridge_module.get_valid_token()
|
||||||
|
assert result == "ya29.valid"
|
||||||
|
|
||||||
|
|
||||||
|
def test_bridge_refreshes_expired_token(bridge_module, tmp_path):
|
||||||
|
"""Expired token triggers a refresh via token_uri."""
|
||||||
|
past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
|
||||||
|
token_path = bridge_module.get_token_path()
|
||||||
|
_write_token(token_path, token="ya29.old", expiry=past)
|
||||||
|
|
||||||
|
mock_resp = MagicMock()
|
||||||
|
mock_resp.read.return_value = json.dumps({
|
||||||
|
"access_token": "ya29.refreshed",
|
||||||
|
"expires_in": 3600,
|
||||||
|
}).encode()
|
||||||
|
mock_resp.__enter__ = lambda s: s
|
||||||
|
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||||
|
|
||||||
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
||||||
|
result = bridge_module.get_valid_token()
|
||||||
|
|
||||||
|
assert result == "ya29.refreshed"
|
||||||
|
# Verify persisted
|
||||||
|
saved = json.loads(token_path.read_text())
|
||||||
|
assert saved["token"] == "ya29.refreshed"
|
||||||
|
|
||||||
|
|
||||||
|
def test_bridge_exits_on_missing_token(bridge_module):
|
||||||
|
"""Missing token file causes exit with code 1."""
|
||||||
with pytest.raises(SystemExit):
|
with pytest.raises(SystemExit):
|
||||||
google_api_module.get_credentials()
|
bridge_module.get_valid_token()
|
||||||
|
|
||||||
err = capsys.readouterr().err
|
|
||||||
assert "missing google workspace scopes" in err.lower()
|
|
||||||
assert "gmail.send" in err
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_credentials_accepts_full_scope_token(google_api_module):
|
def test_bridge_main_injects_token_env(bridge_module, tmp_path):
|
||||||
FakeCredentialsFactory.creds = FakeAuthorizedCredentials(valid=True)
|
"""main() sets GOOGLE_WORKSPACE_CLI_TOKEN in subprocess env."""
|
||||||
_write_token(google_api_module.TOKEN_PATH, list(google_api_module.SCOPES))
|
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
||||||
|
token_path = bridge_module.get_token_path()
|
||||||
|
_write_token(token_path, token="ya29.injected", expiry=future)
|
||||||
|
|
||||||
creds = google_api_module.get_credentials()
|
captured = {}
|
||||||
|
|
||||||
assert creds is FakeCredentialsFactory.creds
|
def capture_run(cmd, **kwargs):
|
||||||
|
captured["cmd"] = cmd
|
||||||
|
captured["env"] = kwargs.get("env", {})
|
||||||
|
return MagicMock(returncode=0)
|
||||||
|
|
||||||
|
with patch.object(sys, "argv", ["gws_bridge.py", "gmail", "+triage"]):
|
||||||
|
with patch.object(subprocess, "run", side_effect=capture_run):
|
||||||
|
with pytest.raises(SystemExit):
|
||||||
|
bridge_module.main()
|
||||||
|
|
||||||
|
assert captured["env"]["GOOGLE_WORKSPACE_CLI_TOKEN"] == "ya29.injected"
|
||||||
|
assert captured["cmd"] == ["gws", "gmail", "+triage"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_api_calendar_list_uses_agenda_by_default(api_module):
|
||||||
|
"""calendar list without dates uses +agenda helper."""
|
||||||
|
captured = {}
|
||||||
|
|
||||||
|
def capture_run(cmd, **kwargs):
|
||||||
|
captured["cmd"] = cmd
|
||||||
|
return MagicMock(returncode=0)
|
||||||
|
|
||||||
|
args = api_module.argparse.Namespace(
|
||||||
|
start="", end="", max=25, calendar="primary", func=api_module.calendar_list,
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(subprocess, "run", side_effect=capture_run):
|
||||||
|
with pytest.raises(SystemExit):
|
||||||
|
api_module.calendar_list(args)
|
||||||
|
|
||||||
|
gws_args = captured["cmd"][2:] # skip python + bridge path
|
||||||
|
assert "calendar" in gws_args
|
||||||
|
assert "+agenda" in gws_args
|
||||||
|
assert "--days" in gws_args
|
||||||
|
|
||||||
|
|
||||||
|
def test_api_calendar_list_respects_date_range(api_module):
|
||||||
|
"""calendar list with --start/--end uses raw events list API."""
|
||||||
|
captured = {}
|
||||||
|
|
||||||
|
def capture_run(cmd, **kwargs):
|
||||||
|
captured["cmd"] = cmd
|
||||||
|
return MagicMock(returncode=0)
|
||||||
|
|
||||||
|
args = api_module.argparse.Namespace(
|
||||||
|
start="2026-04-01T00:00:00Z",
|
||||||
|
end="2026-04-07T23:59:59Z",
|
||||||
|
max=25,
|
||||||
|
calendar="primary",
|
||||||
|
func=api_module.calendar_list,
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(subprocess, "run", side_effect=capture_run):
|
||||||
|
with pytest.raises(SystemExit):
|
||||||
|
api_module.calendar_list(args)
|
||||||
|
|
||||||
|
gws_args = captured["cmd"][2:]
|
||||||
|
assert "events" in gws_args
|
||||||
|
assert "list" in gws_args
|
||||||
|
params_idx = gws_args.index("--params")
|
||||||
|
params = json.loads(gws_args[params_idx + 1])
|
||||||
|
assert params["timeMin"] == "2026-04-01T00:00:00Z"
|
||||||
|
assert params["timeMax"] == "2026-04-07T23:59:59Z"
|
||||||
|
|||||||
Reference in New Issue
Block a user