from __future__ import annotations

import asyncio
import importlib
import importlib.util
import json
import os
import subprocess
from types import SimpleNamespace
from pathlib import Path

import pytest


def test_immediate_postflight_rejects_manifest_scope_mismatch(monkeypatch) -> None:
    web = importlib.import_module("apps.web.vue3.main")
    helper_called = False

    def fake_result_artifacts(*args, **kwargs):
        nonlocal helper_called
        helper_called = True
        return [{"role": "sorted_vcf"}, {"role": "tassel_hapmap"}]

    monkeypatch.setattr(web, "_gwas_fmt_result_artifacts", fake_result_artifacts)
    records, error = web._validate_gwas_fmt_immediate_postflight(
        {"stage": "gwas_fmt", "status": "success", "genotype_scope_id": "genotype:manifest"},
        "genotype:trusted-request",
        Path("/unused"),
    )
    assert records == []
    assert error == "genotype_scope_mismatch"
    assert helper_called is False


def test_immediate_postflight_requires_exact_canonical_roles(monkeypatch) -> None:
    web = importlib.import_module("apps.web.vue3.main")
    monkeypatch.setattr(web, "_gwas_fmt_result_artifacts", lambda *args, **kwargs: [{"role": "sorted_vcf"}])
    records, error = web._validate_gwas_fmt_immediate_postflight(
        {"stage": "gwas_fmt", "status": "success", "genotype_scope_id": "genotype:trusted-request"},
        "genotype:trusted-request",
        Path("/unused"),
    )
    assert records == []
    assert error == "invalid_outputs"


def test_immediate_postflight_accepts_validated_supporting_artifacts(monkeypatch) -> None:
    web = importlib.import_module("apps.web.vue3.main")
    monkeypatch.setattr(web, "_gwas_fmt_result_artifacts", lambda *args, **kwargs: [
        {"role": "sorted_vcf"},
        {"role": "tassel_hapmap"},
        {"role": "report"},
        {"role": "environment"},
    ])
    records, error = web._validate_gwas_fmt_immediate_postflight(
        {"stage": "gwas_fmt", "status": "success", "genotype_scope_id": "genotype:trusted-request"},
        "genotype:trusted-request",
        Path("/workspace/job"),
    )

    assert error == ""
    assert {record["role"] for record in records} == {
        "sorted_vcf", "tassel_hapmap", "report", "environment",
    }


def test_standard_log_lookup_rejects_symlink_escape(tmp_path: Path) -> None:
    web = importlib.import_module("apps.web.vue3.main")
    outside = tmp_path / "outside"
    outside.mkdir()
    (outside / "20260720_120000_gwas_fmt.log").write_text("outside\n", encoding="utf-8")
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    (workspace / "logs").symlink_to(outside, target_is_directory=True)

    assert web._latest_standard_stage_log(workspace, "gwas_fmt") is None


def test_standard_log_lookup_rejects_candidate_symlink(tmp_path: Path) -> None:
    web = importlib.import_module("apps.web.vue3.main")
    logs = tmp_path / "workspace/logs"
    logs.mkdir(parents=True)
    real_log = logs / "real-log-target.txt"
    real_log.write_text("real\n", encoding="utf-8")
    symlink_log = logs / "20260720_130000_gwas_fmt.log"
    symlink_log.symlink_to(real_log)

    assert web._latest_standard_stage_log(tmp_path / "workspace", "gwas_fmt") is None


def test_gwas_precommand_failure_does_not_reuse_or_display_old_log(tmp_path: Path, monkeypatch) -> None:
    web = importlib.import_module("apps.web.vue3.main")
    backend = importlib.import_module("apps.web.pipeline_backend")
    resource_governor = importlib.import_module("gfa_core.mas.resource_governor")
    workspace_root = tmp_path / "workspace"
    input_root = tmp_path / "input"
    job_id = "job-old-log"
    task_id = "task-old-log"
    workspace = workspace_root / job_id
    old_log = workspace / "logs/20260720_110000_gwas_fmt.log"
    old_log.parent.mkdir(parents=True)
    old_log.write_text("old attempt\n", encoding="utf-8")
    (input_root / job_id).mkdir(parents=True)

    class NoopLease:
        def __init__(self, **_kwargs):
            pass

        def wait(self, update_state=None):
            return None

        def release(self, **_kwargs):
            return None

    monkeypatch.setattr(resource_governor, "ResourceLease", NoopLease)
    monkeypatch.setattr(resource_governor, "get_task", lambda _task_id: {})
    monkeypatch.setattr(resource_governor, "snapshot", lambda **_kwargs: {"usage": {}})
    monkeypatch.setattr(web, "WORKSPACE_ROOT", workspace_root)
    monkeypatch.setattr(web, "INPUT_ROOT", input_root)
    monkeypatch.setattr(web.time, "sleep", lambda _seconds: None)
    monkeypatch.setattr(web, "_run_failure_diagnosis_for_task", lambda *args, **kwargs: {})
    monkeypatch.setattr(web, "_attach_single_stage_status", lambda _task: None)
    monkeypatch.setattr(web, "_attach_annotation_progress", lambda _task: None)
    monkeypatch.setattr(backend, "can_start_new_task", lambda: (False, "pre-command failure"))
    web.tasks[task_id] = {
        "id": task_id, "stage": "gwas_fmt", "job_id": job_id, "status": "pending",
        "message": "", "created_at": "2026-07-20T12:00:00",
    }

    result = web.run_stage_sync(task_id, "gwas_fmt", job_id, {
        "prefix": "cohort", "genotypeScopeId": "genotype:scope",
    })

    assert result["success"] is False
    assert result["log_path"] == ""
    assert old_log.read_text(encoding="utf-8") == "old attempt\n"
    task_response = asyncio.run(web.get_task_status(task_id))
    assert task_response["result"]["log_path"] == ""
    assert task_response["log"] == ""


