pelmen/kin
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kin: KIN-BIZ-006 Проверить промпт sysadmin.md на поддержку сценария env_scan

Browse source
This commit is contained in:
Gros Frumos 2026-03-16 19:26:51 +02:00
parent 531275e4ce
commit a58578bb9d
14 changed files with 1619 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

7
tests/conftest.py
View file

@ -4,6 +4,13 @@ import pytest
from unittest.mock import patch
@pytest.fixture(autouse=True)
def _set_kin_secret_key(monkeypatch):
"""Set KIN_SECRET_KEY for all tests (required by _encrypt_auth/_decrypt_auth)."""
from cryptography.fernet import Fernet
monkeypatch.setenv("KIN_SECRET_KEY", Fernet.generate_key().decode())
@pytest.fixture(autouse=True)
def _mock_check_claude_auth():
"""Авто-мок agents.runner.check_claude_auth для всех тестов.

410
tests/test_api.py
View file

@ -1713,4 +1713,412 @@ def test_delete_project_ok(client):
def test_delete_project_not_found(client):
r = client.delete("/api/projects/99999")
assert r.status_code == 404
assert "tasks_count" in data
# ---------------------------------------------------------------------------
# Environments (KIN-087)
# ---------------------------------------------------------------------------
def test_create_environment(client):
r = client.post("/api/projects/p1/environments", json={
"name": "prod",
"host": "10.0.0.1",
"username": "pelmen",
"port": 22,
"auth_type": "password",
"auth_value": "s3cr3t",
"is_installed": False,
})
assert r.status_code == 201
data = r.json()
assert data["name"] == "prod"
assert data["host"] == "10.0.0.1"
assert data["username"] == "pelmen"
# auth_value must be hidden in responses
assert data.get("auth_value") is None
assert "scan_task_id" not in data
def test_create_environment_project_not_found(client):
r = client.post("/api/projects/nope/environments", json={
"name": "prod",
"host": "10.0.0.1",
"username": "root",
})
assert r.status_code == 404
def test_create_environment_invalid_auth_type(client):
r = client.post("/api/projects/p1/environments", json={
"name": "prod",
"host": "10.0.0.1",
"username": "root",
"auth_type": "oauth",
})
assert r.status_code == 422
def test_create_environment_invalid_port(client):
r = client.post("/api/projects/p1/environments", json={
"name": "prod",
"host": "10.0.0.1",
"username": "root",
"port": 99999,
})
assert r.status_code == 422
def test_create_environment_triggers_scan_when_installed(client):
"""is_installed=True на POST должен создать задачу sysadmin и вернуть scan_task_id."""
with patch("subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock(pid=12345)
r = client.post("/api/projects/p1/environments", json={
"name": "prod",
"host": "10.0.0.2",
"username": "pelmen",
"is_installed": True,
})
assert r.status_code == 201
data = r.json()
assert "scan_task_id" in data
task_id = data["scan_task_id"]
# Verify the task exists with sysadmin role
from core.db import init_db
from core import models as m
conn = init_db(api_module.DB_PATH)
task = m.get_task(conn, task_id)
conn.close()
assert task is not None
assert task["assigned_role"] == "sysadmin"
assert task["category"] == "INFRA"
def test_list_environments(client):
client.post("/api/projects/p1/environments", json={
"name": "dev", "host": "10.0.0.10", "username": "dev",
})
client.post("/api/projects/p1/environments", json={
"name": "prod", "host": "10.0.0.11", "username": "prod",
})
r = client.get("/api/projects/p1/environments")
assert r.status_code == 200
data = r.json()
assert len(data) == 2
names = {e["name"] for e in data}
assert names == {"dev", "prod"}
def test_list_environments_project_not_found(client):
r = client.get("/api/projects/nope/environments")
assert r.status_code == 404
def test_patch_environment(client):
r = client.post("/api/projects/p1/environments", json={
"name": "dev", "host": "10.0.0.20", "username": "root",
})
env_id = r.json()["id"]
r = client.patch(f"/api/projects/p1/environments/{env_id}", json={
"host": "10.0.0.99",
})
assert r.status_code == 200
assert r.json()["host"] == "10.0.0.99"
def test_patch_environment_triggers_scan_on_false_to_true(client):
"""PATCH is_installed false→true должен запустить скан."""
r = client.post("/api/projects/p1/environments", json={
"name": "staging", "host": "10.0.0.30", "username": "root", "is_installed": False,
})
env_id = r.json()["id"]
with patch("subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock(pid=22222)
r = client.patch(f"/api/projects/p1/environments/{env_id}", json={
"is_installed": True,
})
assert r.status_code == 200
assert "scan_task_id" in r.json()
def test_patch_environment_no_duplicate_scan(client):
"""Повторный PATCH is_installed=True (true→true) не создаёт новую задачу."""
with patch("subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock(pid=33333)
r = client.post("/api/projects/p1/environments", json={
"name": "prod2", "host": "10.0.0.40", "username": "root", "is_installed": True,
})
first_task_id = r.json().get("scan_task_id")
assert first_task_id is not None
env_id = r.json()["id"]
# Second PATCH with host change — was_installed=True, so no scan triggered
with patch("subprocess.Popen") as mock_popen2:
mock_popen2.return_value = MagicMock(pid=44444)
r2 = client.patch(f"/api/projects/p1/environments/{env_id}", json={
"host": "10.0.0.41",
})
assert r2.status_code == 200
assert "scan_task_id" not in r2.json()
def test_patch_environment_nothing_to_update(client):
r = client.post("/api/projects/p1/environments", json={
"name": "dev", "host": "10.0.0.50", "username": "root",
})
env_id = r.json()["id"]
r = client.patch(f"/api/projects/p1/environments/{env_id}", json={})
assert r.status_code == 400
def test_patch_environment_not_found(client):
r = client.patch("/api/projects/p1/environments/99999", json={"host": "1.2.3.4"})
assert r.status_code == 404
def test_delete_environment(client):
r = client.post("/api/projects/p1/environments", json={
"name": "dev", "host": "10.0.0.60", "username": "root",
})
env_id = r.json()["id"]
r = client.delete(f"/api/projects/p1/environments/{env_id}")
assert r.status_code == 204
# Verify gone
r = client.get("/api/projects/p1/environments")
ids = [e["id"] for e in r.json()]
assert env_id not in ids
def test_delete_environment_not_found(client):
r = client.delete("/api/projects/p1/environments/99999")
assert r.status_code == 404
def test_scan_environment(client):
r = client.post("/api/projects/p1/environments", json={
"name": "prod", "host": "10.0.0.70", "username": "root",
})
env_id = r.json()["id"]
with patch("subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock(pid=55555)
r = client.post(f"/api/projects/p1/environments/{env_id}/scan")
assert r.status_code == 202
data = r.json()
assert data["status"] == "started"
assert "task_id" in data
def test_scan_environment_not_found(client):
r = client.post("/api/projects/p1/environments/99999/scan")
assert r.status_code == 404
# ---------------------------------------------------------------------------
# Environments (KIN-087) — дополнительные тесты по acceptance criteria
# ---------------------------------------------------------------------------
def test_list_environments_auth_value_hidden():
"""GET /environments не должен возвращать auth_value (AC: маскировка)."""
import web.api as api_module2
from pathlib import Path
import tempfile
with tempfile.TemporaryDirectory() as tmp:
db_path = Path(tmp) / "t.db"
api_module2.DB_PATH = db_path
from web.api import app
from fastapi.testclient import TestClient
c = TestClient(app)
c.post("/api/projects", json={"id": "p2", "name": "P2", "path": "/p2"})
c.post("/api/projects/p2/environments", json={
"name": "prod", "host": "1.2.3.4", "username": "root",
"auth_type": "password", "auth_value": "supersecret",
})
r = c.get("/api/projects/p2/environments")
assert r.status_code == 200
for env in r.json():
assert env.get("auth_value") is None
def test_patch_environment_auth_value_hidden(client):
"""PATCH /environments/{id} не должен возвращать auth_value в ответе (AC: маскировка)."""
r = client.post("/api/projects/p1/environments", json={
"name": "masked", "host": "5.5.5.5", "username": "user",
"auth_value": "topsecret",
})
env_id = r.json()["id"]
r = client.patch(f"/api/projects/p1/environments/{env_id}", json={"host": "6.6.6.6"})
assert r.status_code == 200
assert r.json().get("auth_value") is None
def test_is_installed_flag_persisted(client):
"""is_installed=True сохраняется и возвращается в GET-списке (AC: чекбокс работает)."""
with patch("subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock(pid=99001)
r = client.post("/api/projects/p1/environments", json={
"name": "installed_prod", "host": "7.7.7.7", "username": "admin",
"is_installed": True,
})
assert r.status_code == 201
env_id = r.json()["id"]
r = client.get("/api/projects/p1/environments")
envs = {e["id"]: e for e in r.json()}
assert bool(envs[env_id]["is_installed"]) is True
def test_is_installed_false_not_installed(client):
"""is_installed=False по умолчанию сохраняется корректно."""
r = client.post("/api/projects/p1/environments", json={
"name": "notinstalled", "host": "8.8.8.8", "username": "ops",
"is_installed": False,
})
assert r.status_code == 201
env_id = r.json()["id"]
r = client.get("/api/projects/p1/environments")
envs = {e["id"]: e for e in r.json()}
assert not bool(envs[env_id]["is_installed"])
def test_sysadmin_scan_task_has_escalation_in_brief(client):
"""Задача sysadmin должна содержать инструкцию об эскалации при нехватке данных (AC#4)."""
with patch("subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock(pid=99002)
r = client.post("/api/projects/p1/environments", json={
"name": "esc_test", "host": "9.9.9.9", "username": "deploy",
"is_installed": True,
})
task_id = r.json()["scan_task_id"]
from core.db import init_db
from core import models as m
conn = init_db(api_module.DB_PATH)
task = m.get_task(conn, task_id)
conn.close()
brief = task["brief"]
assert isinstance(brief, dict), "brief must be a dict"
text = brief.get("text", "")
assert "эскалация" in text.lower(), (
"Sysadmin task brief must mention escalation to user when data is insufficient"
)
def test_create_environment_key_auth_type(client):
"""auth_type='key' должен быть принят и сохранён (AC: ключ SSH)."""
r = client.post("/api/projects/p1/environments", json={
"name": "ssh_key_env", "host": "10.10.10.10", "username": "git",
"auth_type": "key", "auth_value": "-----BEGIN OPENSSH PRIVATE KEY-----",
})
assert r.status_code == 201
data = r.json()
assert data["auth_type"] == "key"
assert data.get("auth_value") is None
def test_create_environment_duplicate_name_conflict(client):
"""Повторное создание среды с тем же именем в проекте → 409 Conflict."""
client.post("/api/projects/p1/environments", json={
"name": "unique_env", "host": "11.11.11.11", "username": "root",
})
r = client.post("/api/projects/p1/environments", json={
"name": "unique_env", "host": "22.22.22.22", "username": "root",
})
assert r.status_code == 409
def test_patch_environment_empty_auth_value_preserves_stored(client):
"""PATCH с пустым auth_value не стирает сохранённый credential (AC: безопасность)."""
r = client.post("/api/projects/p1/environments", json={
"name": "cred_safe", "host": "33.33.33.33", "username": "ops",
"auth_value": "original_password",
})
env_id = r.json()["id"]
# Patch без auth_value — credential должен сохраниться
r = client.patch(f"/api/projects/p1/environments/{env_id}", json={"host": "44.44.44.44"})
assert r.status_code == 200
# Читаем raw запись из БД (get_environment возвращает obfuscated auth_value)
from core.db import init_db
from core import models as m
conn = init_db(api_module.DB_PATH)
raw = m.get_environment(conn, env_id)
conn.close()
assert raw["auth_value"] is not None, "Stored credential must be preserved after PATCH without auth_value"
decrypted = m._decrypt_auth(raw["auth_value"])
assert decrypted == "original_password", "Stored credential must be decryptable and match original"
# ---------------------------------------------------------------------------
# KIN-088 — POST /run возвращает 409 если задача уже in_progress
# ---------------------------------------------------------------------------
def test_run_returns_409_when_task_already_in_progress(client):
"""KIN-088: повторный POST /run для задачи со статусом in_progress → 409 с task_already_running."""
from core.db import init_db
from core import models as m
conn = init_db(api_module.DB_PATH)
m.update_task(conn, "P1-001", status="in_progress")
conn.close()
r = client.post("/api/tasks/P1-001/run")
assert r.status_code == 409
assert r.json()["error"] == "task_already_running"
def test_run_409_error_key_is_task_already_running(client):
"""KIN-088: тело ответа 409 содержит ключ error='task_already_running'."""
from core.db import init_db
from core import models as m
conn = init_db(api_module.DB_PATH)
m.update_task(conn, "P1-001", status="in_progress")
conn.close()
r = client.post("/api/tasks/P1-001/run")
body = r.json()
assert "error" in body
assert body["error"] == "task_already_running"
def test_run_second_call_does_not_change_status(client):
"""KIN-088: при повторном /run задача остаётся in_progress, статус не сбрасывается."""
from core.db import init_db
from core import models as m
conn = init_db(api_module.DB_PATH)
m.update_task(conn, "P1-001", status="in_progress")
conn.close()
client.post("/api/tasks/P1-001/run") # второй вызов — должен вернуть 409
r = client.get("/api/tasks/P1-001")
assert r.json()["status"] == "in_progress"
def test_run_pending_task_still_returns_202(client):
"""KIN-088: задача со статусом pending запускается без ошибки — 202."""
r = client.post("/api/tasks/P1-001/run")
assert r.status_code == 202
def test_run_kin085_parallel_different_tasks_not_blocked(client):
"""KIN-085: /run для разных задач независимы — in_progress одной не блокирует другую."""
# Создаём вторую задачу
client.post("/api/tasks", json={"project_id": "p1", "title": "Second task"})
# Ставим первую задачу в in_progress
from core.db import init_db
from core import models as m
conn = init_db(api_module.DB_PATH)
m.update_task(conn, "P1-001", status="in_progress")
conn.close()
# Запуск второй задачи должен быть успешным
r = client.post("/api/tasks/P1-002/run")
assert r.status_code == 202