from __future__ import annotations

import gzip
import hashlib
import json
from pathlib import Path
from threading import Barrier, Thread

import pytest


SAMPLES = "wgs-samples:" + "a" * 64
REFERENCE = "wgs-reference:" + "b" * 64
JOINT_ANALYSIS = "wgs-analysis:" + "c" * 64


def _record(workspace: Path, path: Path, role: 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(),
    }


def _write_joint_vcf(path: Path, *, include_indel: bool = True) -> None:
    with gzip.open(path, "wt", encoding="utf-8") as handle:
        handle.write("##fileformat=VCFv4.2\n")
        handle.write("##contig=<ID=chr1,length=1000>\n")
        handle.write(
            "#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT"
            "\tS1\tS2\tS3\n"
        )
        records = [
            "chr1\t1\t.\tA\tC\t50\tPASS\t.\tGT\t0/0\t0/1\t1/1\n",
            "chr1\t3\t.\tAT\tGC\t50\tPASS\t.\tGT\t0/0\t0/1\t1/1\n",
        ]
        if include_indel:
            records.insert(
                1,
                "chr1\t2\t.\tA\tAT\t50\tPASS\t.\tGT\t0/0\t0/1\t1/1\n",
            )
        for record in records:
            handle.write(record)


def _joint_authority(
    workspace: Path, *, include_indel: bool = True
) -> Path:
    bundle = (
        workspace
        / "final"
        / ".wgs_joint_genotyping_runs"
        / "Cohort"
        / ("8" * 32)
    )
    bundle.mkdir(parents=True)
    vcf = bundle / "Cohort.wgs_joint_genotyping.cohort.joint.vcf.gz"
    tbi = Path(f"{vcf}.tbi")
    summary = bundle / "Cohort.wgs_joint_genotyping.cohort.summary.tsv"
    log = bundle / "Cohort.wgs_joint_genotyping.cohort.joint.log"
    _write_joint_vcf(vcf, include_indel=include_indel)
    tbi.write_bytes(b"TBI:joint")
    summary.write_text(
        "CHROM\tPOS\tREF\tALT\tQUAL\nchr1\t1\tA\tC\t50\n",
        encoding="utf-8",
    )
    log.write_text("GenotypeGVCFs exit=0\n", encoding="utf-8")
    authority = (
        workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    )
    authority.write_text(
        json.dumps(
            {
                "manifest_version": "1.0",
                "stage": "wgs_joint_genotyping",
                "status": "success",
                "prefix": "Cohort",
                "job_id": "job-1",
                "run_id": "8" * 32,
                "workspace": str(workspace.resolve()),
                "wgs_sample_set_id": SAMPLES,
                "wgs_reference_id": REFERENCE,
                "wgs_analysis_id": JOINT_ANALYSIS,
                "metrics": {
                    "input_sample_ids": ["S1", "S2", "S3"],
                    "output_sample_ids": ["S1", "S2", "S3"],
                },
                "artifacts": {
                    "joint_vcf": _record(workspace, vcf, "joint_vcf"),
                    "joint_vcf_index": _record(
                        workspace, tbi, "joint_vcf_index"
                    ),
                    "joint_summary": _record(
                        workspace, summary, "joint_summary"
                    ),
                    "joint_log": _record(workspace, log, "joint_log"),
                },
            }
        ),
        encoding="utf-8",
    )
    return authority


