from __future__ import annotations

import importlib.util
import multiprocessing
import sys
import time
from pathlib import Path
from types import SimpleNamespace

import pytest


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


def load_ppi_module():
    module_name = "ppi_skill_task3"
    sys.modules.pop(module_name, None)
    spec = importlib.util.spec_from_file_location(
        module_name, REPO_ROOT / "skills/ppi/run.py"
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module


class FakeStringClient:
    def __init__(self, mapping=None, network=None, network_error=None):
        self.mapping = dict(mapping or {})
        self.network = [] if network is None else network
        self.network_error = network_error
        self.mapping_calls = []
        self.network_calls = []
        self.species_id = "3562"

    def get_string_ids(self, identifiers):
        self.mapping_calls.append(list(identifiers))
        rows = []
        for query_index, identifier in enumerate(identifiers):
            match = self.mapping.get(identifier)
            if match is None:
                continue
            string_id, preferred_name = (
                match if isinstance(match, tuple) else (match, identifier)
            )
            rows.append(
                {
                    "queryIndex": query_index,
                    "queryItem": identifier,
                    "stringId": string_id,
                    "preferredName": preferred_name,
                    "ncbiTaxonId": 3562,
                }
            )
        return rows

    def get_network(self, identifiers, required_score=200, network_type="functional"):
        self.network_calls.append((list(identifiers), required_score, network_type))
        if self.network_error is not None:
            raise self.network_error
        return self.network


class FakeResponse:
    def __init__(self, status_code=200, payload=None):
        self.status_code = status_code
        self._payload = payload

    def json(self):
        if isinstance(self._payload, Exception):
            raise self._payload
        return self._payload


class FakeHttpClient:
    def __init__(self, responses):
        self.responses = list(responses)
        self.calls = []

    def _next(self, method, url, **kwargs):
        self.calls.append((method, url, kwargs))
        if not self.responses:
            raise AssertionError("unexpected HTTP request")
        response = self.responses.pop(0)
        if not hasattr(response, "url"):
            response.url = url
        return response

    def post(self, url, **kwargs):
        return self._next("POST", url, **kwargs)


def write_id_map(tmp_path: Path, rows: list[tuple[str, str, str]]) -> Path:
    path = tmp_path / "soe.pep_cds_idmap.tsv"
    lines = [
        "species_code\tprotein_id\tpep_id\tcds_id\tgene_id\tcds_header\tncbi_gene_id"
    ]
    for accession, gene_id, symbol in rows:
        lines.append(
            "\t".join(
                [
                    "soe",
                    accession,
                    accession,
                    accession,
                    symbol,
                    f"{accession} [gene={symbol}] [db_xref=GeneID:{gene_id}]",
                    gene_id,
                ]
            )
        )
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return path


def origin(module, canonical, reference):
    return module.PpiQueryOrigin(
        canonical_target_id=canonical,
        ortholog_reference_id=reference,
        raw_query_id=reference,
        relationship_source="orthologs.tsv",
    )


def test_parse_ortholog_mapping_preserves_every_exact_relationship(tmp_path):
    module = load_ppi_module()
    mapping = tmp_path / "orthologs.tsv"
    mapping.write_text(
        "Ft1\tXP_000001.1\nFt1\tXP_000002.3\nFt2\tXP_000001.1\n",
        encoding="utf-8",
    )

    origins = module.parse_ortholog_mapping(mapping)

    assert len(origins) == 3
    assert [(item.canonical_target_id, item.ortholog_reference_id) for item in origins] == [
        ("Ft1", "XP_000001.1"),
        ("Ft1", "XP_000002.3"),
        ("Ft2", "XP_000001.1"),
    ]
    assert all(item.raw_query_id == item.ortholog_reference_id for item in origins)


def make_skill(module, client):
    return module.PpiSkill(string_api_factory=lambda _species: client)


def test_cli_forwards_id_map_to_ppi_skill(tmp_path, monkeypatch):
    module = load_ppi_module()
    captured = {}

    class FakePpiSkill:
        def __init__(self, config_path=None):
            captured["config_path"] = config_path

        def run(self, **kwargs):
            captured.update(kwargs)
            return {"status": "success", "manifest_path": "manifest.json", "metrics": {}}

    monkeypatch.setattr(module, "PpiSkill", FakePpiSkill)
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "run.py",
            "--gene-ids",
            str(tmp_path / "genes.txt"),
            "--species",
            "soe",
            "--prefix",
            "Ft.SPL",
            "--output",
            str(tmp_path / "ppi"),
            "--id-map",
            "/tmp/soe.pep_cds_idmap.tsv",
        ],
    )

    with pytest.raises(SystemExit) as exc_info:
        module.main()

    assert exc_info.value.code == 0
    assert captured["id_map_file"] == Path("/tmp/soe.pep_cds_idmap.tsv")


