'''
Shared-memory persistence regressions.

'''
from collections.abc import (
    AsyncIterator,
    Iterator,
)
from contextlib import asynccontextmanager
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import numpy as np
import polars as pl
import pytest

from piker import tsp
from piker.data import def_iohlcv_fields
from piker.storage import cli as storage_cli
from piker.storage.cli import (
    _shm_period_and_invalid_count,
    _summarize_shm_frame,
    ldshm,
)


def test_ldshm_detects_nulls_without_skewing_period() -> None:
    '''
    Null SHM slots must stop persistence without skewing cadence.

    A live QQQ buffer contained an interior run of zero-time rows.
    ``ldshm`` included them in period inference and then sent them to
    NativeDB, which correctly rejected the replacement. Build a
    60-second series with the same interior hole and enough
    bars around the gap. Prove inspection counts every null while
    inferring the published series' 60-second period; the
    command uses that count to skip both persistence and SHM reload.

    '''
    times = np.array([
        60,
        120,
        0,
        0,
        300,
        360,
        420,
    ])

    period_s, invalid_count = _shm_period_and_invalid_count(times)

    assert period_s == 60
    assert invalid_count == 2


def test_ldshm_period_is_an_observed_step() -> None:
    '''
    Sparse timestamps must not invent a storage timeframe.

    The arithmetic median of two steps can produce a cadence that
    occurs nowhere in the source series. Such a value can select the
    wrong NativeDB timeframe and hide the gap. Arrange tied 60 and
    120-second steps and prove the deterministic tie-break chooses
    the smaller observed period.

    '''
    times = np.array([60, 120, 240])

    period_s, invalid_count = _shm_period_and_invalid_count(times)

    assert period_s == 60
    assert invalid_count == 0


def test_ldshm_counts_every_invalid_timestamp() -> None:
    '''
    Corrupt timestamps stop persistence just like null SHM slots.

    Exact zero denotes an unpublished slot, while negative, NaN, and
    infinite values indicate corruption. NativeDB rejects them all,
    but allowing a replacement attempt would still abort ``ldshm``.
    Mix each invalid class into a regular series and prove the same
    snapshot inspection counts all four while preserving cadence.

    '''
    times = np.array([
        60,
        120,
        0,
        -1,
        np.nan,
        np.inf,
        180,
    ])

    period_s, invalid_count = _shm_period_and_invalid_count(times)

    assert period_s == 60
    assert invalid_count == 4


