ysj7705andClaude Opus 5.5
384d0a75f0
Prepare signs in on load (never connects), system CA bundle for HTTPS, root-owned deploy extraction. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
263 lines
9.5 KiB
Python
263 lines
9.5 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from mudfish_decky.client import SigninError
|
|
from mudfish_decky.controller import MSG_MISSING, MSG_RELOGIN, MSG_RTT, MSG_TIMEOUT, Controller
|
|
from mudfish_decky.store import Store
|
|
from tests.test_nodes import STATIC
|
|
|
|
BEFORE = "default via 192.168.1.1 dev wlan0 proto dhcp src 192.168.1.50 metric 600\n"
|
|
NEW_ROUTES = "1.201.174.247 via 192.168.1.1 dev wlan0\n2.58.243.110 via 192.168.1.1 dev wlan0\n"
|
|
|
|
|
|
class FakeSystem:
|
|
def __init__(self):
|
|
self.installed = True
|
|
self.tap = False
|
|
self.routes = BEFORE
|
|
self.deleted = []
|
|
self.daemon_calls = 0
|
|
self.ip = {"ip": "13.73.234.162", "country": "JP"}
|
|
|
|
def find_mudfish(self):
|
|
return ("/opt/mudfish/6.6.0/bin/mudrun-headless", "/opt/mudfish/6.6.0/var") if self.installed else None
|
|
|
|
def ensure_daemon(self, bin_path, is_up):
|
|
self.daemon_calls += 1
|
|
|
|
def tap_up(self):
|
|
return self.tap
|
|
|
|
def route_text(self):
|
|
return self.routes
|
|
|
|
def delete_routes(self, lines):
|
|
self.deleted.extend(lines)
|
|
return len(lines)
|
|
|
|
def public_ip(self):
|
|
return self.ip
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, system):
|
|
self.system = system
|
|
self.st = {"status": 200, "isSignedIn": True, "core_running": False, "core_ready": False,
|
|
"client_ready": False, "jwt": "JWT", "api_server": "api.mudfish.net"}
|
|
self.signin_error = None
|
|
self.rtt_map = {9: 38.5, 11: 31.0}
|
|
self.conf = {"mudfishFullvpnSid": 0, "mudfishFullvpnVid": 0, "mudfishFullvpnExceptions": "", "mudfishMtu": 1350}
|
|
self.saved = []
|
|
self.start_calls = 0
|
|
self.stop_calls = 0
|
|
self.refresh_calls = []
|
|
self.need_restart = True
|
|
self.ready_on_start = True
|
|
|
|
def status(self):
|
|
return dict(self.st)
|
|
|
|
def ensure_signed_in(self):
|
|
if self.signin_error:
|
|
raise self.signin_error
|
|
return dict(self.st)
|
|
|
|
def start(self):
|
|
self.start_calls += 1
|
|
self.st["core_running"] = True
|
|
if self.ready_on_start:
|
|
self.st.update(core_ready=True, client_ready=True)
|
|
self.system.tap = True
|
|
self.system.routes = BEFORE + NEW_ROUTES
|
|
|
|
def stop(self):
|
|
self.stop_calls += 1
|
|
self.st.update(core_running=False, core_ready=False, client_ready=False)
|
|
self.system.tap = False
|
|
|
|
def rtts(self):
|
|
return dict(self.rtt_map)
|
|
|
|
def static_nodes(self, jwt, api):
|
|
return STATIC
|
|
|
|
def prog_conf(self, jwt, api):
|
|
return dict(self.conf)
|
|
|
|
def save_prog_conf(self, jwt, api, conf):
|
|
self.saved.append(dict(conf))
|
|
self.conf = dict(conf)
|
|
return {"status": 200, "need_restart": self.need_restart}
|
|
|
|
def fullvpn_refresh(self, sid, exceptions):
|
|
self.refresh_calls.append((sid, exceptions))
|
|
|
|
|
|
class ControllerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.path = os.path.join(self.tmp.name, "settings.json")
|
|
self.system = FakeSystem()
|
|
self.client = FakeClient(self.system)
|
|
self.now = 1000.0
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def make(self):
|
|
return Controller(self.client, self.system, Store(self.path), clock=lambda: self.now, sleep=lambda s: None)
|
|
|
|
def test_state_missing_when_not_installed(self):
|
|
self.system.installed = False
|
|
st = self.make().state()
|
|
self.assertEqual(st["phase"], "missing")
|
|
self.assertEqual(st["message"], MSG_MISSING)
|
|
|
|
def test_state_off_never_starts_core(self):
|
|
st = self.make().state()
|
|
self.assertEqual(st["phase"], "off")
|
|
self.assertEqual(self.client.start_calls, 0)
|
|
|
|
def test_state_off_shows_auto_pick(self):
|
|
st = self.make().state()
|
|
self.assertEqual(st["node_mode"], "auto")
|
|
self.assertEqual(st["node"], {"sid": 11, "label": "Osaka - SakuraNet 4", "rtt": 31.0})
|
|
|
|
def test_connect_auto_picks_lowest_and_saves_sid(self):
|
|
st = self.make().connect()
|
|
self.assertEqual(self.client.conf["mudfishFullvpnSid"], 11)
|
|
self.assertEqual(self.client.conf["mudfishFullvpnVid"], 0)
|
|
self.assertEqual(self.client.conf["mudfishMtu"], 1350)
|
|
self.assertEqual(st["phase"], "on")
|
|
self.assertEqual(st["connected_since"], 1000.0)
|
|
self.assertEqual((st["ip"], st["country"]), ("13.73.234.162", "JP"))
|
|
|
|
def test_connect_skips_save_when_sid_unchanged(self):
|
|
self.client.conf["mudfishFullvpnSid"] = 11
|
|
self.make().connect()
|
|
self.assertEqual(self.client.saved, [])
|
|
|
|
def test_connect_auto_without_rtts_does_not_start(self):
|
|
self.client.rtt_map = {}
|
|
st = self.make().connect()
|
|
self.assertEqual(st["phase"], "error")
|
|
self.assertEqual(st["message"], MSG_RTT)
|
|
self.assertEqual(self.client.start_calls, 0)
|
|
|
|
def test_connect_uses_cached_rtts_when_measuring(self):
|
|
c = self.make()
|
|
c.state() # caches rtts
|
|
self.client.rtt_map = {} # mid-round: endpoint has no rtts
|
|
self.assertEqual(self.make().connect()["phase"], "on")
|
|
|
|
def test_connect_signin_failure_shows_relogin(self):
|
|
self.client.signin_error = SigninError("401")
|
|
st = self.make().connect()
|
|
self.assertEqual(st["phase"], "error")
|
|
self.assertEqual(st["message"], MSG_RELOGIN)
|
|
self.assertEqual(self.client.start_calls, 0)
|
|
|
|
def test_connect_timeout_stops_core(self):
|
|
self.client.ready_on_start = False
|
|
st = self.make().connect()
|
|
self.assertEqual(self.client.stop_calls, 1)
|
|
self.assertEqual(st["phase"], "error")
|
|
self.assertEqual(st["message"], MSG_TIMEOUT)
|
|
|
|
def test_connect_manual_uses_saved_sid(self):
|
|
c = self.make()
|
|
c.set_node("9")
|
|
c.connect()
|
|
self.assertEqual(self.client.conf["mudfishFullvpnSid"], 9)
|
|
|
|
def test_disconnect_deletes_only_new_routes_and_keeps_sid(self):
|
|
c = self.make()
|
|
c.connect()
|
|
saves = len(self.client.saved)
|
|
st = c.disconnect()
|
|
self.assertEqual(self.system.deleted,
|
|
["1.201.174.247 via 192.168.1.1 dev wlan0", "2.58.243.110 via 192.168.1.1 dev wlan0"])
|
|
self.assertEqual(len(self.client.saved), saves)
|
|
self.assertEqual(self.client.conf["mudfishFullvpnSid"], 11)
|
|
self.assertEqual(st["phase"], "off")
|
|
self.assertIsNone(Store(self.path).load().get("connected_since"))
|
|
|
|
def test_disconnect_without_snapshot_deletes_nothing(self):
|
|
self.client.st.update(core_running=True, client_ready=True)
|
|
self.system.tap = True
|
|
self.system.routes = BEFORE + NEW_ROUTES
|
|
self.make().disconnect()
|
|
self.assertEqual(self.system.deleted, [])
|
|
|
|
def test_state_on_survives_restart_uses_saved_since(self):
|
|
Store(self.path).save({"connected_since": 100.0})
|
|
self.client.st.update(core_running=True, core_ready=True, client_ready=True)
|
|
self.system.tap = True
|
|
st = self.make().state()
|
|
self.assertEqual(st["phase"], "on")
|
|
self.assertEqual(st["connected_since"], 100.0)
|
|
self.assertEqual(self.client.start_calls, 0)
|
|
|
|
def test_state_on_without_saved_since_starts_clock_now(self):
|
|
self.client.st.update(core_running=True, core_ready=True, client_ready=True)
|
|
self.system.tap = True
|
|
self.assertEqual(self.make().state()["connected_since"], 1000.0)
|
|
|
|
def test_state_connecting_when_core_running_not_ready(self):
|
|
self.client.st.update(core_running=True)
|
|
self.assertEqual(self.make().state()["phase"], "connecting")
|
|
|
|
def test_set_node_while_off_only_saves_choice(self):
|
|
st = self.make().set_node("9")
|
|
self.assertEqual(self.client.saved, [])
|
|
self.assertEqual(Store(self.path).load()["manual_sid"], 9)
|
|
self.assertEqual(st["node_mode"], "manual")
|
|
self.assertEqual(st["node"]["sid"], 9)
|
|
|
|
def test_set_node_while_on_refreshes(self):
|
|
c = self.make()
|
|
c.connect()
|
|
self.client.need_restart = False
|
|
st = c.set_node("9")
|
|
self.assertEqual(self.client.refresh_calls, [(9, "")])
|
|
self.assertEqual(self.client.start_calls, 1)
|
|
self.assertEqual(st["node"]["sid"], 9)
|
|
|
|
def test_set_node_while_on_restarts_when_needed(self):
|
|
c = self.make()
|
|
c.connect()
|
|
self.client.need_restart = True
|
|
c.set_node("9")
|
|
self.assertEqual((self.client.stop_calls, self.client.start_calls), (1, 2))
|
|
|
|
def test_set_node_back_to_auto(self):
|
|
c = self.make()
|
|
c.set_node("9")
|
|
st = c.set_node("auto")
|
|
self.assertEqual(st["node_mode"], "auto")
|
|
self.assertEqual(st["node"]["sid"], 11)
|
|
|
|
def test_prepare_signs_in_and_caches_nodes_without_starting(self):
|
|
self.client.st["isSignedIn"] = False
|
|
signed = []
|
|
self.client.ensure_signed_in = lambda: signed.append(1) or dict(self.client.st, isSignedIn=True)
|
|
c = self.make()
|
|
c.prepare()
|
|
self.assertEqual(self.system.daemon_calls, 1)
|
|
self.assertEqual(signed, [1])
|
|
self.assertEqual(self.client.start_calls, 0)
|
|
self.assertEqual([n["sid"] for n in Store(self.path).load()["static_nodes"]], [9, 11, 46, 300])
|
|
|
|
def test_prepare_tolerates_signin_failure(self):
|
|
self.client.signin_error = SigninError("401")
|
|
self.make().prepare()
|
|
self.assertEqual(self.client.start_calls, 0)
|
|
|
|
def test_nodes_lists_japan_by_ping(self):
|
|
self.assertEqual([n["sid"] for n in self.make().nodes()], [11, 9, 46])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|