import json
from pathlib import Path
import subprocess

import pytest
from pydantic import ValidationError

from apps.web.vue3.main import ExecuteRequest


ROOT = Path(__file__).resolve().parents[2]


def test_wgs_qc_api_uses_exact_wgs_sample_config_field_name() -> None:
    request = ExecuteRequest(stage="wgs_qc", wgsSampleConfig="input/wgs_samples.yaml")
    assert request.wgsSampleConfig == "input/wgs_samples.yaml"
    assert "wgsSampleConfig" in ExecuteRequest.model_fields


def test_confirmed_wgs_prefix_becomes_prefix_for_the_next_plan_request() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    start = text.index("const confirmedStageParamsForExecution =")
    end = text.index("const shouldPauseForExpressionDataSource =", start)
    helper = text[start:end]
    script = f"""
const normalizedPrefix = value => String(value || '').trim();
const confirmParams = {{prefix: 'Os', cpu: 8}};
const params = {{prefix: 'At'}};
const batchExecuteParams = {{value: {{prefix: 'At', params: {{prefix: 'At'}}}}}};
const confirmCpuForStage = (_stage, value) => value;
const usesRnaseqSampleParallel = () => false;
const sampleParallelJobsForStage = (_stage, value) => value;
{helper}
const merged = confirmedStageParamsForExecution('wgs_qc', {{prefix: 'At'}});
console.log(JSON.stringify({{merged, params, batch: batchExecuteParams.value}}));
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(result.stdout)
    assert payload["merged"]["prefix"] == "Os"
    assert payload["params"]["prefix"] == "Os"
    assert payload["batch"]["prefix"] == "Os"
    assert payload["batch"]["params"]["prefix"] == "Os"


def test_wgs_index_frontend_uses_reference_and_tool_contract_fields() -> None:
    request = ExecuteRequest(stage="wgs_index", wgsReferenceFasta="input/reference.fasta", wgsSamtools="samtools", wgsBwaMem2="bwa-mem2", wgsGatk="gatk")
    assert request.wgsReferenceFasta == "input/reference.fasta"
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    assert "wgs_index" in text and "wgsReferenceFasta" in text


def test_frontend_binds_wgs_qc_config_and_fastp_parameters_to_execute_request() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    assert "applyWgsExecutionParams" in text
    assert "wgsSampleConfig" in text
    assert "wgsFastp" in text
    assert "wgsMaxJobs" in text
    assert text.count("applyWgsExecutionParams(requestBody, stageName, activeStageParams);") == 1
    assert text.count("applyWgsExecutionParams(requestBody, stageId, stageParams);") == 1
    active_call = text.index("applyWgsExecutionParams(requestBody, stageName, activeStageParams);")
    batch_call = text.index("applyWgsExecutionParams(requestBody, stageId, stageParams);")
    rnaseq_helper = text.index("const applyRnaseqExecutionParams")
    assert active_call > rnaseq_helper and batch_call > rnaseq_helper
    assert "applyWgsExecutionParams(requestBody, stageName, activeStageParams);" in text[active_call - 250:active_call + 100]
    assert "applyWgsExecutionParams(requestBody, stageId, stageParams);" in text[batch_call - 250:batch_call + 100]


def test_wgs_execution_resource_params_fall_back_to_positive_integers() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    apply_start = text.index("const applyWgsExecutionParams =")
    apply_end = text.index("const applyRnaseqExecutionParams =", apply_start)
    script = f"""
const firstDefinedParam = (...values) => {{
  for (const value of values) {{
    if (value !== undefined && value !== null && value !== '') return value;
  }}
  return '';
}};
const numberParam = (value, defaultValue) => {{
  const num = Number(value);
  return Number.isFinite(num) ? num : defaultValue;
}};
const confirmParams = {{}};
{text[apply_start:apply_end]}
const empty = applyWgsExecutionParams({{}}, 'wgs_qc', {{}});
const invalid = applyWgsExecutionParams({{}}, 'wgs_qc', {{
  wgsThreads: 0, wgsMaxJobs: -2
}});
const explicit = applyWgsExecutionParams({{}}, 'wgs_qc', {{
  wgsThreads: 4, wgsMaxJobs: 2
}});
console.log(JSON.stringify({{empty, invalid, explicit}}));
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(result.stdout)
    assert payload["empty"]["wgsThreads"] == 1
    assert payload["empty"]["wgsMaxJobs"] == 1
    assert payload["invalid"]["wgsThreads"] == 1
    assert payload["invalid"]["wgsMaxJobs"] == 1
    assert payload["explicit"]["wgsThreads"] == 4
    assert payload["explicit"]["wgsMaxJobs"] == 2


