Compare commits
2 Commits
6ea5c85a4b
...
3818c2db29
| Author | SHA1 | Date | |
|---|---|---|---|
| 3818c2db29 | |||
| 2b695aed82 |
@@ -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
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
+48
-24
@@ -64,6 +64,8 @@ async def test_analyze_finding_timeout():
|
||||
import guarddog_nexus.config
|
||||
from guarddog_nexus.core.llm import analyze_finding
|
||||
|
||||
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
|
||||
|
||||
@@ -72,8 +74,8 @@ async def test_analyze_finding_timeout():
|
||||
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 = ""
|
||||
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
|
||||
|
||||
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 = ""
|
||||
finally:
|
||||
guarddog_nexus.config.config.llm_api_key = original_api_key
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -96,6 +100,8 @@ async def test_analyze_finding_success():
|
||||
import guarddog_nexus.config
|
||||
from guarddog_nexus.core.llm import analyze_finding
|
||||
|
||||
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
|
||||
|
||||
@@ -117,8 +123,8 @@ async def test_analyze_finding_success():
|
||||
assert result is not None
|
||||
assert result["verdict"] == "safe"
|
||||
assert result["severity_rating"] == "low"
|
||||
|
||||
guarddog_nexus.config.config.llm_api_key = ""
|
||||
finally:
|
||||
guarddog_nexus.config.config.llm_api_key = original_api_key
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -126,6 +132,8 @@ async def test_analyze_finding_markdown_unwrap():
|
||||
import guarddog_nexus.config
|
||||
from guarddog_nexus.core.llm import analyze_finding
|
||||
|
||||
original_api_key = guarddog_nexus.config.config.llm_api_key
|
||||
try:
|
||||
guarddog_nexus.config.config.llm_api_key = "sk-test"
|
||||
|
||||
mock_resp = MagicMock()
|
||||
@@ -145,8 +153,8 @@ async def test_analyze_finding_markdown_unwrap():
|
||||
result = await analyze_finding({"rule": "test"})
|
||||
assert result is not None
|
||||
assert result["verdict"] == "suspicious"
|
||||
|
||||
guarddog_nexus.config.config.llm_api_key = ""
|
||||
finally:
|
||||
guarddog_nexus.config.config.llm_api_key = original_api_key
|
||||
|
||||
|
||||
# --- T1: analyze_finding_htmx endpoint ---
|
||||
@@ -156,45 +164,53 @@ async def test_analyze_finding_markdown_unwrap():
|
||||
async def test_analyze_endpoint_llm_disabled(client, sample_finding):
|
||||
import guarddog_nexus.config
|
||||
|
||||
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
|
||||
finally:
|
||||
guarddog_nexus.config.config.llm_enabled = original_enabled
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_analyze_endpoint_not_found(client):
|
||||
import guarddog_nexus.config
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
original_enabled = guarddog_nexus.config.config.llm_enabled
|
||||
try:
|
||||
guarddog_nexus.config.config.llm_enabled = True
|
||||
|
||||
fake_report = {
|
||||
@@ -211,14 +227,16 @@ async def test_analyze_endpoint_success(client, sample_finding):
|
||||
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
|
||||
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
|
||||
|
||||
original_enabled = guarddog_nexus.config.config.llm_enabled
|
||||
try:
|
||||
guarddog_nexus.config.config.llm_enabled = True
|
||||
|
||||
async def mock_analyze(data):
|
||||
@@ -228,8 +246,8 @@ async def test_analyze_endpoint_failure(client, sample_finding):
|
||||
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
|
||||
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
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
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
|
||||
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,6 +310,8 @@ async def test_analyze_finding_exhausts_all_retries():
|
||||
import guarddog_nexus.config
|
||||
from guarddog_nexus.core.llm import analyze_finding
|
||||
|
||||
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):
|
||||
@@ -296,8 +320,8 @@ async def test_analyze_finding_exhausts_all_retries():
|
||||
|
||||
assert result is None
|
||||
assert mock_sleep.call_count == 1
|
||||
|
||||
guarddog_nexus.config.config.llm_api_key = ""
|
||||
finally:
|
||||
guarddog_nexus.config.config.llm_api_key = original_api_key
|
||||
|
||||
|
||||
# --- LLM lock cleanup ---
|
||||
|
||||
Reference in New Issue
Block a user