Files
last30days-skill/tests/test_setup_openclaw.py
T
Trevin Chow eb2d7a0f37 fix(ci): run full pytest suite, repair 13 rotted tests
CI was running only test_plugin_contract.py and test_version_consistency.py
(2 of 84 test files), masking 13 rotted tests across 4 clusters. The suite is
fully offline-safe (1402 tests in ~7s without network), so the narrow scope
wasn't gating integration flakiness; it was just stale. validate.yml now runs
`uv run pytest` against the full suite.

Engine fix: store.findings_from_report is rerank-first. ranked_candidates is
the primary persistence path; hackernews/polymarket are unconditionally
supplemented from items_by_source because they rank poorly but matter for
watchlists. When ranked_candidates was empty (rerank failed or skipped),
reddit, x, and every other source were silently dropped. The supplement loop
now falls back to all sources only when ranked_candidates is empty; the normal
path is unchanged.

Test repairs:
- test_store.py (6) + test_watchlist_commands.py (2): cascade from the engine fix
- test_get_new_findings_filters_by_date (latent): local-time vs SQLite UTC
  flake — switched to datetime.now(timezone.utc)
- TestPollDeviceAuth (3): mock_time.time side_effect lists too short after
  impl added a last_reminder call — padded timeout test, pinned others to
  return_value=0 (loops terminate via urlopen, not the clock)
- test_bare_run_emits_web_promo: engine reads ~/.config/last30days/.env, so
  a contributor's saved EXA/PARALLEL key made grounding "available" and
  suppressed the web promo. Also missing X made the "x" promo preempt "web".
  Set LAST30DAYS_CONFIG_DIR="", subprocess cwd=tmpdir, XAI_API_KEY stub.
2026-05-16 21:34:22 -07:00

567 lines
21 KiB
Python

