'''
Operator backfill-audit helper regressions.

'''
from hashlib import sha256
import json
from pathlib import Path
import runpy
import subprocess

import numpy as np
import polars as pl
import pytest

from piker.storage._audit import audit_ohlcv_parquet


def load_helpers() -> dict:
    '''
    Load the sourceable xonsh file as its pure-Python namespace.

    '''
    path = (
        Path(__file__).parents[1]
        /'snippets'
        /'nativedb_backfill_audit.xsh'
    )
    return runpy.run_path(path)


def mk_frame(
    times: tuple[int, ...],
    closes: tuple[float, ...]|None = None,
) -> pl.DataFrame:
    '''
    Build canonical history for evidence comparisons.

    '''
    size: int = len(times)
    closes = closes or tuple(float(value) for value in times)
    return pl.DataFrame({
        'index': pl.Series(np.arange(size), dtype=pl.Int64),
        'time': pl.Series(times, dtype=pl.Int64),
        'open': pl.Series(np.arange(size), dtype=pl.Float64),
        'high': pl.Series(np.arange(size) + 1, dtype=pl.Float64),
        'low': pl.Series(np.arange(size), dtype=pl.Float64),
        'close': pl.Series(closes, dtype=pl.Float64),
        'volume': pl.Series(np.arange(size), dtype=pl.Float64),
    })


def init_case(
    helpers: dict,
    run_root: Path,
) -> None:
    '''
    Initialize one fixed synthetic qualification case.

    '''
    helpers['bfq_init'](run_root, validate_piker=False)
    helpers['bfq_record_case'](
        run_root,
        fqme='x.test',
        period_s=60,
        provider='fake',
        symptom='synthetic lifecycle regression',
    )


def write_phase_evidence(
    run_root: Path,
    phase: str,
    path: Path,
    fqme: str = 'x.test',
) -> None:
    '''
    Write checksum-bound report and manual metadata for a snapshot.

    '''
    report: dict = audit_ohlcv_parquet(
        path,
        fqme=fqme,
        period_s=60,
    )
    evidence: Path = run_root / 'evidence'
    report_path: Path = evidence / f'{phase}.json'
    report_path.write_text(
        json.dumps(report, indent=2, sort_keys=True) + '\n'
    )
    manual: dict = {
        'phase': phase,
        'run_root': str(run_root.resolve()),
        'snapshot_path': path.name,
        'report_sha256': sha256(
            report_path.read_bytes()
        ).hexdigest(),
    }
    (evidence / f'{phase}.manual.json').write_text(
        json.dumps(manual, indent=2, sort_keys=True) + '\n'
    )


def test_disposable_seed_is_archived_before_fresh(
    tmp_path: Path,
) -> None:
    '''
    Fresh qualification must never remove the operator's source file.

    The workflow starts from known-bad production evidence but needs
    an empty disposable NativeDB for a full provider backfill.
    Initialize a marked root, seed by copy, then clear for fresh and
    prove both the original and evidence copy survive while only the
    disposable active path is moved into the evidence directory.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    source: Path = tmp_path / 'known-bad.parquet'
    mk_frame((60, 120)).write_parquet(source)

    helpers['bfq_init'](run_root, validate_piker=False)
    active: Path = helpers['bfq_seed'](
        run_root,
        'x.test',
        60,
        source,
    )
    archived: Path = helpers['bfq_clear_for_fresh'](
        run_root,
        'x.test',
        60,
    )

    assert source.is_file()
    assert (run_root / 'evidence' / 'seed.parquet').is_file()
    assert archived.is_file()
    assert not active.exists()


def test_seed_active_copy_comes_from_captured_evidence(
    tmp_path: Path,
) -> None:
    '''
    Source replacement must not split baseline and replay bytes.

    Production writers replace Parquet atomically, so copying the
    source twice can capture one generation as evidence and replay
    another. Replace the source after the first copy, then prove the
    the active disposable file still derives
    from the captured seed rather than reopening the changed source.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    source: Path = tmp_path / 'source.parquet'
    original: bytes
    mk_frame((60, 120)).write_parquet(source)
    original = source.read_bytes()
    helpers['bfq_init'](run_root, validate_piker=False)
    real_copy = helpers['_bfq_copy_exclusive']
    copied: int = 0

    def replace_after_capture(
        src: Path,
        dst: Path,
        mode: int|None = None,
    ) -> None:
        nonlocal copied
        real_copy(src, dst, mode=mode)
        copied += 1
        if copied == 1:
            mk_frame((60, 180)).write_parquet(source)

    helpers['bfq_seed'].__globals__[
        '_bfq_copy_exclusive'
    ] = replace_after_capture
    active: Path = helpers['bfq_seed'](
        run_root,
        'x.test',
        60,
        source,
    )

    assert source.read_bytes() != original
    assert active.read_bytes() == original
    assert (
        run_root / 'evidence' / 'seed.parquet'
    ).read_bytes() == original


