(connection: sqlite3.Connection, args: argparse.Namespace)
| 1826 | |
| 1827 | |
| 1828 | def fail_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]: |
| 1829 | scan_id = require_uuid(args.scan_id, "scan-id") |
| 1830 | connection.execute("BEGIN IMMEDIATE") |
| 1831 | try: |
| 1832 | timestamp = now() |
| 1833 | scan = require_scan(connection, scan_id) |
| 1834 | if scan["status"] == "failed": |
| 1835 | connection.commit() |
| 1836 | return scan_context(connection, scan["id"]) |
| 1837 | if scan["status"] == "complete": |
| 1838 | raise SystemExit("A completed scan cannot be marked failed.") |
| 1839 | updated = connection.execute( |
| 1840 | """ |
| 1841 | UPDATE scans |
| 1842 | SET status = 'failed', failure_message = ?, completed_at = ?, updated_at = ? |
| 1843 | WHERE id = ? AND status = 'running' |
| 1844 | """, |
| 1845 | (optional_text(args.message, maximum=2400), timestamp, timestamp, scan["id"]), |
| 1846 | ) |
| 1847 | if updated.rowcount != 1: |
| 1848 | raise SystemExit("Only a running scan can be marked failed.") |
| 1849 | progress_updated = connection.execute( |
| 1850 | "UPDATE scan_progress SET updated_at = ? WHERE scan_id = ?", |
| 1851 | (timestamp, scan["id"]), |
| 1852 | ) |
| 1853 | if progress_updated.rowcount != 1: |
| 1854 | raise SystemExit("Codex Security scan progress not found.") |
| 1855 | connection.commit() |
| 1856 | except BaseException: |
| 1857 | connection.rollback() |
| 1858 | raise |
| 1859 | return scan_context(connection, scan["id"]) |
| 1860 | |
| 1861 | |
| 1862 | def cancel_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]: |
no test coverage detected