Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions backend/app/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,3 +141,16 @@ class AuditLog(Base):
created_at: Mapped[datetime] = mapped_column(
DateTime, default=lambda: datetime.now(UTC), index=True
)


class Suppression(Base):
__tablename__ = "suppressions"

id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
issue_type: Mapped[str] = mapped_column(String(200), index=True)
line: Mapped[int] = mapped_column(Integer, nullable=True)
reason: Mapped[str] = mapped_column(Text)
scope: Mapped[str] = mapped_column(String(20), default="project", index=True)
created_at: Mapped[datetime] = mapped_column(
DateTime, default=lambda: datetime.now(UTC), index=True
)
113 changes: 111 additions & 2 deletions backend/app/routers/debugging.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,78 @@
"""Debugging router — POST /debugging/"""

from fastapi import APIRouter
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from ..database import Base, get_db
from ..models import Suppression
from ..schemas import CodeRequest, DebuggingResponse
from ..services.code_assistant import detect_language, run_bug_detection

router = APIRouter()

ACTIVE_SUPPRESSION_SCOPES = {"project", "global"}


class SuppressionRequest(BaseModel):
issue_type: str = Field(
..., min_length=1, max_length=200, description="Issue type to suppress"
)
line: int | None = Field(
default=None, gt=0, description="Issue line number if known"
)
reason: str = Field(..., min_length=1, description="Reason for the suppression")
scope: str = Field(default="project", description="Suppression scope")

@field_validator("issue_type")
@classmethod
def validate_issue_type(cls, value: str) -> str:
stripped = value.strip()
if not stripped:
raise ValueError("issue_type must not be empty")
return stripped

@field_validator("reason")
@classmethod
def validate_reason(cls, value: str) -> str:
stripped = value.strip()
if not stripped:
raise ValueError("reason must not be empty")
return stripped

@field_validator("scope")
@classmethod
def validate_scope(cls, value: str) -> str:
normalized = value.strip().lower()
if normalized not in {"project", "global"}:
raise ValueError("scope must be 'project' or 'global'")
return normalized


def _suppression_matches(issue: dict, suppression: Suppression) -> bool:
if suppression.scope not in ACTIVE_SUPPRESSION_SCOPES:
return False
if issue.get("type") != suppression.issue_type:
return False
return suppression.line is None or issue.get("line") == suppression.line


def _filter_suppressed_issues(
issues: list[dict], suppressions: list[Suppression]
) -> list[dict]:
if not issues or not suppressions:
return issues

return [
issue
for issue in issues
if not any(
_suppression_matches(issue, suppression) for suppression in suppressions
)
]


@router.post(
"/",
Expand Down Expand Up @@ -34,9 +100,21 @@
500: {"description": "Internal server error."},
},
)
async def debug(req: CodeRequest):
async def debug(req: CodeRequest, db: Session = Depends(get_db)):
Base.metadata.create_all(bind=db.get_bind())

lang = detect_language(req.code, req.language)
issues = run_bug_detection(req.code, lang)

try:
suppressions = db.execute(select(Suppression)).scalars().all()
except SQLAlchemyError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Could not load suppressions",
) from exc

issues = _filter_suppressed_issues(issues, list(suppressions))
errors = sum(1 for i in issues if i["severity"] == "error")
warnings = sum(1 for i in issues if i["severity"] == "warning")
infos = sum(1 for i in issues if i["severity"] == "info")
Expand All @@ -53,3 +131,34 @@ async def debug(req: CodeRequest):
"info_count": infos,
"code": req.code,
}


@router.post("/suppress", summary="Suppress a debug finding")
async def suppress_issue(req: SuppressionRequest, db: Session = Depends(get_db)):
Base.metadata.create_all(bind=db.get_bind())

suppression = Suppression(
issue_type=req.issue_type,
line=req.line,
reason=req.reason,
scope=req.scope,
)

try:
db.add(suppression)
db.commit()
db.refresh(suppression)
except SQLAlchemyError as exc:
db.rollback()
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Could not persist suppression",
) from exc

return {
"success": True,
"message": "Issue suppressed",
"issue_type": req.issue_type,
"line": req.line,
"scope": req.scope,
}
164 changes: 164 additions & 0 deletions backend/tests/test_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,16 @@
import pytest
from pathlib import Path
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
import sys
import os

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from app import main as app_main
from app.database import Base, get_db
from app.models import Suppression

client = TestClient(app_main.app)

Expand All @@ -27,6 +32,38 @@ def reset_rate_limit_state():
app_main._request_counts.clear()


@pytest.fixture
def isolated_debug_client():
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)

def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()

previous_override = app_main.app.dependency_overrides.get(get_db)
app_main.app.dependency_overrides[get_db] = override_get_db

try:
with TestClient(app_main.app) as test_client:
yield test_client, TestingSessionLocal
finally:
if previous_override is None:
app_main.app.dependency_overrides.pop(get_db, None)
else:
app_main.app.dependency_overrides[get_db] = previous_override
Base.metadata.drop_all(bind=engine)
engine.dispose()