def _tool(tmp_path: Path, name: str) -> Path:
    tool = tmp_path / name
    tool.write_text(
        """#!/usr/bin/env python3
import gzip, json, os, pathlib, sys
name = pathlib.Path(sys.argv[0]).name
args = sys.argv[1:]
if args in (['--version'], ['version']):
    print(('The Genome Analysis Toolkit (GATK) v4.6.0.0'
           if name == 'gatk' else 'bcftools 1.20'))
    raise SystemExit(0)
calls = os.environ.get('WGS_SELECT_CALLS')
if calls:
    with open(calls, 'a', encoding='utf-8') as handle:
        handle.write(json.dumps([name] + args) + '\\n')
def value(flag):
    return args[args.index(flag) + 1]
def records(path):
    with gzip.open(path, 'rt', encoding='utf-8') as handle:
        return list(handle)
def select(source, output, kind):
    lines = records(source)
    with gzip.open(output, 'wt', encoding='utf-8') as handle:
        for line in lines:
            if line.startswith('#'):
                handle.write(line)
                continue
            fields = line.rstrip().split('\\t')
            ref = fields[3]
            alts = fields[4].split(',')
            is_snp = len(ref) == 1 and all(len(alt) == 1 for alt in alts)
            is_indel = any(len(alt) != len(ref) for alt in alts)
            if (kind == 'SNP' and is_snp) or (kind == 'INDEL' and is_indel):
                handle.write(line)
fail = os.environ.get('FAIL_POINT', '')
if name == 'gatk':
    command = args[0]
    if command == 'SelectVariants':
        kind = value('-select-type-to-include')
        if fail == 'gatk:select:' + kind:
            raise SystemExit(23)
        select(pathlib.Path(value('-V')), pathlib.Path(value('-O')), kind)
        raise SystemExit(0)
    if command == 'IndexFeatureFile':
        path = pathlib.Path(value('-I'))
        kind = 'SNP' if '.snp.' in path.name else 'INDEL'
        if fail == 'gatk:index:' + kind:
            raise SystemExit(23)
        marker = (
            b'CORRUPT:gatk'
            if fail == 'gatk:corrupt-index:' + kind
            else b'TBI:gatk'
        )
        pathlib.Path(str(path) + '.tbi').write_bytes(marker)
        raise SystemExit(0)
    if command == 'ValidateVariants':
        path = pathlib.Path(value('-V'))
        kind = ('input' if '.joint.' in path.name else
                ('SNP' if '.snp.' in path.name else 'INDEL'))
        if fail == 'gatk:validate:' + kind:
            raise SystemExit(23)
        index = pathlib.Path(str(path) + '.tbi')
        if ('-L' in args and (
            not index.is_file()
            or index.read_bytes().startswith(b'CORRUPT')
        )):
            raise SystemExit(24)
        raise SystemExit(0)
    raise SystemExit(2)
if args[0] == 'view':
    if '-v' in args:
        raw = value('-v')
        kind = 'SNP' if raw == 'snps' else 'INDEL'
        if fail == 'bcftools:select:' + kind:
            raise SystemExit(23)
        select(pathlib.Path(args[-1]), pathlib.Path(value('-o')), kind)
        raise SystemExit(0)
    path = pathlib.Path(args[-1])
    label = 'input' if '.joint.' in path.name else (
        'SNP' if '.snp.' in path.name else 'INDEL'
    )
    if fail == 'bcftools:validate:' + label:
        raise SystemExit(23)
    index = pathlib.Path(str(path) + '.tbi')
    if not index.is_file() or index.read_bytes().startswith(b'CORRUPT'):
        raise SystemExit(24)
    for line in records(path):
        if not line.startswith('#'):
            sys.stdout.write(line)
    raise SystemExit(0)
if args[0] == 'index':
    path = pathlib.Path(args[-1])
    kind = 'SNP' if '.snp.' in path.name else 'INDEL'
    if '-n' in args:
        if fail == 'bcftools:validate:' + (
            'input' if '.joint.' in path.name else kind
        ):
            raise SystemExit(23)
        index = pathlib.Path(str(path) + '.tbi')
        if not index.is_file() or index.read_bytes().startswith(b'CORRUPT'):
            raise SystemExit(24)
        print('3')
        raise SystemExit(0)
    if fail == 'bcftools:index:' + kind:
        raise SystemExit(23)
    pathlib.Path(str(path) + '.tbi').write_bytes(b'TBI:bcftools')
    raise SystemExit(0)
if args[0] == 'stats':
    path = pathlib.Path(args[-1])
    kind = 'SNP' if '.snp.' in path.name else 'INDEL'
    if fail == 'bcftools:stats:' + kind:
        raise SystemExit(23)
    print('SN\\t0\\tnumber of records:\\t1')
    raise SystemExit(0)
raise SystemExit(2)
""",
        encoding="utf-8",
    )
    tool.chmod(0o755)
    return tool


