tests/test_agent.py

"""Checklist de aceite do endpoint Programa.

Cobre, nesta ordem: /v1/models, chat simples, stream, system, multi-turn,
tools + tool_choice=auto, tool result + continuacao, max_tokens, temperature,
stop, /v1/responses e /v1/messages (somente se o servidor expuser de verdade).

Uso:
  BASE=https://erinaldordso-programa.hf.space/v1 python3 tests/test_agent.py
  BASE=... MODEL=... APIKEY=... python3 tests/test_agent.py

Nao declara sucesso parcial: qualquer FAIL aborta com exit 1.
"""
import json
import os
import sys
import urllib.request

import tiktoken

BASE = os.environ.get("BASE", "https://erinaldordso-minecraft-friends-ai-zerogpu.hf.space/v1").rstrip("/")
MODEL = os.environ.get("MODEL", "granite-4.0-h-tiny")
APIKEY = os.environ.get("APIKEY", "")
PASS, FAIL = "PASS", "FAIL"
results = []


def req(method, path, body=None, stream=False):
    url = BASE + path
    data = json.dumps(body).encode() if body is not None else None
    headers = {"Content-Type": "application/json"}
    if APIKEY:
        headers["Authorization"] = f"Bearer {APIKEY}"
    r = urllib.request.Request(url, data=data, headers=headers, method=method)
    if not stream:
        with urllib.request.urlopen(r, timeout=300) as resp:
            return resp.status, json.loads(resp.read().decode())
    out = []
    with urllib.request.urlopen(r, timeout=300) as resp:
        for line in resp:
            s = line.decode(errors="replace").strip()
            if s.startswith("data:"):
                s = s[5:].strip()
                if s == "[DONE]":
                    break
                try:
                    out.append(json.loads(s))
                except json.JSONDecodeError:
                    pass
    return 200, out


def check(name, cond, extra=""):
    results.append((name, bool(cond)))
    print(f"[{PASS if cond else FAIL}] {name} {extra}")
    if not cond:
        raise SystemExit(f"FAIL em {name}: {extra}")