# ── Fixtures ──────────────────────────────────────────────────────────────────
PHP_CODE = """
<?php
Expand Down Expand Up @@ -371,6 +408,133 @@ def test_debug_detects_zero_division():
assert "ZeroDivisionError" in types


def test_debug_suppression_endpoint_persists_record(isolated_debug_client):
test_client, TestingSessionLocal = isolated_debug_client
payload = {
"issue_type": "ZeroDivisionError",
"line": 1,
"reason": "False positive",
"scope": "project",
}

r = test_client.post("/debugging/suppress", json=payload)

assert r.status_code == 200
data = r.json()
assert data["success"] is True
assert data["message"] == "Issue suppressed"
assert data["issue_type"] == payload["issue_type"]
assert data["line"] == payload["line"]
assert data["scope"] == payload["scope"]

db = TestingSessionLocal()
try:
record = db.execute(select(Suppression)).scalar_one()
assert record.issue_type == payload["issue_type"]
assert record.line == payload["line"]
assert record.reason == payload["reason"]
assert record.scope == payload["scope"]
assert record.created_at is not None
finally:
db.close()


def test_debug_filters_suppressed_findings(isolated_debug_client):
test_client, TestingSessionLocal = isolated_debug_client
db = TestingSessionLocal()
try:
db.add(
Suppression(
issue_type="ZeroDivisionError",
line=1,
reason="Known false positive",
scope="project",
)
)
db.commit()
finally:
db.close()

r = test_client.post(
"/debugging/",
json={
"code": "result = a / b\npassword = \"abc123\"",
"language": "python",
},
)

assert r.status_code == 200
data = r.json()
issue_types = [issue["type"] for issue in data["issues"]]
assert "ZeroDivisionError" not in issue_types
assert "Hardcoded Secret" in issue_types
assert data["clean"] is False
assert data["error_count"] + data["warning_count"] + data["info_count"] == len(
data["issues"]
)
assert data["summary"] == (
f"Found {len(data['issues'])} issue(s): "
f"{data['error_count']} error(s), "
f"{data['warning_count']} warning(s), "
f"{data['info_count']} info."
)


def test_debug_keeps_unsuppressed_findings(isolated_debug_client):
test_client, TestingSessionLocal = isolated_debug_client
db = TestingSessionLocal()
try:
db.add(
Suppression(
issue_type="ZeroDivisionError",
line=2,
reason="Only suppress the second line",
scope="project",
)
)
db.commit()
finally:
db.close()

r = test_client.post(
"/debugging/", json={"code": "result = a / b", "language": "python"}
)

assert r.status_code == 200
issue_types = [issue["type"] for issue in r.json()["issues"]]
assert "ZeroDivisionError" in issue_types


def test_debug_recalculates_counts_after_suppression(isolated_debug_client):
test_client, TestingSessionLocal = isolated_debug_client
db = TestingSessionLocal()
try:
db.add(
Suppression(
issue_type="ZeroDivisionError",
line=1,
reason="Known false positive",
scope="global",
)
)
db.commit()
finally:
db.close()

r = test_client.post(
"/debugging/", json={"code": "result = a / b", "language": "python"}
)

assert r.status_code == 200
data = r.json()
assert data["issues"] == []
assert data["clean"] is True
assert data["error_count"] == 0
assert data["warning_count"] == 0
assert data["info_count"] == 0
assert data["summary"] == "✅ No issues detected!"


def test_debug_detects_hardcoded_secret():
r = client.post(
"/debugging/", json={"code": 'password = "abc123"', "language": "python"}
Expand Down
24 changes: 24 additions & 0 deletions backend/tests/test_schema_smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
FavoriteResult,
QueryHistory,
SharedSnippet,
Suppression,
User,
)

Expand All @@ -52,6 +53,7 @@
"favorite_results",
"digest_subscriptions",
"shares",
"suppressions",
}

# ── database URL (PostgreSQL in CI, SQLite locally) ───────────────────────────
Expand Down Expand Up @@ -230,6 +232,28 @@ def test_digest_subscription_insert(db_session):
assert fetched.is_active is True


def test_suppression_insert(db_session):
"""Suppression must accept a write and preserve the finding details."""
suppression = Suppression(
issue_type="ZeroDivisionError",
line=1,
reason="False positive in generated sample",
scope="project",
)
db_session.add(suppression)
db_session.commit()
db_session.refresh(suppression)

fetched = (
db_session.query(Suppression).filter_by(issue_type="ZeroDivisionError").first()
)
assert fetched is not None
assert fetched.line == 1
assert fetched.reason == "False positive in generated sample"
assert fetched.scope == "project"
assert fetched.created_at is not None


# ── history DB (aiosqlite / FTS5) tests ───────────────────────────────────────
def test_history_db_init():
"""init_db() must create the history and fts_history tables without error."""
Expand Down
Loading
Loading