def test_wgs_bam_qc_frontend_has_both_authorities_and_fixed_metric_name() -> None:
    request = ExecuteRequest(
        stage="wgs_bam_qc",
        wgsMarkdupManifest="final/Cohort.wgs_markdup.manifest.json",
        wgsIndexManifest="final/Cohort.wgs_index.manifest.json",
        wgsIndexAnalysisId="wgs-analysis:" + "a" * 64,
        wgsCoverageThreshold=1,
    )
    assert request.wgsMarkdupManifest.endswith(".wgs_markdup.manifest.json")
    assert request.wgsCoverageThreshold == 1
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    for token in (
        "wgs_bam_qc", "wgsMarkdupManifest", "wgsIndexManifest",
        "wgsIndexAnalysisId", "wgsCoverageThreshold",
        "wgsBamQcHandoffUpdates",
    ):
        assert token in text


def test_wgs_bam_qc_coverage_threshold_is_fixed_to_one() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    apply_start = text.index("const applyWgsExecutionParams =")
    apply_end = text.index("const applyRnaseqExecutionParams =", apply_start)
    script = f"""
const firstDefinedParam = (...values) => {{
  for (const value of values) {{
    if (value !== undefined && value !== null && value !== '') return value;
  }}
  return '';
}};
const numberParam = (value, defaultValue) => {{
  const num = Number(value);
  return Number.isFinite(num) ? num : defaultValue;
}};
const confirmParams = {{}};
{text[apply_start:apply_end]}
const empty = applyWgsExecutionParams({{}}, 'wgs_bam_qc', {{}});
console.log(JSON.stringify(empty));
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    assert json.loads(result.stdout)["wgsCoverageThreshold"] == 1

    with pytest.raises(ValidationError):
        ExecuteRequest(stage="wgs_bam_qc", wgsCoverageThreshold=0)


def test_wgs_downstream_empty_numeric_params_keep_contract_defaults() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    apply_start = text.index("const applyWgsExecutionParams =")
    apply_end = text.index("const applyRnaseqExecutionParams =", apply_start)
    script = f"""
const firstDefinedParam = (...values) => {{
  for (const value of values) {{
    if (value !== undefined && value !== null && value !== '') return value;
  }}
  return '';
}};
const numberParam = (value, defaultValue) => {{
  const num = Number(value);
  return Number.isFinite(num) ? num : defaultValue;
}};
const booleanParam = (_value, defaultValue) => defaultValue;
const confirmParams = {{}};
{text[apply_start:apply_end]}
const haplotype = applyWgsExecutionParams(
  {{}}, 'wgs_haplotype_calling', {{}}
);
const joint = applyWgsExecutionParams(
  {{}}, 'wgs_joint_genotyping', {{}}
);
const impute = applyWgsExecutionParams({{}}, 'wgs_impute', {{}});
const finalFilter = applyWgsExecutionParams(
  {{}}, 'wgs_maf_missing_filter', {{}}
);
console.log(JSON.stringify({{haplotype, joint, impute, finalFilter}}));
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(result.stdout)
    assert payload["haplotype"]["wgsNativePairHmmThreads"] == 1
    assert payload["joint"]["wgsMaxAlternateAlleles"] == 6
    assert payload["joint"]["wgsCallingThreshold"] == 10
    assert payload["impute"]["wgsWindow"] == 40
    assert payload["impute"]["wgsOverlap"] == 2
    assert payload["impute"]["wgsNe"] == 100000
    assert payload["impute"]["wgsIterations"] == 12
    assert payload["impute"]["wgsPhaseStates"] == 280
    assert payload["impute"]["wgsImpStates"] == 1600
    assert payload["impute"]["wgsCluster"] == 0.005
    assert payload["finalFilter"]["wgsMaf"] == 0.05
    assert payload["finalFilter"]["wgsMaxMissing"] == 0.5


@pytest.mark.parametrize(
    ("stage", "field"),
    (
        ("wgs_haplotype_calling", "wgsNativePairHmmThreads"),
        ("wgs_joint_genotyping", "wgsMaxAlternateAlleles"),
        ("wgs_joint_genotyping", "wgsCallingThreshold"),
        ("wgs_impute", "wgsWindow"),
        ("wgs_impute", "wgsOverlap"),
        ("wgs_impute", "wgsNe"),
        ("wgs_impute", "wgsIterations"),
        ("wgs_impute", "wgsPhaseStates"),
        ("wgs_impute", "wgsImpStates"),
        ("wgs_impute", "wgsCluster"),
        ("wgs_maf_missing_filter", "wgsMaf"),
        ("wgs_maf_missing_filter", "wgsMaxMissing"),
    ),
)
def test_wgs_downstream_api_rejects_zero_scientific_params(
    stage: str, field: str
) -> None:
    with pytest.raises(ValidationError):
        ExecuteRequest(stage=stage, **{field: 0})


