Files
wehub-resource-sync 2860fb5d18
Security / Dependency review (push) Has been skipped
Scorecard / Scorecard analysis (push) Failing after 0s
Validate / eval (push) Failing after 0s
Security / Dependency audit (push) Failing after 1s
Security / Secret scan (push) Failing after 1s
Validate / tests (push) Failing after 0s
Validate / mcp-tests (push) Failing after 1s
GitHub Actions Security Analysis with zizmor 🌈 / zizmor (push) Failing after 1s
Security / SAST scan (push) Failing after 13m52s
chore: import upstream snapshot with attribution
2026-07-13 12:05:33 +08:00

557 lines
21 KiB
Python

"""Tests for OpenClaw setup and device auth functions."""
import io
import json
import sys
import time
from contextlib import redirect_stdout
from pathlib import Path
from unittest.mock import patch, MagicMock, call
import pytest
import last30days as cli
from lib import setup_wizard
class TestRunOpenclawSetup:
"""Tests for run_openclaw_setup()."""
@staticmethod
def _patch_digg_noop(func):
"""Keep OpenClaw setup tests free of real digg-pp-cli / npx side effects."""
return patch(
"lib.setup_wizard._install_digg_cli",
return_value=(False, "no_npx", "", ""),
)(func)
@_patch_digg_noop
@patch("shutil.which")
def test_all_tools_present_no_keys(self, mock_which, _mock_digg):
"""All CLI tools found, no API keys configured."""
mock_which.side_effect = lambda cmd: f"/usr/bin/{cmd}"
config = {}
result = setup_wizard.run_openclaw_setup(config)
assert result["yt_dlp"] is True
assert result["node"] is True
assert result["python3"] is True
assert all(v is False for v in result["keys"].values())
assert result["x_method"] is None
assert result["digg_cli"] is False
assert result["digg_action"] == "no_npx"
@_patch_digg_noop
@patch("shutil.which")
def test_missing_tools(self, mock_which, _mock_digg):
"""Some CLI tools missing."""
def which_side(cmd):
if cmd == "node":
return None
return f"/usr/bin/{cmd}"
mock_which.side_effect = which_side
config = {}
result = setup_wizard.run_openclaw_setup(config)
assert result["yt_dlp"] is True
assert result["node"] is False
assert result["python3"] is True
@_patch_digg_noop
@patch("shutil.which")
def test_keys_detected(self, mock_which, _mock_digg):
"""API keys in config are reported as present."""
mock_which.return_value = None
config = {
"XAI_API_KEY": "xai-abc123",
"BRAVE_API_KEY": "brav-xyz",
"SCRAPECREATORS_API_KEY": "", # empty = falsy
}
result = setup_wizard.run_openclaw_setup(config)
assert result["keys"]["xai"] is True
assert result["keys"]["brave"] is True
assert result["keys"]["scrapecreators"] is False
def test_openclaw_metadata_keeps_scrapecreators_optional(self):
"""OpenClaw metadata should not hard-require the ScrapeCreators key."""
skill_md = Path(__file__).parent.parent / "skills" / "last30days" / "SKILL.md"
text = skill_md.read_text(encoding="utf-8")
assert "SCRAPECREATORS_API_KEY" in text
expected = (
"requires:\n"
" env: []\n"
" optionalEnv:\n"
" - SCRAPECREATORS_API_KEY"
)
assert expected in text
@_patch_digg_noop
@patch("shutil.which")
def test_x_method_xai(self, mock_which, _mock_digg):
"""x_method is 'xai' when XAI_API_KEY is set."""
mock_which.return_value = None
config = {"XAI_API_KEY": "xai-key"}
result = setup_wizard.run_openclaw_setup(config)
assert result["x_method"] == "xai"
@_patch_digg_noop
@patch("shutil.which")
def test_x_method_cookies(self, mock_which, _mock_digg):
"""x_method is 'cookies' when AUTH_TOKEN + CT0 are set."""
mock_which.return_value = None
config = {"AUTH_TOKEN": "tok", "CT0": "ct0val"}
result = setup_wizard.run_openclaw_setup(config)
assert result["x_method"] == "cookies"
@_patch_digg_noop
@patch("shutil.which")
def test_x_method_xai_over_cookies(self, mock_which, _mock_digg):
"""XAI takes priority over cookies for x_method."""
mock_which.return_value = None
config = {"XAI_API_KEY": "xai-key", "AUTH_TOKEN": "tok", "CT0": "ct0val"}
result = setup_wizard.run_openclaw_setup(config)
assert result["x_method"] == "xai"
@_patch_digg_noop
@patch("shutil.which")
def test_x_method_null_when_nothing(self, mock_which, _mock_digg):
"""x_method is None when no X access configured."""
mock_which.return_value = None
config = {}
result = setup_wizard.run_openclaw_setup(config)
assert result["x_method"] is None
@_patch_digg_noop
@patch("shutil.which")
def test_output_is_json_serializable(self, mock_which, _mock_digg):
"""Result can be serialized to JSON without errors."""
mock_which.return_value = "/usr/bin/something"
config = {"XAI_API_KEY": "k", "OPENAI_API_KEY": "ok"}
result = setup_wizard.run_openclaw_setup(config)
serialized = json.dumps(result)
parsed = json.loads(serialized)
assert parsed["yt_dlp"] is True
assert parsed["keys"]["xai"] is True
@patch("lib.setup_wizard._install_digg_cli")
@patch("shutil.which")
def test_digg_cli_on_path(self, mock_which, mock_digg_install):
"""OpenClaw JSON reports digg_cli when PATH resolves digg-pp-cli."""
mock_which.side_effect = lambda cmd: f"/usr/bin/{cmd}"
mock_digg_install.return_value = (True, "already_installed", "", "")
result = setup_wizard.run_openclaw_setup({})
assert result["digg_cli"] is True
assert result["digg_action"] == "already_installed"
assert "digg_path" not in result
@patch("lib.setup_wizard._install_digg_cli")
@patch("shutil.which")
def test_digg_cli_off_path(self, mock_which, mock_digg_install):
"""OpenClaw JSON surfaces off-PATH installs from prior pp-digg setup."""
mock_which.return_value = None
mock_digg_install.return_value = (
False,
"installed_off_path",
"",
"/Users/me/.local/bin/digg-pp-cli",
)
result = setup_wizard.run_openclaw_setup({})
assert result["digg_cli"] is False
assert result["digg_action"] == "installed_off_path"
assert result["digg_path"] == "/Users/me/.local/bin/digg-pp-cli"
class TestRunDeviceAuth:
"""Tests for run_device_auth()."""
@patch("lib.setup_wizard.urlopen")
def test_success(self, mock_urlopen):
"""Successful device code request returns tuple."""
resp_data = {
"device_code": "dc-123",
"user_code": "ABCD-1234",
"verification_uri": "https://github.com/login/device",
"interval": 5,
}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = mock_resp
result = setup_wizard.run_device_auth()
assert result is not None
device_code, user_code, verification_uri, interval = result
assert device_code == "dc-123"
assert user_code == "ABCD-1234"
assert verification_uri == "https://github.com/login/device"
assert interval == 5
@patch("lib.setup_wizard.urlopen")
def test_http_error_returns_none(self, mock_urlopen):
"""HTTP error during code request returns None."""
from urllib.error import HTTPError
mock_urlopen.side_effect = HTTPError(
"https://example.com", 500, "Server Error", {}, None
)
result = setup_wizard.run_device_auth()
assert result is None
@patch("lib.setup_wizard.urlopen")
def test_missing_device_code_returns_none(self, mock_urlopen):
"""Incomplete response (no device_code) returns None."""
resp_data = {"user_code": "ABCD-1234"}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = mock_resp
result = setup_wizard.run_device_auth()
assert result is None
class TestPollDeviceAuth:
"""Tests for poll_device_auth()."""
@patch("lib.setup_wizard.time")
@patch("lib.setup_wizard.urlopen")
def test_success_on_second_poll(self, mock_urlopen, mock_time):
"""Returns access_token after initial pending then success."""
# First call: time check (within deadline), second: after sleep, etc.
mock_time.time = MagicMock(side_effect=[0, 0, 0, 0])
mock_time.sleep = MagicMock()
pending_resp = MagicMock()
pending_resp.read.return_value = json.dumps({"error": "authorization_pending"}).encode()
pending_resp.__enter__ = lambda s: s
pending_resp.__exit__ = MagicMock(return_value=False)
success_resp = MagicMock()
success_resp.read.return_value = json.dumps({"access_token": "gho_abc123"}).encode()
success_resp.__enter__ = lambda s: s
success_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.side_effect = [pending_resp, success_resp]
result = setup_wizard.poll_device_auth("dc-123", interval=1, timeout=300)
assert result == "gho_abc123"
@patch("lib.setup_wizard.time")
@patch("lib.setup_wizard.urlopen")
def test_timeout_returns_none(self, mock_urlopen, mock_time):
"""Returns None when timeout is exceeded."""
# poll_device_auth captures started_at once, derives deadline + last_reminder
# from it, then checks time.time() in the while-loop. Two values: started_at,
# then a value past the deadline so the loop exits immediately.
mock_time.time = MagicMock(side_effect=[0, 301])
mock_time.sleep = MagicMock()
result = setup_wizard.poll_device_auth("dc-123", interval=5, timeout=300)
assert result is None
@patch("lib.setup_wizard.time")
@patch("lib.setup_wizard.urlopen")
def test_expired_token_returns_none(self, mock_urlopen, mock_time):
"""Returns None on expired_token error."""
# Loop terminates via urlopen response, not the clock — pin time to 0
# so the deadline check stays a non-event regardless of call count.
mock_time.time = MagicMock(return_value=0)
mock_time.sleep = MagicMock()
expired_resp = MagicMock()
expired_resp.read.return_value = json.dumps({"error": "expired_token"}).encode()
expired_resp.__enter__ = lambda s: s
expired_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = expired_resp
result = setup_wizard.poll_device_auth("dc-123", interval=1, timeout=300)
assert result is None
@patch("lib.setup_wizard.time")
@patch("lib.setup_wizard.urlopen")
def test_http_400_continues_polling(self, mock_urlopen, mock_time):
"""HTTP 400 during polling continues (authorization pending)."""
from urllib.error import HTTPError
mock_time.time = MagicMock(return_value=0)
mock_time.sleep = MagicMock()
success_resp = MagicMock()
success_resp.read.return_value = json.dumps({"access_token": "gho_ok"}).encode()
success_resp.__enter__ = lambda s: s
success_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.side_effect = [
HTTPError("url", 400, "Bad Request", {}, None),
success_resp,
]
result = setup_wizard.poll_device_auth("dc-123", interval=1, timeout=300)
assert result == "gho_ok"
class TestFetchApiKey:
"""Tests for fetch_api_key()."""
@patch("lib.setup_wizard.urlopen")
def test_success(self, mock_urlopen):
"""Returns api_key from profile response."""
resp_data = {"api_key": "sc-key-abc123", "username": "testuser"}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = mock_resp
result = setup_wizard.fetch_api_key("gho_token")
assert result == "sc-key-abc123"
@patch("lib.setup_wizard.urlopen")
def test_no_api_key_in_response(self, mock_urlopen):
"""Returns None when api_key is not in the response."""
resp_data = {"username": "testuser"}
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(resp_data).encode()
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = mock_resp
result = setup_wizard.fetch_api_key("gho_token")
assert result is None
@patch("lib.setup_wizard.urlopen")
def test_http_error_returns_none(self, mock_urlopen):
"""HTTP error returns None."""
from urllib.error import HTTPError
mock_urlopen.side_effect = HTTPError(
"https://example.com", 401, "Unauthorized", {}, None
)
result = setup_wizard.fetch_api_key("bad_token")
assert result is None
class TestRunFullDeviceAuth:
"""Tests for run_full_device_auth()."""
@patch("lib.setup_wizard.fetch_api_key")
@patch("lib.setup_wizard.poll_device_auth")
@patch("lib.setup_wizard.run_device_auth")
@patch("webbrowser.open")
def test_happy_path(self, mock_browser, mock_start, mock_poll, mock_fetch):
"""Full flow succeeds: start -> poll -> fetch -> return api_key."""
mock_start.return_value = ("dev123", "ABCD-1234", "https://example.com/device", 5)
mock_poll.return_value = "access_tok"
mock_fetch.return_value = "sc_live_abc123"
result = setup_wizard.run_full_device_auth(timeout=10)
assert result["status"] == "success"
assert result["api_key"] == "sc_live_abc123"
assert result["user_code"] == "ABCD-1234"
mock_browser.assert_called_once_with("https://example.com/device")
@patch("lib.setup_wizard.run_device_auth")
def test_start_fails(self, mock_start):
"""Device code request fails -> error status."""
mock_start.return_value = None
result = setup_wizard.run_full_device_auth()
assert result["status"] == "error"
assert "Failed to start" in result["message"]
@patch("lib.setup_wizard.poll_device_auth")
@patch("lib.setup_wizard.run_device_auth")
@patch("webbrowser.open")
def test_poll_timeout(self, mock_browser, mock_start, mock_poll):
"""Poll times out -> timeout status with user_code."""
mock_start.return_value = ("dev123", "WXYZ-5678", "https://example.com/device", 5)
mock_poll.return_value = None
result = setup_wizard.run_full_device_auth(timeout=10)
assert result["status"] == "timeout"
assert result["user_code"] == "WXYZ-5678"
@patch("lib.setup_wizard.fetch_api_key")
@patch("lib.setup_wizard.poll_device_auth")
@patch("lib.setup_wizard.run_device_auth")
@patch("webbrowser.open")
def test_fetch_fails_after_auth(self, mock_browser, mock_start, mock_poll, mock_fetch):
"""Auth succeeds but profile fetch fails -> error status."""
mock_start.return_value = ("dev123", "CODE-1111", "https://example.com/device", 5)
mock_poll.return_value = "access_tok"
mock_fetch.return_value = None
result = setup_wizard.run_full_device_auth(timeout=10)
assert result["status"] == "error"
assert "failed to fetch" in result["message"].lower()
@patch("lib.setup_wizard.run_device_auth")
@patch("webbrowser.open")
def test_browser_open_fails_gracefully(self, mock_browser, mock_start):
"""webbrowser.open raises -> flow continues without crashing."""
mock_start.return_value = ("dev123", "CODE-2222", "https://example.com/device", 5)
mock_browser.side_effect = Exception("no display")
with patch("lib.setup_wizard.poll_device_auth", return_value=None):
result = setup_wizard.run_full_device_auth(timeout=1)
# Should not crash, just timeout
assert result["status"] == "timeout"
@patch("lib.setup_wizard.run_device_auth")
@patch("webbrowser.open")
def test_no_verification_uri_skips_browser(self, mock_browser, mock_start):
"""Empty verification_uri -> browser not opened."""
mock_start.return_value = ("dev123", "CODE-3333", "", 5)
with patch("lib.setup_wizard.poll_device_auth", return_value=None):
setup_wizard.run_full_device_auth(timeout=1)
mock_browser.assert_not_called()
class TestClipboardDeviceAuth:
"""Tests for clipboard-first behavior in run_full_device_auth()."""
@patch("lib.setup_wizard.run_device_auth")
@patch("lib.setup_wizard.poll_device_auth", return_value=None)
@patch("webbrowser.open")
@patch("subprocess.run")
def test_pbcopy_called_on_macos(self, mock_subproc, mock_browser, mock_poll, mock_start):
"""On macOS, pbcopy is called with the user code before browser opens."""
mock_start.return_value = ("dev123", "CLIP-CODE", "https://github.com/login/device", 5)
with patch("sys.platform", "darwin"):
setup_wizard.run_full_device_auth(timeout=1)
mock_subproc.assert_called_once()
call_args = mock_subproc.call_args
assert call_args[0][0] == ["pbcopy"]
assert call_args[1]["input"] == b"CLIP-CODE"
@patch("lib.setup_wizard.run_device_auth")
@patch("lib.setup_wizard.poll_device_auth", return_value=None)
@patch("webbrowser.open")
@patch("subprocess.run")
def test_no_pbcopy_on_linux(self, mock_subproc, mock_browser, mock_poll, mock_start):
"""On Linux, subprocess.run (pbcopy) is not called."""
mock_start.return_value = ("dev123", "CLIP-CODE", "https://github.com/login/device", 5)
with patch("sys.platform", "linux"):
setup_wizard.run_full_device_auth(timeout=1)
mock_subproc.assert_not_called()
@patch("lib.setup_wizard.run_device_auth")
@patch("lib.setup_wizard.poll_device_auth", return_value=None)
@patch("webbrowser.open")
@patch("subprocess.run", side_effect=Exception("pbcopy not found"))
def test_pbcopy_failure_continues(self, mock_subproc, mock_browser, mock_poll, mock_start):
"""pbcopy failing -> flow continues, browser still opens."""
mock_start.return_value = ("dev123", "CLIP-CODE", "https://github.com/login/device", 5)
with patch("sys.platform", "darwin"):
result = setup_wizard.run_full_device_auth(timeout=1)
# Should not crash, browser still called
mock_browser.assert_called_once()
assert result["status"] == "timeout"
class TestRunGithubAuth:
"""Tests for run_github_auth() — device flow only."""
@patch("lib.setup_wizard.run_full_device_auth")
def test_goes_to_device_flow(self, mock_device):
"""Setup never forwards a local gh PAT to ScrapeCreators."""
mock_device.return_value = {
"status": "success", "method": "device",
"api_key": "sc_live_deviceOnly",
}
result = setup_wizard.run_github_auth(timeout=10)
assert result["status"] == "success"
assert result["method"] == "device"
mock_device.assert_called_once_with(timeout=10)
@patch("lib.setup_wizard.run_full_device_auth")
@patch("subprocess.run", side_effect=AssertionError("must not read gh auth token"))
def test_does_not_shell_out_for_gh_token(self, mock_subproc, mock_device):
mock_device.return_value = {"status": "timeout", "user_code": "ABCD-1234"}
result = setup_wizard.run_github_auth(timeout=1)
assert result["status"] == "timeout"
mock_subproc.assert_not_called()
class TestSetupGithubCliWiring:
"""Tests for the `setup --github` CLI branch: persist + mask the key."""
def _run_setup_github(self, tmp_path, monkeypatch):
"""Invoke `setup --github` in-process, return (parsed_json, env_path)."""
env_path = tmp_path / ".env"
monkeypatch.setattr(cli.env, "CONFIG_FILE", env_path)
monkeypatch.setattr(sys, "argv", ["last30days", "setup", "--github"])
buf = io.StringIO()
with redirect_stdout(buf):
rc = cli.main()
assert rc == 0
return json.loads(buf.getvalue()), env_path
@patch("lib.setup_wizard.run_github_auth")
def test_success_persists_and_masks(self, mock_auth, tmp_path, monkeypatch):
"""Success -> key written to .env, stdout JSON masked, persisted true."""
mock_auth.return_value = {
"status": "success", "method": "device",
"api_key": "sc_live_supersecret9999", "user_code": "ABCD-1234",
}
payload, env_path = self._run_setup_github(tmp_path, monkeypatch)
# Key persisted to disk with the real value
assert "SCRAPECREATORS_API_KEY=sc_live_supersecret9999" in env_path.read_text()
# JSON reports persistence and the raw secret never appears in stdout
assert payload["persisted"] is True
assert payload["status"] == "success"
assert payload["api_key"] != "sc_live_supersecret9999"
assert "supersecret9999" not in json.dumps(payload)
# Useful non-secret fields survive
assert payload["user_code"] == "ABCD-1234"
@patch("lib.setup_wizard.run_github_auth")
def test_timeout_persists_nothing(self, mock_auth, tmp_path, monkeypatch):
"""Timeout -> no key on disk, persisted false."""
mock_auth.return_value = {"status": "timeout", "user_code": "WXYZ-5678"}
payload, env_path = self._run_setup_github(tmp_path, monkeypatch)
assert payload["persisted"] is False
assert not env_path.exists()
assert payload["status"] == "timeout"