def test_ldshm_invalid_snapshot_never_reaches_storage(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    '''
    Invalid snapshots stop before analyzers and storage mutation.

    The original QQQ failure passed zero-time rows through dedupe and
    into NativeDB. An intermediate fix called null-segment analysis
    before its guard, allowing corrupt indexes to abort the command.
    Supply one immutable Polars snapshot containing that combination,
    then replace dedupe and writes with failing spies. Completing the
    command proves its guard runs first, preventing persistence or
    SHM reload without relying on task timing.

    '''
    shm_df = pl.DataFrame({
        'index': [100, 500, 104],
        'time': [60, 0, 180],
    })

    @asynccontextmanager
    async def open_runtime(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[None]:
        '''
        Yield a runtime-free command context.

        '''
        yield

    async def fail_write(
        *args: Any,
        **kwargs: Any,
    ) -> None:
        '''
        Fail if invalid SHM reaches NativeDB.

        '''
        raise AssertionError('invalid SHM reached NativeDB')

    client = SimpleNamespace(write_ohlcv=fail_write)

    @asynccontextmanager
    async def open_storage(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[tuple[SimpleNamespace, SimpleNamespace]]:
        '''
        Yield storage objects with a failing writer.

        '''
        yield SimpleNamespace(), client

    @asynccontextmanager
    async def open_annotations(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[SimpleNamespace]:
        '''
        Yield an inert annotation controller.

        '''
        yield SimpleNamespace()

    def iter_shms(
        fqme: str,
    ) -> Iterator[tuple[Path, SimpleNamespace, pl.DataFrame]]:
        '''
        Yield the single captured invalid snapshot.

        '''
        yield Path('qqq.rt'), SimpleNamespace(), shm_df

    def dedupe(*args: Any, **kwargs: Any) -> None:
        '''
        Fail if invalid SHM reaches deduplication.

        '''
        raise AssertionError('invalid SHM reached dedupe')

    async def pause() -> None:
        '''
        Replace the command's final interactive pause.

        '''

    monkeypatch.setattr(
        storage_cli,
        'open_piker_runtime',
        open_runtime,
    )
    monkeypatch.setattr(
        storage_cli,
        'open_storage_client',
        open_storage,
    )
    monkeypatch.setattr(tsp, 'iter_dfs_from_shms', iter_shms)
    monkeypatch.setattr(tsp, 'dedupe_ohlcv_smart', dedupe)
    monkeypatch.setattr(storage_cli.tractor, 'pause', pause)

    from piker.ui import _remote_ctl
    monkeypatch.setattr(
        _remote_ctl,
        'open_annot_ctl',
        open_annotations,
    )

    ldshm('qqq.nasdaq.ib')


@pytest.mark.parametrize('write_parquet', [False, True])
def test_ldshm_no_reload_uses_deduped_markup_frame(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    write_parquet: bool,
) -> None:
    '''
    Disabling SHM reload must not require a reload payload.

    ``ldshm`` created ``new`` only in reload but used it for markup.
    No-write and no-reload modes crashed, the latter after a durable
    write. Arrange one gap, exercise
    write modes with reload disabled, and prove markup receives the
    deduplicated frame without any SHM mutation.

    '''
    array = np.zeros(
        3,
        dtype=np.dtype(def_iohlcv_fields),
    )
    array['index'] = np.arange(3)
    array['time'] = [60, 120, 240]
    array['open'] = [10, 11, 12]
    array['high'] = [12, 13, 14]
    array['low'] = [8, 9, 10]
    array['close'] = [11, 12, 13]
    array['volume'] = [100, 101, 102]
    frame = tsp.np2pl(array)
    shm = SimpleNamespace(array=array)
    shmfile = Path('piker.datad[aaaaaaaa-aaaa-aa].x.test.hist')
    writes: list[pl.DataFrame] = []
    markups: list[pl.DataFrame] = []

    @asynccontextmanager
    async def open_runtime(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[None]:
        '''
        Yield a runtime-free command context.

        '''
        yield

    class Client:
        '''
        Record optional durable writes.

        '''
        async def write_ohlcv(
            self,
            fqme: str,
            ohlcv: pl.DataFrame,
            timeframe: int,
        ) -> Path:
            '''
            Persist the frame for the command's read-back step.

            '''
            writes.append(ohlcv)
            path = tmp_path / 'frame.parquet'
            ohlcv.write_parquet(path)
            return path

    @asynccontextmanager
    async def open_storage(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[tuple[SimpleNamespace, Client]]:
        '''
        Yield the recording storage client.

        '''
        yield SimpleNamespace(), Client()

    @asynccontextmanager
    async def open_annotations(
        *args: Any,
        **kwargs: Any,
    ) -> AsyncIterator[SimpleNamespace]:
        '''
        Yield an inert annotation controller.

        '''
        yield SimpleNamespace()

    def iter_shms(
        fqme: str,
    ) -> Iterator[tuple[Path, SimpleNamespace, pl.DataFrame]]:
        '''
        Yield the one valid history snapshot.

        '''
        yield shmfile, shm, frame

    async def markup_gaps(
        fqme: str,
        period_s: int,
        actl: SimpleNamespace,
        new_df: pl.DataFrame,
        step_gaps: pl.DataFrame,
    ) -> dict:
        '''
        Capture the frame used for non-mutating markup.

        '''
        markups.append(new_df)
        return {}

    async def pause() -> None:
        '''
        Replace the command's final interactive pause.

        '''

    monkeypatch.setattr(
        storage_cli,
        'open_piker_runtime',
        open_runtime,
    )
    monkeypatch.setattr(
        storage_cli,
        'open_storage_client',
        open_storage,
    )
    monkeypatch.setattr(tsp, 'iter_dfs_from_shms', iter_shms)
    monkeypatch.setattr(tsp._annotate, 'markup_gaps', markup_gaps)
    monkeypatch.setattr(storage_cli.tractor, 'pause', pause)

    from piker.ui import _remote_ctl
    monkeypatch.setattr(
        _remote_ctl,
        'open_annot_ctl',
        open_annotations,
    )

    ldshm(
        'x.test',
        write_parquet=write_parquet,
        reload_parquet_to_shm=False,
    )

    assert len(writes) == int(write_parquet)
    assert len(markups) == 1
    assert markups[0]['time'].to_list() == [60, 120, 240]


@pytest.mark.parametrize(
    'times',
    [
        np.array([0, 0]),
        np.array([60]),
        np.array([60, 60]),
    ],
)
def test_ldshm_skips_frame_without_positive_cadence(
    times: np.ndarray,
) -> None:
    '''
    Degenerate buffers are skipped before indexing or persistence.

    The old inference indexed final timestamps and left its median
    undefined for short buffers. Fully unpublished, one-row, and
    duplicate-only frames could therefore fail before the command's
    guard. Exercise each shape and prove inspection
    returns no cadence, which directs ``ldshm`` to skip the buffer.

    '''
    period_s, _ = _shm_period_and_invalid_count(times)

    assert period_s is None


def test_shm_buffer_discovery_uses_exact_identities(
    tmp_path: Path,
) -> None:
    '''
    SHM discovery must not mix substring matches or index companions.

    ``ldshm`` globbed ``*fqme*`` and could process stale and
    active objects for another instrument whose name merely contained
    requested text. Create two generations, both OHLCV kinds, index
    companions, and a substring collision. Prove discovery returns
    exact parsed identities in deterministic filename order.

    '''
    names = [
        'piker.datad[bbbbbbbb-bbbb-bb].qqq.nasdaq.ib.rt',
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist',
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist_first',
        'piker.datad[aaaaaaaa-aaaa-aa].myqqq.nasdaq.ib.rt',
        'unrelated.qqq.nasdaq.ib.rt',
    ]
    for name in names:
        (tmp_path / name).touch()

    refs = list(tsp.iter_shm_buffer_refs(
        fqme='qqq.nasdaq.ib',
        shmdir=tmp_path,
    ))

    assert [ref.path.name for ref in refs] == [
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist',
        'piker.datad[bbbbbbbb-bbbb-bb].qqq.nasdaq.ib.rt',
    ]
    assert [
        (ref.service, ref.generation, ref.kind)
        for ref in refs
    ] == [
        ('datad', 'aaaaaaaa-aaaa-aa', 'hist'),
        ('datad', 'bbbbbbbb-bbbb-bb', 'rt'),
    ]


def test_shm_summary_reports_invalid_rows_and_ranked_gaps() -> None:
    '''
    Read-only SHM diagnostics must preserve the live triage evidence.

    The QQQ investigation needed cadence, invalid bounds, and largest
    timestamp jumps without mutation. Build one zero row and two gaps;
    prove summary excludes invalid timestamps and ranks durations.

    '''
    frame = np.zeros(
        5,
        dtype=np.dtype(def_iohlcv_fields),
    )
    frame['index'] = np.arange(100, 105)
    frame['time'] = [1, 0, 3, 4, 10]
    ref = tsp.ShmBufferRef(
        service='datad',
        generation='aaaaaaaa-aaaa-aa',
        fqme='qqq.nasdaq.ib',
        kind='rt',
        path=Path(
            'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt'
        ),
    )

    report = _summarize_shm_frame(ref, frame, max_gaps=2)

    assert report['period_s'] == 1
    assert report['observed_period_s'] == 1
    assert report['invalid_count'] == 1
    assert report['invalid_index_bounds'] == [101, 101]
    assert report['gap_count'] == 2
    assert [
        gap['delta_s']
        for gap in report['largest_gaps']
    ] == [6, 2]


def test_shm_summary_reports_ordering_corruption() -> None:
    '''
    Duplicate and reversed timestamps remain visible in diagnostics.

    Positive-gap detection considers both defects gap-free. Build
    one exact RT snapshot with a duplicate and reversal and prove the
    read-only report distinguishes both without classifying values as
    unpublished slots.

    '''
    frame = np.zeros(
        5,
        dtype=np.dtype(def_iohlcv_fields),
    )
    frame['index'] = np.arange(5)
    frame['time'] = [1, 2, 2, 1, 3]
    ref = tsp.ShmBufferRef(
        service='datad',
        generation='aaaaaaaa-aaaa-aa',
        fqme='qqq.nasdaq.ib',
        kind='rt',
        path=Path(
            'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt'
        ),
    )

    report = _summarize_shm_frame(ref, frame, max_gaps=2)

    assert report['invalid_count'] == 0
    assert report['duplicate_step_count'] == 1
    assert report['reversed_step_count'] == 1


def test_shm_dtype_resolution_matches_allocator_layout(
    tmp_path: Path,
) -> None:
    '''
    Inspector actors must mirror history allocation dtype lookup.

    Allocation reads ``_ohlc_dtype`` from the package; IB does not
    export its API-internal extended dtype there.
    Create four production-shaped canonical rows and prove inspection
    resolves the same layout without searching unrelated submodules.

    '''
    path = tmp_path / (
        'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt'
    )
    path.write_bytes(b'\0' * (4 * 56))
    ref = tsp.ShmBufferRef(
        service='datad',
        generation='aaaaaaaa-aaaa-aa',
        fqme='qqq.nasdaq.ib',
        kind='rt',
        path=path,
    )

    dtype = tsp.resolve_shm_buffer_dtype(ref, size=4)

    assert dtype.itemsize == 56
    assert dtype.names == (
        'index',
        'time',
        'open',
        'high',
        'low',
        'close',
        'volume',
    )


@pytest.mark.parametrize(
    'name',
    [
        'piker..datad[aaaaaaaa-aaaa-aa].x.test.rt',
        'piker.datad[short].x.test.rt',
        'piker.datad[aaaaaaaa-aaaa-aa].x\\test.rt',
        'piker.datad[aaaaaaaa-aaaa-aa].x[y].test.rt',
    ],
)
def test_shm_parser_rejects_forged_identities(
    name: str,
) -> None:
    '''
    SHM selectors must accept only generated Piker identities.

    Reject malformed service, generation, and FQME components so a
    same-user object that merely resembles OHLCV can not become an
    ``ldshm --shm-name`` target.

    '''
    assert tsp.parse_shm_buffer_path(Path(name)) is None