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