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