tests/test_blast_tool.py

"""BLAST, reachable by the agent — and gated, because the sequence leaves.

`server.py` has run blastp against `nr` since the identify feature: background
thread, hash-keyed cache, organism extraction. The agent could not call any of
it, because `agent_tools.py` importing from `server.py` is a circular import
(server imports the agent). So "what IS this sequence?" and "find me homologs"
were unanswerable, and blastn did not exist at all.

The property that matters most here is not speed or parsing. It is that this
is **the only tool in the engine that sends the user's actual residues off the
Space**. Everything else computes locally or transmits at most
(gene_symbol, organism). An audit already forced the REST path to enforce
consent server-side rather than trusting the UI modal ("Audit H1: BLAST
consent must be enforced server-side"); the agent path must be at least as
strict, or the model becomes the hole in the policy.

So these tests are mostly about refusal, disclosure and honest degradation —
not about hits. Nothing here calls NCBI; the network is stubbed, because a
test that depends on a shared public queue is a test that fails on Tuesdays.
"""
import re

import pytest

from dee.core import agent_tools as t
from dee.core import blast
from dee.core import orchestrator as orch

TEM1 = "MSIQHFRVALIPFFAAFCLPVFAHPETLVKVKDAEDQLGARVGYIELDLNSGKILESFRPEERFP"
DNA = "ATGCGTACGATCGATCGGCTAGCTAGCTTAAGGCCTTAAGGATCCGAATTCAAGCTTGCGGCCGC"


@pytest.fixture(autouse=True)
def _clean():
    blast.reset_cache()
    yield
    blast.reset_cache()


@pytest.fixture
def stub(monkeypatch):
    """Answer instantly with a fixed hit set. Never touches the network."""
    calls = []

    def fake(job, seq, hitlist, expect):
        calls.append({"kind": job.kind, "seq": seq, "hitlist": hitlist})
        job.hits = [{"title": "beta-lactamase TEM [Escherichia coli]",
                     "accession": "P62593", "organism": "Escherichia coli",
                     "identity_pct": 99.2, "coverage_pct": 100.0,
                     "evalue": 1e-40, "bit_score": 210.0, "align_length": 65}]
        job.status = "done"
        import time as _t
        job.finished_at = _t.time()
    monkeypatch.setattr(blast, "_run", fake)
    return calls


# --------------------------------------------------------------------------- #
# the consent property — the reason this tool is different
# --------------------------------------------------------------------------- #
def test_blast_is_the_only_tool_that_needs_approval_to_send_data_out():
    """Not a UI nicety. The model must not be able to decide, on a user's
    behalf, to publish their unpublished construct to a public service."""
    assert t._TOOLS["blast_sequence"]["requires_confirm"] is True
    assert orch._requires_confirm("blast_sequence") is True


def test_the_card_says_the_sequence_leaves_and_how_much_of_it():
    """"Allow blast_sequence?" would be a click-through. The user has to learn
    the one fact that distinguishes this tool from every other one."""
    card = orch._confirm_detail("blast_sequence", {"sequence": "M" * 286})
    assert "286" in card
    assert "NCBI" in card
    assert "leave" in card.lower()
    # and it must set the contrast against the rest of the engine
    assert "local" in card.lower()
    assert "blast_sequence" not in card


def test_no_other_read_only_tool_became_gated_by_accident():
    """The gate is meaningful only if it stays rare."""
    gated = [n for n in t._TOOLS if orch._requires_confirm(n)]
    assert set(gated) == {"log_outcome", "edit_sequence", "blast_sequence"}, gated


# --------------------------------------------------------------------------- #
# it actually works
# --------------------------------------------------------------------------- #
def test_a_protein_search_returns_the_hits_verbatim(stub):
    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
    assert out["ok"] is True
    assert out["search"] == "protein"
    assert out["hit_count"] == 1
    assert out["top"]["identity_pct"] == 99.2
    assert out["top"]["organism"] == "Escherichia coli"


def test_dna_goes_to_blastn_not_blastp(stub):
    """blastp on DNA returns confident nonsense, so the alphabet decides."""
    t.execute_tool("blast_sequence", {"sequence": DNA}, auth_anonymous=True)
    assert stub[-1]["kind"] == "nucleotide"


