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
166    from efro.util import utc_now
167
168    have_error = [False]
169
170    # We're running in a background thread but UI stuff needs to run
171    # in the logic thread; give ourself a way to pass stuff to it.
172    def _print(
173        text: str, color: tuple[float, float, float] | None = None
174    ) -> None:
175        def _print_in_logic_thread() -> None:
176            win = weakwin()
177            if win is not None:
178                win.print(text, (1.0, 1.0, 1.0) if color is None else color)
179
180        bui.pushcall(_print_in_logic_thread, from_other_thread=True)
181
182    def _print_test_results(call: Callable[[], Any]) -> bool:
183        """Run the provided call, print result, & return success."""
184        starttime = time.monotonic()
185        try:
186            call()
187            duration = time.monotonic() - starttime
188            _print(f'Succeeded in {duration:.2f}s.', color=(0, 1, 0))
189            return True
190        except Exception as exc:
191            import traceback
192
193            duration = time.monotonic() - starttime
194            msg = (
195                str(exc)
196                if isinstance(exc, CleanError)
197                else traceback.format_exc()
198            )
199            _print(msg, color=(1.0, 1.0, 0.3))
200            _print(f'Failed in {duration:.2f}s.', color=(1, 0, 0))
201            have_error[0] = True
202            return False
203
204    try:
205        plus = bui.app.plus
206        assert plus is not None
207
208        assert bui.app.classic is not None
209
210        _print(
211            f'Running network diagnostics...\n'
212            f'ua: {bui.app.classic.legacy_user_agent_string}\n'
213            f'time: {utc_now()}.'
214        )
215
216        if bool(False):
217            _print('\nRunning dummy success test...')
218            _print_test_results(_dummy_success)
219
220            _print('\nRunning dummy fail test...')
221            _print_test_results(_dummy_fail)
222
223        # V1 ping
224        baseaddr = plus.get_master_server_address(source=0, version=1)
225        _print(f'\nContacting V1 master-server src0 ({baseaddr})...')
226        v1worked = _print_test_results(lambda: _test_fetch(baseaddr))
227
228        # V1 alternate ping (only if primary fails since this often fails).
229        if v1worked:
230            _print('\nSkipping V1 master-server src1 test since src0 worked.')
231        else:
232            baseaddr = plus.get_master_server_address(source=1, version=1)
233            _print(f'\nContacting V1 master-server src1 ({baseaddr})...')
234            _print_test_results(lambda: _test_fetch(baseaddr))
235
236        if 'none succeeded' in bui.app.net.v1_test_log:
237            _print(
238                f'\nV1-test-log failed: {bui.app.net.v1_test_log}',
239                color=(1, 0, 0),
240            )
241            have_error[0] = True
242        else:
243            _print(f'\nV1-test-log ok: {bui.app.net.v1_test_log}')
244
245        for srcid, result in sorted(bui.app.net.v1_ctest_results.items()):
246            _print(f'\nV1 src{srcid} result: {result}')
247
248        curv1addr = plus.get_master_server_address(version=1)
249        _print(f'\nUsing V1 address: {curv1addr}')
250
251        _print('\nRunning V1 transaction...')
252        _print_test_results(_test_v1_transaction)
253
254        # V2 ping
255        baseaddr = plus.get_master_server_address(version=2)
256        _print(f'\nContacting V2 master-server ({baseaddr})...')
257        _print_test_results(lambda: _test_fetch(baseaddr))
258
259        _print('\nComparing local time to V2 server...')
260        _print_test_results(_test_v2_time)
261
262        # Get V2 nearby zone
263        with bui.app.net.zone_pings_lock:
264            zone_pings = copy.deepcopy(bui.app.net.zone_pings)
265        nearest_zone = (
266            None
267            if not zone_pings
268            else sorted(zone_pings.items(), key=lambda i: i[1])[0]
269        )
270
271        if nearest_zone is not None:
272            nearstr = f'{nearest_zone[0]}: {nearest_zone[1]:.0f}ms'
273        else:
274            nearstr = '-'
275        _print(f'\nChecking nearest V2 zone ping ({nearstr})...')
276        _print_test_results(lambda: _test_nearby_zone_ping(nearest_zone))
277
278        _print('\nSending V2 cloud message...')
279        _print_test_results(_test_v2_cloud_message)
280
281        if have_error[0]:
282            _print(
283                '\nDiagnostics complete. Some diagnostics failed.',
284                color=(10, 0, 0),
285            )
286        else:
287            _print(
288                '\nDiagnostics complete. Everything looks good!',
289                color=(0, 1, 0),
290            )
291    except Exception:
292        import traceback
293
294        _print(
295            f'An unexpected error occurred during testing;'
296            f' please report this.\n'
297            f'{traceback.format_exc()}',
298            color=(1, 0, 0),
299        )
300
301
302def _dummy_success() -> None:
303    """Dummy success test."""
304    time.sleep(1.2)
305
306
307def _dummy_fail() -> None:
308    """Dummy fail test case."""
309    raise RuntimeError('fail-test')
310
311
312def _test_v1_transaction() -> None:
313    """Dummy fail test case."""
314    plus = bui.app.plus
315    assert plus is not None
316
317    if plus.get_v1_account_state() != 'signed_in':
318        raise RuntimeError('Not signed in.')
319
320    starttime = time.monotonic()
321
322    # Gets set to True on success or string on error.
323    results: list[Any] = [False]
324
325    def _cb(cbresults: Any) -> None:
326        # Simply set results here; our other thread acts on them.
327        if not isinstance(cbresults, dict) or 'party_code' not in cbresults:
328            results[0] = 'Unexpected transaction response'
329            return
330        results[0] = True  # Success!
331
332    def _do_it() -> None:
333        assert plus is not None
334        # Fire off a transaction with a callback.
335        plus.add_v1_account_transaction(
336            {
337                'type': 'PRIVATE_PARTY_QUERY',
338                'expire_time': time.time() + 20,
339            },
340            callback=_cb,
341        )
342        plus.run_v1_account_transactions()
343
344    bui.pushcall(_do_it, from_other_thread=True)
345
346    while results[0] is False:
347        time.sleep(0.01)
348        if time.monotonic() - starttime > MAX_TEST_SECONDS:
349            raise RuntimeError(
350                f'test timed out after {MAX_TEST_SECONDS} seconds'
351            )
352
353    # If we got left a string, its an error.
354    if isinstance(results[0], str):
355        raise RuntimeError(results[0])
356
357
358def _test_v2_cloud_message() -> None:
359    from dataclasses import dataclass
360    import bacommon.cloud
361
362    @dataclass
363    class _Results:
364        errstr: str | None = None
365        send_time: float | None = None
366        response_time: float | None = None
367
368    results = _Results()
369
370    def _cb(response: bacommon.cloud.PingResponse | Exception) -> None:
371        # Note: this runs in another thread so need to avoid exceptions.
372        results.response_time = time.monotonic()
373        if isinstance(response, Exception):
374            results.errstr = str(response)
375        if not isinstance(response, bacommon.cloud.PingResponse):
376            results.errstr = f'invalid response type: {type(response)}.'
377
378    def _send() -> None:
379        # Note: this runs in another thread so need to avoid exceptions.
380        results.send_time = time.monotonic()
381        assert bui.app.plus is not None
382        bui.app.plus.cloud.send_message_cb(bacommon.cloud.PingMessage(), _cb)
383
384    # This stuff expects to be run from the logic thread.
385    bui.pushcall(_send, from_other_thread=True)
386
387    wait_start_time = time.monotonic()
388    while True:
389        if results.response_time is not None:
390            break
391        time.sleep(0.01)
392        if time.monotonic() - wait_start_time > MAX_TEST_SECONDS:
393            raise RuntimeError(
394                f'Timeout ({MAX_TEST_SECONDS} seconds)'
395                f' waiting for cloud message response'
396            )
397    if results.errstr is not None:
398        raise RuntimeError(results.errstr)
399
400
401def _test_v2_time() -> None:
402    offset = bui.app.net.server_time_offset_hours
403    if offset is None:
404        raise RuntimeError(
405            'no time offset found;'
406            ' perhaps unable to communicate with v2 server?'
407        )
408    if abs(offset) >= 2.0:
409        raise CleanError(
410            f'Your device time is off from world time by {offset:.1f} hours.\n'
411            'This may cause network operations to fail due to your device\n'
412            ' incorrectly treating SSL certificates as not-yet-valid, etc.\n'
413            'Check your device time and time-zone settings to fix this.\n'
414        )
415
416
417def _test_fetch(baseaddr: str) -> None:
418    # pylint: disable=consider-using-with
419    import urllib.request
420
421    assert bui.app.classic is not None
422    response = urllib.request.urlopen(
423        urllib.request.Request(
424            f'{baseaddr}/ping',
425            None,
426            {'User-Agent': bui.app.classic.legacy_user_agent_string},
427        ),
428        context=bui.app.net.sslcontext,
429        timeout=MAX_TEST_SECONDS,
430    )
431    if response.getcode() != 200:
432        raise RuntimeError(
433            f'Got unexpected response code {response.getcode()}.'
434        )
435    data = response.read()
436    if data != b'pong':
437        raise RuntimeError('Got unexpected response data.')
438
439
440def _test_nearby_zone_ping(nearest_zone: tuple[str, float] | None) -> None:
441    """Try to ping nearest v2 zone."""
442    if nearest_zone is None:
443        raise RuntimeError('No nearest zone.')
444    if nearest_zone[1] > 500:
445        raise RuntimeError('Ping too high.')
446
447
448class NetValTestingWindow(TestingWindow):
449    """Window to test network related settings."""
450
451    def __init__(self, transition: str = 'in_right'):
452        entries = [
453            {'name': 'bufferTime', 'label': 'Buffer Time', 'increment': 1.0},
454            {
455                'name': 'delaySampling',
456                'label': 'Delay Sampling',
457                'increment': 1.0,
458            },
459            {
460                'name': 'dynamicsSyncTime',
461                'label': 'Dynamics Sync Time',
462                'increment': 10,
463            },
464            {'name': 'showNetInfo', 'label': 'Show Net Info', 'increment': 1},
465        ]
466        super().__init__(
467            title=bui.Lstr(resource='settingsWindowAdvanced.netTestingText'),
468            entries=entries,
469            transition=transition,
470            back_call=lambda: NetTestingWindow(transition='in_left'),
471        )