def test_owner_taxon_query_uses_normalization_label_for_strict_alias_species(tmp_path):
    module = load_ppi_module()
    id_map = tmp_path / "string_taxon_4555.pep_cds_idmap.tsv"
    id_map.write_text(
        "species_code\tprotein_id\tpep_id\tcds_id\tgene_id\tcds_header\n"
        "string_taxon_4555\tSeita.1\tSeita.1\tSeita.1\tSeita1\tSeita.1 [gene=Seita1]\n",
        encoding="utf-8",
    )

    class OwnerTaxonClient:
        species_id = "4555"

        def get_string_ids(self, identifiers):
            index = identifiers.index("Seita.1")
            return [
                {
                    "queryIndex": index,
                    "queryItem": "Seita.1",
                    "stringId": "4555.Seita.1",
                    "preferredName": "Seita.1",
                    "ncbiTaxonId": 4555,
                }
            ]

    result = make_skill(module, OwnerTaxonClient()).resolve_query_ids(
        [origin(module, "SvGene1", "Seita.1")],
        "4555",
        id_map,
        normalization_species_label="string_taxon_4555",
    )

    assert result.status == "success"
    assert result.selected_string_ids == ["4555.Seita.1"]


def test_raw_xp_failure_falls_back_to_ncbi_geneid_and_selects_string_id(tmp_path):
    module = load_ppi_module()
    id_map = write_id_map(tmp_path, [("XP_000001.1", "1001", "LOC1001")])
    client = FakeStringClient({"1001": ("3562.PROT1", "PROT1")})

    result = make_skill(module, client).resolve_query_ids(
        [origin(module, "Ft1", "XP_000001.1")], "soe", id_map
    )

    assert result.selected_string_ids == ["3562.PROT1"]
    assert result.rows[0].mapping_status == "resolved"
    assert result.rows[0].selected_query_id == "1001"
    assert result.rows[0].relationship_source == "orthologs.tsv"
    assert result.rows[0].evidence_source == str(id_map)
    assert result.metrics["geneid_fallback_resolved_query_count"] == 1


def test_raw_and_geneid_mapping_to_same_string_id_is_not_ambiguous(tmp_path):
    module = load_ppi_module()
    id_map = write_id_map(tmp_path, [("XP_000001.1", "1001", "LOC1001")])
    client = FakeStringClient(
        {"XP_000001.1": "3562.PROT1", "1001": "3562.PROT1"}
    )

    result = make_skill(module, client).resolve_query_ids(
        [origin(module, "Ft1", "XP_000001.1")], "soe", id_map
    )

    assert result.rows[0].mapping_status == "resolved"
    assert result.metrics["ambiguous_member_count"] == 0


def test_one_input_mapping_to_two_distinct_string_ids_is_ambiguous(tmp_path):
    module = load_ppi_module()
    id_map = write_id_map(tmp_path, [("XP_000001.1", "1001", "LOC1001")])
    client = FakeStringClient(
        {"XP_000001.1": "3562.PROT1", "1001": "3562.PROT2"}
    )

    result = make_skill(module, client).resolve_query_ids(
        [origin(module, "Ft1", "XP_000001.1")], "soe", id_map
    )

    assert result.selected_string_ids == []
    assert result.rows[0].mapping_status == "ambiguous"
    assert result.rows[0].reason == "candidate_string_ids_conflict"


def test_partial_resolution_continues_with_explicit_unresolved_rows():
    module = load_ppi_module()
    client = FakeStringClient({"XP_1.1": "3562.PROT1"})
    origins = [
        origin(module, "Ft1", "XP_1.1"),
        origin(module, "Ft2", "XP_2.1"),
    ]

    result = make_skill(module, client).resolve_query_ids(origins, "soe", None)

    assert [row.mapping_status for row in result.rows] == ["resolved", "unresolved"]
    assert result.metrics["probe_status"] == "partial"
    assert result.selected_string_ids == ["3562.PROT1"]


def test_zero_resolved_ids_returns_id_resolution_failure():
    module = load_ppi_module()

    result = make_skill(module, FakeStringClient()).resolve_query_ids(
        [origin(module, "Ft1", "XP_1.1")], "soe", None
    )

    assert result.status == "id_resolution_failure"
    assert result.issue_code == "ppi_string_id_resolution_failed"
    assert result.metrics["probe_status"] == "failed"


