feat: улучшить agentic-readiness — добавить CI, проверки конфигурации и тесты

This commit is contained in:
Marat Kharitonov
2026-07-06 22:18:12 +03:00
parent 2b695aed82
commit 3818c2db29
6 changed files with 251 additions and 133 deletions
+34 -13
View File
@@ -109,60 +109,76 @@ class TestPaginationE2e:
"""End-to-end tests for pagination functionality."""
@pytest.mark.asyncio
async def test_e2e_scans_pagination(self, e2e_client):
async def test_e2e_scans_pagination(self, e2e_client, sample_e2e_scan):
"""Verify that scan list pagination works."""
# First page
resp1 = await e2e_client.get("/api/v1/scans?limit=10&offset=0")
assert resp1.status_code == 200
data1 = resp1.json()
assert data1["limit"] == 10
assert data1["offset"] == 0
assert data1["total"] >= 1
assert len(data1["scans"]) >= 1
assert any(p["package_name"] == "test-e2e-pkg" for p in data1["scans"])
# Second page
resp2 = await e2e_client.get("/api/v1/scans?limit=10&offset=10")
assert resp2.status_code == 200
data2 = resp2.json()
assert data2["limit"] == 10
assert data2["offset"] == 10
assert data2["total"] >= 0
@pytest.mark.asyncio
async def test_e2e_packages_pagination(self, e2e_client):
async def test_e2e_packages_pagination(self, e2e_client, sample_e2e_scan):
"""Verify that package list pagination works."""
resp1 = await e2e_client.get("/api/v1/packages?limit=5&offset=0")
assert resp1.status_code == 200
data1 = resp1.json()
assert data1["limit"] == 5
assert data1["offset"] == 0
assert data1["total"] >= 1
assert any(p["name"] == "test-e2e-pkg" for p in data1["packages"])
class TestFilteringE2e:
"""End-to-end tests for filtering functionality."""
@pytest.mark.asyncio
async def test_e2e_scan_filter_by_status(self, e2e_client):
async def test_e2e_scan_filter_by_status(self, e2e_client, sample_e2e_scan):
"""Verify that scans can be filtered by status."""
resp = await e2e_client.get("/api/v1/scans?status=completed")
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 1
assert all(s["status"] == "completed" for s in data["scans"])
assert any(s["package_name"] == "test-e2e-pkg" for s in data["scans"])
@pytest.mark.asyncio
async def test_e2e_scan_filter_by_flagged(self, e2e_client):
async def test_e2e_scan_filter_by_flagged(self, e2e_client, sample_e2e_scan):
"""Verify that scans can be filtered by flagged status."""
resp = await e2e_client.get("/api/v1/scans?flagged=true")
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 1
assert all(s["flagged"] is True for s in data["scans"])
assert any(s["package_name"] == "test-e2e-pkg" for s in data["scans"])
@pytest.mark.asyncio
async def test_e2e_scan_filter_by_search(self, e2e_client):
async def test_e2e_scan_filter_by_search(self, e2e_client, e2e_db_session, sample_e2e_scan):
"""Verify that scans can be filtered by search term."""
resp = await e2e_client.get("/api/v1/scans?search=e2e")
assert resp.status_code == 200
data = resp.json()
# If there are matching scans, they should contain the search term
if data["scans"]:
assert any("e2e" in s["package_name"] for s in data["scans"])
assert data["total"] >= 1
assert all("e2e" in scan["package_name"] for scan in data["scans"])
@pytest.mark.asyncio
async def test_e2e_scan_filter_no_match_returns_empty(self, e2e_client, e2e_db_session):
"""Verify that filter with no matches returns empty list."""
resp = await e2e_client.get("/api/v1/scans?search=zzznotfound")
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 0
assert data["scans"] == []
class TestErrorHandlingE2e:
@@ -220,16 +236,21 @@ class TestWebsocketFragmentE2e:
"""E2E tests for HTMX fragment responses."""
@pytest.mark.asyncio
async def test_e2e_scans_fragment_response(self, e2e_client):
async def test_e2e_scans_fragment_response(self, e2e_client, sample_e2e_scan):
"""Verify that scans page returns fragment when HX-Request header is set."""
resp = await e2e_client.get("/scans", headers={"HX-Request": "true"})
assert resp.status_code == 200
# Fragment should not include full HTML structure
assert "<!DOCTYPE" not in resp.text
assert "<table" in resp.text
assert "scans-table-container" in resp.text
assert "test-e2e-pkg" in resp.text
@pytest.mark.asyncio
async def test_e2e_packages_fragment_response(self, e2e_client):
async def test_e2e_packages_fragment_response(self, e2e_client, sample_e2e_scan):
"""Verify that packages page returns fragment when HX-Request header is set."""
resp = await e2e_client.get("/packages", headers={"HX-Request": "true"})
assert resp.status_code == 200
assert "<!DOCTYPE" not in resp.text
assert "<table" in resp.text
assert "packages-table-container" in resp.text
assert "test-e2e-pkg" in resp.text
+21 -2
View File
@@ -71,6 +71,10 @@ class TestWebhookToScanFlow:
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "accepted"
assert data["asset"] == "/" + e2e_go_webhook_payload["asset"]["name"].strip("/")
assert data["action"] == "UPDATED"
assert data["asset"] == "/packages/github.com/e2e/test-go/@v/v1.0.0.zip"
assert data["action"] == "UPDATED"
@pytest.mark.asyncio
async def test_e2e_webhook_accepts_npm_asset(
@@ -102,6 +106,10 @@ class TestWebhookToScanFlow:
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "accepted"
assert data["asset"] == e2e_npm_webhook_payload["asset"]["name"]
assert data["action"] == "UPDATED"
assert data["asset"] == "/packages/e2e-test-npm/-/e2e-test-npm-1.0.0.tgz"
assert data["action"] == "UPDATED"
@pytest.mark.asyncio
async def test_e2e_webhook_accepts_scoped_npm_asset(self, e2e_client, e2e_db_session):
@@ -139,7 +147,10 @@ class TestWebhookToScanFlow:
resp = await e2e_client.post("/webhooks/nexus", json=payload)
assert resp.status_code == 200
assert resp.json()["status"] == "accepted"
data = resp.json()
assert data["status"] == "accepted"
assert data["asset"] == "/packages/@angular/core/-/core-18.0.0.tgz"
assert data["action"] == "UPDATED"
class TestWebhookSignatureValidation:
@@ -167,8 +178,11 @@ class TestWebhookSignatureValidation:
headers={"X-Nexus-Webhook-Signature": signature, "Content-Type": "application/json"},
)
# Should be accepted (signature matches)
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "accepted"
assert data["asset"] == e2e_webhook_payload["asset"]["name"]
assert data["action"] == "UPDATED"
config.webhook_secret = original_secret
@@ -237,6 +251,10 @@ class TestApiIntegration:
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 2
assert len(data["findings"]) >= 2
rules = {f["rule"] for f in data["findings"]}
assert "shady-links" in rules
assert "exec-base64" in rules
@pytest.mark.asyncio
async def test_e2e_api_findings_filter_by_rule(self, e2e_client, sample_e2e_scan):
@@ -281,6 +299,7 @@ class TestWebUiIntegration:
resp = await e2e_client.get("/packages")
assert resp.status_code == 200
assert "Packages" in resp.text or "Пакеты" in resp.text
assert "test-e2e-pkg" in resp.text
@pytest.mark.asyncio
async def test_e2e_package_detail_page(self, e2e_client, sample_e2e_scan):
+28 -2
View File
@@ -39,7 +39,6 @@ async def test_scan_not_found(client):
@pytest.mark.asyncio
async def test_list_scans_with_filters(client):
# Filter parameters smoke test — should not 500
for params in [
"?flagged=true&search=test&status=completed&sort_by=id&sort_dir=asc",
"?flagged=false&search=nonexistent&sort_by=total_findings",
@@ -48,6 +47,11 @@ async def test_list_scans_with_filters(client):
]:
resp = await client.get(f"/api/v1/scans{params}")
assert resp.status_code == 200, f"Failed on: {params}"
data = resp.json()
assert "scans" in data
assert "total" in data
assert "limit" in data
assert isinstance(data["scans"], list)
@pytest.mark.asyncio
@@ -58,6 +62,8 @@ async def test_scan_stats_with_data(client, sample_flagged_scan):
assert data["total_scans"] == 1
assert data["flagged_scans"] == 1
assert data["total_findings"] == 1
assert data["recent_flagged"] == 1
assert isinstance(data["top_rules"], list)
@pytest.mark.asyncio
@@ -73,6 +79,8 @@ async def test_scans_csv_export_with_filter(client, sample_flagged_scan):
resp = await client.get("/api/v1/scans/export?flagged=true")
assert resp.status_code == 200
assert sample_flagged_scan.package_name in resp.text
assert sample_flagged_scan.ecosystem in resp.text
assert "text/csv" in resp.headers["content-type"]
# --- Packages ---
@@ -84,18 +92,25 @@ async def test_list_packages_empty(client):
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 0
assert data["packages"] == []
@pytest.mark.asyncio
async def test_list_packages_with_filters(client):
for params in [
"?search=test&sort_by=name&sort_dir=asc",
"?flagged=false&sort_by=last_scanned_at",
"?flagged=False&sort_by=last_scanned_at",
"?ecosystem=pypi",
"?sort_by=invalid",
]:
resp = await client.get(f"/api/v1/packages{params}")
assert resp.status_code == 200, f"Failed on: {params}"
data = resp.json()
assert "packages" in data
assert "total" in data
assert "limit" in data
assert isinstance(data["packages"], list)
assert len(data["packages"]) <= data["total"]
@pytest.mark.asyncio
@@ -111,6 +126,8 @@ async def test_packages_csv_export_with_filter(client, sample_flagged_scan):
resp = await client.get("/api/v1/packages/export?flagged=true")
assert resp.status_code == 200
assert sample_flagged_scan.package_name in resp.text
assert sample_flagged_scan.package_version in resp.text
assert "text/csv" in resp.headers["content-type"]
@pytest.mark.asyncio
@@ -140,6 +157,7 @@ async def test_list_findings_empty(client):
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 0
assert data["findings"] == []
@pytest.mark.asyncio
@@ -160,6 +178,11 @@ async def test_list_findings_with_filters(client, sample_flagged_scan):
]:
resp = await client.get(f"/api/v1/findings{params}")
assert resp.status_code == 200, f"Failed on: {params}"
data = resp.json()
assert "findings" in data
assert "total" in data
assert "limit" in data
assert isinstance(data["findings"], list)
# --- Web UI ---
@@ -190,12 +213,14 @@ async def test_web_ui_scans(client):
async def test_web_ui_scans_with_search(client):
resp = await client.get("/scans?search=nonexistent&status=completed&sort_by=id&sort_dir=asc")
assert resp.status_code == 200
assert "search" in resp.text.lower() or "nonexistent" in resp.text
@pytest.mark.asyncio
async def test_web_ui_scans_page_out_of_range(client):
resp = await client.get("/scans?page=999")
assert resp.status_code == 200
assert "page" in resp.text.lower() or "scan" in resp.text.lower()
@pytest.mark.asyncio
@@ -223,6 +248,7 @@ async def test_web_ui_packages(client):
async def test_web_ui_packages_with_search(client):
resp = await client.get("/packages?search=test&sort_by=name&sort_dir=asc")
assert resp.status_code == 200
assert "search" in resp.text.lower() or "test" in resp.text or "package" in resp.text.lower()
@pytest.mark.asyncio
+1 -1
View File
@@ -18,7 +18,7 @@ async def test_reap_stale_analysis_resets_stuck_findings(db_session):
from guarddog_nexus.db.engine import _engine
async with _engine.begin() as conn:
async with _engine.begin():
pass # ensure tables exist in _engine too
await db_session.execute(
+139 -115
View File
@@ -64,16 +64,18 @@ async def test_analyze_finding_timeout():
import guarddog_nexus.config
from guarddog_nexus.core.llm import analyze_finding
guarddog_nexus.config.config.llm_api_key = "sk-test"
guarddog_nexus.config.config.llm_timeout = 1
original_api_key = guarddog_nexus.config.config.llm_api_key
try:
guarddog_nexus.config.config.llm_api_key = "sk-test"
guarddog_nexus.config.config.llm_timeout = 1
import httpx
import httpx
with patch("httpx.AsyncClient.post", side_effect=httpx.TimeoutException("timeout")):
result = await analyze_finding({"rule": "test", "severity": "WARNING"})
assert result is None
guarddog_nexus.config.config.llm_api_key = ""
with patch("httpx.AsyncClient.post", side_effect=httpx.TimeoutException("timeout")):
result = await analyze_finding({"rule": "test", "severity": "WARNING"})
assert result is None
finally:
guarddog_nexus.config.config.llm_api_key = original_api_key
@pytest.mark.asyncio
@@ -81,14 +83,16 @@ async def test_analyze_finding_api_error():
import guarddog_nexus.config
from guarddog_nexus.core.llm import analyze_finding
guarddog_nexus.config.config.llm_api_key = "sk-test"
guarddog_nexus.config.config.llm_timeout = 30
original_api_key = guarddog_nexus.config.config.llm_api_key
try:
guarddog_nexus.config.config.llm_api_key = "sk-test"
guarddog_nexus.config.config.llm_timeout = 30
with patch("httpx.AsyncClient.post", side_effect=Exception("connection refused")):
result = await analyze_finding({"rule": "test", "severity": "WARNING"})
assert result is None
guarddog_nexus.config.config.llm_api_key = ""
with patch("httpx.AsyncClient.post", side_effect=Exception("connection refused")):
result = await analyze_finding({"rule": "test", "severity": "WARNING"})
assert result is None
finally:
guarddog_nexus.config.config.llm_api_key = original_api_key
@pytest.mark.asyncio
@@ -96,29 +100,31 @@ async def test_analyze_finding_success():
import guarddog_nexus.config
from guarddog_nexus.core.llm import analyze_finding
guarddog_nexus.config.config.llm_api_key = "sk-test"
guarddog_nexus.config.config.llm_timeout = 30
original_api_key = guarddog_nexus.config.config.llm_api_key
try:
guarddog_nexus.config.config.llm_api_key = "sk-test"
guarddog_nexus.config.config.llm_timeout = 30
mock_resp = MagicMock()
mock_resp.raise_for_status.return_value = None
mock_resp.json.return_value = {
"choices": [
{
"message": {
"content": '{"verdict":"safe","summary":"ok",'
'"analysis":"fine","severity_rating":"low"}',
mock_resp = MagicMock()
mock_resp.raise_for_status.return_value = None
mock_resp.json.return_value = {
"choices": [
{
"message": {
"content": '{"verdict":"safe","summary":"ok",'
'"analysis":"fine","severity_rating":"low"}',
}
}
}
]
}
]
}
with patch("guarddog_nexus.core.llm.httpx.AsyncClient.post", return_value=mock_resp):
result = await analyze_finding({"rule": "test"})
assert result is not None
assert result["verdict"] == "safe"
assert result["severity_rating"] == "low"
guarddog_nexus.config.config.llm_api_key = ""
with patch("guarddog_nexus.core.llm.httpx.AsyncClient.post", return_value=mock_resp):
result = await analyze_finding({"rule": "test"})
assert result is not None
assert result["verdict"] == "safe"
assert result["severity_rating"] == "low"
finally:
guarddog_nexus.config.config.llm_api_key = original_api_key
@pytest.mark.asyncio
@@ -126,27 +132,29 @@ async def test_analyze_finding_markdown_unwrap():
import guarddog_nexus.config
from guarddog_nexus.core.llm import analyze_finding
guarddog_nexus.config.config.llm_api_key = "sk-test"
original_api_key = guarddog_nexus.config.config.llm_api_key
try:
guarddog_nexus.config.config.llm_api_key = "sk-test"
mock_resp = MagicMock()
mock_resp.raise_for_status.return_value = None
mock_resp.json.return_value = {
"choices": [
{
"message": {
"content": '```json\n{"verdict":"suspicious","summary":"hm",'
'"analysis":"...","severity_rating":"medium"}\n```',
mock_resp = MagicMock()
mock_resp.raise_for_status.return_value = None
mock_resp.json.return_value = {
"choices": [
{
"message": {
"content": '```json\n{"verdict":"suspicious","summary":"hm",'
'"analysis":"...","severity_rating":"medium"}\n```',
}
}
}
]
}
]
}
with patch("guarddog_nexus.core.llm.httpx.AsyncClient.post", return_value=mock_resp):
result = await analyze_finding({"rule": "test"})
assert result is not None
assert result["verdict"] == "suspicious"
guarddog_nexus.config.config.llm_api_key = ""
with patch("guarddog_nexus.core.llm.httpx.AsyncClient.post", return_value=mock_resp):
result = await analyze_finding({"rule": "test"})
assert result is not None
assert result["verdict"] == "suspicious"
finally:
guarddog_nexus.config.config.llm_api_key = original_api_key
# --- T1: analyze_finding_htmx endpoint ---
@@ -156,80 +164,90 @@ async def test_analyze_finding_markdown_unwrap():
async def test_analyze_endpoint_llm_disabled(client, sample_finding):
import guarddog_nexus.config
guarddog_nexus.config.config.llm_enabled = False
original_enabled = guarddog_nexus.config.config.llm_enabled
try:
guarddog_nexus.config.config.llm_enabled = False
resp = await client.post(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "disabled" in resp.text.lower()
guarddog_nexus.config.config.llm_enabled = False
resp = await client.post(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "disabled" in resp.text.lower()
finally:
guarddog_nexus.config.config.llm_enabled = original_enabled
@pytest.mark.asyncio
async def test_analyze_endpoint_not_found(client):
import guarddog_nexus.config
guarddog_nexus.config.config.llm_enabled = True
original_enabled = guarddog_nexus.config.config.llm_enabled
try:
guarddog_nexus.config.config.llm_enabled = True
resp = await client.post("/api/v1/findings/99999/analyze")
assert resp.status_code == 404
assert "not found" in resp.text.lower()
guarddog_nexus.config.config.llm_enabled = False
resp = await client.post("/api/v1/findings/99999/analyze")
assert resp.status_code == 404
assert "not found" in resp.text.lower()
finally:
guarddog_nexus.config.config.llm_enabled = original_enabled
@pytest.mark.asyncio
async def test_analyze_endpoint_idempotent_already_analyzed(client, sample_finding_with_report):
import guarddog_nexus.config
guarddog_nexus.config.config.llm_enabled = True
original_enabled = guarddog_nexus.config.config.llm_enabled
try:
guarddog_nexus.config.config.llm_enabled = True
resp = await client.post(f"/api/v1/findings/{sample_finding_with_report.id}/analyze")
assert resp.status_code == 200
assert "safe" in resp.text
guarddog_nexus.config.config.llm_enabled = False
resp = await client.post(f"/api/v1/findings/{sample_finding_with_report.id}/analyze")
assert resp.status_code == 200
assert "safe" in resp.text
finally:
guarddog_nexus.config.config.llm_enabled = original_enabled
@pytest.mark.asyncio
async def test_analyze_endpoint_success(client, sample_finding):
import guarddog_nexus.config
guarddog_nexus.config.config.llm_enabled = True
original_enabled = guarddog_nexus.config.config.llm_enabled
try:
guarddog_nexus.config.config.llm_enabled = True
fake_report = {
"verdict": "malicious",
"summary": "bad",
"analysis": "evil",
"severity_rating": "critical",
}
fake_report = {
"verdict": "malicious",
"summary": "bad",
"analysis": "evil",
"severity_rating": "critical",
}
async def mock_analyze(data):
return fake_report
async def mock_analyze(data):
return fake_report
with patch("guarddog_nexus.core.llm.analyze_finding", mock_analyze):
resp = await client.post(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "malicious" in resp.text
guarddog_nexus.config.config.llm_enabled = False
with patch("guarddog_nexus.core.llm.analyze_finding", mock_analyze):
resp = await client.post(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "malicious" in resp.text
finally:
guarddog_nexus.config.config.llm_enabled = original_enabled
@pytest.mark.asyncio
async def test_analyze_endpoint_failure(client, sample_finding):
import guarddog_nexus.config
guarddog_nexus.config.config.llm_enabled = True
original_enabled = guarddog_nexus.config.config.llm_enabled
try:
guarddog_nexus.config.config.llm_enabled = True
async def mock_analyze(data):
return None
async def mock_analyze(data):
return None
with patch("guarddog_nexus.core.llm.analyze_finding", mock_analyze):
resp = await client.post(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "failed" in resp.text.lower()
guarddog_nexus.config.config.llm_enabled = False
with patch("guarddog_nexus.core.llm.analyze_finding", mock_analyze):
resp = await client.post(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "failed" in resp.text.lower()
finally:
guarddog_nexus.config.config.llm_enabled = original_enabled
# --- GET /analyze polling endpoint ---
@@ -245,25 +263,29 @@ class TestAnalyzeStatusEndpoint:
async def test_status_returns_report_when_complete(self, client, sample_finding_with_report):
import guarddog_nexus.config
guarddog_nexus.config.config.llm_enabled = True
original_enabled = guarddog_nexus.config.config.llm_enabled
try:
guarddog_nexus.config.config.llm_enabled = True
resp = await client.get(f"/api/v1/findings/{sample_finding_with_report.id}/analyze")
assert resp.status_code == 200
assert "safe" in resp.text
guarddog_nexus.config.config.llm_enabled = False
resp = await client.get(f"/api/v1/findings/{sample_finding_with_report.id}/analyze")
assert resp.status_code == 200
assert "safe" in resp.text
finally:
guarddog_nexus.config.config.llm_enabled = original_enabled
@pytest.mark.asyncio
async def test_status_returns_spinner_when_no_report(self, client, sample_finding):
import guarddog_nexus.config
guarddog_nexus.config.config.llm_enabled = True
original_enabled = guarddog_nexus.config.config.llm_enabled
try:
guarddog_nexus.config.config.llm_enabled = True
resp = await client.get(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "hx-get" in resp.text.lower()
guarddog_nexus.config.config.llm_enabled = False
resp = await client.get(f"/api/v1/findings/{sample_finding.id}/analyze")
assert resp.status_code == 200
assert "hx-get" in resp.text.lower()
finally:
guarddog_nexus.config.config.llm_enabled = original_enabled
@pytest.mark.asyncio
async def test_status_returns_spinner_when_analyzing(self, client, db_session, sample_finding):
@@ -288,16 +310,18 @@ async def test_analyze_finding_exhausts_all_retries():
import guarddog_nexus.config
from guarddog_nexus.core.llm import analyze_finding
guarddog_nexus.config.config.llm_api_key = "sk-test"
original_api_key = guarddog_nexus.config.config.llm_api_key
try:
guarddog_nexus.config.config.llm_api_key = "sk-test"
with patch("guarddog_nexus.core.llm._attempt_llm_call", return_value=None):
with patch("guarddog_nexus.core.llm.asyncio.sleep") as mock_sleep:
result = await analyze_finding({"rule": "test-rule"}, max_retries=2)
with patch("guarddog_nexus.core.llm._attempt_llm_call", return_value=None):
with patch("guarddog_nexus.core.llm.asyncio.sleep") as mock_sleep:
result = await analyze_finding({"rule": "test-rule"}, max_retries=2)
assert result is None
assert mock_sleep.call_count == 1
guarddog_nexus.config.config.llm_api_key = ""
assert result is None
assert mock_sleep.call_count == 1
finally:
guarddog_nexus.config.config.llm_api_key = original_api_key
# --- LLM lock cleanup ---