hermes-kanban-miniapp/backend/tests/test_api.py

import hashlib
import hmac
import json
import time
from urllib.parse import quote, urlencode

import pytest
from fastapi.testclient import TestClient

from app.auth import validate_telegram_init_data
from app.config import Settings
from app.main import create_app


def signed_init_data(bot_token: str, user_id: int = 123) -> str:
    pairs = {
        "query_id": "AAEAAAE",
        "user": json.dumps({"id": user_id, "first_name": "Test"}, separators=(",", ":")),
        "auth_date": str(int(time.time())),
    }
    data_check_string = "\n".join(f"{k}={v}" for k, v in sorted(pairs.items()))
    secret_key = hmac.new(b"WebAppData", bot_token.encode(), hashlib.sha256).digest()
    pairs["hash"] = hmac.new(secret_key, data_check_string.encode(), hashlib.sha256).hexdigest()
    return urlencode(pairs, quote_via=quote)


class FakeAdapter:
    def __init__(self):
        self.calls = []
        self.tasks = [
            {"id": "t_1", "title": "Ready task", "status": "ready", "assignee": "dev", "priority": 7},
            {"id": "t_2", "title": "Blocked task", "status": "blocked", "assignee": "worker", "priority": 1},
        ]

    def boards(self):
        return [{"slug": "default", "current": True}]

    def list_tasks(self, board=None):
        self.calls.append(("list", board))
        return self.tasks

    def show_task(self, task_id, board=None):
        return {"id": task_id, "title": "Ready task", "body": "Full body", "status": "ready", "comments": [], "events": [], "runs": []}

    def task_events(self, task_id, board=None):
        return [{"kind": "created"}]

    def task_runs(self, task_id, board=None):
        return [{"id": 1, "status": "completed"}]

    def task_log(self, task_id, board=None, lines=200):
        return {"task_id": task_id, "log": "", "missing": True}

    def create_task(self, payload, board=None):
        self.calls.append(("create", payload.title))
        return {"id": "t_new", "title": payload.title, "status": "todo"}

    def comment(self, task_id, body, board=None):
        self.calls.append(("comment", task_id, body))
        return {"ok": True}

    def assign(self, task_id, assignee, board=None):
        self.calls.append(("assign", task_id, assignee))
        return {"ok": True}

    def promote(self, task_id, board=None):
        self.calls.append(("promote", task_id))
        return {"ok": True}

    def block(self, task_id, reason, board=None):
        self.calls.append(("block", task_id, reason))
        return {"ok": True}

    def unblock(self, task_id, board=None):
        self.calls.append(("unblock", task_id))
        return {"ok": True}

    def archive(self, task_id, board=None):
        self.calls.append(("archive", task_id))
        return {"ok": True}

    def dispatch(self, board=None):
        self.calls.append(("dispatch", board))
        return {"ok": True}


@pytest.fixture
def fake_adapter():
    return FakeAdapter()


@pytest.fixture
def client(fake_adapter):
    app = create_app(Settings(auth_mode="none", allowed_origins=["http://localhost:5173"]), fake_adapter)
    return TestClient(app)


def test_health_returns_ok(client):
    assert client.get("/api/health").json()["status"] == "ok"


def test_telegram_init_data_accepts_correct_signature():
    data = signed_init_data("token123", 123)
    user = validate_telegram_init_data(data, "token123", max_age_seconds=3600)
    assert user["id"] == 123


def test_telegram_init_data_rejects_wrong_hash():
    data = signed_init_data("token123", 123).replace("hash=", "hash=bad")
    with pytest.raises(ValueError):
        validate_telegram_init_data(data, "token123", max_age_seconds=3600)


def test_allowed_user_filter_rejects_unauthorized(fake_adapter):
    settings = Settings(auth_mode="telegram", telegram_bot_token="token123", telegram_allowed_user_ids=[999])
    app = create_app(settings, fake_adapter)
    response = TestClient(app).get("/api/tasks", headers={"X-Telegram-Init-Data": signed_init_data("token123", 123)})
    assert response.status_code == 403


def test_env_list_settings_accept_json_strings():
    settings = Settings(
        telegram_allowed_user_ids='[123, "456"]',
        allowed_origins='["https://miniapp.example", "http://localhost:5173"]',
    )
    assert settings.telegram_allowed_user_ids == [123, 456]
    assert settings.allowed_origins == ["https://miniapp.example", "http://localhost:5173"]


def test_env_list_settings_accept_comma_separated_strings():
    settings = Settings(
        telegram_allowed_user_ids="123,456",
        allowed_origins="https://miniapp.example,http://localhost:5173",
    )
    assert settings.telegram_allowed_user_ids == [123, 456]
    assert settings.allowed_origins == ["https://miniapp.example", "http://localhost:5173"]


def test_static_frontend_can_be_served_from_backend(tmp_path, fake_adapter):
    (tmp_path / "index.html").write_text("<h1>Kanban Mini App</h1>", encoding="utf-8")
    settings = Settings(auth_mode="none", frontend_dist_dir=str(tmp_path))
    app = create_app(settings, fake_adapter)

    response = TestClient(app).get("/")

    assert response.status_code == 200
    assert "Kanban Mini App" in response.text


def test_task_list_returns_normalized_tasks_grouped_by_status(client):
    data = client.get("/api/tasks").json()
    assert data["groups"]["ready"][0]["id"] == "t_1"
    assert data["groups"]["blocked"][0]["status"] == "blocked"


def test_create_task_validates_required_title(client):
    response = client.post("/api/tasks", json={"title": "   "})
    assert response.status_code == 422


def test_create_task_accepts_safe_payload(client):
    response = client.post("/api/tasks", json={"title": "New task", "body": "Body", "assignee": "developer", "priority": 3})
    assert response.status_code == 200
    assert response.json()["task"]["id"] == "t_new"


def test_action_endpoints_call_fixed_adapter_methods(client, fake_adapter):
    assert client.post("/api/tasks/t_1/comment", json={"body": "hello"}).status_code == 200
    assert client.post("/api/tasks/t_1/assign", json={"assignee": "worker"}).status_code == 200
    assert client.post("/api/tasks/t_1/promote").status_code == 200
    assert client.post("/api/tasks/t_1/block", json={"reason": "need input"}).status_code == 200
    assert client.post("/api/tasks/t_1/unblock").status_code == 200
    assert client.post("/api/tasks/t_1/archive").status_code == 200
    assert ("comment", "t_1", "hello") in fake_adapter.calls
    assert ("assign", "t_1", "worker") in fake_adapter.calls
    assert ("promote", "t_1") in fake_adapter.calls


def test_log_endpoint_handles_missing_log_gracefully(client):
    response = client.get("/api/tasks/t_1/log")
    assert response.status_code == 200
    assert response.json()["missing"] is True


def test_adapter_failure_returns_structured_error_without_secret(client, fake_adapter):
    def broken(*args, **kwargs):
        raise RuntimeError("failed with secret token value")

    fake_adapter.list_tasks = broken
    response = client.get("/api/tasks")
    assert response.status_code == 502
    body = response.json()["detail"]
    assert body["error"] == "adapter_error"
    assert "secret" not in body["message"].lower()