Compare commits

...

2 Commits

7 changed files with 272 additions and 133 deletions
@@ -0,0 +1,28 @@
name: "Continuous Integration"
on:
push:
branches:
- main
pull_request:
jobs:
checks:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Set up Python 3.12
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install dependencies
run: make install dev
- name: Lint (ruff)
run: make lint
- name: Test (pytest)
run: make test
+21
View File
@@ -252,6 +252,27 @@ curl -X POST http://localhost:8080/webhooks/nexus \
---
## Verification patterns
- Always run tests after making code changes. Report exact result: pass/fail count and any failure messages.
- Never say "tests pass" without running `make test` yourself.
- When adding a feature, also test boundary cases and error paths, not just the happy path.
- After fixing a bug, add a regression test that reproduces the original failure. The regression test must fail before your fix and pass after. If you cannot write such a test, your fix may not be addressing the root cause.
- When a test fails, do NOT delete, skip, or weaken it. Investigate: is the test wrong (rewrite it) or is your code wrong (fix the code)?
- Use `make check` (lint + format + typecheck + test) as the final gate before considering a task complete.
---
## Bug Fixing Process
- Reproduce first: identify the exact failing behavior. If possible, capture it as a failing test.
- Minimal change: edit only the lines required to fix the bug. Do not rewrite larger sections "for cleanliness" unless you also rewrite the tests.
- After fixing, verify with the same command that failed before (e.g., `make test`, a curl command, or a manual check).
- Check for side effects: does your fix change behavior in an unrelated area? Run the full `make test` to be sure — partial test runs may miss regressions in other modules.
- Document the fix: after resolving, briefly note what went wrong and how you diagnosed it. This becomes part of the team knowledge base for future reference.
---
## Notes
- **AI-generated code:** all code in this repository was generated by an AI assistant (Claude). Review carefully before production use.
+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 ---