operator
123 lines
4.7 KiB
Python
123 lines
4.7 KiB
Python
#!/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)
|
|
|
|
@patch("invite.redeem_invite")
|
|
@patch("subprocess.run")
|
|
@patch("invite_handler.send_loopback_notice")
|
|
def test_finish_onboarding_redemption_failure_with_loopback(self, mock_notice, mock_subproc, mock_redeem):
|
|
mock_redeem.return_value = {"ok": False, "reason": "already_redeemed", "detail": "Already redeemed"}
|
|
mock_notice.return_value = True
|
|
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"], "auth_active_redemption_warning")
|
|
self.assertIn("Redemption failed: already_redeemed", state.detail)
|
|
mock_notice.assert_called_once()
|
|
kwargs = mock_notice.call_args.kwargs
|
|
self.assertEqual(kwargs.get("recipient"), "646")
|
|
self.assertEqual(kwargs.get("target"), "646 tasks")
|
|
self.assertIn("already_redeemed", kwargs.get("message", ""))
|
|
|
|
|
|
def test_get_all_connects_fast(self):
|
|
from onboard_pipeline import get_all_connects
|
|
connects = get_all_connects(fast=True)
|
|
self.assertIsInstance(connects, list)
|
|
self.assertGreaterEqual(len(connects), 6)
|
|
nodes = {c["node"] for c in connects}
|
|
for expected in ("muse", "pip", "646", "opm", "dev", "def"):
|
|
self.assertIn(expected, nodes)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|