File size: 3,167 Bytes
3ed6b66 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """The deploy descriptor, the endpoint table and the gr.api surface must agree (names are frozen)."""
from __future__ import annotations
import inspect
import re
from pathlib import Path
import yaml
_EXPECTED = [
"drugs_for_targets",
"trials_for_genes",
"annotate_genes",
"pathways_for_genes",
"targets_for_disease",
]
_REPO = Path(__file__).resolve().parent.parent
def _descriptor() -> dict:
return yaml.safe_load((_REPO / "deploy" / "orchestrator_registration.yaml").read_text())
def test_descriptor_names_and_params_match_the_endpoint_table():
from src import config
from src.endpoints import ENDPOINTS
descriptor = _descriptor()
assert descriptor["id"] == "pdac_knowledge_agent"
assert descriptor["dispatch_kind"] == "tool_server"
assert descriptor["space"] == "anne-voigt/pdac-knowledge-agent"
assert descriptor["space_dev"] == "anne-voigt/pdac-knowledge-agent-dev"
names = [t["name"] for t in descriptor["tools"]]
assert names == _EXPECTED == list(ENDPOINTS) == list(config.ENDPOINTS)
for tool in descriptor["tools"]:
assert tool["api_name"] == f"/{tool['name']}"
assert tuple(tool["params"]) == ENDPOINTS[tool["name"]][1]
def test_gradio_functions_exist_with_string_defaults_and_trailing_request():
import gradio_ui
from src.endpoints import ENDPOINTS
for name in _EXPECTED:
params = list(inspect.signature(getattr(gradio_ui, name)).parameters.values())
assert params[-1].name == "request"
assert [p.name for p in params[:-1]] == list(ENDPOINTS[name][1])
assert all(p.default == "" for p in params[:-1])
def test_every_endpoint_is_registered_with_gr_api():
import gradio_ui
registered = {getattr(fn, "api_name", None) for fn in gradio_ui.demo.fns.values()}
assert set(_EXPECTED) <= registered
def test_the_boundary_sentence_is_the_adrs_verbatim():
adr = (_REPO / "docs/adr/ADR-0001-knowledge-specialist-contract.md").read_text()
quoted = re.search(
r'reused verbatim in the descriptor and the routing prompt: \*"(.+?)"\*', adr, re.DOTALL
)
sentence = " ".join(quoted.group(1).split())
assert " ".join(_descriptor()["description"].split()) == sentence
def test_descriptor_states_the_caps_the_code_enforces():
from src.endpoints import drugs, targets
from src.identifiers import MAX_IDENTIFIERS
described = {t["name"]: " ".join(t["description"].split()) for t in _descriptor()["tools"]}
assert (
f"for the {drugs.MAX_MECHANISM_LOOKUPS} most advanced drugs"
in described["drugs_for_targets"]
)
assert (
f"`size` is 1-{targets.MAX_SIZE} (default {targets.DEFAULT_SIZE})"
in described["targets_for_disease"]
)
for name in ("drugs_for_targets", "annotate_genes", "pathways_for_genes"):
assert f"at most {MAX_IDENTIFIERS}" in described[name], name
assert "MEMBERSHIP, NOT ENRICHMENT" in described["pathways_for_genes"]
def test_annotate_genes_carries_its_longer_timeout():
[annotate] = [t for t in _descriptor()["tools"] if t["name"] == "annotate_genes"]
assert annotate["timeout_seconds"] == 300
|