def test_gwas_failure_formatter_deduplicates_short_terminal_error(tmp_path: Path) -> None:
    frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper_source = frontend[
        frontend.index("const compactTechnicalError ="):
        frontend.index("const showFailureDiagnosis =")
    ]
    log_path = "/workspace/job/logs/20260720_120000_gwas_fmt.log"
    reason = "invalid_outputs"
    payload = {
        "stage": "gwas_fmt",
        "message": f"gwas_fmt 结果未通过 manifest 校验。 Log: {log_path}",
        "result": {"success": False, "stage": "gwas_fmt", "error": reason, "log_path": log_path},
        "log": f"tool output\nERROR: {reason}\n",
    }
    node_script = tmp_path / "format-short-gwas-error.cjs"
    node_script.write_text(
        "const getStageLabel=(stage)=>stage;\n"
        + helper_source
        + "\nconst data=" + json.dumps(payload) + ";"
        + "process.stdout.write(formatStageFailureMessage('gwas_fmt',data));",
        encoding="utf-8",
    )

    completed = subprocess.run(["node", str(node_script)], text=True, capture_output=True, check=False)

    assert completed.returncode == 0, completed.stderr
    assert f"原因：{reason}" in completed.stdout
    assert completed.stdout.count(reason) == 1
    assert "技术细节：" in completed.stdout
    assert "tool output" in completed.stdout
    assert f"日志：{log_path}" in completed.stdout


def test_confirmation_fail_policy_is_submitted_by_shared_binding(tmp_path: Path) -> None:
    frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    assert 'v-model="confirmParams.gwasUnsupportedMarkerPolicy"' in frontend
    start = frontend.index("const applyGwasExecutionParams =")
    end = frontend.index("const executeSingleStageWithParams =", start)
    node_script = tmp_path / "confirm-policy.cjs"
    node_script.write_text(
        "const batchExecuteParams={value:{}};"
        "const confirmParams={gwasVcf:'cohort.vcf.gz',tasselPipeline:'run_pipeline.pl',tasselXmx:'10g',gwasUnsupportedMarkerPolicy:'fail'};"
        + frontend[start:end]
        + "const submitted=applyGwasExecutionParams({},'gwas_fmt',{}); process.stdout.write(submitted.gwasUnsupportedMarkerPolicy);",
        encoding="utf-8",
    )
    completed = subprocess.run(["node", str(node_script)], text=True, capture_output=True, check=False)
    assert completed.returncode == 0, completed.stderr
    assert completed.stdout == "fail"


