operator
313 lines
12 KiB
Python
313 lines
12 KiB
Python
#!/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
|
||
import invite_handler
|
||
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"])
|
||
self.assertTrue(res.get("field_missing"))
|
||
|
||
def test_salvage_blocked_node_with_loopback_dm(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",
|
||
)
|
||
with patch("invite_handler.send_loopback_notice", return_value=True) as mock_notice:
|
||
res = salvage_blocked_node("646", notify_target="heartbeat-opm", notify_agent="opm")
|
||
self.assertFalse(res["success"])
|
||
self.assertTrue(res.get("field_missing"))
|
||
self.assertTrue(res.get("loopback_notified"))
|
||
mock_notice.assert_called_once()
|
||
call_args = mock_notice.call_args[0]
|
||
self.assertEqual(call_args[0], "opm")
|
||
self.assertEqual(call_args[1], "heartbeat-opm")
|
||
self.assertIn("REDCJ7", call_args[2])
|
||
|
||
@patch("invite_handler.SettingsRPA")
|
||
def test_redeem_code_dom_missing_row_api_fallback_success(self, mock_rpa_cls):
|
||
mock_rpa = MagicMock()
|
||
mock_rpa.open_settings_dialog.return_value = True
|
||
mock_rpa.select_tab.return_value = True
|
||
mock_rpa_cls.return_value.__enter__.return_value = mock_rpa
|
||
|
||
h = InviteHandler("dev")
|
||
h.ws = MagicMock()
|
||
with patch.object(h, "connect"), patch("time.sleep", return_value=None):
|
||
with patch("invite_handler.cdp_evaluate", return_value={"found": False, "text": "General"}):
|
||
with patch.object(h, "redeem_code_api") as mock_api:
|
||
mock_api.return_value = RedemptionResult(
|
||
target_node="dev",
|
||
code="REDCJ7",
|
||
success=True,
|
||
status="redeemed",
|
||
reason=None,
|
||
detail="API success",
|
||
method_used="api",
|
||
)
|
||
res = h.redeem_code_dom("REDCJ7")
|
||
self.assertTrue(res.success)
|
||
self.assertEqual(res.status, "redeemed")
|
||
self.assertIn("Redeem field was missing in Settings DOM; redeemed successfully via API fallback!", res.detail)
|
||
mock_rpa.close_settings_dialog.assert_called_once()
|
||
|
||
@patch("invite_handler.SettingsRPA")
|
||
def test_redeem_code_dom_missing_row_already_redeemed(self, mock_rpa_cls):
|
||
mock_rpa = MagicMock()
|
||
mock_rpa.open_settings_dialog.return_value = True
|
||
mock_rpa.select_tab.return_value = True
|
||
mock_rpa_cls.return_value.__enter__.return_value = mock_rpa
|
||
|
||
h = InviteHandler("646")
|
||
h.ws = MagicMock()
|
||
with patch.object(h, "connect"), patch("time.sleep", return_value=None):
|
||
with patch("invite_handler.cdp_evaluate", return_value={"found": False, "has_additional": True, "text": "Additional tokens"}):
|
||
with patch.object(h, "redeem_code_api") as mock_api:
|
||
mock_api.return_value = RedemptionResult(
|
||
target_node="646",
|
||
code="REDCJ7",
|
||
success=False,
|
||
status="already_redeemed",
|
||
reason="already_redeemed",
|
||
detail="Already redeemed",
|
||
method_used="api",
|
||
)
|
||
res = h.redeem_code_dom("REDCJ7")
|
||
self.assertFalse(res.success)
|
||
self.assertEqual(res.status, "already_redeemed")
|
||
self.assertTrue(res.field_missing)
|
||
self.assertIn("already redeemed an invite code", res.detail)
|
||
|
||
@patch("invite_handler.SettingsRPA")
|
||
def test_redeem_code_dom_missing_input_box_with_loopback(self, mock_rpa_cls):
|
||
mock_rpa = MagicMock()
|
||
mock_rpa.open_settings_dialog.return_value = True
|
||
mock_rpa.select_tab.return_value = True
|
||
mock_rpa_cls.return_value.__enter__.return_value = mock_rpa
|
||
|
||
h = InviteHandler("pip")
|
||
h.ws = MagicMock()
|
||
|
||
def mock_eval(ws, js, *args, **kwargs):
|
||
if "Redeem invite code" in js:
|
||
return {"found": True}
|
||
if "found_input" in js:
|
||
return {"found_input": False}
|
||
return True
|
||
|
||
with patch.object(h, "connect"), patch("time.sleep", return_value=None):
|
||
with patch("invite_handler.cdp_evaluate", side_effect=mock_eval), patch("invite_handler.cdp_send_escape"):
|
||
with patch.object(h, "redeem_code_api") as mock_api:
|
||
mock_api.return_value = RedemptionResult(
|
||
target_node="pip",
|
||
code="TEST01",
|
||
success=False,
|
||
status="dom_input_missing",
|
||
reason="input_field_missing",
|
||
detail="API fallback also failed",
|
||
method_used="api",
|
||
)
|
||
with patch("invite_handler.send_loopback_notice", return_value=True) as mock_notice:
|
||
res = h.redeem_code_dom("TEST01", notify_target="pip tasks", notify_agent="pip")
|
||
self.assertFalse(res.success)
|
||
self.assertTrue(res.field_missing)
|
||
self.assertEqual(res.reason, "input_field_missing")
|
||
self.assertTrue(res.loopback_notified)
|
||
mock_notice.assert_called_once()
|
||
self.assertIn("pip tasks", res.loopback_detail)
|
||
|
||
@patch("subprocess.run")
|
||
def test_send_loopback_notice(self, mock_subproc):
|
||
mock_subproc.return_value = MagicMock(returncode=0)
|
||
ok = invite_handler.send_loopback_notice("opm", "646 tasks", "Test alert")
|
||
self.assertTrue(ok)
|
||
mock_subproc.assert_called_once()
|
||
cmd = mock_subproc.call_args[0][0]
|
||
self.assertIn("send", cmd)
|
||
self.assertIn("646 tasks", cmd)
|
||
self.assertIn("Test alert", cmd)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|