262 lines
10 KiB
Python
262 lines
10 KiB
Python
|
"""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()
|