def test_wgs_haplotype_frontend_uses_three_authorities_and_bam_qc_handoff() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    apply_start = text.index("const applyWgsExecutionParams =")
    apply_end = text.index("const applyRnaseqExecutionParams =", apply_start)
    handoff_start = text.index("const wgsHaplotypeCallingHandoffUpdates =")
    handoff_end = text.index(
        "const handoffWgsHaplotypeCallingProducerAuthority =", handoff_start
    )
    samples = "wgs-samples:" + "a" * 64
    reference = "wgs-reference:" + "b" * 64
    analysis = "wgs-analysis:" + "c" * 64
    script = f"""
const firstDefinedParam = (...values) => values.find(v => v !== undefined && v !== null && v !== '');
const numberParam = (value, fallback) => Number(value || fallback);
const confirmParams = {{}};
{text[apply_start:apply_end]}
{text[handoff_start:handoff_end]}
const body = applyWgsExecutionParams({{}}, 'wgs_haplotype_calling', {{
  wgsMarkdupManifest: 'markdup.json', wgsBamQcManifest: 'bam-qc.json',
  wgsIndexManifest: 'index.json', wgsNativePairHmmThreads: 3
}});
const handoff = wgsHaplotypeCallingHandoffUpdates('wgs_bam_qc', {{
  result: {{manifest_path: '/workspace/job/final/Cohort.wgs_bam_qc.manifest.json',
    stage_status: {{completed: true, stage: 'wgs_bam_qc', prefix: 'Cohort',
      wgs_sample_set_id: '{samples}', wgs_reference_id: '{reference}',
      wgs_analysis_id: '{analysis}'}}}}
}}, 'Cohort');
console.log(JSON.stringify({{body, handoff}}));
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(result.stdout)
    assert payload["body"]["wgsBamQcManifest"] == "bam-qc.json"
    assert payload["body"]["wgsNativePairHmmThreads"] == 3
    assert payload["handoff"]["wgsBamQcAnalysisId"] == analysis


def test_wgs_gvcf_combine_frontend_uses_two_authorities_and_handoff() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    apply_start = text.index("const applyWgsExecutionParams =")
    apply_end = text.index("const applyRnaseqExecutionParams =", apply_start)
    handoff_start = text.index("const wgsGvcfCombineHandoffUpdates =")
    handoff_end = text.index(
        "const handoffWgsGvcfCombineProducerAuthority =", handoff_start
    )
    samples = "wgs-samples:" + "a" * 64
    reference = "wgs-reference:" + "b" * 64
    analysis = "wgs-analysis:" + "c" * 64
    script = f"""
const firstDefinedParam = (...values) => values.find(v => v !== undefined && v !== null && v !== '');
const numberParam = (value, fallback) => Number(value || fallback);
const confirmParams = {{}};
{text[apply_start:apply_end]}
{text[handoff_start:handoff_end]}
const body = applyWgsExecutionParams({{}}, 'wgs_gvcf_combine', {{
  wgsHaplotypeManifest: 'haplotype.json', wgsIndexManifest: 'index.json',
  wgsGatk: '/opt/gatk'
}});
const handoff = wgsGvcfCombineHandoffUpdates('wgs_haplotype_calling', {{
  result: {{manifest_path: '/workspace/job/final/Cohort.wgs_haplotype_calling.manifest.json',
    stage_status: {{completed: true, stage: 'wgs_haplotype_calling', prefix: 'Cohort',
      wgs_sample_set_id: '{samples}', wgs_reference_id: '{reference}',
      wgs_analysis_id: '{analysis}'}}}}
}}, 'Cohort');
console.log(JSON.stringify({{body, handoff}}));
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(result.stdout)
    assert payload["body"]["wgsHaplotypeManifest"] == "haplotype.json"
    assert payload["body"]["wgsIndexManifest"] == "index.json"
    assert payload["body"]["wgsGatk"] == "/opt/gatk"
    assert payload["handoff"]["wgsHaplotypeAnalysisId"] == analysis


