from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
import hashlib
import json
from pathlib import Path
from threading import Event

import pytest

from apps.cli.run_wgs_markdup import build_parser, main
from apps.cli import run_wgs_markdup


ALIGN_RUN = "3" * 32


def _record(workspace: Path, path: Path, role: str, sample_id: str) -> dict:
    return {
        "artifact": role,
        "role": role,
        "path": str(path.relative_to(workspace)),
        "full_path": str(path.resolve()),
        "size": path.stat().st_size,
        "size_bytes": path.stat().st_size,
        "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
        "sample_id": sample_id,
    }


def _alignment_authority(workspace: Path) -> Path:
    final = workspace / "final"
    bundle = final / ".wgs_align_runs" / "Cohort" / ALIGN_RUN
    bundle.mkdir(parents=True)
    artifacts = {}
    for sample in ("S1", "S2", "S3"):
        bam = bundle / f"Cohort.wgs_align.{sample}.sorted.bam"
        bam.write_text(
            "@HD\tVN:1.6\tSO:coordinate\n"
            "@SQ\tSN:chr1\tLN:8\n"
            f"@RG\tID:{sample}\tSM:{sample}\tPL:illumina\n"
            f"{sample}.read\t0\tchr1\t1\t60\t4M\t*\t0\t0\tACGT\tIIII\n"
        )
        bai = Path(f"{bam}.bai")
        bai.write_bytes(b"BAI")
        artifacts[f"sorted_bam.{sample}"] = _record(workspace, bam, "sorted_bam", sample)
        artifacts[f"sorted_bam_index.{sample}"] = _record(
            workspace, bai, "sorted_bam_index", sample
        )
    manifest = {
        "manifest_version": "1.0",
        "stage": "wgs_align",
        "status": "success",
        "prefix": "Cohort",
        "job_id": "job-1",
        "run_id": ALIGN_RUN,
        "workspace": str(workspace.resolve()),
        "wgs_sample_set_id": "wgs-samples:" + "a" * 64,
        "wgs_reference_id": "wgs-reference:" + "b" * 64,
        "wgs_analysis_id": "wgs-analysis:" + "c" * 64,
        "artifacts": artifacts,
    }
    authority = final / "Cohort.wgs_align.manifest.json"
    authority.write_text(json.dumps(manifest))
    return authority


