304 lines
11 KiB
Python
304 lines
11 KiB
Python
"""
|
||
KIN-090: Integration tests for task attachment API endpoints.
|
||
|
||
Tests cover:
|
||
AC1 — upload saves file to {project_path}/.kin/attachments/{task_id}/
|
||
AC3 — file available for download via GET /api/attachments/{id}/file
|
||
AC4 — data persists in SQLite
|
||
Integration: upload → list → verify agent context (build_context)
|
||
"""
|
||
|
||
import io
|
||
import pytest
|
||
from pathlib import Path
|
||
from fastapi.testclient import TestClient
|
||
|
||
import web.api as api_module
|
||
|
||
|
||
@pytest.fixture
|
||
def client(tmp_path):
|
||
"""TestClient with isolated DB and a seeded project+task.
|
||
|
||
Project path set to tmp_path so attachment dirs are created there
|
||
and cleaned up automatically.
|
||
"""
|
||
db_path = tmp_path / "test.db"
|
||
api_module.DB_PATH = db_path
|
||
|
||
from web.api import app
|
||
c = TestClient(app)
|
||
|
||
project_path = str(tmp_path / "myproject")
|
||
c.post("/api/projects", json={
|
||
"id": "prj",
|
||
"name": "My Project",
|
||
"path": project_path,
|
||
})
|
||
c.post("/api/tasks", json={"project_id": "prj", "title": "Fix login bug"})
|
||
return c
|
||
|
||
|
||
def _png_bytes() -> bytes:
|
||
"""Minimal valid 1x1 PNG image."""
|
||
import base64
|
||
# 1x1 red pixel PNG (base64-encoded)
|
||
data = (
|
||
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
|
||
b"\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f\x00"
|
||
b"\x00\x01\x01\x00\x05\x18\xd8N\x00\x00\x00\x00IEND\xaeB`\x82"
|
||
)
|
||
return data
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Upload
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_upload_attachment_returns_201(client):
|
||
"""KIN-090: POST /api/tasks/{id}/attachments возвращает 201 и данные вложения."""
|
||
r = client.post(
|
||
"/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("bug.png", io.BytesIO(_png_bytes()), "image/png")},
|
||
)
|
||
assert r.status_code == 201
|
||
data = r.json()
|
||
assert data["task_id"] == "PRJ-001"
|
||
assert data["filename"] == "bug.png"
|
||
assert data["mime_type"] == "image/png"
|
||
assert data["size"] == len(_png_bytes())
|
||
assert data["id"] is not None
|
||
|
||
|
||
def test_upload_attachment_saves_file_to_correct_path(client, tmp_path):
|
||
"""KIN-090: AC1 — файл сохраняется в {project_path}/.kin/attachments/{task_id}/."""
|
||
r = client.post(
|
||
"/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("shot.png", io.BytesIO(_png_bytes()), "image/png")},
|
||
)
|
||
assert r.status_code == 201
|
||
saved_path = Path(r.json()["path"])
|
||
|
||
# Path structure: <project_path>/.kin/attachments/PRJ-001/shot.png
|
||
assert saved_path.name == "shot.png"
|
||
assert saved_path.parent.name == "PRJ-001"
|
||
assert saved_path.parent.parent.name == "attachments"
|
||
assert saved_path.parent.parent.parent.name == ".kin"
|
||
assert saved_path.exists()
|
||
|
||
|
||
def test_upload_attachment_file_content_matches(client):
|
||
"""KIN-090: содержимое сохранённого файла совпадает с загруженным."""
|
||
content = _png_bytes()
|
||
r = client.post(
|
||
"/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("img.png", io.BytesIO(content), "image/png")},
|
||
)
|
||
assert r.status_code == 201
|
||
saved_path = Path(r.json()["path"])
|
||
assert saved_path.read_bytes() == content
|
||
|
||
|
||
def test_upload_attachment_persists_in_sqlite(client, tmp_path):
|
||
"""KIN-090: AC4 — запись о вложении сохраняется в SQLite и доступна через list."""
|
||
client.post(
|
||
"/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("db_test.png", io.BytesIO(_png_bytes()), "image/png")},
|
||
)
|
||
# Verify via list endpoint (reads from DB)
|
||
r = client.get("/api/tasks/PRJ-001/attachments")
|
||
assert r.status_code == 200
|
||
assert any(a["filename"] == "db_test.png" for a in r.json())
|
||
|
||
|
||
def test_upload_attachment_task_not_found_returns_404(client):
|
||
"""KIN-090: 404 если задача не существует."""
|
||
r = client.post(
|
||
"/api/tasks/PRJ-999/attachments",
|
||
files={"file": ("x.png", io.BytesIO(_png_bytes()), "image/png")},
|
||
)
|
||
assert r.status_code == 404
|
||
|
||
|
||
def test_upload_attachment_operations_project_returns_400(client, tmp_path):
|
||
"""KIN-090: 400 для operations-проекта (нет project path)."""
|
||
db_path = tmp_path / "test2.db"
|
||
api_module.DB_PATH = db_path
|
||
|
||
from web.api import app
|
||
c = TestClient(app)
|
||
c.post("/api/projects", json={
|
||
"id": "ops",
|
||
"name": "Ops Server",
|
||
"project_type": "operations",
|
||
"ssh_host": "10.0.0.1",
|
||
})
|
||
c.post("/api/tasks", json={"project_id": "ops", "title": "Reboot server"})
|
||
|
||
r = c.post(
|
||
"/api/tasks/OPS-001/attachments",
|
||
files={"file": ("x.png", io.BytesIO(_png_bytes()), "image/png")},
|
||
)
|
||
assert r.status_code == 400
|
||
|
||
|
||
def test_upload_oversized_file_returns_413(client):
|
||
"""KIN-090: 413 если файл превышает 10 MB."""
|
||
big_content = b"x" * (10 * 1024 * 1024 + 1)
|
||
r = client.post(
|
||
"/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("huge.png", io.BytesIO(big_content), "image/png")},
|
||
)
|
||
assert r.status_code == 413
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# List
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_list_attachments_empty_for_new_task(client):
|
||
"""KIN-090: GET /api/tasks/{id}/attachments возвращает [] для задачи без вложений."""
|
||
r = client.get("/api/tasks/PRJ-001/attachments")
|
||
assert r.status_code == 200
|
||
assert r.json() == []
|
||
|
||
|
||
def test_list_attachments_returns_all_uploaded(client):
|
||
"""KIN-090: список содержит все загруженные вложения."""
|
||
client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("a.png", io.BytesIO(_png_bytes()), "image/png")})
|
||
client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("b.jpg", io.BytesIO(_png_bytes()), "image/jpeg")})
|
||
|
||
r = client.get("/api/tasks/PRJ-001/attachments")
|
||
assert r.status_code == 200
|
||
filenames = {a["filename"] for a in r.json()}
|
||
assert "a.png" in filenames
|
||
assert "b.jpg" in filenames
|
||
|
||
|
||
def test_list_attachments_task_not_found_returns_404(client):
|
||
"""KIN-090: 404 если задача не существует."""
|
||
r = client.get("/api/tasks/PRJ-999/attachments")
|
||
assert r.status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Delete
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_delete_attachment_returns_204(client):
|
||
"""KIN-090: DELETE возвращает 204."""
|
||
r = client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("del.png", io.BytesIO(_png_bytes()), "image/png")})
|
||
att_id = r.json()["id"]
|
||
|
||
r = client.delete(f"/api/tasks/PRJ-001/attachments/{att_id}")
|
||
assert r.status_code == 204
|
||
|
||
|
||
def test_delete_attachment_removes_from_list(client):
|
||
"""KIN-090: после удаления вложение не появляется в списке."""
|
||
r = client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("rm.png", io.BytesIO(_png_bytes()), "image/png")})
|
||
att_id = r.json()["id"]
|
||
client.delete(f"/api/tasks/PRJ-001/attachments/{att_id}")
|
||
|
||
attachments = client.get("/api/tasks/PRJ-001/attachments").json()
|
||
assert not any(a["id"] == att_id for a in attachments)
|
||
|
||
|
||
def test_delete_attachment_removes_file_from_disk(client):
|
||
"""KIN-090: удаление вложения удаляет файл с диска."""
|
||
r = client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("disk.png", io.BytesIO(_png_bytes()), "image/png")})
|
||
saved_path = Path(r.json()["path"])
|
||
att_id = r.json()["id"]
|
||
|
||
assert saved_path.exists()
|
||
client.delete(f"/api/tasks/PRJ-001/attachments/{att_id}")
|
||
assert not saved_path.exists()
|
||
|
||
|
||
def test_delete_attachment_not_found_returns_404(client):
|
||
"""KIN-090: 404 если вложение не существует."""
|
||
r = client.delete("/api/tasks/PRJ-001/attachments/99999")
|
||
assert r.status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Download
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_download_attachment_file_returns_correct_content(client):
|
||
"""KIN-090: AC3 — GET /api/attachments/{id}/file возвращает содержимое файла."""
|
||
content = _png_bytes()
|
||
r = client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("get.png", io.BytesIO(content), "image/png")})
|
||
att_id = r.json()["id"]
|
||
|
||
r = client.get(f"/api/attachments/{att_id}/file")
|
||
assert r.status_code == 200
|
||
assert r.content == content
|
||
|
||
|
||
def test_download_attachment_file_returns_correct_content_type(client):
|
||
"""KIN-090: AC3 — Content-Type соответствует mime_type вложения."""
|
||
r = client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("ct.png", io.BytesIO(_png_bytes()), "image/png")})
|
||
att_id = r.json()["id"]
|
||
|
||
r = client.get(f"/api/attachments/{att_id}/file")
|
||
assert r.status_code == 200
|
||
assert "image/png" in r.headers["content-type"]
|
||
|
||
|
||
def test_download_attachment_not_found_returns_404(client):
|
||
"""KIN-090: 404 если вложение не существует."""
|
||
r = client.get("/api/attachments/99999/file")
|
||
assert r.status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Integration: upload → list → agent context (AC2)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_integration_upload_list_agent_context(client, tmp_path):
|
||
"""KIN-090: Интеграционный тест: upload → list → build_context включает вложения.
|
||
|
||
Проверяет AC1 (путь), AC3 (доступен для скачивания), AC4 (SQLite),
|
||
и AC2 (агенты получают вложения через build_context).
|
||
"""
|
||
# Step 1: Upload image
|
||
content = _png_bytes()
|
||
r = client.post("/api/tasks/PRJ-001/attachments",
|
||
files={"file": ("integration.png", io.BytesIO(content), "image/png")})
|
||
assert r.status_code == 201
|
||
att = r.json()
|
||
|
||
# Step 2: AC1 — file is at correct path inside project
|
||
saved_path = Path(att["path"])
|
||
assert saved_path.exists()
|
||
assert "PRJ-001" in str(saved_path)
|
||
assert ".kin/attachments" in str(saved_path)
|
||
|
||
# Step 3: List confirms persistence (AC4)
|
||
r = client.get("/api/tasks/PRJ-001/attachments")
|
||
assert r.status_code == 200
|
||
assert len(r.json()) == 1
|
||
|
||
# Step 4: Download works (AC3)
|
||
r = client.get(f"/api/attachments/{att['id']}/file")
|
||
assert r.status_code == 200
|
||
assert r.content == content
|
||
|
||
# Step 5: AC2 — agent context includes attachment path
|
||
from core.db import init_db
|
||
from core.context_builder import build_context
|
||
conn = init_db(api_module.DB_PATH)
|
||
ctx = build_context(conn, "PRJ-001", "debugger", "prj")
|
||
conn.close()
|
||
|
||
assert "attachments" in ctx
|
||
paths = [a["path"] for a in ctx["attachments"]]
|
||
assert att["path"] in paths
|