def _argv(
    workspace: Path,
    joint: Path,
    gatk: Path,
    bcftools: Path,
    method: str = "gatk",
) -> list[str]:
    return [
        "--workspace",
        str(workspace),
        "--job-id",
        "job-1",
        "--prefix",
        "Cohort",
        "--joint-manifest",
        str(joint),
        "--method",
        method,
        "--gatk",
        str(gatk),
        "--bcftools",
        str(bcftools),
    ]


def _variant_kinds(path: Path) -> set[str]:
    observed = set()
    with gzip.open(path, "rt", encoding="utf-8") as handle:
        for line in handle:
            if line.startswith("#"):
                continue
            fields = line.rstrip().split("\t")
            ref, alts = fields[3], fields[4].split(",")
            if len(ref) == 1 and all(len(alt) == 1 for alt in alts):
                observed.add("SNP")
            elif any(len(alt) != len(ref) for alt in alts):
                observed.add("INDEL")
            else:
                observed.add("OTHER")
    return observed


def test_wgs_variant_select_exposes_import_safe_cli() -> None:
    from apps.cli.run_wgs_variant_select import build_parser, main

    assert callable(build_parser)
    assert callable(main)


def test_parser_defaults_to_gatk_and_accepts_only_two_methods() -> None:
    from apps.cli.run_wgs_variant_select import build_parser

    parser = build_parser()
    base = [
        "--workspace", "w", "--job-id", "j", "--prefix", "P",
        "--joint-manifest", "joint.json",
    ]
    assert parser.parse_args(base).method == "gatk"
    assert parser.parse_args(base + ["--method", "bcftools"]).method == "bcftools"
    with pytest.raises(SystemExit):
        parser.parse_args(base + ["--method", "auto"])


def test_both_methods_publish_same_roles_and_names_but_distinct_authorities(
    tmp_path: Path,
) -> None:
    from apps.cli.run_wgs_variant_select import main
    from gfa_core.wgs_upstream_manifest import (
        WgsUpstreamManifestError,
        resolve_wgs_artifacts,
    )

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    gatk = _tool(tmp_path, "gatk")
    bcftools = _tool(tmp_path, "bcftools")
    manifests = {}
    for method in ("gatk", "bcftools"):
        assert main(_argv(workspace, joint, gatk, bcftools, method)) == 0
        authority = (
            workspace
            / "final"
            / f"Cohort.wgs_variant_select.{method}.manifest.json"
        )
        manifests[method] = json.loads(authority.read_text())
        resolved = resolve_wgs_artifacts(
            authority,
            "wgs_variant_select",
            ("selected_snp_vcf", "selected_indel_vcf"),
            expected_sample_set_id=SAMPLES,
            expected_reference_id=REFERENCE,
            expected_analysis_id=manifests[method]["wgs_analysis_id"],
            expected_method=method,
        )
        assert set(resolved) == {"selected_snp_vcf", "selected_indel_vcf"}
        other_method = "bcftools" if method == "gatk" else "gatk"
        with pytest.raises(WgsUpstreamManifestError, match="method mismatch"):
            resolve_wgs_artifacts(
                authority,
                "wgs_variant_select",
                ("selected_snp_vcf",),
                expected_method=other_method,
            )
    expected_roles = {
        "selected_snp_vcf",
        "selected_snp_vcf_index",
        "selected_indel_vcf",
        "selected_indel_vcf_index",
        "variant_select_report",
    }
    assert set(manifests["gatk"]["artifacts"]) == expected_roles
    assert set(manifests["bcftools"]["artifacts"]) == expected_roles
    assert manifests["gatk"]["parameters"]["method"] == "gatk"
    assert manifests["bcftools"]["parameters"]["method"] == "bcftools"
    assert (
        manifests["gatk"]["wgs_analysis_id"]
        != manifests["bcftools"]["wgs_analysis_id"]
    )
    names = {}
    for method, manifest in manifests.items():
        names[method] = {
            role: Path(record["full_path"]).name
            for role, record in manifest["artifacts"].items()
        }
        snp = Path(manifest["artifacts"]["selected_snp_vcf"]["full_path"])
        indel = Path(manifest["artifacts"]["selected_indel_vcf"]["full_path"])
        assert _variant_kinds(snp) == {"SNP"}
        assert _variant_kinds(indel) == {"INDEL"}
        assert Path(
            manifest["artifacts"]["selected_snp_vcf_index"]["full_path"]
        ) == Path(f"{snp}.tbi")
        assert Path(
            manifest["artifacts"]["selected_indel_vcf_index"]["full_path"]
        ) == Path(f"{indel}.tbi")
        assert manifest["metrics"]["input_sample_ids"] == ["S1", "S2", "S3"]
        assert manifest["metrics"]["output_sample_ids"] == ["S1", "S2", "S3"]
        report = Path(
            manifest["artifacts"]["variant_select_report"]["full_path"]
        ).read_text()
        assert f"method: {method}" in report
    assert names["gatk"] == names["bcftools"]


