from collections.abc import Callable
from pathlib import Path
import socket
from time import monotonic
from types import SimpleNamespace

import numpy as np
from PyQt6.QtWidgets import QApplication
import pytest
from pytestqt.qtbot import QtBot
import tractor
from tractor.ipc import _shm

from piker import config
from piker.brokers import get_brokermod
from piker.ui._app import _async_main
from piker.ui._exec import start_qtractor
from piker.ui._widget import GodWidget


def test_kraken_chart_boots_and_closes(
    request: pytest.FixtureRequest,
    isolated_ui_state: SimpleNamespace,
    qapp: QApplication,
    qtbot: QtBot,
    reg_addr: tuple[str, int],
    monkeypatch: pytest.MonkeyPatch,
    record_property: Callable[[str, object], None],

) -> None:
    '''
    Qualify production chart boot against public Kraken, not mocks.

    A blocking Qt host and hidden Trio errors previously prevented
    testing the composed chart. Explicit selection starts GodWidget
    and `_async_main` with paper EMS, an empty owned broker config,
    and the test's unique registry. No orders or input are submitted.
    Readiness races guest completion so a provider/startup exception
    is unwrapped rather than replaced by a readiness timeout.

    Nonempty historical/realtime visualizations and a later price
    change prove actual feed consumption. Registry snapshots check
    paperboi and absence of brokerd; exact chart SHM names and actor
    endpoints are retained for non-mutating teardown checks. These
    snapshots cannot rule out short-lived actors between polls or
    account for SHM never published to a chart before early failure.

    Every exit requests cancellation and pumps Qt to guest completion
    before pytest-qt may destroy widgets. Cleanup and guest failures
    are retained together rather than hiding either original error.

    '''
    if request.config.getoption('--live-provider') != 'kraken':
        pytest.skip('requires --live-provider=kraken')

    fqme: str = 'xbtusd.spot.kraken'
    confdir: Path = isolated_ui_state.test_config_dir
    # A missing brokers.toml loads templates with example accounts.
    (confdir / 'brokers.toml').write_text('', encoding='utf-8')
    for name in ('_config_dir', '_click_config_dir'):
        monkeypatch.setattr(config, name, confdir)
    monkeypatch.setattr(
        config,
        '_watchlists_data_path',
        confdir / 'watchlists.json',
    )
    assert config.load('brokers')[0] == {}

    owned_actors: dict[tuple[str, str], list] = {}
    shm_names: set[str] = set()
    errors: list[BaseException] = []
    started: float = monotonic()
    session = start_qtractor(
        func=_async_main,
        args=(
            [fqme],
            {'kraken': get_brokermod('kraken')},
            'warning',
            'paper',
        ),
        main_widget_type=GodWidget,
        tractor_kwargs={
            'registry_addrs': [reg_addr],
            'tractor_runtime_overrides': {
                'piker_test_dir': str(confdir),
            },
            'loglevel': 'warning',
        },
        app=qapp,
    )

    def observe_runtime() -> None:
        '''
        Retain identities from this guest's local Registrar only.

        '''
        actor = tractor.current_actor(err_on_no_runtime=False)
        if actor is not None:
            assert actor.is_registrar
            assert reg_addr in actor.accept_addrs
            owned_actors.update(actor._registry)
            assert not any(
                uid[0].startswith('brokerd.')
                for uid in owned_actors
            )
        for attr in ('hist_linked', 'rt_linked'):
            linked = getattr(session.main_widget, attr, None)
            chart = getattr(linked, 'chart', None)
            viz = getattr(chart, 'main_viz', None)
            if viz is not None:
                token: dict = viz.shm.token
                shm_names.update(
                    token[key] for key in (
                        'shm_name',
                        'shm_first_index_name',
                        'shm_last_index_name',
                    )
                )

    def boot_finished() -> bool:
        observe_runtime()
        return session.is_ready or session.outcome is not None

    try:
        qtbot.addWidget(session.window)
        qtbot.waitUntil(boot_finished, timeout=120_000)
        record_property('boot_seconds', monotonic() - started)
        if session.outcome is None:
            godw = session.main_widget
            assert session.is_ready
            assert godw.exec_mode == 'paper'
            assert config.get_conf_dir() == confdir
            assert tractor.runtime._state._runtime_vars[
                'piker_vars'
            ]['piker_test_dir'] == str(confdir)
            names: set[str] = {uid[0] for uid in owned_actors}
            assert {
                'pikerd', 'datad.kraken', 'samplerd',
                'emsd', 'paperboi.kraken',
            } <= names
            assert set(godw.rt_linked.mode.trackers) == {'paper'}
            assert not godw.rt_linked.mode.client._sent_orders
            assert session.window.windowTitle().startswith(fqme)
            for linked in (godw.hist_linked, godw.rt_linked):
                assert linked.mkt.fqme == fqme
                viz = linked.chart.main_viz
                data = viz.shm.array
                assert len(data) > 1
                assert np.isfinite(data['close']).all()
                assert (data['close'] > 0).any()

            rt_shm = godw.rt_linked.chart.main_viz.shm
            before = rt_shm.array[-1].copy()

            def live_update_or_exit() -> bool:
                observe_runtime()
                if session.outcome is not None:
                    return True
                # A sampler clock tick alone must not satisfy this.
                return bool(
                    rt_shm.array[-1]['close'] != before['close']
                )

            qtbot.waitUntil(live_update_or_exit, timeout=60_000)
    except BaseException as exc:
        errors.append(exc)
    finally:
        session.request_shutdown()
        try:
            qtbot.waitUntil(
                lambda: session.outcome is not None,
                timeout=60_000,
            )
            session.outcome.unwrap()
        except BaseException as exc:
            errors.append(exc)
        if session.outcome is not None:
            # Early failure can leave these outside Qt parent trees.
            for attr in ('search', 'hist_linked', 'rt_linked'):
                widget = getattr(session.main_widget, attr, None)
                if widget is not None and widget.parent() is None:
                    qtbot.addWidget(widget)
        record_property('owned_actors', repr(owned_actors))
        record_property('chart_shm_names', sorted(shm_names))

    # Check exact survivors even when boot assertions already failed.
    if tractor.current_actor(err_on_no_runtime=False) is not None:
        errors.append(AssertionError('Guest runtime survived'))
    for name in shm_names:
        try:
            segment = _shm.SharedMemory(name=name, create=False)
        except FileNotFoundError:
            continue
        segment.close()
        errors.append(AssertionError(
            f'Chart SHM survived guest completion: {name}'
        ))
    endpoints: set[tuple] = {reg_addr}
    for addrs in owned_actors.values():
        endpoints.update(map(tuple, addrs))
    for addr in endpoints:
        with socket.socket() as sock:
            sock.settimeout(1)
            if sock.connect_ex(addr) == 0:
                errors.append(AssertionError(
                    f'Owned actor endpoint survived: {addr}'
                ))

    if errors:
        if len(errors) == 1:
            raise errors[0]
        raise BaseExceptionGroup('Chart and shutdown failed', errors)
    assert session.is_ready, 'guest returned before readiness'