def test_network_empty_list_is_success_empty_network():
    module = load_ppi_module()
    client = FakeStringClient(
        {"XP_1.1": "3562.PROT1"}, network=[]
    )

    result = make_skill(module, client).analyze_network(
        "soe",
        origins=[origin(module, "Ft1", "XP_1.1")],
        canonical_target_ids=["Ft1"],
    )

    assert result["interactions"] == []
    assert result["stats"]["network_status"] == "empty_network"


def test_network_http_error_is_external_api_failure():
    module = load_ppi_module()
    error = module.StringApiError("network failed", kind="http", status_code=503)
    client = FakeStringClient(
        {"XP_1.1": "3562.PROT1"}, network_error=error
    )

    with pytest.raises(module.StringApiError) as exc_info:
        make_skill(module, client).analyze_network(
            "soe",
            origins=[origin(module, "Ft1", "XP_1.1")],
            canonical_target_ids=["Ft1"],
        )

    assert exc_info.value.kind == "http"
    assert exc_info.value.status_code == 503


def test_get_network_classifies_http_response_failure():
    module = load_ppi_module()
    api = module.StringAPI(
        "soe",
        http_client=FakeHttpClient([FakeResponse(status_code=503)]),
        stable_api_base="https://stable.string-db.org/api",
        rate_gate=SimpleNamespace(wait=lambda: None),
    )

    with pytest.raises(module.StringApiError) as exc_info:
        api.get_network(["3562.PROT1"])

    assert exc_info.value.kind == "http"
    assert exc_info.value.status_code == 503


def test_get_network_classifies_json_decode_failure():
    module = load_ppi_module()
    api = module.StringAPI(
        "soe",
        http_client=FakeHttpClient([FakeResponse(payload=ValueError("invalid JSON"))]),
        stable_api_base="https://stable.string-db.org/api",
        rate_gate=SimpleNamespace(wait=lambda: None),
    )

    with pytest.raises(module.StringApiError) as exc_info:
        api.get_network(["3562.PROT1"])

    assert exc_info.value.kind == "json"


def test_get_string_ids_uses_query_index_not_response_position(tmp_path):
    module = load_ppi_module()
    response = FakeResponse(
        payload=[
            {
                "queryIndex": 1,
                "queryItem": "B",
                "stringId": "3562.B",
                "preferredName": "B",
                "ncbiTaxonId": 3562,
            },
            {
                "queryIndex": 0,
                "queryItem": "A",
                "stringId": "3562.A",
                "preferredName": "A",
                "ncbiTaxonId": 3562,
            },
        ]
    )
    http = FakeHttpClient([response])
    api = module.StringAPI(
        "soe",
        http_client=http,
        stable_api_base="https://stable.string-db.org/api",
        rate_gate=SimpleNamespace(wait=lambda: None),
    )

    rows = api.get_string_ids(["A", "B"])

    assert [row["queryItem"] for row in rows] == ["A", "B"]
    method, url, kwargs = http.calls[0]
    assert method == "POST"
    assert url == "https://stable.string-db.org/api/json/get_string_ids"
    assert kwargs["data"]["echo_query"] == 1
    assert kwargs["data"]["caller_identity"] == "PlantAI"
    assert kwargs["timeout"] == 30


def test_24_origins_collapse_to_12_queries_without_losing_provenance(tmp_path):
    module = load_ppi_module()
    aliases = []
    origins = []
    mapping = {}
    for index in range(12):
        accession = f"XP_{index:06d}.1"
        gene_id = str(1000 + index)
        aliases.append((accession, gene_id, f"LOC{gene_id}"))
        string_id = f"3562.PROT{index}"
        mapping[gene_id] = string_id
        if index < 5:
            mapping[accession] = string_id
        origins.extend(
            [
                origin(module, f"Ft{index * 2 + 1}", accession),
                origin(module, f"Ft{index * 2 + 2}", accession),
            ]
        )
    id_map = write_id_map(tmp_path, aliases)
    client = FakeStringClient(mapping, network=[])

    result = make_skill(module, client).resolve_query_ids(origins, "soe", id_map)

    assert len(result.rows) == 24
    assert [
        (row.canonical_target_id, row.ortholog_reference_id) for row in result.rows
    ] == [
        (item.canonical_target_id, item.ortholog_reference_id) for item in origins
    ]
    assert len(client.mapping_calls) == 1
    assert result.metrics == {
        "input_member_count": 24,
        "relationship_count": 24,
        "unique_reference_query_count": 12,
        "raw_resolved_query_count": 5,
        "geneid_fallback_resolved_query_count": 7,
        "resolved_reference_query_count": 12,
        "selected_string_query_count": 12,
        "resolved_member_count": 24,
        "unresolved_member_count": 0,
        "ambiguous_member_count": 0,
        "member_mapping_rate": 1.0,
        "query_mapping_rate": 1.0,
        "probe_status": "complete",
    }