"""Tests for OpenClaw setup and device auth functions."""
import json
import sys
import time
from pathlib import Path
from unittest.mock import patch, MagicMock, call
import pytest
# Add scripts dir to path
SCRIPTS_DIR = Path(__file__).parent.parent / "skills" / "last30days" / "scripts"
sys.path.insert(0, str(SCRIPTS_DIR))
from lib import setup_wizard
class TestRunOpenclawSetup:
"""Tests for run_openclaw_setup()."""
@patch("shutil.which")
def test_all_tools_present_no_keys(self, mock_which):
"""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
@patch("shutil.which")
def test_missing_tools(self, mock_which):
"""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("shutil.which")
def test_keys_detected(self, mock_which):
"""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
@patch("shutil.which")
def test_x_method_xai(self, mock_which):
"""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("shutil.which")
def test_x_method_cookies(self, mock_which):
"""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("shutil.which")
def test_x_method_xai_over_cookies(self, mock_which):
"""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("shutil.which")
def test_x_method_null_when_nothing(self, mock_which):
"""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("shutil.which")
def test_output_is_json_serializable(self, mock_which):
"""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
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 calls time.time() for deadline init, last_reminder init,
# then once per while-loop iteration. Three values are enough for one check
# that exceeds the deadline.
mock_time.time = MagicMock(side_effect=[0, 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 TestAuthWithPat:
"""Tests for auth_with_pat()."""
@patch("lib.setup_wizard.urlopen")
def test_success(self, mock_urlopen):
"""Valid PAT -> returns dict with api_key."""
resp = MagicMock()
resp.read.return_value = json.dumps({
"api_key": "sc_live_test123",
"github_username": "testuser",
"credits_remaining": 100,
}).encode()
resp.__enter__ = lambda s: s
resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = resp
result = setup_wizard.auth_with_pat("gho_validtoken")
assert result is not None
assert result["api_key"] == "sc_live_test123"
assert result["github_username"] == "testuser"
@patch("lib.setup_wizard.urlopen")
def test_invalid_token_returns_none(self, mock_urlopen):
"""HTTP 401 (invalid token) -> returns None."""
from urllib.error import HTTPError
mock_urlopen.side_effect = HTTPError(
url="https://api.scrapecreators.com/v1/github/pat/auth",
code=401, msg="Unauthorized", hdrs={}, fp=None,
)
result = setup_wizard.auth_with_pat("gho_badtoken")
assert result is None
@patch("lib.setup_wizard.urlopen")
def test_insufficient_scope_returns_none(self, mock_urlopen):
"""HTTP 422 (insufficient scope) -> returns None."""
from urllib.error import HTTPError
mock_urlopen.side_effect = HTTPError(
url="https://api.scrapecreators.com/v1/github/pat/auth",
code=422, msg="Unprocessable", hdrs={}, fp=None,
)
result = setup_wizard.auth_with_pat("gho_noscope")
assert result is None
@patch("lib.setup_wizard.urlopen")
def test_network_error_returns_none(self, mock_urlopen):
"""URLError -> returns None."""
from urllib.error import URLError
mock_urlopen.side_effect = URLError("Connection refused")
result = setup_wizard.auth_with_pat("gho_anytoken")
assert result is None
@patch("lib.setup_wizard.urlopen")
def test_no_api_key_in_response(self, mock_urlopen):
"""Response without api_key -> returns None."""
resp = MagicMock()
resp.read.return_value = json.dumps({"error": "something"}).encode()
resp.__enter__ = lambda s: s
resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = resp
result = setup_wizard.auth_with_pat("gho_validtoken")
assert result is None
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() — PAT-first with device fallback."""
@patch("lib.setup_wizard.auth_with_pat")
@patch("subprocess.run")
@patch("shutil.which", return_value="/usr/local/bin/gh")
def test_pat_success(self, mock_which, mock_subproc, mock_pat):
"""gh found + valid token + PAT endpoint success -> pat method."""
mock_subproc.return_value = MagicMock(
returncode=0, stdout="gho_testtoken123\n",
)
mock_pat.return_value = {
"api_key": "sc_live_fromPAT",
"github_username": "testuser",
}
result = setup_wizard.run_github_auth(timeout=10)
assert result["status"] == "success"
assert result["method"] == "pat"
assert result["api_key"] == "sc_live_fromPAT"
@patch("lib.setup_wizard.run_full_device_auth")
@patch("lib.setup_wizard.auth_with_pat", return_value=None)
@patch("subprocess.run")
@patch("shutil.which", return_value="/usr/local/bin/gh")
def test_pat_fails_falls_to_device(self, mock_which, mock_subproc, mock_pat, mock_device):
"""gh found + PAT endpoint fails -> falls through to device flow."""
mock_subproc.return_value = MagicMock(
returncode=0, stdout="gho_badtoken\n",
)
mock_device.return_value = {
"status": "success", "method": "device",
"api_key": "sc_live_fromDevice",
}
result = setup_wizard.run_github_auth(timeout=10)
assert result["status"] == "success"
assert result["method"] == "device"
mock_device.assert_called_once()
@patch("lib.setup_wizard.run_full_device_auth")
@patch("shutil.which", return_value=None)
def test_no_gh_goes_to_device(self, mock_which, mock_device):
"""gh not installed -> straight to device flow."""
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"
@patch("lib.setup_wizard.run_full_device_auth")
@patch("subprocess.run")
@patch("shutil.which", return_value="/usr/local/bin/gh")
def test_gh_not_logged_in_falls_to_device(self, mock_which, mock_subproc, mock_device):
"""gh exists but not logged in (exit code 1) -> device flow."""
mock_subproc.return_value = MagicMock(
returncode=1, stdout="", stderr="not logged in",
)
mock_device.return_value = {
"status": "success", "method": "device",
"api_key": "sc_live_fallback",
}
result = setup_wizard.run_github_auth(timeout=10)
assert result["status"] == "success"
assert result["method"] == "device"