@pytest.mark.parametrize(
    ("mode", "provenance_kind"),
    [
        ("nonzero", "absolute"),
        ("postflight", "parent"),
        ("postflight", "missing"),
        ("postflight", "stale"),
        ("postflight", "no_backend_log"),
    ],
)
def test_gwas_task_failures_use_backend_log_not_manifest_provenance(
    tmp_path: Path, monkeypatch, mode: str, provenance_kind: str,
) -> None:
    web = importlib.import_module("apps.web.vue3.main")
    backend = importlib.import_module("apps.web.pipeline_backend")
    resource_governor = importlib.import_module("gfa_core.mas.resource_governor")
    stage_identity = importlib.import_module("gfa_core.mas.tools.stage_identity")
    workspace_root = tmp_path / "workspace"
    input_root = tmp_path / "input"
    job_id = f"job-{mode}-{provenance_kind}"
    task_id = f"task-{mode}-{provenance_kind}"
    workspace = workspace_root / job_id
    log_path = workspace / "logs" / "20260720_120000_gwas_fmt.log"
    manifest_path = workspace / "final" / "cohort.gwas_fmt.manifest.json"
    (input_root / job_id).mkdir(parents=True)
    vcf = tmp_path / f"{mode}-{provenance_kind}.vcf"
    tassel = tmp_path / f"run-{mode}-{provenance_kind}.pl"

    helper_path = Path(__file__).with_name("test_gwas_fmt_stage18.py")
    spec = importlib.util.spec_from_file_location(f"gwas_stage18_{mode}_{provenance_kind}", helper_path)
    assert spec and spec.loader
    helpers = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(helpers)
    helpers._write_vcf(vcf)
    helpers._write_fake_tassel(tassel, fail=mode == "nonzero")

    class NoopLease:
        def __init__(self, **_kwargs):
            pass

        def wait(self, update_state=None):
            return None

        def release(self, **_kwargs):
            return None

    monkeypatch.setattr(resource_governor, "ResourceLease", NoopLease)
    monkeypatch.setattr(web, "WORKSPACE_ROOT", workspace_root)
    monkeypatch.setattr(web, "INPUT_ROOT", input_root)
    monkeypatch.setattr(web.time, "sleep", lambda _seconds: None)
    monkeypatch.setattr(web, "_run_failure_diagnosis_for_task", lambda *args, **kwargs: {})
    monkeypatch.setattr(web, "_attach_single_stage_status", lambda _task: None)
    monkeypatch.setattr(web, "_attach_annotation_progress", lambda _task: None)
    monkeypatch.setattr(resource_governor, "get_task", lambda _task_id: {})
    monkeypatch.setattr(resource_governor, "snapshot", lambda **_kwargs: {"usage": {}})
    monkeypatch.setattr(backend, "_now_tag", lambda: "20260720_120000")

    real_run_stage = backend.run_stage
    outside_log = tmp_path / "outside_gwas_fmt.log"
    outside_log.write_text("outside\n", encoding="utf-8")

    def run_stage_with_untrusted_provenance(**kwargs):
        result_message, result_manifest = real_run_stage(**kwargs)
        assert result_manifest == manifest_path
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
        if mode == "postflight":
            manifest["error"] = "stale_success_manifest_error"
        if provenance_kind == "absolute":
            untrusted_path = str(outside_log)
        elif provenance_kind == "parent":
            escape_log = workspace / "escape_gwas_fmt.log"
            escape_log.write_text("escape\n", encoding="utf-8")
            untrusted_path = "../escape_gwas_fmt.log"
        elif provenance_kind == "missing":
            untrusted_path = "logs/missing_gwas_fmt.log"
        elif provenance_kind == "stale":
            stale_log = workspace / "logs/19990101_000000_gwas_fmt.log"
            stale_log.write_text("stale\n", encoding="utf-8")
            os.utime(stale_log, (1, 1))
            untrusted_path = "logs/19990101_000000_gwas_fmt.log"
        else:
            untrusted_path = str(outside_log)
            log_path.unlink()
        manifest.setdefault("provenance", {})["log_path"] = untrusted_path
        manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
        return result_message, result_manifest

    monkeypatch.setattr(backend, "run_stage", run_stage_with_untrusted_provenance)

    frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper_source = frontend[
        frontend.index("const compactTechnicalError ="):
        frontend.index("const showFailureDiagnosis =")
    ]

    completed_status = mode == "postflight"
    expected_error = "invalid_outputs" if completed_status else "TASSEL execution failed with exit code 3"
    monkeypatch.setattr(
        stage_identity,
        "get_stage_status_for_identity",
        lambda **_kwargs: SimpleNamespace(
            completed=completed_status,
            to_dict=lambda: {"completed": completed_status},
        ),
    )
    monkeypatch.setattr(
        web,
        "_validate_gwas_fmt_immediate_postflight",
        lambda *_args: ([], "invalid_outputs" if completed_status else ""),
    )
    web.tasks[task_id] = {
        "id": task_id,
        "stage": "gwas_fmt",
        "job_id": job_id,
        "status": "pending",
        "message": "",
        "created_at": "2026-07-20T12:00:00",
    }

    result = web.run_stage_sync(task_id, "gwas_fmt", job_id, {
        "prefix": "cohort",
        "genotypeScopeId": "genotype:scope",
        "gwasVcf": str(vcf),
        "tasselPipeline": str(tassel),
        "tasselXmx": "1g",
    })

    task = web.tasks[task_id]
    assert result["success"] is False
    assert result["error"] == expected_error
    expected_log_path = "" if provenance_kind == "no_backend_log" else str(log_path)
    assert result["log_path"] == expected_log_path
    assert task["message"].count(str(log_path)) == (0 if not expected_log_path else 1)
    assert expected_error not in task["message"]
    log_text = log_path.read_text(encoding="utf-8") if expected_log_path else ""
    if expected_log_path:
        assert log_text.endswith(f"ERROR: {expected_error}\n")
        assert log_text.count(f"ERROR: {expected_error}") == 1

    task_response = asyncio.run(web.get_task_status(task_id))
    assert task_response["result"]["log_path"] == expected_log_path
    assert task_response["log"] == log_text

    node_script = tmp_path / f"format-{mode}-{provenance_kind}.cjs"
    node_script.write_text(
        "const getStageLabel=(stage)=>stage;\n"
        + helper_source
        + "\nconst data=" + json.dumps(task_response) + ";"
        + "process.stdout.write(formatStageFailureMessage('gwas_fmt',data));",
        encoding="utf-8",
    )
    completed = subprocess.run(["node", str(node_script)], text=True, capture_output=True, check=False)
    assert completed.returncode == 0, completed.stderr
    assert all(label in completed.stdout for label in ("原因：", "建议：", "技术细节："))
    assert ("日志：" in completed.stdout) is bool(expected_log_path)
    assert completed.stdout.count(str(log_path)) == (0 if not expected_log_path else 1)
    assert completed.stdout.count(expected_error) == 1