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