def test_wgs_joint_genotyping_frontend_uses_two_authorities_parameters_and_handoff() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    apply_start = text.index("const applyWgsExecutionParams =")
    apply_end = text.index("const applyRnaseqExecutionParams =", apply_start)
    handoff_start = text.index("const wgsJointGenotypingHandoffUpdates =")
    handoff_end = text.index(
        "const handoffWgsJointGenotypingProducerAuthority =", handoff_start
    )
    samples = "wgs-samples:" + "a" * 64
    reference = "wgs-reference:" + "b" * 64
    analysis = "wgs-analysis:" + "c" * 64
    script = f"""
const firstDefinedParam = (...values) => values.find(v => v !== undefined && v !== null && v !== '');
const numberParam = (value, fallback) => Number(value || fallback);
const booleanParam = (value, fallback) => value === undefined || value === null || value === '' ? fallback : Boolean(value);
const confirmParams = {{}};
{text[apply_start:apply_end]}
{text[handoff_start:handoff_end]}
const body = applyWgsExecutionParams({{}}, 'wgs_joint_genotyping', {{
  wgsCombineManifest: 'combine.json', wgsIndexManifest: 'index.json',
  wgsGatk: '/opt/gatk', wgsMaxAlternateAlleles: 6,
  wgsCallingThreshold: 10, wgsIncludeNonVariantSites: false
}});
const handoff = wgsJointGenotypingHandoffUpdates('wgs_gvcf_combine', {{
  result: {{manifest_path: '/workspace/job/final/Cohort.wgs_gvcf_combine.manifest.json',
    stage_status: {{completed: true, stage: 'wgs_gvcf_combine', prefix: 'Cohort',
      wgs_sample_set_id: '{samples}', wgs_reference_id: '{reference}',
      wgs_analysis_id: '{analysis}'}}}}
}}, 'Cohort');
console.log(JSON.stringify({{body, handoff}}));
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(result.stdout)
    assert payload["body"]["wgsCombineManifest"] == "combine.json"
    assert payload["body"]["wgsIndexManifest"] == "index.json"
    assert payload["body"]["wgsMaxAlternateAlleles"] == 6
    assert payload["body"]["wgsCallingThreshold"] == 10
    assert payload["body"]["wgsIncludeNonVariantSites"] is False
    assert payload["handoff"]["wgsCombineAnalysisId"] == analysis


def test_wgs_variant_select_frontend_has_confirmed_fallback_contract() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    for token in (
        "wgs_variant_select",
        "wgsJointManifest",
        "wgsMethod",
        "wgsBcftools",
        "wgsJointAnalysisId",
        "confirmWgsFallback",
        "cancelWgsFallback",
    ):
        assert token in text
    assert "wgsMethod: pendingWgsFallback.fallback_method" in text


def test_wgs_variant_select_modal_uses_real_reason_cancel_zero_confirm_once() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    start = text.index("const wgsFallbackDetails =")
    end = text.index("const waitForDialogTransition =", start)
    script = f"""
{text[start:end]}
let requests = [];
let shown = [];
const error = {{responseData: {{result: {{
  error: 'GATK SelectVariants SNP failed with exit code 23',
  fallback_method: 'bcftools'
}}}}}};
const executeFn = async (stage, params) => {{
  requests.push({{stage, params}});
  return {{ok: true}};
}};
(async () => {{
  await confirmWgsFallback(error, 'wgs_variant_select', {{wgsMethod: 'gatk'}},
    executeFn, async (message) => {{ shown.push(message); throw new Error('cancel'); }});
  const afterCancel = requests.length;
  await confirmWgsFallback(error, 'wgs_variant_select', {{wgsMethod: 'gatk'}},
    executeFn, async (message) => {{ shown.push(message); }});
  console.log(JSON.stringify({{afterCancel, requests, shown}}));
}})();
"""
    result = subprocess.run(
        ["/usr/bin/node", "-e", script],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(result.stdout)
    assert payload["afterCancel"] == 0
    assert len(payload["requests"]) == 1
    assert payload["requests"][0]["stage"] == "wgs_variant_select"
    assert payload["requests"][0]["params"]["wgsMethod"] == "bcftools"
    assert all(
        "GATK SelectVariants SNP failed with exit code 23" in message
        for message in payload["shown"]
    )


def test_active_foreground_batch_guards_wgs_out_of_background_executor() -> None:
    text = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    start = text.index("const plannedExecutionStages =")
    end = text.index("// 存储批量执行信息", start)
    scheduling_branch = text[start:end]
    background_call = scheduling_branch.index("startBackgroundBatchExecution(")

    assert "wgs_" in scheduling_branch[:background_call].lower()
    assert "startBackgroundBatchExecution(" in scheduling_branch