(
connection: sqlite3.Connection,
scan: sqlite3.Row,
*,
occurrence_id: str | None = None,
)
| 3048 | |
| 3049 | |
| 3050 | def scan_result( |
| 3051 | connection: sqlite3.Connection, |
| 3052 | scan: sqlite3.Row, |
| 3053 | *, |
| 3054 | occurrence_id: str | None = None, |
| 3055 | ) -> dict[str, Any]: |
| 3056 | backfill_legacy_finding_details(connection, scan) |
| 3057 | progress = connection.execute( |
| 3058 | "SELECT * FROM scan_progress WHERE scan_id = ?", (scan["id"],) |
| 3059 | ).fetchone() |
| 3060 | artifact_rows = connection.execute( |
| 3061 | "SELECT kind, path FROM scan_artifacts WHERE scan_id = ?", (scan["id"],) |
| 3062 | ) |
| 3063 | artifacts = {} |
| 3064 | for row in artifact_rows: |
| 3065 | if row["kind"] not in ARTIFACTS: |
| 3066 | continue |
| 3067 | path = available_artifact_path(Path(scan["scan_dir"]), Path(row["path"])) |
| 3068 | if path is not None: |
| 3069 | artifacts[row["kind"]] = str(path) |
| 3070 | sarif_path = available_artifact_path( |
| 3071 | Path(scan["scan_dir"]), Path(scan["scan_dir"]) / "exports" / "results.sarif" |
| 3072 | ) |
| 3073 | if sarif_path is not None: |
| 3074 | artifacts["sarifReport"] = str(sarif_path) |
| 3075 | occurrence_rows = finding_occurrence_rows( |
| 3076 | connection, scan["id"], offset=0, limit=FINDINGS_RESULT_LIMIT |
| 3077 | ) |
| 3078 | if occurrence_id is not None and all(row["id"] != occurrence_id for row in occurrence_rows): |
| 3079 | occurrence = require_occurrence(connection, occurrence_id) |
| 3080 | if occurrence["scan_id"] != scan["id"]: |
| 3081 | raise SystemExit("This finding does not belong to the selected scan.") |
| 3082 | occurrence_rows.append(occurrence) |
| 3083 | finding_count = connection.execute( |
| 3084 | "SELECT COUNT(*) FROM finding_occurrences WHERE scan_id = ?", (scan["id"],) |
| 3085 | ).fetchone()[0] |
| 3086 | severity_counts = { |
| 3087 | row["severity"]: row["count"] |
| 3088 | for row in connection.execute( |
| 3089 | """ |
| 3090 | SELECT severity, COUNT(*) AS count |
| 3091 | FROM finding_occurrences |
| 3092 | WHERE scan_id = ? |
| 3093 | GROUP BY severity |
| 3094 | """, |
| 3095 | (scan["id"],), |
| 3096 | ) |
| 3097 | } |
| 3098 | remediation_available, remediation_unavailable_reason = remediation_availability(scan) |
| 3099 | return { |
| 3100 | "artifacts": artifacts, |
| 3101 | "canceledAt": scan["canceled_at"], |
| 3102 | "contract": scan_contract(scan), |
| 3103 | "failureMessage": scan["failure_message"], |
| 3104 | "findings": [finding_result(connection, scan, row) for row in occurrence_rows], |
| 3105 | "findingCount": finding_count, |
| 3106 | "findingsTruncated": finding_count > len(occurrence_rows), |
| 3107 | "severityCounts": severity_counts, |
no test coverage detected