@pytest.mark.parametrize("method", ("gatk", "bcftools"))
def test_zero_record_variant_class_still_publishes_valid_indexed_output(
    tmp_path: Path, method: str
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace, include_indel=False)
    assert main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            _tool(tmp_path, "bcftools"),
            method,
        )
    ) == 0
    manifest = json.loads(
        (
            workspace
            / "final"
            / f"Cohort.wgs_variant_select.{method}.manifest.json"
        ).read_text()
    )
    assert manifest["metrics"]["indel_records"] == 0
    indel = Path(
        manifest["artifacts"]["selected_indel_vcf"]["full_path"]
    )
    assert indel.is_file()
    assert Path(f"{indel}.tbi").is_file()


def test_zero_record_gatk_output_still_queries_and_rejects_corrupt_tbi(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace, include_indel=False)
    monkeypatch.setenv("FAIL_POINT", "gatk:corrupt-index:INDEL")
    assert main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            Path("/bin/false"),
            "gatk",
        )
    ) == 1


def test_stale_publication_failure_never_advertises_fallback(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli import run_wgs_variant_select as stage

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    monkeypatch.setattr(stage, "_publish", lambda *args, **kwargs: False)
    assert stage.main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            Path("/bin/false"),
            "gatk",
        )
    ) == 1
    failure = json.loads(
        (
            workspace
            / "final"
            / "Cohort.wgs_variant_select.gatk.manifest.json"
        ).read_text()
    )
    assert failure["artifacts"] == {}
    assert "fallback_method" not in failure


