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