def test_run_resolves_relationships_submits_selected_string_ids_and_writes_provenance(
    tmp_path,
):
    module = load_ppi_module()
    gene_file = tmp_path / "members.id"
    gene_file.write_text("Ft1\nFt2\n", encoding="utf-8")
    ortholog_file = tmp_path / "orthologs.tsv"
    ortholog_file.write_text(
        "Ft1\tXP_000001.1\nFt2\tXP_000001.1\n", encoding="utf-8"
    )
    id_map = write_id_map(tmp_path, [("XP_000001.1", "1001", "LOC1001")])
    client = FakeStringClient({"1001": ("3562.PROT1", "PROT1")}, network=[])
    output_dir = tmp_path / "ppi"

    result = module.PpiSkill(
        string_api_factory=lambda _species: client
    ).run(
        gene_ids_file=gene_file,
        species="soe",
        prefix="Ft.SPL",
        output_dir=output_dir,
        required_score=200,
        mapping_file=ortholog_file,
        id_map_file=id_map,
    )

    assert result["status"] == "success"
    assert client.network_calls == [(["3562.PROT1"], 200, "functional")]
    assert result["metrics"]["network_status"] == "empty_network"
    assert result["metrics"]["id_resolution"]["relationship_count"] == 2
    assert (output_dir / "Ft.SPL_network.tsv").read_text(encoding="utf-8") == (
        "protein1\tprotein2\tscore\n"
    )
    report = output_dir / "Ft.SPL_id_resolution.tsv"
    lines = report.read_text(encoding="utf-8").splitlines()
    assert lines[0].split("\t") == [
        "canonical_target_id",
        "ortholog_reference_id",
        "raw_query_id",
        "candidate_gene_id",
        "selected_query_id",
        "string_id",
        "string_preferred_name",
        "mapping_status",
        "method",
        "evidence_source",
        "relationship_source",
        "species",
        "taxon_id",
        "reason",
    ]
    assert [line.split("\t")[:3] for line in lines[1:]] == [
        ["Ft1", "XP_000001.1", "XP_000001.1"],
        ["Ft2", "XP_000001.1", "XP_000001.1"],
    ]
    assert all(line.split("\t")[10] == str(ortholog_file) for line in lines[1:])


def test_run_direct_target_raw_candidate_uses_canonical_member_evidence(tmp_path):
    module = load_ppi_module()
    gene_file = tmp_path / "members.id"
    gene_file.write_text("DIRECT1\n", encoding="utf-8")
    client = FakeStringClient({"DIRECT1": "3562.DIRECT1"}, network=[])
    output_dir = tmp_path / "direct"

    result = module.PpiSkill(string_api_factory=lambda _species: client).run(
        gene_ids_file=gene_file,
        species="soe",
        prefix="Soe.Direct",
        output_dir=output_dir,
    )

    assert result["status"] == "success"
    report_lines = (output_dir / "Soe.Direct_id_resolution.tsv").read_text(
        encoding="utf-8"
    ).splitlines()
    header = report_lines[0].split("\t")
    row = dict(zip(header, report_lines[1].split("\t")))
    assert row["ortholog_reference_id"] == ""
    assert row["evidence_source"] == "canonical_member_input"
    assert row["relationship_source"] == str(gene_file)


def test_run_counts_canonical_member_missing_ortholog_as_unresolved(tmp_path):
    module = load_ppi_module()
    gene_file = tmp_path / "members.id"
    gene_file.write_text("Ft1\nFt2\n", encoding="utf-8")
    ortholog_file = tmp_path / "orthologs.tsv"
    ortholog_file.write_text("Ft1\tXP_000001.1\n", encoding="utf-8")
    id_map = write_id_map(tmp_path, [("XP_000001.1", "1001", "LOC1001")])
    client = FakeStringClient({"1001": "3562.PROT1"}, network=[])

    result = module.PpiSkill(string_api_factory=lambda _species: client).run(
        gene_ids_file=gene_file,
        species="soe",
        prefix="Ft.SPL",
        output_dir=tmp_path / "partial",
        mapping_file=ortholog_file,
        id_map_file=id_map,
    )

    metrics = result["metrics"]["id_resolution"]
    assert metrics["input_member_count"] == 2
    assert metrics["relationship_count"] == 1
    assert metrics["resolved_member_count"] == 1
    assert metrics["unresolved_member_count"] == 1
    assert metrics["member_mapping_rate"] == 0.5