def test_the_caller_can_override_the_guess(stub):
    t.execute_tool("blast_sequence", {"sequence": DNA, "kind": "protein"},
                   auth_anonymous=True)
    assert stub[-1]["kind"] == "protein"


def test_kind_detection_errs_toward_nucleotide():
    """Only letters that CANNOT be bases make it protein. A short ACGT-only
    peptide is genuinely ambiguous, and blastn on a peptide finds nothing —
    which is a safe failure — while blastp on DNA invents a story."""
    assert blast.guess_kind(TEM1) == "protein"
    assert blast.guess_kind(DNA) == "nucleotide"
    assert blast.guess_kind("ACGTACGTACGT") == "nucleotide"


def test_repeat_searches_reuse_the_job(stub):
    """NCBI's queue is shared and slow; asking twice for the same sequence
    must not queue twice."""
    a = blast.submit(TEM1)
    b = blast.submit(TEM1)
    assert a.job_id == b.job_id
    assert len(stub) == 1


# --------------------------------------------------------------------------- #
# honest failure
# --------------------------------------------------------------------------- #
def test_no_hits_is_reported_as_a_finding_not_a_gap(stub, monkeypatch):
    def empty(job, seq, hitlist, expect):
        job.hits = []
        job.status = "done"
        import time as _t
        job.finished_at = _t.time()
    monkeypatch.setattr(blast, "_run", empty)
    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
    assert out["ok"] is True and out["hit_count"] == 0
    assert "novel or synthetic" in out["next"]
    assert "guessing" in out["next"]


def test_a_still_running_search_says_so_instead_of_hanging(monkeypatch):
    """NCBI can take minutes. A tool that blocks the conversation reads as
    broken; one that reports a job id is merely slow."""
    monkeypatch.setattr(blast, "_run", lambda *a, **k: None)   # never finishes
    # Readable at call time now — see blast.wait(). Setting it as a
    # signature default froze it at import and this test waited 75s.
    monkeypatch.setattr(blast, "AGENT_WAIT_SECONDS", 0.2)
    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
    assert out["ok"] is False
    assert out["kind"] == "still_running"
    assert out["job_id"]
    assert "never invent hits" in out["next"].lower()


def test_an_ncbi_failure_does_not_become_a_remembered_answer(stub, monkeypatch):
    def boom(job, seq, hitlist, expect):
        job.error = "NCBI BLAST did not return a result (URLError)."
        job.status = "error"
        import time as _t
        job.finished_at = _t.time()
    monkeypatch.setattr(blast, "_run", boom)
    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
    assert out["ok"] is False and out["kind"] == "blast_failed"
    assert "from memory" in out["next"]


def test_a_too_short_query_is_refused():
    out = t.execute_tool("blast_sequence", {"sequence": "MSIQ"},
                         auth_anonymous=True)
    assert out["ok"] is False and "12" in out["error"]


def test_an_unknown_kind_is_refused():
    out = t.execute_tool("blast_sequence", {"sequence": TEM1, "kind": "rna"},
                         auth_anonymous=True)
    assert out["ok"] is False and "protein" in out["error"]


# --------------------------------------------------------------------------- #
# wiring
# --------------------------------------------------------------------------- #
def test_the_engine_lives_where_both_callers_can_reach_it():
    """The reason this was unreachable: agent_tools importing server.py is a
    circular import. Same extraction scoring.py already did."""
    src = open("dee/core/agent_tools.py", encoding="utf-8").read()
    assert "from dee.core import blast" in src
    assert "import server" not in src


def test_the_summary_leads_with_what_it_found(stub):
    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
    line = orch._summarize("blast_sequence", out)
    assert "1 hit" in line and "99.2" in line
    assert "Escherichia coli" in line


def test_the_description_warns_that_the_sequence_is_sent(stub):
    spec = next(s for s in orch.TOOL_SPECS
                if s["function"]["name"] == "blast_sequence")
    d = spec["function"]["description"]
    assert "SENDS THE SEQUENCE" in d
    assert "may decline" in d
    assert "still_running" in d