@pytest.mark.parametrize("method", ("gatk", "bcftools"))
def test_all_input_and_output_indices_are_queried_by_actual_method(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, method: str
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    calls = tmp_path / "calls.jsonl"
    monkeypatch.setenv("WGS_SELECT_CALLS", str(calls))
    gatk = _tool(tmp_path, "gatk")
    bcftools = _tool(tmp_path, "bcftools")
    assert main(_argv(workspace, joint, gatk, bcftools, method)) == 0
    observed = [json.loads(line) for line in calls.read_text().splitlines()]
    if method == "gatk":
        assert {call[0] for call in observed} == {"gatk"}
        validations = [
            call for call in observed if len(call) > 1
            and call[1] == "ValidateVariants"
        ]
        assert validations
        assert all("REF" in call for call in validations)
        assert sum(
            call[1] == "ValidateVariants" for call in observed
        ) == 3
    else:
        assert {call[0] for call in observed} == {"bcftools"}
        assert sum(
            call[1] == "view" and "-R" in call for call in observed
        ) == 3
        assert sum(call[1] == "stats" for call in observed) == 2


@pytest.mark.parametrize(
    ("method", "inactive_tool"),
    (("gatk", "bcftools"), ("bcftools", "gatk")),
)
def test_method_does_not_execute_or_require_inactive_fallback_tool(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    method: str,
    inactive_tool: str,
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    calls = tmp_path / "calls.jsonl"
    monkeypatch.setenv("WGS_SELECT_CALLS", str(calls))
    gatk = Path("/bin/false") if inactive_tool == "gatk" else _tool(
        tmp_path, "gatk"
    )
    bcftools = (
        Path("/bin/false")
        if inactive_tool == "bcftools"
        else _tool(tmp_path, "bcftools")
    )
    assert main(_argv(workspace, joint, gatk, bcftools, method)) == 0
    observed = [json.loads(line) for line in calls.read_text().splitlines()]
    assert {call[0] for call in observed} == {method}


@pytest.mark.parametrize(
    ("method", "failure"),
    (
        ("gatk", "gatk:select:SNP"),
        ("gatk", "gatk:index:INDEL"),
        ("gatk", "gatk:validate:SNP"),
        ("gatk", "gatk:validate:input"),
        ("bcftools", "bcftools:select:SNP"),
        ("bcftools", "bcftools:index:INDEL"),
        ("bcftools", "bcftools:validate:SNP"),
        ("bcftools", "bcftools:stats:INDEL"),
    ),
)
def test_each_select_index_validate_and_stat_failure_preserves_method_authority(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    method: str,
    failure: str,
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    authority = (
        workspace
        / "final"
        / f"Cohort.wgs_variant_select.{method}.manifest.json"
    )
    authority.write_text(
        json.dumps(
            {
                "stage": "wgs_variant_select",
                "status": "success",
                "method": method,
                "sentinel": True,
            }
        )
    )
    before = hashlib.sha256(authority.read_bytes()).hexdigest()
    monkeypatch.setenv("FAIL_POINT", failure)
    assert main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            _tool(tmp_path, "bcftools"),
            method,
        )
    ) == 1
    assert hashlib.sha256(authority.read_bytes()).hexdigest() == before
    sidecars = list(
        (workspace / "final").glob(
            f"Cohort.wgs_variant_select.{method}.failed.*.manifest.json"
        )
    )
    assert len(sidecars) == 1
    failure_manifest = json.loads(sidecars[0].read_text())
    assert failure_manifest["artifacts"] == {}
    assert failure_manifest["method"] == method


def test_gatk_failure_does_not_automatically_run_bcftools_selection(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    calls = tmp_path / "calls.jsonl"
    monkeypatch.setenv("WGS_SELECT_CALLS", str(calls))
    monkeypatch.setenv("FAIL_POINT", "gatk:select:SNP")
    assert main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            _tool(tmp_path, "bcftools"),
            "gatk",
        )
    ) == 1
    observed = [json.loads(line) for line in calls.read_text().splitlines()]
    assert not any(
        call[0] == "bcftools"
        and call[1] == "view"
        and "-v" in call
        for call in observed
    )


@pytest.mark.parametrize(
    "mutation", ("analysis", "hash", "path", "symlink", "roles", "sidecar")
)
def test_joint_authority_is_strictly_bound(
    tmp_path: Path, mutation: str
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    payload = json.loads(joint.read_text())
    if mutation == "analysis":
        payload.pop("wgs_analysis_id")
    elif mutation == "hash":
        payload["artifacts"]["joint_vcf"]["sha256"] = "f" * 64
    elif mutation == "path":
        payload["artifacts"]["joint_vcf"]["path"] = "../escape.vcf.gz"
    elif mutation == "symlink":
        record = payload["artifacts"]["joint_vcf"]
        original = Path(record["full_path"])
        link = original.with_name("linked.vcf.gz")
        link.symlink_to(original)
        record["path"] = str(link.relative_to(workspace))
        record["full_path"] = str(link)
    elif mutation == "roles":
        payload["artifacts"].pop("joint_vcf_index")
    else:
        record = payload["artifacts"]["joint_vcf_index"]
        unrelated = Path(record["full_path"]).with_name("unrelated.tbi")
        unrelated.write_bytes(b"TBI:unrelated")
        record.update(_record(workspace, unrelated, "joint_vcf_index"))
    joint.write_text(json.dumps(payload))
    assert main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            _tool(tmp_path, "bcftools"),
        )
    ) == 1