TOOLS = [{"type": "function", "function": {
    "name": "read_file",
    "description": "Read a file from the project",
    "parameters": {"type": "object",
                   "properties": {"path": {"type": "string"}},
                   "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "list_directory",
        "description": "List a directory",
        "parameters": {"type": "object",
                       "properties": {"path": {"type": "string"}},
                       "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "execute_command",
        "description": "Run a shell command",
        "parameters": {"type": "object",
                       "properties": {"command": {"type": "string"}},
                       "required": ["command"]}}}]


def main():
    print(f"BASE={BASE} MODEL={MODEL}")

    st, models = req("GET", "/models")
    ids = [m.get("id", "") for m in models.get("data", [])]
    check("models", st == 200 and any(MODEL in i or "granite" in i.lower() for i in ids), f"ids={ids[:4]}")

    st, emb = req("POST", "/embeddings", {"model": "text-embedding-ada-002", "input": ["mine wood", "craft pickaxe"]})
    vecs = emb.get("data", [])
    check("embeddings Voyager", st == 200 and len(vecs) == 2 and all(isinstance(x.get("embedding"), list) and len(x.get("embedding")) > 0 for x in vecs), f"vectors={len(vecs)}")

    enc = tiktoken.get_encoding("cl100k_base")
    st, emb = req("POST", "/embeddings", {"model": "text-embedding-ada-002", "input": [enc.encode("mine wood"), enc.encode("craft pickaxe")]})
    vecs = emb.get("data", [])
    check("embeddings token input", st == 200 and len(vecs) == 2 and all(isinstance(x.get("embedding"), list) and len(x.get("embedding")) > 0 for x in vecs), f"vectors={len(vecs)}")

    st, r = req("POST", "/chat/completions", {
        "model": MODEL, "messages": [{"role": "user", "content": "diga ok"}],
        "max_tokens": 16, "temperature": 0})
    txt = r["choices"][0]["message"].get("content", "")
    check("chat stream=false", st == 200 and isinstance(txt, str), f"len={len(txt)}")

    st, chunks = req("POST", "/chat/completions", {
        "model": MODEL, "messages": [{"role": "user", "content": "conte ate 3"}],
        "max_tokens": 32, "temperature": 0, "stream": True}, stream=True)
    joined = "".join((c.get("choices", [{}])[0].get("delta", {}) or {}).get("content") or "" for c in chunks)
    check("chat stream=true", st == 200 and len(chunks) > 0, f"chunks={len(chunks)} chars={len(joined)}")

    st, r = req("POST", "/chat/completions", {
        "model": MODEL,
        "messages": [{"role": "system", "content": "You are an autonomous software engineering agent."},
                     {"role": "user", "content": "diga ok"}],
        "max_tokens": 16, "temperature": 0})
    check("system message", st == 200, "")

    st, r = req("POST", "/chat/completions", {
        "model": MODEL,
        "messages": [{"role": "user", "content": "meu projeto usa npm"},
                     {"role": "assistant", "content": "anotado"},
                     {"role": "user", "content": "qual gerenciador citei?"}],
        "max_tokens": 32, "temperature": 0})
    check("multi-turn", st == 200 and "npm" in json.dumps(r).lower(), "")

    st, r = req("POST", "/chat/completions", {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": "You are an autonomous software engineering agent. Use tools instead of describing hypothetical actions. Inspect files before modifying them."},
            {"role": "user", "content": "Leia package.json e diga qual comando executa os testes."}],
        "tools": TOOLS[:1], "tool_choice": "auto",
        "max_tokens": 256, "temperature": 0})
    msg = r["choices"][0]["message"]
    tcs = msg.get("tool_calls") or []
    ok = (r["choices"][0].get("finish_reason") == "tool_calls" and tcs
          and tcs[0]["function"]["name"] == "read_file"
          and "package.json" in tcs[0]["function"].get("arguments", ""))
    check("tools read_file", ok, f"finish={r['choices'][0].get('finish_reason')} calls={json.dumps(tcs)[:200]}")

    tc = tcs[0]
    st, r = req("POST", "/chat/completions", {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": "You are an autonomous software engineering agent."},
            {"role": "user", "content": "Leia package.json e diga qual comando executa os testes."},
            {"role": "assistant", "tool_calls": [{
                "id": tc.get("id", "call_1"), "type": "function",
                "function": {"name": "read_file", "arguments": tc["function"]["arguments"]}}]},
            {"role": "tool", "tool_call_id": tc.get("id", "call_1"),
             "content": "{\"name\": \"demo\", \"scripts\": {\"test\": \"npm test\"}}"}],
        "max_tokens": 128, "temperature": 0})
    txt2 = r["choices"][0]["message"].get("content", "")
    check("tool result + continuacao", st == 200 and "npm test" in txt2, f"resp={txt2[:160]}")

    st, r = req("POST", "/chat/completions", {
        "model": MODEL, "messages": [{"role": "user", "content": "fale muito"}],
        "max_tokens": 8, "temperature": 0})
    check("max_tokens respeitado", st == 200, "")

    st, r = req("POST", "/chat/completions", {
        "model": MODEL, "messages": [{"role": "user", "content": "diga ok"}],
        "max_tokens": 16, "temperature": 0.7, "stop": ["<|im_end|>"]})
    check("temperature+stop", st == 200, "")

    for path, label in [("/responses", "responses"), ("/messages", "messages")]:
        try:
            if label == "responses":
                st, r = req("POST", path, {"model": MODEL, "input": "diga ok", "max_output_tokens": 16})
            else:
                st, r = req("POST", path, {"model": MODEL, "max_tokens": 16,
                                           "messages": [{"role": "user", "content": "diga ok"}]})
            check(label, st in (200, 201), f"status={st}")
        except Exception as e:  # noqa: BLE001
            print(f"[SKIP] {label} ausente neste build ({str(e)[:120]})")

    print(f"\nTodos os checks obrigatorios passaram ({len(results)}).")


if __name__ == "__main__":
    main()