"""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()