def test_seed_copy_failure_removes_partial_active_file(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    '''
    Failed replay copy must not leave an active partial Parquet.

    Seed evidence can complete before the disposable active copy runs
    out of space. A partial active path blocks retry and lets a chart
    mistake truncated bytes for history. Fail the second copy
    after a short write and prove active storage is removed while the
    completed seed remains exact.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    source: Path = tmp_path / 'source.parquet'
    mk_frame((60, 120)).write_parquet(source)
    before: bytes = source.read_bytes()
    helpers['bfq_init'](run_root, validate_piker=False)
    real_copy = helpers['shutil'].copyfileobj
    copies: int = 0

    def fail_second(source_file, destination_file) -> None:
        nonlocal copies
        copies += 1
        if copies == 2:
            destination_file.write(source_file.read(10))
            raise OSError('simulated full filesystem')
        real_copy(source_file, destination_file)

    monkeypatch.setattr(
        helpers['shutil'],
        'copyfileobj',
        fail_second,
    )
    with pytest.raises(OSError, match='full filesystem'):
        helpers['bfq_seed'](
            run_root,
            'x.test',
            60,
            source,
        )

    active: Path = (
        run_root
        /'xdg/piker/nativedb/x.test.ohlcv60s.parquet'
    )
    assert not active.exists()
    assert (
        run_root / 'evidence' / 'seed.parquet'
    ).read_bytes() == before


def test_helpers_reject_paths_outside_marked_root(
    tmp_path: Path,
) -> None:
    '''
    A marker must not authorize traversal or symlinked paths.

    FQME text becomes part of the active Parquet filename and fresh
    reset moves that path. Prove parent traversal is rejected, then
    replace the evidence directory with a symlink and prove path
    construction refuses to cross the disposable-root boundary.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    helpers['bfq_init'](run_root, validate_piker=False)

    with pytest.raises(ValueError, match='Unsafe FQME'):
        helpers['_bfq_paths'](
            run_root,
            '../../outside',
            60,
        )

    evidence: Path = run_root / 'evidence'
    outside: Path = tmp_path / 'outside'
    outside.mkdir()
    evidence.rmdir()
    evidence.symlink_to(outside, target_is_directory=True)
    with pytest.raises(RuntimeError, match='symlinked path'):
        helpers['bfq_init'](
            run_root,
            validate_piker=False,
        )
    with pytest.raises(RuntimeError, match='escaped run root'):
        helpers['_bfq_paths'](run_root, 'x.test', 60)
    with pytest.raises(RuntimeError, match='Unsafe evidence'):
        helpers['bfq_compare'](run_root, 'before', 'after')


def test_helpers_reject_dangling_evidence_symlinks(
    tmp_path: Path,
) -> None:
    '''
    Final evidence names must not follow dangling external symlinks.

    ``Path.exists()`` is false for a dangling symlink, which can make
    a check-then-write path create its external target. Link the case
    record and seed outside a marked root. Prove both writers reject
    the final component before creating external bytes.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    helpers['bfq_init'](run_root, validate_piker=False)
    outside_case: Path = tmp_path / 'outside-case.json'
    case_path: Path = run_root / 'case.json'
    case_path.symlink_to(outside_case)
    with pytest.raises(RuntimeError, match='symlinked case'):
        helpers['bfq_record_case'](
            run_root,
            fqme='x.test',
            period_s=60,
            provider='fake',
            symptom='unsafe path test',
        )
    assert not outside_case.exists()

    case_path.unlink()
    source: Path = tmp_path / 'source.parquet'
    mk_frame((60, 120)).write_parquet(source)
    outside_seed: Path = tmp_path / 'outside-seed.parquet'
    seed: Path = run_root / 'evidence' / 'seed.parquet'
    seed.symlink_to(outside_seed)
    with pytest.raises(RuntimeError, match='symlinked output'):
        helpers['bfq_seed'](
            run_root,
            'x.test',
            60,
            source,
        )
    assert not outside_seed.exists()


def test_bfq_use_updates_xonsh_subprocess_environment(
    tmp_path: Path,
) -> None:
    '''
    Disposable XDG selection must reach commands launched by xonsh.

    Updating only ``os.environ`` does not mutate xonsh's environment
    mapping, so manual chart commands previously retained normal user
    storage. Source the real snippet in xonsh, initialize a run root,
    and assert a subprocess sees the disposable XDG path.

    '''
    repo: Path = Path(__file__).parents[1]
    snippet: Path = repo / 'snippets' / 'nativedb_backfill_audit.xsh'
    run_root: Path = tmp_path / 'run'
    code: str = (
        f'source {snippet}; '
        f'bfq_init(p\'{run_root}\', validate_piker=False); '
        'env'
    )
    result = subprocess.run(
        ['xonsh', '-c', code],
        check=False,
        capture_output=True,
        cwd=repo,
        text=True,
    )

    assert result.returncode == 0, result.stderr
    expected: str = f'XDG_CONFIG_HOME={run_root / "xdg"}'
    assert result.stdout.count(expected) == 2


def test_init_validates_console_script_against_worktree(
    tmp_path: Path,
) -> None:
    '''
    An installed console script must import the reviewed worktree.

    A shared virtualenv's ``piker`` entry point can resolve its
    editable root checkout, where a new audit command is absent, even
    while the operator is in this linked worktree. Run initialization
    and prove its PYTHONPATH-bound preflight sees the local command
    before it records the executable and repository identity.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'

    helpers['bfq_init'](run_root)

    marker: dict = json.loads(
        (run_root / '.piker-backfill-qualification').read_text()
    )
    assert Path(marker['repo_root']) == Path(__file__).parents[1]


def test_phase_compare_detects_preservation_and_seam_changes(
    tmp_path: Path,
) -> None:
    '''
    Lifecycle comparison must preserve data without hiding seams.

    Restart and append may replace a matching provider seam while old
    timestamps remain. Save two read-only snapshots where the second
    appends a row and updates a close. Prove checks pass while the
    the changed timestamp stays explicit operator evidence.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    init_case(helpers, run_root)
    evidence: Path = run_root / 'evidence'
    before_path: Path = evidence / 'fresh.parquet'
    after_path: Path = evidence / 'restart.parquet'
    mk_frame((60, 120), (1, 2)).write_parquet(before_path)
    mk_frame((60, 120, 180), (1, 20, 3)).write_parquet(after_path)

    for phase, path in (
        ('fresh', before_path),
        ('restart', after_path),
    ):
        write_phase_evidence(run_root, phase, path)

    comparison: dict = helpers['bfq_compare'](
        run_root,
        'fresh',
        'restart',
        require_newer=True,
    )

    assert comparison['passed'] is True
    assert comparison['missing_timestamp_count'] == 0
    assert comparison['changed_common_row_count'] == 1
    assert comparison['changed_common_timestamps'] == [120]
    assert comparison['checks']['maximum_advanced'] is True
    with pytest.raises(ValueError, match='phases must differ'):
        helpers['bfq_compare'](run_root, 'fresh', 'fresh')


def test_repair_compare_allows_invalid_and_duplicate_removal(
    tmp_path: Path,
) -> None:
    '''
    Repair may remove invalid rows but must preserve valid times.

    Known-bad history may contain zero-epoch and duplicate rows. Both
    should disappear during canonical rewrite. Compare
    that baseline to clean extended history and prove preservation
    ignores invalid epoch and duplicate multiplicity while recording
    their before and after counts.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    init_case(helpers, run_root)
    evidence: Path = run_root / 'evidence'
    before_path: Path = evidence / 'baseline.parquet'
    after_path: Path = evidence / 'repair.parquet'
    mk_frame((0, 60, 60, 120)).write_parquet(before_path)
    mk_frame((60, 120, 180)).write_parquet(after_path)

    for phase, path in (
        ('baseline', before_path),
        ('repair', after_path),
    ):
        write_phase_evidence(run_root, phase, path)

    comparison: dict = helpers['bfq_compare'](
        run_root,
        'baseline',
        'repair',
        require_newer=True,
    )

    assert comparison['passed'] is True
    assert comparison['missing_timestamp_count'] == 0
    assert comparison['before_duplicate_excess'] == 1
    assert comparison['after_duplicate_excess'] == 0


def test_malformed_baseline_produces_failed_comparison_evidence(
    tmp_path: Path,
) -> None:
    '''
    Comparison must report unauditable baselines instead of crashing.

    Raw audit accepts malformed Parquet so humans can save evidence
    before repair. Write a baseline with textual timestamps and no
    close column, compare it with canonical history, and prove the
    helper writes limitations plus failed checks before raising its
    qualification assertion.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    init_case(helpers, run_root)
    evidence: Path = run_root / 'evidence'
    before_path: Path = evidence / 'broken.parquet'
    after_path: Path = evidence / 'repair.parquet'
    mk_frame((60, 120)).drop('close').with_columns(
        pl.Series('time', ['60', '120'])
    ).write_parquet(before_path)
    mk_frame((60, 120, 180)).write_parquet(after_path)

    for phase, path in (
        ('broken', before_path),
        ('repair', after_path),
    ):
        write_phase_evidence(run_root, phase, path)

    with pytest.raises(AssertionError):
        helpers['bfq_compare'](
            run_root,
            'broken',
            'repair',
        )

    comparison: dict = json.loads(
        (
            evidence
            /'compare-broken-to-repair.json'
        ).read_text()
    )
    assert comparison['passed'] is False
    assert comparison['checks']['no_timestamps_lost'] is False
    assert 'before_time_nonnumeric' in comparison['limitations']
    assert (
        'common_value_comparison_unavailable'
        in comparison['limitations']
    )


def test_compare_rejects_modified_or_cross_case_evidence(
    tmp_path: Path,
) -> None:
    '''
    Lifecycle checks must not combine unrelated or altered artifacts.

    JSON can remain plausible when a phase snapshot is replaced or a
    report was generated for another market. Bind each report to its
    checksum and fixed case metadata. Prove both changes abort before
    abort before a comparison result can be accepted.

    '''
    helpers: dict = load_helpers()
    run_root: Path = tmp_path / 'run'
    init_case(helpers, run_root)
    evidence: Path = run_root / 'evidence'
    before: Path = evidence / 'before.parquet'
    after: Path = evidence / 'after.parquet'
    mk_frame((60, 120)).write_parquet(before)
    mk_frame((60, 120, 180)).write_parquet(after)
    write_phase_evidence(run_root, 'before', before)
    write_phase_evidence(run_root, 'after', after)

    after_report: Path = evidence / 'after.json'
    report_bytes: bytes = after_report.read_bytes()
    after_report.write_bytes(report_bytes + b'\n')
    with pytest.raises(RuntimeError, match='report checksum'):
        helpers['bfq_compare'](run_root, 'before', 'after')
    after_report.write_bytes(report_bytes)

    mk_frame((60, 180)).write_parquet(before)
    with pytest.raises(RuntimeError, match='snapshot checksum'):
        helpers['bfq_compare'](run_root, 'before', 'after')

    other: Path = evidence / 'other.parquet'
    mk_frame((60, 120)).write_parquet(other)
    write_phase_evidence(
        run_root,
        'other',
        other,
        fqme='other.test',
    )
    with pytest.raises(RuntimeError, match='does not match'):
        helpers['bfq_compare'](run_root, 'other', 'after')