def test_many_canonical_targets_sharing_one_reference_keep_separate_rows():
    module = load_ppi_module()
    client = FakeStringClient({"XP_1.1": "3562.PROT1"})
    origins = [origin(module, f"Ft{i}", "XP_1.1") for i in range(4)]

    result = make_skill(module, client).resolve_query_ids(origins, "soe", None)

    assert len(result.rows) == 4
    assert {row.canonical_target_id for row in result.rows} == {
        "Ft0",
        "Ft1",
        "Ft2",
        "Ft3",
    }
    assert client.mapping_calls == [["XP_1.1", "XP_1"]]


def test_one_canonical_source_with_conflicting_references_is_excluded():
    module = load_ppi_module()
    client = FakeStringClient(
        {"XP_1.1": "3562.PROT1", "XP_2.1": "3562.PROT2"}
    )
    origins = [
        origin(module, "Ft1", "XP_1.1"),
        origin(module, "Ft1", "XP_2.1"),
    ]

    result = make_skill(module, client).resolve_query_ids(origins, "soe", None)

    assert [row.mapping_status for row in result.rows] == [
        "ambiguous_ortholog",
        "ambiguous_ortholog",
    ]
    assert result.selected_string_ids == []


def test_mapping_404_and_200_empty_both_mean_zero_resolution():
    module = load_ppi_module()
    for response in [FakeResponse(status_code=404), FakeResponse(payload=[])]:
        http = FakeHttpClient([response])
        api = module.StringAPI(
            "soe",
            http_client=http,
            stable_api_base="https://stable.string-db.org/api",
            rate_gate=SimpleNamespace(wait=lambda: None),
        )
        assert api.get_string_ids(["missing"]) == []


def test_request_throttle_uses_injected_clock_and_sleeper(tmp_path):
    from gfa_core.string_rate_gate import StringRateGate

    class Clock:
        now = 100.0

        def __init__(self):
            self.sleeps = []

        def time(self):
            return self.now

        def sleep(self, seconds):
            self.sleeps.append(seconds)
            self.now += seconds

    clock = Clock()
    gate = StringRateGate(
        path=tmp_path / "gate.timestamp",
        min_interval=1.0,
        clock=clock.time,
        sleeper=clock.sleep,
    )

    gate.wait()
    gate.wait()

    assert clock.sleeps == [pytest.approx(1.0)]


def test_two_reference_queries_converging_on_one_string_id_are_both_resolved():
    module = load_ppi_module()
    client = FakeStringClient(
        {"XP_1.1": "3562.PROT1", "XP_2.1": "3562.PROT1"}
    )
    origins = [
        origin(module, "Ft1", "XP_1.1"),
        origin(module, "Ft1", "XP_2.1"),
    ]

    result = make_skill(module, client).resolve_query_ids(origins, "soe", None)

    assert [row.mapping_status for row in result.rows] == ["resolved", "resolved"]
    assert result.selected_string_ids == ["3562.PROT1"]
    assert result.metrics["resolved_reference_query_count"] == 2
    assert result.metrics["selected_string_query_count"] == 1


def _gate_worker(path, interval, ready, output):
    from gfa_core.string_rate_gate import StringRateGate

    gate = StringRateGate(path=Path(path), min_interval=interval)
    ready.wait()
    gate.wait()
    output.put(time.time())


def test_interprocess_rate_gate_serializes_concurrent_clients(tmp_path):
    context = multiprocessing.get_context("fork")
    ready = context.Event()
    output = context.Queue()
    interval = 0.12
    processes = [
        context.Process(
            target=_gate_worker,
            args=(str(tmp_path / "shared.timestamp"), interval, ready, output),
        )
        for _ in range(2)
    ]
    for process in processes:
        process.start()
    ready.set()
    timestamps = sorted(output.get(timeout=5) for _ in processes)
    for process in processes:
        process.join(timeout=5)
        assert process.exitcode == 0

    assert timestamps[1] - timestamps[0] >= interval * 0.85