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