def _fake_tools(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[Path, Path]:
    tool_dir = tmp_path / "bin"
    tool_dir.mkdir()
    argv_log = tmp_path / "gatk.argv.jsonl"
    gatk = tool_dir / "gatk"
    gatk.write_text(
        """#!/usr/bin/env python3
import json, os, pathlib, sys
with open(os.environ['GATK_ARGV_LOG'], 'a') as out:
    out.write(json.dumps(sys.argv[1:]) + '\\n')
if sys.argv[1:] == ['--version']:
    print('The Genome Analysis Toolkit (GATK) v4.6.1.0')
    raise SystemExit(0)
if not sys.argv[1:] or sys.argv[1] != 'MarkDuplicates':
    raise SystemExit(2)
values = dict(item.split('=', 1) for item in sys.argv[2:] if '=' in item)
source = pathlib.Path(values['INPUT'])
output = pathlib.Path(values['OUTPUT'])
metrics = pathlib.Path(values['METRICS_FILE'])
sample = source.name.split('.')[2]
if os.environ.get('WGS_FAIL_SAMPLE') == sample:
    print('controlled markdup failure', file=sys.stderr)
    raise SystemExit(23)
data = source.read_text()
if os.environ.get('WGS_BAD_RG_SAMPLE') == sample:
    data = data.replace(f'SM:{sample}', 'SM:Wrong')
output.write_text(data)
if os.environ.get('WGS_BAD_INDEX_SAMPLE') != sample:
    pathlib.Path(f'{output}.bai').write_bytes(b'BAI')
if os.environ.get('WGS_BAD_METRICS_SAMPLE') == sample:
    metrics.write_text('old or malformed metrics\\n')
else:
    metrics.write_text(
        '## METRICS CLASS\\tpicard.sam.DuplicationMetrics\\n'
        'LIBRARY\\tUNPAIRED_READS_EXAMINED\\tREAD_PAIRS_EXAMINED\\tPERCENT_DUPLICATION\\n'
        f'{sample}\\t0\\t1\\t0.0\\n'
    )
print(f'completed {sample}')
"""
    )
    samtools = tool_dir / "samtools"
    samtools.write_text(
        """#!/usr/bin/env python3
import json, os, pathlib, sys
with open(os.environ['SAMTOOLS_ARGV_LOG'], 'a') as out:
    out.write(json.dumps(sys.argv[1:]) + '\\n')
if sys.argv[1:] == ['--version']:
    print('samtools 1.17')
    raise SystemExit(0)
cmd = sys.argv[1]
bam = pathlib.Path(sys.argv[-1])
sample = bam.name.split('.')[2]
if cmd == 'quickcheck':
    raise SystemExit(9 if os.environ.get('WGS_BAD_QUICKCHECK_SAMPLE') == sample else 0)
if cmd == 'idxstats':
    index = pathlib.Path(f'{bam}.bai')
    if not index.is_file() or not index.stat().st_size:
        raise SystemExit(4)
    print('chr1\\t8\\t1\\t0')
    raise SystemExit(0)
if cmd == 'view' and '-H' in sys.argv:
    print('\\n'.join(line for line in bam.read_text().splitlines() if line.startswith('@')))
    raise SystemExit(0)
raise SystemExit(2)
"""
    )
    gatk.chmod(0o755)
    samtools.chmod(0o755)
    monkeypatch.setenv("GATK_ARGV_LOG", str(argv_log))
    monkeypatch.setenv("SAMTOOLS_ARGV_LOG", str(tmp_path / "samtools.argv.jsonl"))
    monkeypatch.setenv("PATH", f"{tool_dir}:{Path('/usr/bin')}")
    return gatk, argv_log


def _argv(
    workspace: Path,
    authority: Path,
    gatk: Path,
    samtools: Path | None = None,
) -> list[str]:
    return [
        "--workspace",
        str(workspace),
        "--job-id",
        "job-1",
        "--prefix",
        "Cohort",
        "--align-manifest",
        str(authority),
        "--gatk",
        str(gatk),
        "--samtools",
        str(samtools or gatk.parent / "samtools"),
        "--max-jobs",
        "2",
        "--remove-duplicates=true",
    ]


def test_wgs_markdup_exposes_import_safe_cli() -> None:
    assert callable(build_parser)
    assert callable(main)


def test_parser_requires_one_unambiguous_true_boolean() -> None:
    parser = build_parser()
    args = parser.parse_args(
        [
            "--workspace",
            "w",
            "--job-id",
            "j",
            "--prefix",
            "p",
            "--align-manifest",
            "a",
            "--gatk",
            "g",
            "--samtools",
            "s",
            "--max-jobs",
            "1",
            "--remove-duplicates=true",
        ]
    )
    assert args.remove_duplicates == "true"
    with pytest.raises(SystemExit):
        parser.parse_args(
            [
                "--workspace",
                "w",
                "--job-id",
                "j",
                "--prefix",
                "p",
                "--align-manifest",
                "a",
                "--gatk",
                "g",
                "--samtools",
                "s",
                "--max-jobs",
                "1",
                "--remove-duplicates=false",
            ]
        )


def test_three_samples_publish_four_valid_roles_and_exact_gatk_argv(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    workspace = tmp_path / "workspace"
    authority = _alignment_authority(workspace)
    gatk, argv_log = _fake_tools(tmp_path, monkeypatch)
    assert main(_argv(workspace, authority, gatk)) == 0

    manifest = json.loads(
        (workspace / "final" / "Cohort.wgs_markdup.manifest.json").read_text()
    )
    assert manifest["status"] == "success"
    assert manifest["metrics"]["input_sample_ids"] == ["S1", "S2", "S3"]
    assert manifest["metrics"]["output_sample_ids"] == ["S1", "S2", "S3"]
    assert manifest["wgs_sample_set_id"] == "wgs-samples:" + "a" * 64
    assert manifest["wgs_reference_id"] == "wgs-reference:" + "b" * 64
    assert manifest["wgs_analysis_id"].startswith("wgs-analysis:")
    assert manifest["parameters"]["samtools"] == str(gatk.parent / "samtools")
    assert manifest["software"]["samtools"]["version"] == "samtools 1.17"
    assert len(manifest["artifacts"]) == 12
    assert {
        item["role"] for item in manifest["artifacts"].values()
    } == {"markdup_bam", "markdup_bam_index", "markdup_metrics", "markdup_log"}
    assert {
        item["sample_id"] for item in manifest["artifacts"].values()
    } == {"S1", "S2", "S3"}
    run_root = (
        workspace
        / "final"
        / ".wgs_markdup_runs"
        / "Cohort"
        / manifest["run_id"]
    )
    assert {
        Path(item["full_path"]).parent for item in manifest["artifacts"].values()
    } == {run_root}
    calls = [
        value
        for value in map(json.loads, argv_log.read_text().splitlines())
        if value != ["--version"]
    ]
    assert len(calls) == 3
    for call in calls:
        assert call[0] == "MarkDuplicates"
        assert "REMOVE_DUPLICATES=true" in call
        assert "CREATE_INDEX=true" in call
        assert "VALIDATION_STRINGENCY=LENIENT" in call
        assert all(not token.startswith((";", "|", "&&")) for token in call)

    samtools_calls = [
        value
        for value in map(
            json.loads,
            (tmp_path / "samtools.argv.jsonl").read_text().splitlines(),
        )
        if value != ["--version"]
    ]
    assert len(samtools_calls) == 12


def test_custom_samtools_path_and_version_affect_analysis_identity(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    workspace = tmp_path / "workspace"
    authority = _alignment_authority(workspace)
    gatk, _ = _fake_tools(tmp_path, monkeypatch)
    samtools = gatk.parent / "samtools"
    monkeypatch.setenv("PATH", "/usr/bin")
    assert main(_argv(workspace, authority, gatk, samtools)) == 0
    manifest_path = workspace / "final" / "Cohort.wgs_markdup.manifest.json"
    first = json.loads(manifest_path.read_text())

    samtools.write_text(
        samtools.read_text().replace("samtools 1.17", "samtools 1.18")
    )
    assert main(_argv(workspace, authority, gatk, samtools)) == 0
    second = json.loads(manifest_path.read_text())
    assert first["wgs_analysis_id"] != second["wgs_analysis_id"]
    assert second["software"]["samtools"] == {
        "path": str(samtools),
        "version": "samtools 1.18",
    }


@pytest.mark.parametrize(
    ("environment", "sample", "message"),
    [
        ("WGS_FAIL_SAMPLE", "S2", "exit code 23"),
        ("WGS_BAD_QUICKCHECK_SAMPLE", "S1", "quickcheck"),
        ("WGS_BAD_INDEX_SAMPLE", "S3", "index"),
        ("WGS_BAD_METRICS_SAMPLE", "S2", "metrics"),
        ("WGS_BAD_RG_SAMPLE", "S1", "read group"),
    ],
)
def test_any_sample_or_postflight_failure_publishes_no_artifacts(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
    environment: str,
    sample: str,
    message: str,
) -> None:
    workspace = tmp_path / "workspace"
    authority = _alignment_authority(workspace)
    gatk, _ = _fake_tools(tmp_path, monkeypatch)
    monkeypatch.setenv(environment, sample)
    assert main(_argv(workspace, authority, gatk)) == 1
    failure = json.loads(
        (workspace / "final" / "Cohort.wgs_markdup.manifest.json").read_text()
    )
    assert failure["status"] == "failed"
    assert failure["artifacts"] == {}
    assert message in capsys.readouterr().err


def test_old_metrics_cannot_satisfy_failure_and_old_authority_survives(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    workspace = tmp_path / "workspace"
    align = _alignment_authority(workspace)
    final = workspace / "final"
    old_bundle = final / ".wgs_markdup_runs" / "Cohort" / ("4" * 32)
    old_bundle.mkdir(parents=True)
    old_metrics = old_bundle / "Cohort.wgs_markdup.S2.markdup.metrics.txt"
    old_metrics.write_text("valid old metrics must not count")
    authority = final / "Cohort.wgs_markdup.manifest.json"
    authority.write_text(
        json.dumps(
            {
                "stage": "wgs_markdup",
                "status": "success",
                "start_time": "2026-01-01T00:00:00+00:00",
                "artifacts": {"old": {"full_path": str(old_metrics)}},
            }
        )
    )
    authority_hash = hashlib.sha256(authority.read_bytes()).hexdigest()
    metrics_hash = hashlib.sha256(old_metrics.read_bytes()).hexdigest()
    gatk, _ = _fake_tools(tmp_path, monkeypatch)
    monkeypatch.setenv("WGS_BAD_METRICS_SAMPLE", "S2")
    assert main(_argv(workspace, align, gatk)) == 1
    assert hashlib.sha256(authority.read_bytes()).hexdigest() == authority_hash
    assert hashlib.sha256(old_metrics.read_bytes()).hexdigest() == metrics_hash
    sidecars = list(final.glob("Cohort.wgs_markdup.failed.*.manifest.json"))
    assert len(sidecars) == 1
    assert json.loads(sidecars[0].read_text())["artifacts"] == {}


def test_concurrent_attempts_leave_one_complete_authority_bundle(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    workspace = tmp_path / "workspace"
    align = _alignment_authority(workspace)
    gatk, _ = _fake_tools(tmp_path, monkeypatch)
    argv = _argv(workspace, align, gatk)
    with ThreadPoolExecutor(max_workers=2) as pool:
        results = [future.result() for future in (pool.submit(main, argv), pool.submit(main, argv))]
    assert 0 in results
    manifest = json.loads(
        (workspace / "final" / "Cohort.wgs_markdup.manifest.json").read_text()
    )
    assert manifest["status"] == "success"
    assert len(manifest["artifacts"]) == 12
    assert {
        Path(item["full_path"]).parent for item in manifest["artifacts"].values()
    } == {
        workspace
        / "final"
        / ".wgs_markdup_runs"
        / "Cohort"
        / manifest["run_id"]
    }


def test_failing_attempt_holds_identity_lock_while_deciding_failure_destination(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    workspace = tmp_path / "workspace"
    align = _alignment_authority(workspace)
    gatk, _ = _fake_tools(tmp_path, monkeypatch)
    authority = workspace / "final" / "Cohort.wgs_markdup.manifest.json"
    failure_ready = Event()
    release_failure = Event()
    success_done = Event()
    original_atomic = run_wgs_markdup._atomic_json

    def delayed_failure_write(path: Path, payload: dict) -> None:
        if payload.get("status") == "failed" and path == authority:
            failure_ready.set()
            assert release_failure.wait(5)
        original_atomic(path, payload)

    monkeypatch.setattr(run_wgs_markdup, "_atomic_json", delayed_failure_write)
    monkeypatch.setenv("WGS_FAIL_SAMPLE", "S1")
    argv = _argv(workspace, align, gatk)

    def successful_attempt() -> int:
        try:
            return main(argv)
        finally:
            success_done.set()

    with ThreadPoolExecutor(max_workers=2) as pool:
        failing = pool.submit(main, argv)
        assert failure_ready.wait(5)
        monkeypatch.delenv("WGS_FAIL_SAMPLE")
        successful = pool.submit(successful_attempt)
        success_was_blocked = not success_done.wait(0.5)
        release_failure.set()
        assert failing.result() == 1
        assert successful.result() == 0
    assert success_was_blocked
    assert json.loads(authority.read_text())["status"] == "success"


def test_authority_analysis_swap_between_header_and_resolver_fails_closed(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys
) -> None:
    workspace = tmp_path / "workspace"
    align = _alignment_authority(workspace)
    gatk, _ = _fake_tools(tmp_path, monkeypatch)
    original_header = run_wgs_markdup._authority_header

    def swap_after_header(*args, **kwargs):
        header = original_header(*args, **kwargs)
        swapped = json.loads(align.read_text())
        swapped["wgs_analysis_id"] = "wgs-analysis:" + "d" * 64
        align.write_text(json.dumps(swapped))
        return header

    monkeypatch.setattr(run_wgs_markdup, "_authority_header", swap_after_header)
    assert main(_argv(workspace, align, gatk)) == 1
    assert "analysis identity mismatch" in capsys.readouterr().err