bastd.ui.settings.nettesting

  1# Released under the MIT License. See LICENSE for details.
  2#
  3"""Provides ui for network related testing."""
  4
  5from __future__ import annotations
  6
  7import time
  8import copy
  9import weakref
 10from threading import Thread
 11from typing import TYPE_CHECKING
 12
 13from efro.error import CleanError
 14import _ba
 15import ba
 16from bastd.ui.settings.testing import TestingWindow
 17
 18if TYPE_CHECKING:
 19    from typing import Callable, Any
 20
 21
 22class NetTestingWindow(ba.Window):
 23    """Window that runs a networking test suite to help diagnose issues."""
 24
 25    def __init__(self, transition: str = 'in_right'):
 26        self._width = 820
 27        self._height = 500
 28        self._printed_lines: list[str] = []
 29        uiscale = ba.app.ui.uiscale
 30        super().__init__(root_widget=ba.containerwidget(
 31            size=(self._width, self._height),
 32            scale=(1.56 if uiscale is ba.UIScale.SMALL else
 33                   1.2 if uiscale is ba.UIScale.MEDIUM else 0.8),
 34            stack_offset=(0.0, -7 if uiscale is ba.UIScale.SMALL else 0.0),
 35            transition=transition))
 36        self._done_button = ba.buttonwidget(parent=self._root_widget,
 37                                            position=(40, self._height - 77),
 38                                            size=(120, 60),
 39                                            scale=0.8,
 40                                            autoselect=True,
 41                                            label=ba.Lstr(resource='doneText'),
 42                                            on_activate_call=self._done)
 43
 44        self._copy_button = ba.buttonwidget(parent=self._root_widget,
 45                                            position=(self._width - 200,
 46                                                      self._height - 77),
 47                                            size=(100, 60),
 48                                            scale=0.8,
 49                                            autoselect=True,
 50                                            label=ba.Lstr(resource='copyText'),
 51                                            on_activate_call=self._copy)
 52
 53        self._settings_button = ba.buttonwidget(
 54            parent=self._root_widget,
 55            position=(self._width - 100, self._height - 77),
 56            size=(60, 60),
 57            scale=0.8,
 58            autoselect=True,
 59            label=ba.Lstr(value='...'),
 60            on_activate_call=self._show_val_testing)
 61
 62        twidth = self._width - 450
 63        ba.textwidget(
 64            parent=self._root_widget,
 65            position=(self._width * 0.5, self._height - 55),
 66            size=(0, 0),
 67            text=ba.Lstr(resource='settingsWindowAdvanced.netTestingText'),
 68            color=(0.8, 0.8, 0.8, 1.0),
 69            h_align='center',
 70            v_align='center',
 71            maxwidth=twidth)
 72
 73        self._scroll = ba.scrollwidget(parent=self._root_widget,
 74                                       position=(50, 50),
 75                                       size=(self._width - 100,
 76                                             self._height - 140),
 77                                       capture_arrows=True,
 78                                       autoselect=True)
 79        self._rows = ba.columnwidget(parent=self._scroll)
 80
 81        ba.containerwidget(edit=self._root_widget,
 82                           cancel_button=self._done_button)
 83
 84        # Now kick off the tests.
 85        # Pass a weak-ref to this window so we don't keep it alive
 86        # if we back out before it completes. Also set is as daemon
 87        # so it doesn't keep the app running if the user is trying to quit.
 88        Thread(
 89            daemon=True,
 90            target=ba.Call(_run_diagnostics, weakref.ref(self)),
 91        ).start()
 92
 93    def print(self, text: str, color: tuple[float, float, float]) -> None:
 94        """Print text to our console thingie."""
 95        for line in text.splitlines():
 96            txt = ba.textwidget(parent=self._rows,
 97                                color=color,
 98                                text=line,
 99                                scale=0.75,
100                                flatness=1.0,
101                                shadow=0.0,
102                                size=(0, 20))
103            ba.containerwidget(edit=self._rows, visible_child=txt)
104            self._printed_lines.append(line)
105
106    def _copy(self) -> None:
107        if not ba.clipboard_is_supported():
108            ba.screenmessage('Clipboard not supported on this platform.',
109                             color=(1, 0, 0))
110            return
111        ba.clipboard_set_text('\n'.join(self._printed_lines))
112        ba.screenmessage(f'{len(self._printed_lines)} lines copied.')
113
114    def _show_val_testing(self) -> None:
115        ba.app.ui.set_main_menu_window(NetValTestingWindow().get_root_widget())
116        ba.containerwidget(edit=self._root_widget, transition='out_left')
117
118    def _done(self) -> None:
119        # pylint: disable=cyclic-import
120        from bastd.ui.settings.advanced import AdvancedSettingsWindow
121        ba.app.ui.set_main_menu_window(
122            AdvancedSettingsWindow(transition='in_left').get_root_widget())
123        ba.containerwidget(edit=self._root_widget, transition='out_right')
124
125
126def _run_diagnostics(weakwin: weakref.ref[NetTestingWindow]) -> None:
127    # pylint: disable=too-many-statements
128
129    from efro.util import utc_now
130
131    have_error = [False]
132
133    # We're running in a background thread but UI stuff needs to run
134    # in the logic thread; give ourself a way to pass stuff to it.
135    def _print(text: str,
136               color: tuple[float, float, float] | None = None) -> None:
137
138        def _print_in_logic_thread() -> None:
139            win = weakwin()
140            if win is not None:
141                win.print(text, (1.0, 1.0, 1.0) if color is None else color)
142
143        ba.pushcall(_print_in_logic_thread, from_other_thread=True)
144
145    def _print_test_results(call: Callable[[], Any]) -> None:
146        """Run the provided call; return success/fail text & color."""
147        starttime = time.monotonic()
148        try:
149            call()
150            duration = time.monotonic() - starttime
151            _print(f'Succeeded in {duration:.2f}s.', color=(0, 1, 0))
152        except Exception as exc:
153            import traceback
154            duration = time.monotonic() - starttime
155            msg = (str(exc)
156                   if isinstance(exc, CleanError) else traceback.format_exc())
157            _print(msg, color=(1.0, 1.0, 0.3))
158            _print(f'Failed in {duration:.2f}s.', color=(1, 0, 0))
159            have_error[0] = True
160
161    try:
162        _print(f'Running network diagnostics...\n'
163               f'ua: {_ba.app.user_agent_string}\n'
164               f'time: {utc_now()}.')
165
166        if bool(False):
167            _print('\nRunning dummy success test...')
168            _print_test_results(_dummy_success)
169
170            _print('\nRunning dummy fail test...')
171            _print_test_results(_dummy_fail)
172
173        # V1 ping
174        baseaddr = _ba.get_master_server_address(source=0, version=1)
175        _print(f'\nContacting V1 master-server src0 ({baseaddr})...')
176        _print_test_results(lambda: _test_fetch(baseaddr))
177
178        # V1 alternate ping
179        baseaddr = _ba.get_master_server_address(source=1, version=1)
180        _print(f'\nContacting V1 master-server src1 ({baseaddr})...')
181        _print_test_results(lambda: _test_fetch(baseaddr))
182
183        _print(f'\nV1-test-log: {ba.app.net.v1_test_log}')
184
185        for srcid, result in sorted(ba.app.net.v1_ctest_results.items()):
186            _print(f'\nV1 src{srcid} result: {result}')
187
188        curv1addr = _ba.get_master_server_address(version=1)
189        _print(f'\nUsing V1 address: {curv1addr}')
190
191        _print('\nRunning V1 transaction...')
192        _print_test_results(_test_v1_transaction)
193
194        # V2 ping
195        baseaddr = _ba.get_master_server_address(version=2)
196        _print(f'\nContacting V2 master-server ({baseaddr})...')
197        _print_test_results(lambda: _test_fetch(baseaddr))
198
199        _print('\nComparing local time to V2 server...')
200        _print_test_results(_test_v2_time)
201
202        # Get V2 nearby zone
203        with ba.app.net.zone_pings_lock:
204            zone_pings = copy.deepcopy(ba.app.net.zone_pings)
205        nearest_zone = (None if not zone_pings else sorted(
206            zone_pings.items(), key=lambda i: i[1])[0])
207
208        if nearest_zone is not None:
209            nearstr = f'{nearest_zone[0]}: {nearest_zone[1]:.0f}ms'
210        else:
211            nearstr = '-'
212        _print(f'\nChecking nearest V2 zone ping ({nearstr})...')
213        _print_test_results(lambda: _test_nearby_zone_ping(nearest_zone))
214
215        _print('\nSending V2 cloud message...')
216        _print_test_results(_test_v2_cloud_message)
217
218        if have_error[0]:
219            _print('\nDiagnostics complete. Some diagnostics failed.',
220                   color=(10, 0, 0))
221        else:
222            _print('\nDiagnostics complete. Everything looks good!',
223                   color=(0, 1, 0))
224    except Exception:
225        import traceback
226        _print(
227            f'An unexpected error occurred during testing;'
228            f' please report this.\n'
229            f'{traceback.format_exc()}',
230            color=(1, 0, 0))
231
232
233def _dummy_success() -> None:
234    """Dummy success test."""
235    time.sleep(1.2)
236
237
238def _dummy_fail() -> None:
239    """Dummy fail test case."""
240    raise RuntimeError('fail-test')
241
242
243def _test_v1_transaction() -> None:
244    """Dummy fail test case."""
245    if _ba.get_v1_account_state() != 'signed_in':
246        raise RuntimeError('Not signed in.')
247
248    starttime = time.monotonic()
249
250    # Gets set to True on success or string on error.
251    results: list[Any] = [False]
252
253    def _cb(cbresults: Any) -> None:
254        # Simply set results here; our other thread acts on them.
255        if not isinstance(cbresults, dict) or 'party_code' not in cbresults:
256            results[0] = 'Unexpected transaction response'
257            return
258        results[0] = True  # Success!
259
260    def _do_it() -> None:
261        # Fire off a transaction with a callback.
262        _ba.add_transaction(
263            {
264                'type': 'PRIVATE_PARTY_QUERY',
265                'expire_time': time.time() + 20,
266            },
267            callback=_cb,
268        )
269        _ba.run_transactions()
270
271    ba.pushcall(_do_it, from_other_thread=True)
272
273    while results[0] is False:
274        time.sleep(0.01)
275        if time.monotonic() - starttime > 10.0:
276            raise RuntimeError('timed out')
277
278    # If we got left a string, its an error.
279    if isinstance(results[0], str):
280        raise RuntimeError(results[0])
281
282
283def _test_v2_cloud_message() -> None:
284    from dataclasses import dataclass
285    import bacommon.cloud
286
287    @dataclass
288    class _Results:
289        errstr: str | None = None
290        send_time: float | None = None
291        response_time: float | None = None
292
293    results = _Results()
294
295    def _cb(response: bacommon.cloud.PingResponse | Exception) -> None:
296        # Note: this runs in another thread so need to avoid exceptions.
297        results.response_time = time.monotonic()
298        if isinstance(response, Exception):
299            results.errstr = str(response)
300        if not isinstance(response, bacommon.cloud.PingResponse):
301            results.errstr = f'invalid response type: {type(response)}.'
302
303    def _send() -> None:
304        # Note: this runs in another thread so need to avoid exceptions.
305        results.send_time = time.monotonic()
306        ba.app.cloud.send_message_cb(bacommon.cloud.PingMessage(), _cb)
307
308    # This stuff expects to be run from the logic thread.
309    ba.pushcall(_send, from_other_thread=True)
310
311    wait_start_time = time.monotonic()
312    while True:
313        if results.response_time is not None:
314            break
315        time.sleep(0.01)
316        if time.monotonic() - wait_start_time > 10.0:
317            raise RuntimeError('Timeout waiting for cloud message response')
318    if results.errstr is not None:
319        raise RuntimeError(results.errstr)
320
321
322def _test_v2_time() -> None:
323    offset = ba.app.net.server_time_offset_hours
324    if offset is None:
325        raise RuntimeError('no time offset found;'
326                           ' perhaps unable to communicate with v2 server?')
327    if abs(offset) >= 2.0:
328        raise CleanError(
329            f'Your device time is off from world time by {offset:.1f} hours.\n'
330            'This may cause network operations to fail due to your device\n'
331            ' incorrectly treating SSL certificates as not-yet-valid, etc.\n'
332            'Check your device time and time-zone settings to fix this.\n')
333
334
335def _test_fetch(baseaddr: str) -> None:
336    # pylint: disable=consider-using-with
337    import urllib.request
338    response = urllib.request.urlopen(
339        urllib.request.Request(f'{baseaddr}/ping', None,
340                               {'User-Agent': _ba.app.user_agent_string}),
341        context=ba.app.net.sslcontext,
342        timeout=10.0,
343    )
344    if response.getcode() != 200:
345        raise RuntimeError(
346            f'Got unexpected response code {response.getcode()}.')
347    data = response.read()
348    if data != b'pong':
349        raise RuntimeError('Got unexpected response data.')
350
351
352def _test_nearby_zone_ping(nearest_zone: tuple[str, float] | None) -> None:
353    """Try to ping nearest v2 zone."""
354    if nearest_zone is None:
355        raise RuntimeError('No nearest zone.')
356    if nearest_zone[1] > 500:
357        raise RuntimeError('Ping too high.')
358
359
360class NetValTestingWindow(TestingWindow):
361    """Window to test network related settings."""
362
363    def __init__(self, transition: str = 'in_right'):
364
365        entries = [
366            {
367                'name': 'bufferTime',
368                'label': 'Buffer Time',
369                'increment': 1.0
370            },
371            {
372                'name': 'delaySampling',
373                'label': 'Delay Sampling',
374                'increment': 1.0
375            },
376            {
377                'name': 'dynamicsSyncTime',
378                'label': 'Dynamics Sync Time',
379                'increment': 10
380            },
381            {
382                'name': 'showNetInfo',
383                'label': 'Show Net Info',
384                'increment': 1
385            },
386        ]
387        super().__init__(
388            title=ba.Lstr(resource='settingsWindowAdvanced.netTestingText'),
389            entries=entries,
390            transition=transition,
391            back_call=lambda: NetTestingWindow(transition='in_left'))