def test_corrupt_nonempty_joint_tbi_is_rejected_by_real_query_boundary(
    tmp_path: Path,
) -> None:
    from apps.cli.run_wgs_variant_select import main

    workspace = tmp_path / "workspace"
    joint = _joint_authority(workspace)
    payload = json.loads(joint.read_text())
    record = payload["artifacts"]["joint_vcf_index"]
    tbi = Path(record["full_path"])
    tbi.write_bytes(b"CORRUPT:input")
    record.update(_record(workspace, tbi, "joint_vcf_index"))
    joint.write_text(json.dumps(payload))
    assert main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            _tool(tmp_path, "bcftools"),
        )
    ) == 1


def test_method_publish_lock_rejects_equal_or_older_attempt(
    tmp_path: Path,
) -> None:
    from apps.cli.run_wgs_variant_select import _publish

    authority = tmp_path / "final" / "Cohort.wgs_variant_select.gatk.manifest.json"
    authority.parent.mkdir()
    timestamp = "2026-07-23T12:00:00Z"
    authority.write_text(
        json.dumps({"status": "success", "start_time": timestamp})
    )
    for number, started_at in enumerate(
        ("2026-07-23T11:00:00Z", timestamp)
    ):
        bundle = tmp_path / f"bundle-{number}"
        bundle.mkdir()
        manifest = tmp_path / f"manifest-{number}.json"
        manifest.write_text("{}")
        assert not _publish(
            bundle,
            tmp_path / "runs" / str(number),
            manifest,
            authority,
            started_at,
        )


def test_concurrent_publish_keeps_newest_complete_method_authority(
    tmp_path: Path,
) -> None:
    from apps.cli.run_wgs_variant_select import _publish

    authority = tmp_path / "final" / "Cohort.wgs_variant_select.gatk.manifest.json"
    authority.parent.mkdir()
    barrier = Barrier(2)

    def worker(name: str, started_at: str) -> None:
        bundle = tmp_path / f"bundle-{name}"
        bundle.mkdir()
        manifest = tmp_path / f"manifest-{name}.json"
        manifest.write_text(
            json.dumps(
                {"status": "success", "start_time": started_at, "winner": name}
            )
        )
        barrier.wait()
        _publish(
            bundle,
            tmp_path / "runs" / name,
            manifest,
            authority,
            started_at,
        )

    threads = [
        Thread(target=worker, args=("a", "2026-07-23T11:00:00Z")),
        Thread(target=worker, args=("b", "2026-07-23T12:00:00Z")),
    ]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert json.loads(authority.read_text())["winner"] == "b"


def test_status_and_results_select_exact_method_authority_and_show_method(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import asyncio

    from apps.cli.run_wgs_variant_select import main
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools.stage_status import get_stage_status

    workspace_root = tmp_path / "workspaces"
    workspace = workspace_root / "job-1"
    joint = _joint_authority(workspace)
    assert main(
        _argv(
            workspace,
            joint,
            _tool(tmp_path, "gatk"),
            Path("/bin/false"),
            "gatk",
        )
    ) == 0
    authority = (
        workspace
        / "final"
        / "Cohort.wgs_variant_select.gatk.manifest.json"
    )
    manifest = json.loads(authority.read_text())
    analysis_id = manifest["wgs_analysis_id"]
    status = get_stage_status(
        "job-1",
        "Cohort",
        "wgs_variant_select",
        workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id=analysis_id,
    )
    assert status.completed
    assert status.manifest_path == authority
    assert status.manifest["method"] == "gatk"
    assert not get_stage_status(
        "job-1",
        "Cohort",
        "wgs_variant_select",
        workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id="wgs-analysis:" + "f" * 64,
    ).completed
    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
    results = asyncio.run(
        web_main.get_results(
            job_id="job-1",
            stage="wgs_variant_select",
            wgs_sample_set_id=SAMPLES,
            wgs_reference_id=REFERENCE,
            wgs_analysis_id=analysis_id,
        )
    )
    serialized = json.dumps(results)
    assert '"method": "gatk"' in serialized
    assert "selected_snp_vcf" in serialized
    assert "selected_indel_vcf" in serialized