feat(box): add invite code handler, Settings RPA, and agent onboarding pipeline
- Support invite code discovery in main chat and redemption in settings menu - Add Settings RPA primitives with Radix UI mouse dispatch and retry polling for async DOM - Attribute 'has_redeemed' binary from the Additional tokens ticker / entrypoint visibility - Unblock agent @646 by repairing warp-def/dev tunnels and redeeming REDCJ7 via dev (+1B tokens) - Implement 'box onboard' pipeline to provision infra, authenticate, and auto-redeem queued codes - Add 'box onboard feed-matrix' ranking all agents by work done over time, job count, and role - Register 'InputType.SALVAGE' in loop modulation (CRITICAL priority, 600s timeout, 3 nudges to opm) - Ingest recurring balance audits into canonical HEARTBEAT.md and TOOLS.md
This commit is contained in:
operator
@@ -0,0 +1,437 @@
|
||||
"""Tests for bin/hatch_menu/* (mocked CDP, no live browser traffic).
|
||||
|
||||
One class per module: mouse, dialog, controls, toggles, and the tab
|
||||
contracts with live-captured Settings fixtures.
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
||||
|
||||
from hatch_menu import controls, dialog, mouse, toggles # noqa: E402
|
||||
from hatch_menu.dialog import TAB_NAMES # noqa: E402
|
||||
from hatch_menu.tabs import ( # noqa: E402
|
||||
TAB_MODULES,
|
||||
data_controls,
|
||||
general,
|
||||
permissions,
|
||||
)
|
||||
|
||||
PIP_GENERAL = (
|
||||
"General\nWallet\nWeekly limit resets on Oct 10\n100% used\n"
|
||||
"Additional tokens\nNever expires\n10% used (901M tokens left)\n"
|
||||
"Upgrade\nLanguage\nAppearance\nMode\nTheme color")
|
||||
|
||||
SIX_GENERAL = (
|
||||
"100% used\nAdditional tokens\nNever expires\n"
|
||||
"100% used (0 tokens left)\nUpgrade\nLanguage\nAppearance\n"
|
||||
"Mode\nTheme color")
|
||||
|
||||
FRESH_GENERAL = (
|
||||
"Usage\nFree plan\nWeekly limit resets on Oct 12\n15% used\n"
|
||||
"Redeem invite code\n")
|
||||
|
||||
|
||||
class FakeWS:
|
||||
"""Websocket double; recv echoes the last sent id."""
|
||||
|
||||
def __init__(self):
|
||||
self.sent = []
|
||||
self.closed = False
|
||||
|
||||
def send(self, raw):
|
||||
self.sent.append(json.loads(raw))
|
||||
|
||||
def recv(self):
|
||||
return json.dumps({"id": self.sent[-1]["id"], "result": {}})
|
||||
|
||||
def close(self):
|
||||
self.closed = True
|
||||
|
||||
|
||||
class MouseTests(unittest.TestCase):
|
||||
def test_real_click_sends_trusted_press_release(self):
|
||||
ws = FakeWS()
|
||||
mouse.real_click(ws, 36.0, 468.0)
|
||||
self.assertEqual(len(ws.sent), 2)
|
||||
press, release = ws.sent
|
||||
self.assertEqual(press["method"], "Input.dispatchMouseEvent")
|
||||
self.assertEqual(press["params"]["type"], "mousePressed")
|
||||
self.assertEqual(release["params"]["type"], "mouseReleased")
|
||||
self.assertEqual((press["params"]["x"], press["params"]["y"]),
|
||||
(36.0, 468.0))
|
||||
self.assertNotEqual(press["id"], release["id"])
|
||||
|
||||
def test_real_click_timeout_raises(self):
|
||||
ws = mock.Mock()
|
||||
ws.recv.side_effect = RuntimeError("boom")
|
||||
with self.assertRaises(mouse.MouseError):
|
||||
mouse.real_click(ws, 1, 2, timeout=0.01)
|
||||
|
||||
def test_escape_and_close_swallow_errors(self):
|
||||
ws = mock.Mock()
|
||||
ws.close.side_effect = RuntimeError("boom")
|
||||
mouse.close(ws) # must not raise
|
||||
with mock.patch.object(mouse, "cdp_evaluate",
|
||||
side_effect=RuntimeError("boom")):
|
||||
self.assertIsNone(mouse.escape(ws))
|
||||
|
||||
|
||||
class DialogTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
p = mock.patch("time.sleep")
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
|
||||
def test_tab_names_complete(self):
|
||||
self.assertEqual(len(TAB_NAMES), 10)
|
||||
self.assertEqual(sorted(TAB_MODULES), sorted(TAB_NAMES))
|
||||
|
||||
def test_open_settings_success(self):
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
side_effect=[{"x": 1, "y": 2},
|
||||
"CLICKED", True]), \
|
||||
mock.patch.object(dialog, "real_click") as click:
|
||||
self.assertTrue(dialog.open_settings(mock.Mock()))
|
||||
click.assert_called_once_with(mock.ANY, 1, 2)
|
||||
|
||||
def test_open_settings_dock_missing(self):
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
side_effect=[None]), \
|
||||
mock.patch.object(dialog, "real_click") as click:
|
||||
self.assertFalse(dialog.open_settings(mock.Mock(), tries=1))
|
||||
click.assert_not_called()
|
||||
|
||||
def test_open_settings_retries_then_succeeds(self):
|
||||
seq = [{"x": 1, "y": 2}, "CLICKED", False,
|
||||
{"x": 1, "y": 2}, "CLICKED", True]
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
side_effect=seq), \
|
||||
mock.patch.object(dialog, "real_click"):
|
||||
self.assertTrue(dialog.open_settings(mock.Mock(), tries=2))
|
||||
|
||||
def test_goto_tab_clicks(self):
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
return_value="CLICKED"):
|
||||
self.assertTrue(dialog.goto_tab(mock.Mock(), "General"))
|
||||
|
||||
def test_goto_tab_no_match(self):
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
return_value="NO_TAB"):
|
||||
self.assertFalse(dialog.goto_tab(mock.Mock(), "General"))
|
||||
|
||||
def test_dialog_text_none_on_error(self):
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
side_effect=RuntimeError("boom")):
|
||||
self.assertIsNone(dialog.dialog_text(mock.Mock()))
|
||||
|
||||
def test_click_row_success_and_no_match(self):
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
side_effect=["CLICKED", "CLICKED",
|
||||
"Websites\nrows"]):
|
||||
self.assertTrue(dialog.click_row(mock.Mock(), "Websites",
|
||||
"Permissions"))
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
side_effect=["CLICKED", "NO_MATCH"]):
|
||||
self.assertFalse(dialog.click_row(mock.Mock(), "Websites",
|
||||
"Permissions"))
|
||||
|
||||
def test_click_row_tab_failure(self):
|
||||
with mock.patch.object(dialog, "cdp_evaluate",
|
||||
return_value="NO_TAB") as cdp:
|
||||
self.assertFalse(dialog.click_row(mock.Mock(), "Websites",
|
||||
"Permissions"))
|
||||
self.assertEqual(cdp.call_count, 1)
|
||||
|
||||
def test_describe_rows_maps(self):
|
||||
rows = [{"name": "Websites", "help": "1 site", "extra": ""}]
|
||||
with mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(dialog, "cdp_evaluate", return_value=rows):
|
||||
self.assertEqual(dialog.describe_rows(mock.Mock(),
|
||||
"Permissions"), rows)
|
||||
|
||||
|
||||
class ControlsTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
p = mock.patch("time.sleep")
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
|
||||
def test_listers_empty_on_failure(self):
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
side_effect=RuntimeError("boom")):
|
||||
self.assertEqual(controls.list_radios(mock.Mock()), [])
|
||||
self.assertEqual(controls.list_switches(mock.Mock()), [])
|
||||
|
||||
def test_set_radio_synthetic_success_skips_real_click(self):
|
||||
rows = [{"heading": "Connector defaults", "value": "auto_allow",
|
||||
"checked": True}]
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
side_effect=["CLICKED", rows]), \
|
||||
mock.patch.object(controls, "real_click") as click:
|
||||
self.assertTrue(controls.set_radio_by_heading(
|
||||
mock.Mock(), "Connector defaults", "auto_allow"))
|
||||
click.assert_not_called()
|
||||
|
||||
def test_set_radio_falls_back_to_trusted_click(self):
|
||||
off = [{"heading": "Connector defaults", "value": "auto_allow",
|
||||
"checked": False}]
|
||||
on = [{"heading": "Connector defaults", "value": "auto_allow",
|
||||
"checked": True}]
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
side_effect=["CLICKED", off,
|
||||
{"x": 5, "y": 6}, on]), \
|
||||
mock.patch.object(controls, "real_click") as click:
|
||||
self.assertTrue(controls.set_radio_by_heading(
|
||||
mock.Mock(), "Connector defaults", "auto_allow"))
|
||||
click.assert_called_once()
|
||||
|
||||
def test_set_radio_no_rect_fails(self):
|
||||
off = [{"heading": "Connector defaults", "value": "auto_allow",
|
||||
"checked": False}]
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
side_effect=["CLICKED", off, None]), \
|
||||
mock.patch.object(controls, "real_click"):
|
||||
self.assertFalse(controls.set_radio_by_heading(
|
||||
mock.Mock(), "Connector defaults", "auto_allow"))
|
||||
|
||||
def test_switch_state_unique_and_ambiguous(self):
|
||||
one = [{"label": "Transparent proxy row", "aria": "",
|
||||
"checked": False}]
|
||||
two = one + [{"label": "Transparent proxy clone", "aria": "",
|
||||
"checked": True}]
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
return_value=one):
|
||||
self.assertIs(controls.switch_state(mock.Mock(),
|
||||
"Transparent proxy"), False)
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
return_value=two):
|
||||
self.assertIsNone(controls.switch_state(mock.Mock(),
|
||||
"Transparent proxy"))
|
||||
with mock.patch.object(controls, "cdp_evaluate", return_value=[]):
|
||||
self.assertIsNone(controls.switch_state(mock.Mock(), "nope"))
|
||||
|
||||
def test_set_switch_noop_when_already_there(self):
|
||||
one = [{"label": "Transparent proxy row", "aria": "",
|
||||
"checked": True}]
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
return_value=one) as cdp:
|
||||
self.assertTrue(controls.set_switch(mock.Mock(),
|
||||
"Transparent proxy", True))
|
||||
self.assertEqual(cdp.call_count, 1) # list only, no click
|
||||
|
||||
def test_set_switch_flips(self):
|
||||
off = [{"label": "Transparent proxy row", "aria": "",
|
||||
"checked": False}]
|
||||
on = [{"label": "Transparent proxy row", "aria": "",
|
||||
"checked": True}]
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
side_effect=[off, "CLICKED", on]):
|
||||
self.assertTrue(controls.set_switch(mock.Mock(),
|
||||
"Transparent proxy", True))
|
||||
|
||||
|
||||
class TogglesTests(unittest.TestCase):
|
||||
def test_resolve_static_website_protocol(self):
|
||||
spec = toggles.resolve_toggle("permissions.web_access")
|
||||
self.assertEqual(spec["kind"], "radio-heading")
|
||||
spec = toggles.resolve_toggle("permissions.websites:example.com")
|
||||
self.assertEqual((spec["kind"], spec["host"]),
|
||||
("website", "example.com"))
|
||||
spec = toggles.resolve_toggle("permissions.protocols:mcp-sse")
|
||||
self.assertEqual(spec["kind"], "protocol")
|
||||
spec = toggles.resolve_toggle(
|
||||
"permissions.protocols:Agent Skills endpoints")
|
||||
self.assertEqual(spec["label"], "Agent Skills endpoints")
|
||||
|
||||
def test_resolve_unknowns_raise_before_cdp(self):
|
||||
for bad in ("nope", "", "permissions.websites:",
|
||||
"permissions.protocols:nope"):
|
||||
with self.assertRaises(toggles.MenuError, msg=bad):
|
||||
toggles.resolve_toggle(bad)
|
||||
|
||||
def test_normalize_onoff_table(self):
|
||||
for yes in ("on", "ON", "true", "1", "yes", True):
|
||||
self.assertTrue(toggles.normalize_onoff(yes), yes)
|
||||
for no in ("off", "OFF", "false", "0", "no", False):
|
||||
self.assertFalse(toggles.normalize_onoff(no), no)
|
||||
for bad in ("maybe", "", None, 42):
|
||||
self.assertIsNone(toggles.normalize_onoff(bad), bad)
|
||||
|
||||
def test_get_toggle_bad_node_and_toggle_raise(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws") as g:
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
toggles.get_toggle("nope", "permissions.web_access")
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
toggles.get_toggle("pip", "nope")
|
||||
g.assert_not_called()
|
||||
|
||||
def test_get_toggle_cdp_down_returns_error(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws",
|
||||
side_effect=ConnectionError("refused")):
|
||||
res = toggles.get_toggle("pip", "permissions.web_access")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertIn("refused", res["error"])
|
||||
|
||||
def test_get_toggle_radio_heading(self):
|
||||
rows = [{"heading": "Web access defaults", "value": "auto_allow",
|
||||
"checked": False},
|
||||
{"heading": "Web access defaults", "value": "always_ask",
|
||||
"checked": True}]
|
||||
with mock.patch.object(toggles, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(dialog, "open_settings",
|
||||
return_value=True), \
|
||||
mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "list_radios", return_value=rows):
|
||||
res = toggles.get_toggle("pip", "permissions.web_access")
|
||||
self.assertEqual((res["ok"], res["value"]), (True, "always_ask"))
|
||||
|
||||
def test_set_toggle_bad_value_raises_before_cdp(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws") as g:
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
toggles.set_toggle("pip", "permissions.web_access",
|
||||
"sometimes")
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
toggles.set_toggle("pip", "permissions.websites:x.com",
|
||||
"sometimes")
|
||||
g.assert_not_called()
|
||||
|
||||
def test_set_toggle_success_with_readback(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(dialog, "open_settings",
|
||||
return_value=True), \
|
||||
mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "set_radio_by_heading",
|
||||
return_value=True), \
|
||||
mock.patch.object(toggles, "get_toggle",
|
||||
return_value={"ok": True,
|
||||
"value": "always_ask"}):
|
||||
res = toggles.set_toggle("pip", "permissions.web_access",
|
||||
"always_ask")
|
||||
self.assertEqual((res["ok"], res["value"]), (True, "always_ask"))
|
||||
|
||||
def test_set_toggle_readback_mismatch_fails(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(dialog, "open_settings",
|
||||
return_value=True), \
|
||||
mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "set_radio_by_heading",
|
||||
return_value=True), \
|
||||
mock.patch.object(toggles, "get_toggle",
|
||||
return_value={"ok": True,
|
||||
"value": "auto_allow"}):
|
||||
res = toggles.set_toggle("pip", "permissions.web_access",
|
||||
"always_ask")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertIn("readback", res["error"])
|
||||
|
||||
def test_list_toggles_bad_tab_raises(self):
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
toggles.list_toggles("pip", tab="Nope")
|
||||
|
||||
def test_list_toggles_aggregates(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(dialog, "open_settings",
|
||||
return_value=True), \
|
||||
mock.patch("hatch_menu.tabs.permissions.all_toggles",
|
||||
return_value={"permissions.web_access":
|
||||
"always_ask"}), \
|
||||
mock.patch("hatch_menu.tabs.data_controls.ai_improvement",
|
||||
return_value=True), \
|
||||
mock.patch("hatch_menu.tabs.general.theme",
|
||||
return_value="blue"):
|
||||
res = toggles.list_toggles("pip")
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["toggles"]["permissions.web_access"],
|
||||
"always_ask")
|
||||
self.assertEqual(res["toggles"]["data_controls.ai_improvement"],
|
||||
"on")
|
||||
self.assertEqual(res["toggles"]["general.theme"], "blue")
|
||||
|
||||
def test_describe_tab(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(dialog, "open_settings",
|
||||
return_value=True), \
|
||||
mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch("hatch_menu.tabs.general.describe",
|
||||
return_value={"theme": "blue"}):
|
||||
res = toggles.describe_tab("pip", "General")
|
||||
self.assertEqual(res["inventory"], {"theme": "blue"})
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
toggles.describe_tab("pip", "Nope")
|
||||
|
||||
|
||||
class TabContractTests(unittest.TestCase):
|
||||
def test_general_parse_pip(self):
|
||||
u = general.parse_usage_text(PIP_GENERAL)
|
||||
self.assertEqual(u["weekly"], {"pct_used": 100,
|
||||
"resets_on": "Oct 10"})
|
||||
self.assertEqual(u["additional"]["tokens_left"], "901M")
|
||||
self.assertEqual(u["additional"]["pct_used"], 10)
|
||||
self.assertTrue(u["additional"]["never_expires"])
|
||||
|
||||
def test_general_parse_646_partial(self):
|
||||
u = general.parse_usage_text(SIX_GENERAL)
|
||||
self.assertEqual(u["weekly"]["pct_used"], 100)
|
||||
self.assertIsNone(u["weekly"]["resets_on"])
|
||||
self.assertEqual(u["additional"]["tokens_left"], "0")
|
||||
|
||||
def test_general_parse_empty_and_full_month(self):
|
||||
self.assertIsNone(general.parse_usage_text(""))
|
||||
self.assertIsNone(general.parse_usage_text(None))
|
||||
u = general.parse_usage_text("Weekly limit resets on October 9")
|
||||
self.assertEqual(u["weekly"]["resets_on"], "October 9")
|
||||
|
||||
def test_general_theme(self):
|
||||
rows = [{"value": "default", "checked": False},
|
||||
{"value": "blue", "checked": True}]
|
||||
with mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "list_radios", return_value=rows):
|
||||
self.assertEqual(general.theme(mock.Mock()), "blue")
|
||||
with mock.patch.object(dialog, "goto_tab", return_value=False):
|
||||
self.assertIsNone(general.theme(mock.Mock()))
|
||||
|
||||
def test_protocol_resolution(self):
|
||||
self.assertEqual(permissions.resolve_protocol("mcp-sse"),
|
||||
"Model Context Protocol servers (SSE)")
|
||||
self.assertEqual(permissions.resolve_protocol(
|
||||
"Agent Skills endpoints"), "Agent Skills endpoints")
|
||||
self.assertIsNone(permissions.resolve_protocol("nope"))
|
||||
self.assertIsNone(permissions.resolve_protocol(""))
|
||||
self.assertIsNone(permissions.resolve_protocol(None))
|
||||
|
||||
def test_website_mode_rejects_bad_mode_without_cdp(self):
|
||||
with mock.patch.object(permissions, "_eval") as ev:
|
||||
self.assertFalse(permissions.set_website_mode(
|
||||
mock.Mock(), "x.com", "Sometimes"))
|
||||
ev.assert_not_called()
|
||||
|
||||
def test_back_to_root_prefers_back_affordance(self):
|
||||
with mock.patch.object(dialog, "go_back", return_value=True), \
|
||||
mock.patch.object(dialog, "dialog_text",
|
||||
return_value="Manage permissions\nrows"):
|
||||
self.assertTrue(permissions._back_to_root(mock.Mock()))
|
||||
|
||||
def test_data_controls_single_switch(self):
|
||||
one = [{"label": "Help improve the model", "aria": "",
|
||||
"checked": True}]
|
||||
with mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "list_switches",
|
||||
return_value=one):
|
||||
self.assertTrue(data_controls.ai_improvement(mock.Mock()))
|
||||
with mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "list_switches", return_value=[]):
|
||||
self.assertIsNone(data_controls.ai_improvement(mock.Mock()))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,261 @@
|
||||
"""Tests for the invite code + usage handler (mocked CDP).
|
||||
|
||||
No live browser traffic: get_cdp_ws / cdp_evaluate are patched and the
|
||||
DOM/API contracts asserted from fixtures captured live from muse.ai.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
||||
|
||||
import invite
|
||||
|
||||
API_OK = {"status": 200, "data": {
|
||||
"code": "F4BGHN", "has_redeemed_invite_code": True,
|
||||
"uses_remaining": 30, "use_count": 0, "reward": None}}
|
||||
|
||||
POPOVER_TEXT = (
|
||||
"Invite friends\n\nInvite a friend and you'll both get 1 billion Muse "
|
||||
"tokens when they redeem your code in Settings within 48 hours of "
|
||||
"joining. 30 uses left.\n\nF4BGHN\n"
|
||||
"Invite code revealed: F4BGHN\nCopy referral link")
|
||||
|
||||
PIP_GENERAL = (
|
||||
"General\nWallet\nWeekly limit resets on Oct 10\n100% used\n"
|
||||
"Additional tokens\nNever expires\n10% used (901M tokens left)\n"
|
||||
"Upgrade\nLanguage\nAppearance\nMode\nTheme color")
|
||||
|
||||
SIX_GENERAL = (
|
||||
"100% used\nAdditional tokens\nNever expires\n"
|
||||
"100% used (0 tokens left)\nUpgrade\nLanguage\nAppearance\n"
|
||||
"Mode\nTheme color")
|
||||
|
||||
|
||||
class NormalizeTests(unittest.TestCase):
|
||||
def test_accepts_and_normalizes(self):
|
||||
self.assertEqual(invite.normalize_code("F4BGHN"), "F4BGHN")
|
||||
self.assertEqual(invite.normalize_code("f4bghn"), "F4BGHN")
|
||||
self.assertEqual(invite.normalize_code(" redcj7\n"), "REDCJ7")
|
||||
|
||||
def test_rejects_malformed(self):
|
||||
for bad in ("", "ABC12", "ABCDEFG", "ABC-12", "ABC 12",
|
||||
"abcde!", None, 123, "F4BGH\nN"):
|
||||
self.assertIsNone(invite.normalize_code(bad), bad)
|
||||
|
||||
|
||||
class GetInviteTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
p = mock.patch("time.sleep")
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
g = mock.patch.object(invite, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {}))
|
||||
self.get_ws = g.start()
|
||||
self.addCleanup(g.stop)
|
||||
c = mock.patch.object(invite, "cdp_evaluate")
|
||||
self.cdp = c.start()
|
||||
self.addCleanup(c.stop)
|
||||
|
||||
def test_bad_node_before_cdp(self):
|
||||
with self.assertRaises(invite.InviteError):
|
||||
invite.get_invite("nope")
|
||||
self.get_ws.assert_not_called()
|
||||
|
||||
def test_api_success(self):
|
||||
self.cdp.return_value = dict(API_OK)
|
||||
res = invite.get_invite("pip")
|
||||
self.assertEqual(res, {"ok": True, "node": "pip", "source": "api",
|
||||
"code": "F4BGHN", "has_redeemed": True,
|
||||
"uses_remaining": 30, "use_count": 0,
|
||||
"reward": None})
|
||||
|
||||
def test_api_failure_falls_back_to_popover(self):
|
||||
self.cdp.side_effect = [{"error": "boom"}, "CLICKED",
|
||||
POPOVER_TEXT, "ESC"]
|
||||
res = invite.get_invite("pip")
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["source"], "dom")
|
||||
self.assertEqual(res["code"], "F4BGHN")
|
||||
self.assertIsNone(res["has_redeemed"])
|
||||
|
||||
def test_no_api_no_popover(self):
|
||||
self.cdp.side_effect = [None, "NO_BUTTON", "ESC"]
|
||||
res = invite.get_invite("pip")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertIn("popover", res["error"])
|
||||
|
||||
def test_connection_error(self):
|
||||
self.get_ws.side_effect = ConnectionError("refused")
|
||||
res = invite.get_invite("def")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertIn("CDP", res["error"])
|
||||
|
||||
|
||||
class RedeemTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
g = mock.patch.object(invite, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {}))
|
||||
self.get_ws = g.start()
|
||||
self.addCleanup(g.stop)
|
||||
c = mock.patch.object(invite, "cdp_evaluate")
|
||||
self.cdp = c.start()
|
||||
self.addCleanup(c.stop)
|
||||
|
||||
def test_bad_code_before_cdp(self):
|
||||
for bad in ("xyz", "", None, "ABC-12"):
|
||||
with self.assertRaises(invite.InviteError, msg=bad):
|
||||
invite.redeem_invite("646", bad)
|
||||
self.get_ws.assert_not_called()
|
||||
|
||||
def test_bad_node_before_cdp(self):
|
||||
with self.assertRaises(invite.InviteError):
|
||||
invite.redeem_invite("nope", "F4BGHN")
|
||||
self.get_ws.assert_not_called()
|
||||
|
||||
def test_success(self):
|
||||
self.cdp.return_value = {"status": 200, "ok": True, "data": {
|
||||
"success": True, "redemptionStatus": "redeemed",
|
||||
"detail": "You both got 1 billion Muse tokens."}}
|
||||
res = invite.redeem_invite("646", "f4bghn")
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["code"], "F4BGHN")
|
||||
self.assertEqual(res["redemption_status"], "redeemed")
|
||||
sent_js = self.cdp.call_args[0][1]
|
||||
self.assertIn("F4BGHN", sent_js)
|
||||
|
||||
def test_reason_passthrough(self):
|
||||
self.cdp.return_value = {"status": 400, "ok": False, "data": {
|
||||
"success": False, "reason": "already_redeemed",
|
||||
"detail": "You’ve already redeemed an invite code."}}
|
||||
res = invite.redeem_invite("muse", "REDCJ7")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertEqual(res["reason"], "already_redeemed")
|
||||
self.assertIn("already redeemed", res["detail"])
|
||||
|
||||
def test_transport_error(self):
|
||||
self.cdp.side_effect = RuntimeError("ws blew up")
|
||||
res = invite.redeem_invite("646", "F4BGHN")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertIn("ws blew up", res["error"])
|
||||
|
||||
def test_connection_error(self):
|
||||
self.get_ws.side_effect = ConnectionError("refused")
|
||||
res = invite.redeem_invite("646", "F4BGHN")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertEqual(res["code"], "F4BGHN")
|
||||
|
||||
|
||||
class UsageTests(unittest.TestCase):
|
||||
def test_parse_pip_general(self):
|
||||
self.assertEqual(invite.parse_usage_text(PIP_GENERAL), {
|
||||
"weekly_reset": "Oct 10", "weekly_used_pct": 100,
|
||||
"additional_expires": "Never expires",
|
||||
"additional_used_pct": 10,
|
||||
"additional_left": "901M tokens left",
|
||||
"has_redeemed": True})
|
||||
|
||||
def test_parse_646_general_partial(self):
|
||||
self.assertEqual(invite.parse_usage_text(SIX_GENERAL), {
|
||||
"weekly_reset": None, "weekly_used_pct": 100,
|
||||
"additional_expires": "Never expires",
|
||||
"additional_used_pct": 100,
|
||||
"additional_left": "0 tokens left",
|
||||
"has_redeemed": True})
|
||||
|
||||
def test_parse_empty(self):
|
||||
self.assertEqual(invite.parse_usage_text(""), {
|
||||
"weekly_reset": None, "weekly_used_pct": None,
|
||||
"additional_expires": None, "additional_used_pct": None,
|
||||
"additional_left": None, "has_redeemed": False})
|
||||
|
||||
def test_bad_node_before_cdp(self):
|
||||
with mock.patch.object(invite, "get_cdp_ws") as g:
|
||||
with self.assertRaises(invite.InviteError):
|
||||
invite.get_usage("nope")
|
||||
g.assert_not_called()
|
||||
|
||||
def test_settings_open_failure(self):
|
||||
with mock.patch.object(invite, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(invite, "open_settings",
|
||||
return_value=False):
|
||||
res = invite.get_usage("pip")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertIn("did not open", res["error"])
|
||||
|
||||
def test_success(self):
|
||||
with mock.patch.object(invite, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(invite, "open_settings",
|
||||
return_value=True), \
|
||||
mock.patch("hatch_menu.dialog.goto_tab",
|
||||
return_value=True), \
|
||||
mock.patch("hatch_menu.dialog.dialog_text",
|
||||
return_value=PIP_GENERAL):
|
||||
res = invite.get_usage("pip")
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["weekly_reset"], "Oct 10")
|
||||
self.assertEqual(res["additional_left"], "901M tokens left")
|
||||
self.assertTrue(res["has_redeemed"])
|
||||
|
||||
def test_general_tab_failure(self):
|
||||
with mock.patch.object(invite, "get_cdp_ws",
|
||||
return_value=(mock.Mock(), {})), \
|
||||
mock.patch.object(invite, "open_settings",
|
||||
return_value=True), \
|
||||
mock.patch("hatch_menu.dialog.goto_tab",
|
||||
return_value=False):
|
||||
res = invite.get_usage("pip")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertIn("General tab", res["error"])
|
||||
|
||||
|
||||
class FleetTests(unittest.TestCase):
|
||||
def test_invite_status_aggregates_and_never_raises(self):
|
||||
with mock.patch.object(
|
||||
invite, "get_invite",
|
||||
side_effect=[{"ok": True, "node": "646"},
|
||||
invite.InviteError("bad")]):
|
||||
res = invite.fleet_invite_status(["646", "bogus"])
|
||||
self.assertTrue(res["646"]["ok"])
|
||||
self.assertFalse(res["bogus"]["ok"])
|
||||
|
||||
def test_usage_aggregates(self):
|
||||
with mock.patch.object(
|
||||
invite, "get_usage",
|
||||
return_value={"ok": True, "node": "pip"}):
|
||||
res = invite.fleet_usage(["pip"])
|
||||
self.assertTrue(res["pip"]["ok"])
|
||||
|
||||
|
||||
class ShimTests(unittest.TestCase):
|
||||
"""open_settings/click_settings_tab delegate to the tree (single copy).
|
||||
|
||||
Trusted-click behavior itself is covered by tests/test_hatch_menu.py.
|
||||
"""
|
||||
|
||||
def test_open_settings_delegates(self):
|
||||
ws = mock.Mock()
|
||||
with mock.patch("hatch_menu.dialog.open_settings",
|
||||
return_value=True) as m:
|
||||
self.assertTrue(invite.open_settings(ws))
|
||||
m.assert_called_once_with(ws)
|
||||
|
||||
def test_open_settings_failure_passthrough(self):
|
||||
with mock.patch("hatch_menu.dialog.open_settings",
|
||||
return_value=False):
|
||||
self.assertFalse(invite.open_settings(mock.Mock()))
|
||||
|
||||
def test_click_settings_tab_delegates(self):
|
||||
ws = mock.Mock()
|
||||
with mock.patch("hatch_menu.dialog.goto_tab",
|
||||
return_value=True) as m:
|
||||
self.assertTrue(invite.click_settings_tab(ws, "General"))
|
||||
m.assert_called_once_with(ws, "General")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,182 @@
|
||||
#!/usr/bin/env python3
|
||||
"""test_invite_handler.py: Unit tests for invite code handler, usage parser, and settings RPA."""
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
||||
|
||||
import invite
|
||||
from invite_handler import InviteCodeInfo, InviteHandler, RedemptionResult, salvage_blocked_node
|
||||
from settings_rpa import NodeUsage, SettingsRPA
|
||||
|
||||
|
||||
class TestInviteCodeValidation(unittest.TestCase):
|
||||
def test_normalize_valid_codes(self):
|
||||
self.assertEqual(invite.normalize_code("REDCJ7"), "REDCJ7")
|
||||
self.assertEqual(invite.normalize_code(" redcj7 "), "REDCJ7")
|
||||
self.assertEqual(invite.normalize_code("81mdir"), "81MDIR")
|
||||
self.assertEqual(invite.normalize_code("F4BGHN"), "F4BGHN")
|
||||
self.assertEqual(invite.normalize_code("14ogf2"), "14OGF2")
|
||||
|
||||
def test_normalize_invalid_codes(self):
|
||||
self.assertIsNone(invite.normalize_code("SHORT"))
|
||||
self.assertIsNone(invite.normalize_code("TOOLONG123"))
|
||||
self.assertIsNone(invite.normalize_code("INV@L!"))
|
||||
self.assertIsNone(invite.normalize_code(""))
|
||||
self.assertIsNone(invite.normalize_code(None))
|
||||
self.assertIsNone(invite.normalize_code(123456))
|
||||
|
||||
|
||||
class TestUsageParser(unittest.TestCase):
|
||||
def test_parse_usage_blocked_node(self):
|
||||
sample_646 = """
|
||||
Settings
|
||||
General
|
||||
Meta Account
|
||||
Password, security, personal details
|
||||
Usage
|
||||
Free plan
|
||||
Weekly limit resets on Oct 9
|
||||
100% used
|
||||
Additional tokens
|
||||
Never expires
|
||||
100% used (0 tokens left)
|
||||
Upgrade
|
||||
Language
|
||||
Appearance
|
||||
"""
|
||||
parsed = invite.parse_usage_text(sample_646)
|
||||
self.assertEqual(parsed["weekly_reset"], "Oct 9")
|
||||
self.assertEqual(parsed["weekly_used_pct"], 100)
|
||||
self.assertEqual(parsed["additional_expires"], "Never expires")
|
||||
self.assertEqual(parsed["additional_used_pct"], 100)
|
||||
self.assertEqual(parsed["additional_left"], "0 tokens left")
|
||||
|
||||
def test_parse_usage_active_pip(self):
|
||||
sample_pip = """
|
||||
Usage
|
||||
Free plan
|
||||
Weekly limit resets on Oct 10
|
||||
100% used
|
||||
Additional tokens
|
||||
Never expires
|
||||
10% used (901M tokens left)
|
||||
Upgrade
|
||||
"""
|
||||
parsed = invite.parse_usage_text(sample_pip)
|
||||
self.assertEqual(parsed["weekly_reset"], "Oct 10")
|
||||
self.assertEqual(parsed["weekly_used_pct"], 100)
|
||||
self.assertEqual(parsed["additional_expires"], "Never expires")
|
||||
self.assertEqual(parsed["additional_used_pct"], 10)
|
||||
self.assertEqual(parsed["additional_left"], "901M tokens left")
|
||||
self.assertTrue(parsed["has_redeemed"])
|
||||
|
||||
def test_parse_usage_fresh_account_no_extra_tokens(self):
|
||||
sample_fresh = """
|
||||
Usage
|
||||
Free plan
|
||||
Weekly limit resets on Oct 12
|
||||
15% used
|
||||
Redeem invite code
|
||||
"""
|
||||
parsed = invite.parse_usage_text(sample_fresh)
|
||||
self.assertEqual(parsed["weekly_reset"], "Oct 12")
|
||||
self.assertEqual(parsed["weekly_used_pct"], 15)
|
||||
self.assertIsNone(parsed["additional_left"])
|
||||
self.assertFalse(parsed["has_redeemed"])
|
||||
|
||||
def test_parse_usage_empty(self):
|
||||
parsed = invite.parse_usage_text("")
|
||||
self.assertIsNone(parsed["weekly_reset"])
|
||||
self.assertIsNone(parsed["weekly_used_pct"])
|
||||
self.assertIsNone(parsed["additional_left"])
|
||||
self.assertFalse(parsed["has_redeemed"])
|
||||
|
||||
|
||||
class TestInviteHandlerMocked(unittest.TestCase):
|
||||
@patch("invite.get_cdp_ws")
|
||||
@patch("invite.cdp_evaluate")
|
||||
def test_get_invite_api_success(self, mock_eval, mock_ws):
|
||||
mock_ws.return_value = (MagicMock(), MagicMock())
|
||||
mock_eval.return_value = {
|
||||
"status": 200,
|
||||
"data": {
|
||||
"code": "REDCJ7",
|
||||
"uses_remaining": 30,
|
||||
"use_count": 0,
|
||||
"has_redeemed_invite_code": True,
|
||||
"reward": None,
|
||||
},
|
||||
}
|
||||
|
||||
res = invite.get_invite("646")
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["code"], "REDCJ7")
|
||||
self.assertTrue(res["has_redeemed"])
|
||||
self.assertEqual(res["uses_remaining"], 30)
|
||||
|
||||
@patch("invite.get_cdp_ws")
|
||||
@patch("invite.cdp_evaluate")
|
||||
def test_redeem_invite_already_redeemed(self, mock_eval, mock_ws):
|
||||
mock_ws.return_value = (MagicMock(), MagicMock())
|
||||
mock_eval.return_value = {
|
||||
"status": 400,
|
||||
"ok": False,
|
||||
"data": {
|
||||
"success": False,
|
||||
"reason": "already_redeemed",
|
||||
"detail": "You’ve already redeemed an invite code.",
|
||||
},
|
||||
}
|
||||
|
||||
res = invite.redeem_invite("muse", "REDCJ7")
|
||||
self.assertFalse(res["ok"])
|
||||
self.assertEqual(res["reason"], "already_redeemed")
|
||||
self.assertIn("already redeemed", res["detail"])
|
||||
|
||||
@patch("invite.get_cdp_ws")
|
||||
@patch("invite.cdp_evaluate")
|
||||
def test_redeem_invite_success(self, mock_eval, mock_ws):
|
||||
mock_ws.return_value = (MagicMock(), MagicMock())
|
||||
mock_eval.return_value = {
|
||||
"status": 200,
|
||||
"ok": True,
|
||||
"data": {
|
||||
"success": True,
|
||||
"redemptionStatus": "redeemed",
|
||||
"detail": "Congratulations!",
|
||||
},
|
||||
}
|
||||
|
||||
res = invite.redeem_invite("def", "REDCJ7")
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["redemption_status"], "redeemed")
|
||||
|
||||
def test_salvage_blocked_node_diagnosis(self):
|
||||
with patch.object(InviteHandler, "find_code") as mock_find:
|
||||
mock_find.return_value = InviteCodeInfo(
|
||||
node="646",
|
||||
code="REDCJ7",
|
||||
uses_remaining=30,
|
||||
use_count=0,
|
||||
max_uses=30,
|
||||
has_redeemed=True,
|
||||
invite_state="ELIGIBLE",
|
||||
reward=None,
|
||||
method_used="api",
|
||||
)
|
||||
# All peers already redeemed
|
||||
res = salvage_blocked_node("646")
|
||||
self.assertFalse(res["success"])
|
||||
self.assertEqual(res["invite_code"], "REDCJ7")
|
||||
self.assertIn("An onboarding agent or fresh client profile must redeem this code", res["error"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,88 @@
|
||||
#!/usr/bin/env python3
|
||||
"""test_onboard_pipeline.py — Unit tests for the onboarding and invite salvage pipeline."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
||||
|
||||
from onboard_pipeline import (
|
||||
OnboardState,
|
||||
finish_onboarding_redemption,
|
||||
select_urgent_beneficiary,
|
||||
start_onboarding,
|
||||
submit_onboarding_otp,
|
||||
)
|
||||
|
||||
|
||||
class TestOnboardPipeline(unittest.TestCase):
|
||||
def test_select_urgent_beneficiary_explicit(self):
|
||||
b_node, code, reason = select_urgent_beneficiary("dev", "TEST01")
|
||||
self.assertEqual(b_node, "dev")
|
||||
self.assertEqual(code, "TEST01")
|
||||
self.assertIn("Explicit", reason)
|
||||
|
||||
@patch("invite.fleet_invite_status")
|
||||
@patch("invite.fleet_usage")
|
||||
def test_select_urgent_beneficiary_blocked_first(self, mock_usage, mock_invite):
|
||||
mock_usage.return_value = {
|
||||
"646": {"weekly_used_pct": 100, "additional_used_pct": 100, "additional_left": "0 tokens left"},
|
||||
"pip": {"weekly_used_pct": 100, "additional_used_pct": 10, "additional_left": "890M tokens left"},
|
||||
"muse": {"weekly_used_pct": 50, "additional_used_pct": 0, "additional_left": "950M tokens left"},
|
||||
}
|
||||
mock_invite.return_value = {
|
||||
"646": {"code": "REDCJ7"},
|
||||
"pip": {"code": "F4BGHN"},
|
||||
"muse": {"code": "81MDIR"},
|
||||
}
|
||||
|
||||
b_node, code, reason = select_urgent_beneficiary()
|
||||
self.assertEqual(b_node, "646")
|
||||
self.assertEqual(code, "REDCJ7")
|
||||
self.assertIn("BLOCKED", reason)
|
||||
|
||||
@patch("onboard_pipeline.provision_node_infra")
|
||||
@patch("onboard_pipeline.CredClient")
|
||||
@patch("onboard_pipeline.select_urgent_beneficiary")
|
||||
def test_start_onboarding_awaiting_otp(self, mock_select, mock_cred_cls, mock_infra):
|
||||
mock_select.return_value = ("646", "REDCJ7", "Blocked test")
|
||||
mock_infra.return_value = {"ok": True}
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.initiate.return_value = {
|
||||
"status": "awaiting_otp",
|
||||
"message": "OTP sent",
|
||||
}
|
||||
mock_cred_cls.return_value = mock_client
|
||||
|
||||
res = start_onboarding("testnode", "client@test.com")
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["status"], "awaiting_otp")
|
||||
self.assertEqual(res["beneficiary_node"], "646")
|
||||
self.assertEqual(res["invite_code_queued"], "REDCJ7")
|
||||
|
||||
@patch("invite.redeem_invite")
|
||||
@patch("subprocess.run")
|
||||
def test_finish_onboarding_redemption_success(self, mock_subproc, mock_redeem):
|
||||
mock_redeem.return_value = {"ok": True, "redemption_status": "redeemed"}
|
||||
state = OnboardState(
|
||||
node="testnode",
|
||||
email="client@test.com",
|
||||
beneficiary_node="646",
|
||||
invite_code="REDCJ7",
|
||||
stage="auth_active",
|
||||
)
|
||||
|
||||
res = finish_onboarding_redemption(state)
|
||||
self.assertTrue(res["ok"])
|
||||
self.assertEqual(res["status"], "completed")
|
||||
self.assertEqual(res["beneficiary_node"], "646")
|
||||
mock_redeem.assert_called_once_with("testnode", "REDCJ7", timeout=20.0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,110 @@
|
||||
#!/usr/bin/env python3
|
||||
"""test_settings_rpa.py: Unit tests for Settings RPA substrate."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
||||
|
||||
from settings_rpa import NodeUsage, SettingsRPA, cdp_dispatch_mouse_click, cdp_send_command
|
||||
|
||||
|
||||
class TestSettingsRPAPrimitives(unittest.TestCase):
|
||||
def test_cdp_send_command_matches_id(self):
|
||||
mock_ws = MagicMock()
|
||||
mock_ws.recv.return_value = json.dumps({"id": 10000, "result": {"success": True}})
|
||||
# Mock _REQ_COUNTER next value
|
||||
with patch("settings_rpa._REQ_COUNTER", iter([10000])):
|
||||
res = cdp_send_command(mock_ws, "Test.method", {"param": 1})
|
||||
self.assertEqual(res, {"success": True})
|
||||
mock_ws.send.assert_called_once()
|
||||
sent_msg = json.loads(mock_ws.send.call_args[0][0])
|
||||
self.assertEqual(sent_msg["method"], "Test.method")
|
||||
|
||||
def test_cdp_dispatch_mouse_click(self):
|
||||
mock_ws = MagicMock()
|
||||
with patch("settings_rpa.cdp_send_command") as mock_send:
|
||||
mock_send.return_value = {}
|
||||
res = cdp_dispatch_mouse_click(mock_ws, 100.0, 200.0)
|
||||
self.assertTrue(res)
|
||||
self.assertEqual(mock_send.call_count, 2)
|
||||
# Check mousePressed and mouseReleased
|
||||
call_methods = [c[0][1] for c in mock_send.call_args_list]
|
||||
self.assertEqual(call_methods, ["Input.dispatchMouseEvent", "Input.dispatchMouseEvent"])
|
||||
|
||||
@patch("settings_rpa.get_cdp_ws")
|
||||
@patch("settings_rpa.cdp_evaluate")
|
||||
def test_settings_dialog_check(self, mock_eval, mock_ws):
|
||||
mock_ws.return_value = (MagicMock(), MagicMock())
|
||||
mock_eval.return_value = True
|
||||
|
||||
rpa = SettingsRPA("646")
|
||||
rpa.connect()
|
||||
self.assertTrue(rpa.is_settings_dialog_open())
|
||||
|
||||
@patch("settings_rpa.get_cdp_ws")
|
||||
@patch("settings_rpa.cdp_evaluate")
|
||||
def test_read_usage_blocked(self, mock_eval, mock_ws):
|
||||
mock_ws.return_value = (MagicMock(), MagicMock())
|
||||
# Provide responses: open check (True), tab open check (True), click tab (True), usage dict
|
||||
usage_payload = {
|
||||
"dialogText": "Free plan\nWeekly limit resets on Oct 9\n100% used\nAdditional tokens\n100% used (0 tokens left)",
|
||||
"bars": [
|
||||
{"label": "Free plan", "percent": 100, "raw_text": "Free plan\nWeekly limit resets on Oct 9\n100% used"},
|
||||
{"label": "Additional tokens", "percent": 100, "raw_text": "Additional tokens\n100% used (0 tokens left)"},
|
||||
],
|
||||
"plan": "Free plan",
|
||||
"resetText": "Weekly limit resets on Oct 9",
|
||||
"tokensLeft": "0 tokens left",
|
||||
}
|
||||
mock_eval.side_effect = [
|
||||
True, # is_settings_dialog_open
|
||||
True, # is_settings_dialog_open in select_tab
|
||||
True, # click General tab
|
||||
usage_payload, # read_usage evaluation
|
||||
]
|
||||
|
||||
with SettingsRPA("646") as rpa:
|
||||
usage = rpa.read_usage(keep_dialog_open=True)
|
||||
self.assertEqual(usage.node, "646")
|
||||
self.assertEqual(usage.weekly_percent_used, 100)
|
||||
self.assertEqual(usage.extra_percent_used, 100)
|
||||
self.assertTrue(usage.is_blocked)
|
||||
self.assertEqual(usage.extra_tokens_remaining, "0 tokens left")
|
||||
self.assertTrue(usage.has_redeemed)
|
||||
|
||||
@patch("settings_rpa.get_cdp_ws")
|
||||
@patch("settings_rpa.cdp_evaluate")
|
||||
def test_read_usage_not_redeemed(self, mock_eval, mock_ws):
|
||||
mock_ws.return_value = (MagicMock(), MagicMock())
|
||||
usage_payload = {
|
||||
"dialogText": "Free plan\nWeekly limit resets on Oct 9\n20% used\nRedeem invite code",
|
||||
"bars": [
|
||||
{"label": "Free plan", "percent": 20, "raw_text": "Free plan\nWeekly limit resets on Oct 9\n20% used"},
|
||||
],
|
||||
"plan": "Free plan",
|
||||
"resetText": "Weekly limit resets on Oct 9",
|
||||
"tokensLeft": "",
|
||||
}
|
||||
mock_eval.side_effect = [
|
||||
True, # is_settings_dialog_open
|
||||
True, # is_settings_dialog_open in select_tab
|
||||
True, # click General tab
|
||||
usage_payload, # read_usage evaluation
|
||||
]
|
||||
|
||||
with SettingsRPA("646") as rpa:
|
||||
usage = rpa.read_usage(keep_dialog_open=True)
|
||||
self.assertEqual(usage.node, "646")
|
||||
self.assertEqual(usage.weekly_percent_used, 20)
|
||||
self.assertEqual(usage.extra_percent_used, 0)
|
||||
self.assertFalse(usage.is_blocked)
|
||||
self.assertFalse(usage.has_redeemed)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user