diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index cc2c052..4861b47 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -12,10 +12,23 @@ permissions: jobs: test: runs-on: ubuntu-latest + env: + SECRET_KEY: test-secret-key-for-github-actions strategy: matrix: python-version: ["3.10", "3.11", "3.12"] + services: + postgres: + image: postgres:15 + env: + POSTGRES_PASSWORD: pixelprobe_test + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready --health-interval 10s + --health-timeout 5s --health-retries 5 + steps: - uses: actions/checkout@v4 @@ -46,10 +59,15 @@ jobs: - name: Run tests with pytest env: - SECRET_KEY: test-secret-key-for-github-actions + PIXELPROBE_TEST_POSTGRES_URI: postgresql://postgres:pixelprobe_test@localhost:5432/postgres run: | pytest -m "not real_media" --cov=pixelprobe --cov-report=xml --cov-report=term + - name: Run real-media corruption detection tests + if: matrix.python-version == '3.12' + run: | + pytest -m real_media -v + - name: Upload coverage to Codecov uses: codecov/codecov-action@v5 with: diff --git a/CHANGELOG.MD b/CHANGELOG.MD index 0d4147e..b8a86ab 100644 --- a/CHANGELOG.MD +++ b/CHANGELOG.MD @@ -5,6 +5,27 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0). +## [2.8.5] - 2026-08-19 + +### Fixed + +- **Valid oversized images are no longer reported corrupted.** Any image past Pillow's decompression-bomb threshold (~358MP at defaults) fell through to ImageMagick, whose `cache resources exhausted` exit landed in the generic corruption branch, so a valid 625MP scan was flagged the same as real pixel damage. Both guards are resource limits, not corruption evidence; they now produce a warning verdict following the existing HEIC libheif pattern. +- **Parallel file-list scans no longer share one database connection across worker threads.** The scan-files API path (default `num_workers=4`) fanned a single PixelProbe instance across a thread pool while its engine used StaticPool, one raw psycopg2 connection that psycopg2 forbids using concurrently; the symptom was sporadic save and cache failures. The worker engine now uses a QueuePool sized to the worker count, and a postgres-backed regression test fails against the old pool. + +### Changed + +- **Frame integrity check: packets first, decode only to confirm, warning verdict.** Stage 1 paid a full sequential `-count_frames` decode on every video and then discarded the result - ffprobe's csv column order never matched what the parser expected under ffmpeg 8, so the check has been silently inert while costing minutes on large files. It now compares the near-free packet count first and runs the full decode only on a mismatch. A confirmed mismatch produces a warning, not a corruption verdict: container framerate metadata lies on sparse-video files (a 240s QuickTime fixture with 244 real frames declares 25fps), and real decode damage is caught by the deep-decode stage. + +### Removed + +- Dead `ScanExecutor`/`BatchProcessor` (`pixelprobe/services/scan_executor.py`), superseded by the Celery task path; its only callers were its own tests. + +### Tests + +- **Fixture corpus repaired.** Four committed "valid" samples (3gp, flv, mpg, wmv) were 189-byte HTML error pages from failed downloads; valid.mkv failed h264 decode under ffmpeg 8; valid.webp failed both PIL and ImageMagick; six "corrupted" samples (mp3, aiff, jpg, png, gif, bmp) decode cleanly under modern tools because the FFmpeg-bug-tracker bugs they exercised were in FFmpeg, not the files. All are replaced by a committed deterministic generator script, with per-format detection expectations documented (mpg and gif are warning-level by design). +- **real_media suite made runnable and wired into CI.** The scan fixture rescanned every sample per test and blew each test's timeout, so the suite could never complete - and CI always deselected the marker, which is how the fixture rot went unnoticed. Scans are now session-cached, the tautological `scan_status == 'completed'` assertion clause is gone, and CI runs the suite on the Python 3.12 leg with a PostgreSQL service for the concurrency test. +- New synthetic corruption matrix (truncation at header/mid/tail, zero-byte files, PNG-renamed-to-JPG, scattered mdat damage, SVG/PSD/progressive-JPEG/animated-WebP, symlink-loop discovery) plus oversized-image regression tests. + ## [2.8.4] - 2026-08-18 ### Fixed diff --git a/docs/glossary.md b/docs/glossary.md index 00f7bc5..89e9be4 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -18,15 +18,15 @@ Every term PixelProbe uses, defined once and linked to the doc that covers it. ## Validation verdicts - **Healthy** - The file decoded and validated without corruption signals. Benign decoder noise (NAL unit warnings, DTS/PTS timestamp warnings, ffmpeg 8 Opus EOF parse notices) does not affect this verdict. -- **Corrupted** - A corruption signal with a verdict fired: FFmpeg validation failure, frame integrity mismatch, JPEG pixel corruption, or a decode error flood. See [Scan Types](scan-types.md). -- **Warning** - A signal that is informative but does not prove damage: freeze events, elevated TOUT or VREP, strict-decode notices. Warning files play back fine in most cases. See [Scan Types](scan-types.md). +- **Corrupted** - A corruption signal with a verdict fired: FFmpeg validation failure, JPEG pixel corruption, or a decode error flood. See [Scan Types](scan-types.md). +- **Warning** - A signal that is informative but does not prove damage: freeze events, frame-count mismatches, elevated TOUT or VREP, strict-decode notices, tool resource limits on oversized images. Warning files play back fine in most cases. See [Scan Types](scan-types.md). - **Marked as good** - A manual override: the file keeps its scan history but is treated as healthy in stats and filters. - **Error** - The file could not be read or scanned at all (permissions, I/O failure, unreadable media). ## Deep checks - **Enhanced corruption analysis** - The staged deep check for video files: Stage 1 frame integrity, Stage 2 temporal outliers, Stage 3 multi-point sampling, Stage 4 strict error detection. See [How It Works](how-it-works.md). -- **Frame integrity check (Stage 1)** - Compares the decodable frame count against the expected count from duration and frame rate. +- **Frame integrity check (Stage 1)** - Compares the counted packets (confirmed by a decode when they disagree by more than 5%) against the count expected from duration and frame rate. Warning-only: container metadata is unreliable on sparse-video and variable-frame-rate files. - **Temporal outlier check (Stage 2)** - Samples three 10-second windows at 25/50/75% of the file and computes signalstats TOUT/VREP percentages. Warning-only. - **TOUT (temporal outliers)** - A signalstats metric flagging pixels that differ from both temporal neighbors. Film grain triggers it on clean content, so it warns rather than condemns. - **VREP (vertical line repetition)** - A signalstats metric from analog-tape QC; high values are normal in flat or graphic digital content, so it warns rather than condemns. diff --git a/docs/how-it-works.md b/docs/how-it-works.md index 9e52f7d..fb9120b 100644 --- a/docs/how-it-works.md +++ b/docs/how-it-works.md @@ -157,7 +157,6 @@ api/ - `MaintenanceService`: Database maintenance - `NotificationService`: Notification provider dispatch and rule evaluation - `HealthcheckService`: Outbound healthcheck pings for scheduled scans -- `ScanExecutor`: Bounded thread-pool execution for selected-file rescans - `scan_engine`: Celery-free core of the chunk-distributed scan engine (scan-slot claim, chunk building, finalization) - `scan_reporting`: Scan report creation @@ -429,7 +428,7 @@ Each file is validated in `pixelprobe/media_checker.py` (`PixelProbe.scan_file`) - ffprobe metadata probe (stream presence, codec, duration) - Full remux validation: FFmpeg reads the ENTIRE file with `-map 0 -c copy -f null -` and aggressive error detection to validate container integrity across all streams - Enhanced corruption analysis, run for every video: - - **Stage 1 - Frame integrity** (always): frame count verified against duration and framerate via `ffprobe -count_frames`; this is the authoritative deep corruption check + - **Stage 1 - Frame integrity** (always, warning-only): the packet count from a demux-only `ffprobe -count_packets` pass is compared against duration and framerate; a mismatch above 5% is confirmed with a full `-count_frames` decode before a warning is recorded. Never a corruption verdict - container framerate metadata lies on sparse-video and VFR files - **Stage 2 - Temporal outlier detection** (files > 1GB): sampled decode windows checked for timing anomalies; can mark corrupt or warn - **Stage 3 - Multi-point sampling** (files > 5GB): decodes 10s samples at beginning, middle, and end; NEVER marks a file corrupted (seeking produces FFmpeg-version-dependent false positives), results are informational - **Stage 4 - Strict error detection** (warnings only): `-err_detect crccheck+bitstream+buffer+explode` over the first 30 seconds; findings are container/muxing warnings, never corruption verdicts diff --git a/docs/project-structure.md b/docs/project-structure.md index d853664..1af4ec8 100644 --- a/docs/project-structure.md +++ b/docs/project-structure.md @@ -46,7 +46,6 @@ PixelProbe/ | | +-- maintenance_service.py | | +-- notification_service.py | | +-- scan_engine.py # Chunk building, scan finalization -| | +-- scan_executor.py | | +-- scan_reporting.py # Scan reports, batch file inserts | | +-- scan_service.py | | `-- stats_service.py diff --git a/docs/testing-guide.md b/docs/testing-guide.md index aac2f8b..977bc41 100644 --- a/docs/testing-guide.md +++ b/docs/testing-guide.md @@ -23,9 +23,11 @@ tests/ |-- test_migration_lock.py # Advisory-lock migration coordination |-- test_performance.py # Performance-oriented tests |-- test_read_timeout.py # Unreadable-file / read timeout handling +|-- test_postgres_concurrency.py # Marker-gated parallel-scan test against real Postgres |-- test_real_media_samples.py # Marker-gated real media parser tests |-- test_scheduler.py # Scheduled scan management |-- test_security_fixes.py # Security regression tests +|-- test_synthetic_corruption.py # Marker-gated synthetic damage matrix (truncation, mid-stream, format confusion) |-- unit/ # Unit tests for individual components | |-- test_bitrot_classification.py | |-- test_celery_settings.py @@ -57,10 +59,13 @@ tests/ Two files deserve a call-out: -- `test_real_media_samples.py` is gated behind the `real_media` marker. It - exercises the FFmpeg/ImageMagick stderr parsers against the real sample - corpus and is sensitive to tool versions, so it is excluded from the - default local run and executed in CI inside the Docker image instead. +- `test_real_media_samples.py` and `test_synthetic_corruption.py` are gated + behind the `real_media` marker. They exercise the FFmpeg/ImageMagick + validation paths against the sample corpus (plus fixtures synthesized at + test time) and are sensitive to tool versions, so they are excluded from + the default local run; CI runs them on the Python 3.12 matrix leg. + Committed synthesized fixtures are regenerated with + `tests/fixtures/media_samples/generate_corrupted_fixtures.py`. - `test_frontend_build.py` actually runs `npm install` and `npm run build`, so it needs Node.js 20 and npm available. @@ -107,6 +112,7 @@ Markers are declared in `pytest.ini`: | Marker | Meaning | |--------|---------| | `real_media` | Requires the real media sample corpus and matching tool versions; deselect locally with `-m "not real_media"` | +| `postgres` | Requires a live PostgreSQL; set `PIXELPROBE_TEST_POSTGRES_URI`, otherwise skipped | | `slow` | Long-running tests; deselect with `-m "not slow"` | | `integration` | Integration tests | | `timeout` | Sets a per-test execution timeout | diff --git a/pixelprobe/api/scan_routes.py b/pixelprobe/api/scan_routes.py index 7cf6de6..4dde82b 100644 --- a/pixelprobe/api/scan_routes.py +++ b/pixelprobe/api/scan_routes.py @@ -912,7 +912,14 @@ def scan_files_parallel(): data = request.get_json() or {} force_rescan = data.get('force_rescan', False) - num_workers = data.get('num_workers', 4) + # Cap caller-supplied worker counts: num_workers sizes thread pools and + # the checker's DB connection pool, so an uncapped value translates + # directly into PostgreSQL connections + try: + num_workers = int(data.get('num_workers', 4)) + except (TypeError, ValueError): + num_workers = 4 + num_workers = max(1, min(num_workers, current_app.config.get('MAX_WORKERS', 10))) scan_dirs = data.get('directories', []) file_paths = data.get('file_paths', []) diff --git a/pixelprobe/media_checker.py b/pixelprobe/media_checker.py index f3a5f10..44fe10c 100644 --- a/pixelprobe/media_checker.py +++ b/pixelprobe/media_checker.py @@ -288,18 +288,31 @@ def _init_database_connection(self): try: from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker - from sqlalchemy.pool import StaticPool + from sqlalchemy.pool import QueuePool, StaticPool from pixelprobe.config import PG_SESSION_TZ_UTC - # Thread-local instances use StaticPool with single persistent connection - # This avoids both connection pool exhaustion AND NullPool's concurrent operation errors - if self.database_path.startswith('postgresql://'): - # Use StaticPool with single connection per worker - # With MAX_WORKERS=10: 10 workers × 1 connection = 10 connections from workers - # Main app pool: 20+40=60, workers: 10, total: 70 (well under PostgreSQL default 100) - # NullPool caused "concurrent operations not permitted" errors during progress updates + # Match driver-qualified URLs too (postgresql+psycopg2://...) + if self.database_path.startswith(('postgresql://', 'postgresql+')): + # QueuePool sized to the worker count: each scan thread checks out + # its own connection. StaticPool shared one raw psycopg2 connection + # across ThreadPoolExecutor workers, which psycopg2 forbids using + # concurrently (sporadic save/cache failures under num_workers > 1). + # NullPool caused "concurrent operations not permitted" errors during + # progress updates. Budget: main app pool 20+40=60, workers + # max_workers+10 overflow ceiling, still under PostgreSQL's + # default 100 for any MAX_WORKERS the config allows. self._db_engine = create_engine( self.database_path, - poolclass=StaticPool, # Single persistent connection per engine + poolclass=QueuePool, + pool_size=self.max_workers, + # Overflow headroom so a caller that forgets to size + # max_workers to its thread count degrades to on-demand + # connections instead of a 30s checkout timeout + max_overflow=10, + # No pre-ping: the scan hot path checks out a session + # per cache read and per save, and a per-checkout + # SELECT 1 across a million-file scan is real traffic; + # recycle covers long-lived staleness instead + pool_recycle=3600, connect_args={ 'connect_timeout': 10, 'application_name': 'pixelprobe_worker', @@ -309,13 +322,19 @@ def _init_database_connection(self): } ) else: - # SQLite or other database - also use StaticPool for consistency + # SQLite (tests only; production is PostgreSQL-only since v2.2.0). + # StaticPool is required for :memory: databases to share one + # connection; check_same_thread=False permits cross-thread use + # (and is an invalid connect arg for any other driver). + connect_args = {'check_same_thread': False} \ + if self.database_path.startswith('sqlite') else {} self._db_engine = create_engine( self.database_path, - poolclass=StaticPool + poolclass=StaticPool, + connect_args=connect_args ) self._db_session_factory = sessionmaker(bind=self._db_engine) - logger.info("Thread-local database connection initialized (StaticPool with 1 connection)") + logger.info(f"Worker database engine initialized (pool sized for {self.max_workers} workers)") except Exception as e: logger.error(f"Failed to initialize database connection: {e}") self._db_engine = None @@ -326,6 +345,11 @@ def _get_db_session(self): if self._db_session_factory: return self._db_session_factory() return None + + def dispose_database_connection(self): + """Release pooled connections deterministically when a scan finishes""" + if self._db_engine: + self._db_engine.dispose() def discover_media_files(self, directories, max_files=None, existing_files=None, batch_check_callback=None, progress_callback=None): """Phase 1: Discover all supported files and return their paths (parallel version) @@ -1084,6 +1108,7 @@ def _check_image_corruption(self, file_path): pil_load_failed = False pil_load_error = None + pil_size_limited = False # Pillow bomb guard fired: a size limit, not a PIL failure # File size needed for JPEG pixel analysis guard and ImageMagick timeout try: @@ -1107,9 +1132,8 @@ def _check_image_corruption(self, file_path): else: scan_output.append(f"Image dimensions: {img.size[0]}x{img.size[1]}") - # Note: After load(), tile data is consumed and cleared in PIL - this is normal behavior - # Removed incorrect tile data check that was causing false positives - + # Note: After load(), tile data is consumed and cleared in PIL - this is normal behavior + # Removed incorrect tile data check that was causing false positives img.transpose(Image.Transpose.FLIP_LEFT_RIGHT) scan_output.append("Transform test: PASSED") @@ -1142,7 +1166,22 @@ def _check_image_corruption(self, file_path): elif is_heic and 'cannot identify image file' in error_lower: logger.info(f"HEIC PIL load failed (may be libheif limitation) for {file_path}: {str(e)}") # Don't mark as corrupted yet - ImageMagick will provide the definitive answer - + # Pillow's decompression-bomb guard is a pixel-count limit, not corruption evidence + elif isinstance(e, Image.DecompressionBombError): + pil_size_limited = True + warning_details.append("Image exceeds Pillow pixel-count limit; PIL validation skipped (file likely valid)") + scan_output.append("PIL load test: SKIPPED (exceeds pixel-count limit)") + logger.info(f"Pillow decompression-bomb guard for {file_path}: {str(e)}") + + if pil_size_limited: + # The shipped image's ImageMagick policy (256MP area) is below + # Pillow's bomb threshold (~358MP), so the convert can only grind + # its pixel cache and fail at a resource limit; the warning is + # already recorded, so skip the subprocess outright + scan_output.append("ImageMagick convert: SKIPPED (image exceeds pixel-count limits)") + logger.info(f"Skipping ImageMagick for {file_path}: exceeds Pillow pixel-count limit") + return is_corrupted, corruption_details, "pil", scan_output, warning_details + logger.info(f"Starting ImageMagick verification for: {file_path}") # Scale timeout with file size, no artificial limit for large files @@ -1189,6 +1228,12 @@ def _check_image_corruption(self, file_path): scan_output.append(f"Note: iOS 18+ HEIC files may use features not supported by older libheif versions") logger.info(f"HEIC libheif limitation (not corruption) for {file_path}: {result.stderr[:100]}") # Don't mark as corrupted - this is a tool limitation, not file corruption + # ImageMagick resource/policy limits (cache resources exhausted, width/height/area + # exceeds user limit) are tool limits on oversized-but-valid images, not corruption + elif 'cache resources exhausted' in stderr_lower or 'exceeds user limit' in stderr_lower: + warning_details.append("Image validation skipped: exceeds ImageMagick resource limit (file likely valid)") + scan_output.append("ImageMagick convert: SKIPPED (resource limit - not corruption)") + logger.info(f"ImageMagick resource limit (not corruption) for {file_path}: {result.stderr[:100]}") else: corruption_details.append("ImageMagick pixel validation failed") is_corrupted = True @@ -1196,7 +1241,7 @@ def _check_image_corruption(self, file_path): else: # Check if PIL passed before marking as corrupted # ImageMagick might fail due to missing delegates/decoders - if not pil_failed and not pil_load_failed: + if pil_size_limited or (not pil_failed and not pil_load_failed): # PIL passed, so file is likely OK - ImageMagick issue warning_details.append("ImageMagick convert failed (but PIL passed - likely decoder issue)") scan_output.append("Note: ImageMagick failed but PIL verified OK") @@ -1251,8 +1296,10 @@ def _check_image_corruption(self, file_path): # Only mark as corrupted if other tools also failed timeout_msg = f"ImageMagick convert timeout ({imagemagick_timeout}s) - file may be very complex" - # If PIL passed, treat timeout as warning rather than corruption - if not pil_failed and not pil_load_failed: + # If PIL passed (or only hit its size guard), treat timeout as + # warning rather than corruption - a huge valid image can outlast + # the size-scaled timeout before hitting ImageMagick's cache limit + if pil_size_limited or (not pil_failed and not pil_load_failed): warning_details.append(timeout_msg) scan_output.append("ImageMagick identify: TIMEOUT (treating as warning - PIL verification passed)") logger.warning(f"ImageMagick timeout for {file_path} - treating as warning since PIL passed") @@ -2209,12 +2256,18 @@ def _enhanced_corruption_check(self, file_path, file_size_gb, duration=None): enhanced_output.append(f"=== Enhanced Corruption Analysis for {file_size_gb:.2f}GB file ===") # Stage 1: Frame count verification - frame_corrupted, frame_details = self._check_frame_integrity(file_path) + frame_corrupted, frame_details, frame_warnings, frame_notes = self._check_frame_integrity(file_path) enhanced_output.append("Stage 1: Frame integrity check") if frame_corrupted: is_corrupted = True corruption_details.extend(frame_details) enhanced_output.append(f" Result: FAILED - {'; '.join(frame_details)}") + elif frame_warnings: + warning_details.extend([f"Stage 1: {detail}" for detail in frame_warnings]) + enhanced_output.append(f" Result: WARNING - {'; '.join(frame_warnings)}") + elif frame_notes: + # Never claim PASSED for a check that did not measure the file + enhanced_output.append(f" Result: INCOMPLETE - {'; '.join(frame_notes)}") else: enhanced_output.append(" Result: PASSED") @@ -2273,59 +2326,112 @@ def _enhanced_corruption_check(self, file_path, file_size_gb, duration=None): enhanced_output.append(f"=== Enhanced Analysis Complete: {'CORRUPTED' if is_corrupted else 'CLEAN'} ===") return is_corrupted, corruption_details, enhanced_output, warning_details, has_notes + def _probe_stream_counts(self, file_path, count_flag, count_key, timeout): + """One ffprobe pass; returns (framerate, count, duration) or None if unavailable. + + Uses key=value output because ffprobe orders csv columns by its own + section layout, not the requested order (the old csv parser expected a + column count ffmpeg 8 never emits, so the check silently never ran). + The video stream's own duration is preferred (a container can carry + audio longer than the video track); the format duration is only the + fallback for Matroska streams that report duration=N/A. avg_frame_rate + is used instead of r_frame_rate so variable-frame-rate content does + not overshoot the expected count. + """ + result = safe_subprocess_run([ + 'ffprobe', + '-show_entries', f'stream=avg_frame_rate,duration,{count_key}:format=duration', + '-select_streams', 'v:0', + count_flag, + '-of', 'default=noprint_wrappers=1', + '-v', 'quiet', + ensure_cli_safe_path(file_path) + ], capture_output=True, text=True, timeout=timeout) + if result.returncode != 0 or not result.stdout.strip(): + return None + values = {} + for line in result.stdout.strip().splitlines(): + key, sep, value = line.partition('=') + if sep and value and value != 'N/A' and key not in values: + values[key] = value + try: + framerate_str = values['avg_frame_rate'] + if '/' in framerate_str: + num, den = map(float, framerate_str.split('/')) + framerate = num / den if den != 0 else 0.0 + else: + framerate = float(framerate_str) + count = int(values[count_key]) + duration = float(values['duration']) + except (KeyError, ValueError): + return None + if framerate <= 0 or duration <= 0: + return None + return framerate, count, duration + + @staticmethod + def _frame_mismatch(framerate, count, duration): + """Return (expected, diff, diff_percent) for a counted stream""" + expected = int(framerate * duration) + diff = abs(expected - count) + return expected, diff, (diff / expected * 100) if expected > 0 else 0 + def _check_frame_integrity(self, file_path): - """Verify frame count matches expected count based on duration and framerate""" + """Compare decodable frame count against container metadata expectations. + + Returns (is_corrupted, corruption_details, warning_details). A confirmed + mismatch is a warning, never a corruption verdict: container framerate + lies on sparse-video and variable-frame-rate files (a 240s QuickTime + with 244 real frames declares 25fps), and real decode damage is caught + by the err_detect deep-decode stage. Only an ffprobe process crash + marks corruption here. + """ corruption_details = [] + warning_details = [] + info_notes = [] is_corrupted = False - + try: logger.info(f"Checking frame integrity for {file_path}") - result = safe_subprocess_run([ - 'ffprobe', - '-show_entries', 'stream=r_frame_rate,nb_read_frames,duration', - '-select_streams', 'v:0', - '-count_frames', - '-of', 'csv=p=0', - '-v', 'quiet', - file_path - ], capture_output=True, text=True, timeout=120) - - if result.returncode == 0 and result.stdout.strip(): - lines = result.stdout.strip().split('\n') - if lines: - # Parse: stream,framerate,frame_count,duration - parts = lines[0].split(',') - if len(parts) >= 4: - framerate_str = parts[1] - frame_count_str = parts[2] - duration_str = parts[3] - - if framerate_str and frame_count_str and duration_str: - # Calculate expected vs actual frames - if '/' in framerate_str: - num, den = map(float, framerate_str.split('/')) - framerate = num / den if den != 0 else 0 - else: - framerate = float(framerate_str) - - actual_frames = int(frame_count_str) if frame_count_str.isdigit() else 0 - duration = float(duration_str) - expected_frames = int(framerate * duration) - - frame_diff = abs(expected_frames - actual_frames) - frame_diff_percent = (frame_diff / expected_frames * 100) if expected_frames > 0 else 0 - - logger.info(f"Frame analysis: Expected {expected_frames}, Found {actual_frames}, Diff: {frame_diff} ({frame_diff_percent:.1f}%)") - - # Consider significant frame loss as corruption (>5% missing) - if frame_diff_percent > 5.0: - corruption_details.append(f"Significant frame loss: {frame_diff} frames missing ({frame_diff_percent:.1f}%)") - is_corrupted = True - elif frame_diff_percent > 1.0: - corruption_details.append(f"Minor frame inconsistency: {frame_diff} frames ({frame_diff_percent:.1f}%)") - + # Cheap pass first: -count_packets does no decode. The full + # -count_frames decode is sequential and can take minutes on large + # files, so it only runs to confirm an ambiguous packet count. + # (Header metadata such as nb_frames cannot stand in for this pass: + # a truncated mdat with an intact moov still claims the full sample + # count, so only counting what is actually present has signal.) + packets = self._probe_stream_counts( + file_path, '-count_packets', 'nb_read_packets', timeout=120) + if packets: + framerate, packet_count, duration = packets + expected, diff, diff_percent = self._frame_mismatch(framerate, packet_count, duration) + logger.info(f"Frame analysis (packets): expected {expected}, " + f"found {packet_count}, diff {diff} ({diff_percent:.1f}%)") + if diff_percent > 5.0: + # Packets and frames can legitimately differ; confirm with a decode + decoded = self._probe_stream_counts( + file_path, '-count_frames', 'nb_read_frames', timeout=120) + if decoded: + framerate, frame_count, duration = decoded + expected, diff, diff_percent = self._frame_mismatch(framerate, frame_count, duration) + logger.info(f"Frame analysis (decoded): expected {expected}, " + f"found {frame_count}, diff {diff} ({diff_percent:.1f}%)") + if diff_percent > 5.0: + warning_details.append( + f"Frame count differs from container metadata by {diff} frames " + f"({diff_percent:.1f}%) - possible missing frames or " + f"sparse/variable frame rate content") + else: + # Heavily damaged files are exactly where the confirm + # decode errors out; the packet evidence must not + # vanish with it + warning_details.append( + f"Packet count differs from container metadata by {diff} frames " + f"({diff_percent:.1f}%) and decode confirmation failed") + except subprocess.TimeoutExpired: - corruption_details.append("Frame integrity check timeout") + # Operational outcome, not a file signal (Stage 2's pattern): + # surfaced as INCOMPLETE, never as warning status + info_notes.append("Frame integrity check timed out; result inconclusive") except OSError as e: # OSError includes SIGBUS and other memory-related errors logger.error(f"FFprobe process crashed with OS error for {file_path}: {str(e)}") @@ -2333,8 +2439,8 @@ def _check_frame_integrity(self, file_path): is_corrupted = True except Exception as e: logger.debug(f"Frame integrity check error: {str(e)}") - - return is_corrupted, corruption_details + + return is_corrupted, corruption_details, warning_details, info_notes def _check_temporal_outliers(self, file_path, duration=None): """Detect temporal outliers that indicate visual corruption using signalstats diff --git a/pixelprobe/services/scan_executor.py b/pixelprobe/services/scan_executor.py deleted file mode 100644 index 6509a6b..0000000 --- a/pixelprobe/services/scan_executor.py +++ /dev/null @@ -1,276 +0,0 @@ -""" -Unified Scan Executor - Implements DRY principles for scan operations -Part of P2 implementation from audit plan -""" - -import logging -import os -from typing import List, Dict, Optional, Callable, Any -from concurrent.futures import ThreadPoolExecutor, as_completed -from datetime import datetime, timezone -import threading - -logger = logging.getLogger(__name__) - - -class ScanExecutor: - """Generic scan executor pattern to eliminate code duplication across scan methods""" - - def __init__(self, scan_type: str, batch_size: int = 100, max_workers: int = None): - """ - Initialize the scan executor - - Args: - scan_type: Type of scan (full, parallel, pending, file_changes, orphan) - batch_size: Number of items to process in each batch - max_workers: Maximum number of parallel workers - """ - self.scan_type = scan_type - self.batch_size = batch_size - self.max_workers = max_workers or int(os.environ.get('MAX_WORKERS', 10)) - self.progress_callback = None - self.cancel_event = threading.Event() - self.stats = { - 'total_items': 0, - 'processed_items': 0, - 'failed_items': 0, - 'start_time': None, - 'end_time': None - } - - def set_progress_callback(self, callback: Callable[[Dict], None]): - """Set a callback function for progress updates""" - self.progress_callback = callback - - def cancel(self): - """Cancel the ongoing scan""" - self.cancel_event.set() - logger.info(f"Scan {self.scan_type} cancellation requested") - - def _batch_items(self, items: List[Any]) -> List[List[Any]]: - """Split items into batches for processing""" - for i in range(0, len(items), self.batch_size): - if self.cancel_event.is_set(): - break - yield items[i:i + self.batch_size] - - def _process_batch(self, batch: List[Any], process_func: Callable) -> Dict: - """ - Process a batch of items - - Args: - batch: List of items to process - process_func: Function to process each item - - Returns: - Dict with processing results - """ - results = { - 'successful': 0, - 'failed': 0, - 'errors': [] - } - - for item in batch: - if self.cancel_event.is_set(): - break - - try: - process_func(item) - results['successful'] += 1 - except Exception as e: - logger.error(f"Error processing item {item}: {e}") - results['failed'] += 1 - results['errors'].append(str(e)) - - return results - - def _update_progress(self, batch_results: Dict): - """Update progress statistics and notify callback""" - self.stats['processed_items'] += batch_results['successful'] - self.stats['failed_items'] += batch_results['failed'] - - if self.progress_callback: - progress_data = { - 'scan_type': self.scan_type, - 'total': self.stats['total_items'], - 'processed': self.stats['processed_items'], - 'failed': self.stats['failed_items'], - 'percentage': (self.stats['processed_items'] / self.stats['total_items'] * 100) - if self.stats['total_items'] > 0 else 0, - 'is_cancelled': self.cancel_event.is_set() - } - self.progress_callback(progress_data) - - def execute(self, items: List[Any], process_func: Callable, parallel: bool = True) -> Dict: - """ - Execute scan on items with generic processing - - Args: - items: List of items to scan - process_func: Function to process each item - parallel: Whether to use parallel processing - - Returns: - Dict with execution statistics - """ - self.stats['total_items'] = len(items) - self.stats['start_time'] = datetime.now(timezone.utc) - - logger.info(f"Starting {self.scan_type} scan of {len(items)} items " - f"(batch_size={self.batch_size}, parallel={parallel})") - - try: - if parallel: - # Parallel execution using ThreadPoolExecutor - with ThreadPoolExecutor(max_workers=self.max_workers) as executor: - futures = [] - - for batch in self._batch_items(items): - if self.cancel_event.is_set(): - break - future = executor.submit(self._process_batch, batch, process_func) - futures.append(future) - - # Wait for all futures to complete - for future in as_completed(futures): - if self.cancel_event.is_set(): - # Cancel remaining futures - for f in futures: - f.cancel() - break - - try: - batch_results = future.result(timeout=300) - self._update_progress(batch_results) - except Exception as e: - logger.error(f"Batch processing failed: {e}") - self._update_progress({'successful': 0, 'failed': self.batch_size}) - else: - # Sequential execution - for batch in self._batch_items(items): - if self.cancel_event.is_set(): - break - batch_results = self._process_batch(batch, process_func) - self._update_progress(batch_results) - - except Exception as e: - logger.error(f"Scan execution failed: {e}") - self.stats['error'] = str(e) - - finally: - self.stats['end_time'] = datetime.now(timezone.utc) - duration = (self.stats['end_time'] - self.stats['start_time']).total_seconds() - self.stats['duration_seconds'] = duration - - if self.cancel_event.is_set(): - self.stats['status'] = 'cancelled' - logger.info(f"{self.scan_type} scan cancelled after {duration:.2f} seconds") - else: - self.stats['status'] = 'completed' - logger.info(f"{self.scan_type} scan completed in {duration:.2f} seconds") - - return self.stats - - def execute_with_phases(self, phases: List[Dict]) -> Dict: - """ - Execute scan with multiple phases (discovery, adding, scanning) - - Args: - phases: List of phase configurations with: - - name: Phase name - - items_func: Function to get items for this phase - - process_func: Function to process each item - - parallel: Whether to use parallel processing - - Returns: - Dict with execution statistics for all phases - """ - overall_stats = { - 'phases': {}, - 'total_duration': 0, - 'status': 'completed' - } - - for phase_config in phases: - if self.cancel_event.is_set(): - overall_stats['status'] = 'cancelled' - break - - phase_name = phase_config['name'] - logger.info(f"Starting phase: {phase_name}") - - # Get items for this phase - items = phase_config['items_func']() - - # Execute the phase - phase_stats = self.execute( - items=items, - process_func=phase_config['process_func'], - parallel=phase_config.get('parallel', True) - ) - - overall_stats['phases'][phase_name] = phase_stats - overall_stats['total_duration'] += phase_stats.get('duration_seconds', 0) - - # If phase failed, stop execution - if phase_stats.get('status') != 'completed': - overall_stats['status'] = phase_stats['status'] - break - - return overall_stats - - -class BatchProcessor: - """Utility class for efficient batch processing operations""" - - @staticmethod - def process_in_chunks(items: List, chunk_size: int, process_func: Callable) -> List[Any]: - """ - Process items in chunks and collect results - - Args: - items: Items to process - chunk_size: Size of each chunk - process_func: Function to process each chunk - - Returns: - List of results from processing each chunk - """ - results = [] - for i in range(0, len(items), chunk_size): - chunk = items[i:i + chunk_size] - chunk_result = process_func(chunk) - results.append(chunk_result) - return results - - @staticmethod - def parallel_map(func: Callable, items: List, max_workers: int = None) -> List[Any]: - """ - Apply a function to items in parallel - - Args: - func: Function to apply to each item - items: Items to process - max_workers: Maximum number of parallel workers - - Returns: - List of results in the same order as input items - """ - max_workers = max_workers or int(os.environ.get('MAX_WORKERS', 10)) - - with ThreadPoolExecutor(max_workers=max_workers) as executor: - # Submit all tasks - futures = {executor.submit(func, item): i for i, item in enumerate(items)} - - # Collect results in order - results = [None] * len(items) - for future in as_completed(futures): - index = futures[future] - try: - results[index] = future.result() - except Exception as e: - logger.error(f"Error processing item at index {index}: {e}") - results[index] = None - - return results \ No newline at end of file diff --git a/pixelprobe/services/scan_service.py b/pixelprobe/services/scan_service.py index f9e092b..c7074ee 100644 --- a/pixelprobe/services/scan_service.py +++ b/pixelprobe/services/scan_service.py @@ -130,6 +130,7 @@ def scan_single_file(self, file_path: str, force_rescan: bool = False, # Create scan thread def run_scan(): + checker = None # Set up Flask app context for the thread with app.app_context(): try: @@ -179,6 +180,8 @@ def run_scan(): self.update_progress(1, 1, file_path, 'error') raise finally: + if checker is not None: + checker.dispose_database_connection() # Clear thread reference to allow new scans self.current_scan_thread = None logger.debug("Single file scan thread cleaned up") @@ -258,6 +261,7 @@ def scan_files(self, file_paths: List[str], force_rescan: bool = False, # Create scan thread def run_scan(): + checker = None with app.app_context(): try: # Get fresh ScanState object in worker thread @@ -269,11 +273,12 @@ def run_scan(): excluded_paths, excluded_extensions, excluded_patterns = load_exclusions_with_patterns() checker = PixelProbe( database_path=self.database_uri, + max_workers=num_workers, # sizes the checker's DB connection pool excluded_paths=excluded_paths, excluded_extensions=excluded_extensions, excluded_patterns=excluded_patterns ) - + # Skip discovery phase - we already have the files total_files = len(valid_files) logger.info(f"Scanning {total_files} specific files") @@ -348,6 +353,8 @@ def run_scan(): db.session.commit() raise finally: + if checker is not None: + checker.dispose_database_connection() # Clear thread reference to allow new scans self.current_scan_thread = None logger.info("File scan thread cleaned up") diff --git a/pixelprobe/version.py b/pixelprobe/version.py index c20be29..49bae89 100644 --- a/pixelprobe/version.py +++ b/pixelprobe/version.py @@ -4,7 +4,7 @@ # Default version - this is the single source of truth -_DEFAULT_VERSION = '2.8.4' +_DEFAULT_VERSION = '2.8.5' # Allow override via environment variable for CI/CD, but default to the hardcoded version diff --git a/pytest.ini b/pytest.ini index 847376e..6a17914 100644 --- a/pytest.ini +++ b/pytest.ini @@ -4,4 +4,5 @@ markers = slow: marks tests as slow (deselect with '-m "not slow"') integration: marks tests as integration tests real_media: marks tests that require real media files (deselect with '-m "not real_media"') + postgres: marks tests that require a live PostgreSQL (set PIXELPROBE_TEST_POSTGRES_URI) timeout: sets a timeout for test execution \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index e09f69c..7ce5cb4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -385,65 +385,59 @@ def mock_scan_result(db): return result +@pytest.fixture(scope='session') +def real_scan_data(): + """Scan every media fixture once per session. + + Scanning all fixtures takes minutes; the previous function-scoped fixture + rescanned everything for every test and blew each test's timeout, so the + real_media suite could never complete. Files are scanned in place (the + checker has no database configured, so scans are read-only). + """ + from pixelprobe.media_checker import PixelProbe + + checker = PixelProbe() + samples_dir = os.path.join(os.path.dirname(__file__), 'fixtures', 'media_samples') + data = [] + for filename in sorted(os.listdir(samples_dir)): + if filename.startswith(('valid.', 'corrupted.')): + path = os.path.join(samples_dir, filename) + scan = checker.scan_file(path) + if scan: + data.append((path, scan)) + return data + + @pytest.fixture -def real_scan_results(db, test_data_dir): - """Scan real media files into test database""" +def real_scan_results(db, real_scan_data): + """Insert the session-cached scan results into this test's database""" from pixelprobe.models import ScanResult - from pixelprobe.media_checker import PixelProbe from datetime import datetime, timezone - - checker = PixelProbe() + results = [] - - # Scan valid files - for key, path in test_data_dir.items(): - if key.startswith('valid_') and os.path.exists(path): - scan_data = checker.scan_file(path) - if scan_data: - result = ScanResult( - file_path=path, - file_size=scan_data.get('file_size', 0), - file_type=scan_data.get('file_type', ''), - file_hash=scan_data.get('file_hash', ''), - scan_date=datetime.now(timezone.utc), - scan_status='completed', - is_corrupted=scan_data.get('is_corrupted', False), - error_message=scan_data.get('error_message'), - corruption_details=scan_data.get('corruption_details'), - scan_tool=scan_data.get('scan_tool', 'ffmpeg'), - scan_output=scan_data.get('scan_output'), - warning_details=scan_data.get('warning_details'), - marked_as_good=False - ) - db.session.add(result) - results.append(result) - - # Scan corrupted files - for key, path in test_data_dir.items(): - if key.startswith('corrupted_') and os.path.exists(path): - scan_data = checker.scan_file(path) - if scan_data: - result = ScanResult( - file_path=path, - file_size=scan_data.get('file_size', 0), - file_type=scan_data.get('file_type', ''), - file_hash=scan_data.get('file_hash', ''), - scan_date=datetime.now(timezone.utc), - scan_status='completed', - is_corrupted=scan_data.get('is_corrupted', False), - error_message=scan_data.get('error_message'), - corruption_details=scan_data.get('corruption_details'), - scan_tool=scan_data.get('scan_tool', 'ffmpeg'), - scan_output=scan_data.get('scan_output'), - warning_details=scan_data.get('warning_details'), - marked_as_good=False - ) - db.session.add(result) - results.append(result) - + for path, scan_data in real_scan_data: + result = ScanResult( + file_path=path, + file_size=scan_data.get('file_size', 0), + file_type=scan_data.get('file_type', ''), + file_hash=scan_data.get('file_hash', ''), + scan_date=datetime.now(timezone.utc), + scan_status='completed', + is_corrupted=scan_data.get('is_corrupted', False), + error_message=scan_data.get('error_message'), + corruption_details=scan_data.get('corruption_details'), + scan_tool=scan_data.get('scan_tool', 'ffmpeg'), + scan_output=scan_data.get('scan_output'), + warning_details=scan_data.get('warning_details'), + marked_as_good=False + ) + db.session.add(result) + results.append(result) + db.session.commit() return results + @pytest.fixture def mock_corrupted_result(db): """Create a mock corrupted scan result""" diff --git a/tests/fixtures/media_samples/README.md b/tests/fixtures/media_samples/README.md index 4efe71d..4e5350b 100644 --- a/tests/fixtures/media_samples/README.md +++ b/tests/fixtures/media_samples/README.md @@ -1,12 +1,14 @@ # Test Media Samples -This directory contains real media files from FFmpeg samples for testing PixelProbe's corruption detection. +This directory contains media files for testing PixelProbe's corruption +detection: real samples from the FFmpeg sample corpus plus locally +synthesized files (see "Synthesized fixtures" below). ## Valid Files (18 formats) ### Video - `valid.mp4` - Apple iTunes Video (turn-on-off.mp4) - `valid.avi` - 320x240 uncompressed AVI (dance1.avi) -- `valid.mkv` - Matroska video container +- `valid.mkv` - Matroska container, H.264 + AAC (synthesized; the original sample failed to decode under ffmpeg 8) - `valid.mov` - QuickTime movie with IMA ADPCM audio - `valid.webm` - WebM video container - `valid.hevc` - HEVC/H.265 video stream @@ -17,7 +19,7 @@ This directory contains real media files from FFmpeg samples for testing PixelPr - `valid.gif` - Animated GIF (synthetic) - `valid.bmp` - 447x335 24-bit bitmap - `valid.tiff` - 12-bit RGB TIFF image -- `valid.webp` - WebP image (Big Buck Bunny title) +- `valid.webp` - WebP image (synthesized; the original sample failed PIL/ImageMagick decode) ### Audio - `valid.flac` - 16-bit stereo FLAC (Yesterday) @@ -28,32 +30,56 @@ This directory contains real media files from FFmpeg samples for testing PixelPr - `valid.ogg` - Ogg Vorbis audio - `valid.wma` - Windows Media Audio - `valid.opus` - Opus audio codec -- `valid.aiff` - Audio Interchange File Format +- `valid.aiff` - 16-bit PCM AIFF (synthesized) ## Corrupted/Problematic Files (17 formats) ### Video Issues - `corrupted.mp4` - MP4 from ticket #5522 - `corrupted.avi` - AVI with msmpeg4 bug -- `corrupted.mkv` - Corrupted Matroska container +- `corrupted.mkv` - Scattered random damage over the synthesized valid.mkv - `corrupted.mov` - MOV with ADPCM bug - `corrupted.webm` - WebM from roundup issue ### Image Issues -- `corrupted.jpg` - Small/incomplete JPEG (authentica.jpg) -- `corrupted.png` - PNG with known bug (pngbug_001) -- `corrupted.gif` - Broken GIF (ban4[1].gif) -- `corrupted.bmp` - BMP from bug #874 (correct_rgb_image.bmp) +- `corrupted.jpg` - valid.jpg truncated at 50% +- `corrupted.png` - valid.png truncated at 50% +- `corrupted.gif` - valid.gif truncated at 50% (warning-level: GIF header issues are deliberately demoted) +- `corrupted.bmp` - valid.bmp truncated at 50% - `corrupted.tiff` - TIFF with invalid strip offset size ### Audio Issues - `corrupted.flac` - FLAC from bug #810 (milk_30sec.flac) - `corrupted.wav` - WAV with format 0x1501 -- `corrupted.mp3` - MP3 with broken first frame +- `corrupted.mp3` - valid.mp3 truncated at 30% - `corrupted.aac` - AAC with decoding errors - `corrupted.m4a` - M4A from issue #1254 - `corrupted.ogg` - Ogg Vorbis with bad loop (Lumme-Badloop) - `corrupted.wma` - Broken WMA2 file -- `corrupted.aiff` - Invalid AIFF with no common chunk +- `corrupted.aiff` - valid.aiff with randomized header (first 256 bytes) + +## Synthesized fixtures + +The originally committed `valid.3gp`, `valid.flv`, `valid.mpg`, and +`valid.wmv` were 189-byte HTML error pages from failed downloads. They and +their corrupted counterparts (plus `corrupted.mkv`, `corrupted.opus`, +`corrupted.heic`, `corrupted.heif`) are now generated locally by +`generate_corrupted_fixtures.py`. The corrupted mp3/aiff/jpg/png/gif/bmp +samples originally came from the FFmpeg bug tracker, but the bugs they +exercised were in old FFmpeg rather than in the files - modern decoders +accept all six without error - so they are also generated locally by +`generate_corrupted_fixtures.py` with deterministic damage recipes; run it +from the repository root to regenerate. `valid.3g2`, `valid.mpe`, +`valid.mpeg`, and `valid.mpv` remain symlinks to their sibling formats. + +Detection expectations: every synthesized `corrupted.*` file produces a +corruption verdict except `corrupted.mpg` - MPEG-1 decoders conceal even +heavy scattered damage and exit cleanly, so PixelProbe's only signal for +that format is the frame-count-vs-metadata warning. + +`valid.mov` is a sparse-video QuickTime sample: 244 real video frames over +240 seconds while container metadata declares 25fps. It is kept +deliberately as the regression case proving frame-count mismatches must +never produce a corruption verdict. ## Sources - Valid files: https://samples.ffmpeg.org/ diff --git a/tests/fixtures/media_samples/corrupted.3gp b/tests/fixtures/media_samples/corrupted.3gp new file mode 100644 index 0000000..ab4763f Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.3gp differ diff --git a/tests/fixtures/media_samples/corrupted.aiff b/tests/fixtures/media_samples/corrupted.aiff index c4c606e..e884583 100644 Binary files a/tests/fixtures/media_samples/corrupted.aiff and b/tests/fixtures/media_samples/corrupted.aiff differ diff --git a/tests/fixtures/media_samples/corrupted.bmp b/tests/fixtures/media_samples/corrupted.bmp index 35643c4..b52a36c 100644 Binary files a/tests/fixtures/media_samples/corrupted.bmp and b/tests/fixtures/media_samples/corrupted.bmp differ diff --git a/tests/fixtures/media_samples/corrupted.flv b/tests/fixtures/media_samples/corrupted.flv new file mode 100644 index 0000000..7602ef4 Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.flv differ diff --git a/tests/fixtures/media_samples/corrupted.gif b/tests/fixtures/media_samples/corrupted.gif index f543210..5e47b27 100644 Binary files a/tests/fixtures/media_samples/corrupted.gif and b/tests/fixtures/media_samples/corrupted.gif differ diff --git a/tests/fixtures/media_samples/corrupted.heic b/tests/fixtures/media_samples/corrupted.heic new file mode 100644 index 0000000..5116c28 Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.heic differ diff --git a/tests/fixtures/media_samples/corrupted.heif b/tests/fixtures/media_samples/corrupted.heif new file mode 100644 index 0000000..5116c28 Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.heif differ diff --git a/tests/fixtures/media_samples/corrupted.jpg b/tests/fixtures/media_samples/corrupted.jpg index f037ad4..06be6fa 100644 Binary files a/tests/fixtures/media_samples/corrupted.jpg and b/tests/fixtures/media_samples/corrupted.jpg differ diff --git a/tests/fixtures/media_samples/corrupted.mkv b/tests/fixtures/media_samples/corrupted.mkv new file mode 100644 index 0000000..03c45c7 Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.mkv differ diff --git a/tests/fixtures/media_samples/corrupted.mp3 b/tests/fixtures/media_samples/corrupted.mp3 index 63ca6e4..5b9b85f 100644 Binary files a/tests/fixtures/media_samples/corrupted.mp3 and b/tests/fixtures/media_samples/corrupted.mp3 differ diff --git a/tests/fixtures/media_samples/corrupted.mpg b/tests/fixtures/media_samples/corrupted.mpg new file mode 100644 index 0000000..f33980d Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.mpg differ diff --git a/tests/fixtures/media_samples/corrupted.opus b/tests/fixtures/media_samples/corrupted.opus new file mode 100644 index 0000000..3cf1183 Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.opus differ diff --git a/tests/fixtures/media_samples/corrupted.png b/tests/fixtures/media_samples/corrupted.png index 087dff6..df548bf 100644 Binary files a/tests/fixtures/media_samples/corrupted.png and b/tests/fixtures/media_samples/corrupted.png differ diff --git a/tests/fixtures/media_samples/corrupted.wmv b/tests/fixtures/media_samples/corrupted.wmv new file mode 100644 index 0000000..831335c Binary files /dev/null and b/tests/fixtures/media_samples/corrupted.wmv differ diff --git a/tests/fixtures/media_samples/generate_corrupted_fixtures.py b/tests/fixtures/media_samples/generate_corrupted_fixtures.py new file mode 100644 index 0000000..15b540e --- /dev/null +++ b/tests/fixtures/media_samples/generate_corrupted_fixtures.py @@ -0,0 +1,126 @@ +"""Regenerate the synthesized fixtures in this directory. + +Valid fixtures for formats whose original downloads were broken (the +committed files were HTML error pages) are produced with ffmpeg/PIL; their +corrupted counterparts are derived from them with deterministic damage +recipes tuned so the scanner genuinely detects each one. + +The corrupted.mp3/aiff/jpg/png/gif/bmp samples originally came from the +FFmpeg bug tracker, but the bugs they exercised were in old FFmpeg, not in +the files: modern decoders accept all six without a single error, so they +could never fail a detection test. They are likewise replaced with derived +damage from the valid samples. + +Detection expectations per format (verified against ffmpeg 6/8 and +ImageMagick 6/7): +- mkv, opus, wmv, 3gp, flv, heic, heif, mp3, aiff, jpg, png, bmp: + corruption verdict +- mpg: warning verdict only. MPEG-1 decoders conceal even heavy scattered + damage and exit cleanly, so the frame-count-vs-metadata warning is the + only signal PixelProbe can produce for this format. +- gif: warning verdict only. GIF header issues are deliberately demoted to + warnings (historical false positives on playable files), and GIF pixel + damage decodes without errors. + +Run from the repository root: + python tests/fixtures/media_samples/generate_corrupted_fixtures.py +""" + +import os +import random +import subprocess + +from PIL import Image + +SAMPLES_DIR = os.path.dirname(os.path.abspath(__file__)) + + +def _path(name): + return os.path.join(SAMPLES_DIR, name) + + +def generate_valid_fixtures(): + """Synthesize small genuine media files for formats with broken samples.""" + ffmpeg = ['ffmpeg', '-v', 'error', '-y'] + testsrc = 'testsrc=duration=2:size=320x240:rate={rate}' + commands = [ + ffmpeg + ['-f', 'lavfi', '-i', 'testsrc=duration=2:size=176x144:rate=15', + '-f', 'lavfi', '-i', 'sine=frequency=440:duration=2', + '-c:v', 'h263', '-c:a', 'aac', '-shortest', _path('valid.3gp')], + ffmpeg + ['-f', 'lavfi', '-i', testsrc.format(rate=15), + '-c:v', 'flv1', '-an', _path('valid.flv')], + ffmpeg + ['-f', 'lavfi', '-i', testsrc.format(rate=25), + '-c:v', 'mpeg1video', '-b:v', '800k', '-an', _path('valid.mpg')], + ffmpeg + ['-f', 'lavfi', '-i', testsrc.format(rate=15), + '-c:v', 'wmv2', '-an', _path('valid.wmv')], + ffmpeg + ['-f', 'lavfi', '-i', testsrc.format(rate=25), + '-f', 'lavfi', '-i', 'sine=frequency=440:duration=2', + '-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-c:a', 'aac', + '-shortest', _path('valid.mkv')], + ffmpeg + ['-f', 'lavfi', '-i', 'sine=frequency=440:duration=2', + '-c:a', 'pcm_s16be', _path('valid.aiff')], + ] + for cmd in commands: + subprocess.run(cmd, check=True) + Image.new('RGB', (320, 240), (60, 120, 180)).save(_path('valid.webp'), quality=80) + + +def corrupt_range(src, dest, offset, span, seed=42): + """Overwrite [offset, offset+span) with deterministic pseudo-random bytes.""" + rng = random.Random(seed) + data = bytearray(open(src, 'rb').read()) + for i in range(offset, min(offset + span, len(data))): + data[i] = rng.randrange(256) + open(dest, 'wb').write(data) + + +def corrupt_fraction(src, dest, offset_frac, span_frac, seed=42): + size = os.path.getsize(src) + corrupt_range(src, dest, int(size * offset_frac), max(1, int(size * span_frac)), seed) + + +def truncate(src, dest, fraction): + data = open(src, 'rb').read() + open(dest, 'wb').write(data[:int(len(data) * fraction)]) + + +def scattered_bytes(data, stride, span, seed=42, skip_head=8192): + """Damage `span` bytes every `stride` bytes, leaving the header intact.""" + rng = random.Random(seed) + out = bytearray(data) + pos = skip_head + while pos < len(out): + for i in range(pos, min(pos + span, len(out))): + out[i] = rng.randrange(256) + pos += stride + return bytes(out) + + +def scattered(src, dest, stride, span, seed=42, skip_head=8192): + open(dest, 'wb').write(scattered_bytes(open(src, 'rb').read(), stride, span, seed, skip_head)) + + +def generate_corrupted_fixtures(): + # Recipes tuned until the scanner reports each file (see module docstring) + corrupt_fraction(_path('valid.mkv'), _path('corrupted.mkv'), 0.05, 0.80) + corrupt_range(_path('valid.opus'), _path('corrupted.opus'), 0, 256) + corrupt_range(_path('valid.wmv'), _path('corrupted.wmv'), 0, 256) + truncate(_path('valid.heic'), _path('corrupted.heic'), 0.6) + truncate(_path('valid.heif'), _path('corrupted.heif'), 0.6) + truncate(_path('valid.3gp'), _path('corrupted.3gp'), 0.5) + truncate(_path('valid.flv'), _path('corrupted.flv'), 0.15) + # MPEG-1 conceals damage; this still only yields a frame-count warning + scattered(_path('valid.mpg'), _path('corrupted.mpg'), stride=4096, span=2048) + truncate(_path('valid.mp3'), _path('corrupted.mp3'), 0.3) + corrupt_range(_path('valid.aiff'), _path('corrupted.aiff'), 0, 256) + truncate(_path('valid.jpg'), _path('corrupted.jpg'), 0.5) + truncate(_path('valid.png'), _path('corrupted.png'), 0.5) + truncate(_path('valid.bmp'), _path('corrupted.bmp'), 0.5) + # GIF header issues are deliberately demoted; truncation yields a warning + truncate(_path('valid.gif'), _path('corrupted.gif'), 0.5) + + +if __name__ == '__main__': + generate_valid_fixtures() + generate_corrupted_fixtures() + print('Fixtures regenerated in', SAMPLES_DIR) diff --git a/tests/fixtures/media_samples/valid.3gp b/tests/fixtures/media_samples/valid.3gp index 9a31a28..7a11b2a 100644 Binary files a/tests/fixtures/media_samples/valid.3gp and b/tests/fixtures/media_samples/valid.3gp differ diff --git a/tests/fixtures/media_samples/valid.aiff b/tests/fixtures/media_samples/valid.aiff new file mode 100644 index 0000000..df6dd76 Binary files /dev/null and b/tests/fixtures/media_samples/valid.aiff differ diff --git a/tests/fixtures/media_samples/valid.flv b/tests/fixtures/media_samples/valid.flv index 9a31a28..595c75b 100644 Binary files a/tests/fixtures/media_samples/valid.flv and b/tests/fixtures/media_samples/valid.flv differ diff --git a/tests/fixtures/media_samples/valid.mkv b/tests/fixtures/media_samples/valid.mkv index aac3550..a3c7953 100644 Binary files a/tests/fixtures/media_samples/valid.mkv and b/tests/fixtures/media_samples/valid.mkv differ diff --git a/tests/fixtures/media_samples/valid.mpg b/tests/fixtures/media_samples/valid.mpg index 9a31a28..6c7f766 100644 Binary files a/tests/fixtures/media_samples/valid.mpg and b/tests/fixtures/media_samples/valid.mpg differ diff --git a/tests/fixtures/media_samples/valid.webp b/tests/fixtures/media_samples/valid.webp index 7ff72a1..eb78040 100644 Binary files a/tests/fixtures/media_samples/valid.webp and b/tests/fixtures/media_samples/valid.webp differ diff --git a/tests/fixtures/media_samples/valid.wmv b/tests/fixtures/media_samples/valid.wmv index 9a31a28..3c56b2f 100644 Binary files a/tests/fixtures/media_samples/valid.wmv and b/tests/fixtures/media_samples/valid.wmv differ diff --git a/tests/integration/test_scan_launch.py b/tests/integration/test_scan_launch.py index d3e3a34..8fa03ff 100644 --- a/tests/integration/test_scan_launch.py +++ b/tests/integration/test_scan_launch.py @@ -89,3 +89,31 @@ def test_scan_parallel_v2_response_omits_directories(self, authenticated_client, 'pixelprobe.api.scan_routes_parallel') assert response.status_code == 200 assert 'directories' not in response.get_json() + + +class TestNumWorkersCap: + + def test_api_num_workers_is_capped(self, authenticated_client, app, db, + monkeypatch, tmp_path): + """Caller-supplied num_workers sizes thread pools and the checker's + DB connection pool, so the route must clamp it to MAX_WORKERS.""" + with app.app_context(): + captured = {} + + def fake_scan_files(file_paths, force_rescan=False, num_workers=1, **kwargs): + captured['num_workers'] = num_workers + return {'status': 'started', 'num_workers': num_workers} + + monkeypatch.setattr(app.scan_service, 'scan_files', fake_scan_files) + monkeypatch.setattr('pixelprobe.api.scan_routes.check_celery_available', + lambda: False) + media = tmp_path / 'clip.mp4' + media.write_bytes(b'\x00' * 128) + + response = authenticated_client.post('/api/scan-files-parallel', json={ + 'file_paths': [str(media)], + 'num_workers': 999, + }) + assert response.status_code == 200 + assert captured['num_workers'] <= app.config.get('MAX_WORKERS', 10) + assert captured['num_workers'] >= 1 diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 1934b54..d20ddbd 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -302,33 +302,3 @@ def update_state(): # Check for inconsistencies assert len(inconsistencies) == 0, f"State inconsistencies detected: {inconsistencies}" - - def test_file_discovery_deduplication(self, app): - """Test that file discovery properly deduplicates across parallel workers""" - from pixelprobe.services.scan_executor import BatchProcessor - - test_files = [f'/test/file_{i}.mp4' for i in range(100)] - - # Simulate parallel discovery with overlapping results - def discover_files(start, end): - # Intentionally create overlaps - return test_files[max(0, start-5):min(100, end+5)] - - # Run parallel discovery - results = BatchProcessor.parallel_map( - lambda r: discover_files(r[0], r[1]), - [(0, 25), (20, 50), (45, 75), (70, 100)], - max_workers=4 - ) - - # Flatten and deduplicate - all_files = [] - for batch in results: - if batch: - all_files.extend(batch) - - unique_files = list(set(all_files)) - - # Verify deduplication works - assert len(unique_files) == len(test_files), \ - f"Expected {len(test_files)} unique files, got {len(unique_files)}" \ No newline at end of file diff --git a/tests/test_media_checker.py b/tests/test_media_checker.py index 7915088..db6941c 100644 --- a/tests/test_media_checker.py +++ b/tests/test_media_checker.py @@ -864,7 +864,7 @@ def test_stage2_warning_routed_to_warnings(self, mock_run, mock_duration): mock_run.return_value = self._mock_result([(0.5, 0.0)] * 200) checker = PixelProbe() - with patch.object(checker, '_check_frame_integrity', return_value=(False, [])), \ + with patch.object(checker, '_check_frame_integrity', return_value=(False, [], [], [])), \ patch.object(checker, '_check_strict_error_detection', return_value=(False, [])): is_corrupted, details, output, warnings, has_notes = checker._enhanced_corruption_check( '/fake/video.mkv', file_size_gb=2.0 @@ -886,7 +886,7 @@ def test_stage2_timeout_notes_stay_out_of_warnings(self, mock_run, mock_duration ] checker = PixelProbe() - with patch.object(checker, '_check_frame_integrity', return_value=(False, [])), \ + with patch.object(checker, '_check_frame_integrity', return_value=(False, [], [], [])), \ patch.object(checker, '_check_strict_error_detection', return_value=(False, [])): is_corrupted, details, output, warnings, has_notes = checker._enhanced_corruption_check( '/fake/video.mkv', file_size_gb=2.0 @@ -1018,3 +1018,138 @@ def test_opus_parse_error_flood_stays_corrupted(self): assert is_corrupted is True assert any('Opus' in d for d in corruption_details) + + +class TestWorkerDatabasePool: + """Worker DB engine must not share one raw connection across scan threads""" + + def test_postgres_engine_uses_queuepool_sized_to_workers(self): + from sqlalchemy.pool import QueuePool + # create_engine connects lazily, so a bogus URI is safe here + checker = PixelProbe(database_path='postgresql://u:p@localhost:5/db', max_workers=6) + assert checker._db_engine is not None + assert isinstance(checker._db_engine.pool, QueuePool) + assert checker._db_engine.pool.size() == 6 + + def test_driver_qualified_postgres_url_not_treated_as_sqlite(self): + # postgresql+psycopg2:// URLs pass app.py's startswith('postgresql') + # check and must not fall into the SQLite branch, whose + # check_same_thread connect arg psycopg2 rejects + from sqlalchemy.pool import QueuePool + checker = PixelProbe(database_path='postgresql+psycopg2://u:p@localhost:5/db', max_workers=3) + assert checker._db_engine is not None + assert isinstance(checker._db_engine.pool, QueuePool) + + +class TestFrameIntegrityPacketFirst: + """Stage 1 frame check: cheap -count_packets pass before full-decode -count_frames""" + + def _proc(self, framerate, count_key, count, duration, rc=0): + m = MagicMock() + m.returncode = rc + m.stdout = f"avg_frame_rate={framerate}\n{count_key}={count}\nduration={duration}\n" + m.stderr = '' + return m + + def test_packet_count_match_skips_frame_decode(self): + checker = PixelProbe() + with patch('pixelprobe.media_checker.safe_subprocess_run') as run: + # 25fps * 10s = 250 expected; 250 packets reported + run.return_value = self._proc('25/1', 'nb_read_packets', 250, '10.000000') + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mp4') + assert is_corrupted is False + assert details == [] + assert warnings == [] + assert run.call_count == 1 + assert '-count_packets' in run.call_args_list[0][0][0] + + def test_small_packet_diff_skips_decode(self): + # A 1-5% packet diff can never produce the >5% warning, so the + # expensive confirm decode must not run for it + checker = PixelProbe() + with patch('pixelprobe.media_checker.safe_subprocess_run') as run: + run.return_value = self._proc('25/1', 'nb_read_packets', 240, '10.000000') # 4% diff + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mp4') + assert is_corrupted is False + assert warnings == [] + assert run.call_count == 1 + + def test_packet_mismatch_falls_back_to_frame_count(self): + checker = PixelProbe() + with patch('pixelprobe.media_checker.safe_subprocess_run') as run: + run.side_effect = [ + self._proc('25/1', 'nb_read_packets', 200, '10.000000'), # 20% short: ambiguous + self._proc('25/1', 'nb_read_frames', 248, '10.000000'), # decode: 0.8% diff, fine + ] + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mp4') + assert is_corrupted is False + assert details == [] + assert warnings == [] + assert run.call_count == 2 + assert '-count_frames' in run.call_args_list[1][0][0] + + def test_confirmed_frame_mismatch_is_warning_not_corruption(self): + # Container metadata lies on sparse-video/VFR files, so even a + # decode-confirmed mismatch must never produce a corruption verdict + checker = PixelProbe() + with patch('pixelprobe.media_checker.safe_subprocess_run') as run: + run.side_effect = [ + self._proc('25/1', 'nb_read_packets', 200, '10.000000'), + self._proc('25/1', 'nb_read_frames', 200, '10.000000'), # decode confirms 20% diff + ] + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mp4') + assert is_corrupted is False + assert details == [] + assert any('frame count differs' in w.lower() for w in warnings) + + def test_confirm_probe_failure_degrades_to_packet_warning(self): + # Heavily damaged files are where -count_frames errors out; the + # packet-pass evidence must survive as a warning, not vanish + checker = PixelProbe() + with patch('pixelprobe.media_checker.safe_subprocess_run') as run: + run.side_effect = [ + self._proc('25/1', 'nb_read_packets', 200, '10.000000'), # 20% short + self._proc('25/1', 'nb_read_frames', 0, '10.000000', rc=1), + ] + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mp4') + assert is_corrupted is False + assert any('decode confirmation failed' in w for w in warnings) + + def test_stream_duration_preferred_over_format_duration(self): + # A container carrying audio longer than its video track must be + # measured against the video stream's own duration + checker = PixelProbe() + m = MagicMock() + m.returncode = 0 + m.stdout = ("avg_frame_rate=25/1\nduration=10.000000\n" + "nb_read_packets=250\nduration=120.000000\n") + m.stderr = '' + with patch('pixelprobe.media_checker.safe_subprocess_run', return_value=m) as run: + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mp4') + assert is_corrupted is False + assert warnings == [] + assert run.call_count == 1 # 250 packets vs 25fps*10s: no decode needed + + def test_timeout_is_inconclusive_note_not_pass(self): + import subprocess as sp + checker = PixelProbe() + with patch('pixelprobe.media_checker.safe_subprocess_run', + side_effect=sp.TimeoutExpired(cmd='ffprobe', timeout=120)): + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mkv') + assert is_corrupted is False + assert warnings == [] + assert any('timed out' in n for n in notes) + + def test_unavailable_metadata_skips_check(self): + checker = PixelProbe() + with patch('pixelprobe.media_checker.safe_subprocess_run') as run: + m = MagicMock() + m.returncode = 0 + m.stdout = "avg_frame_rate=N/A\nnb_read_packets=200\nduration=N/A\n" + m.stderr = '' + run.return_value = m + is_corrupted, details, warnings, notes = checker._check_frame_integrity('/fake.mkv') + assert is_corrupted is False + assert details == [] + assert warnings == [] + assert run.call_count == 1 diff --git a/tests/test_performance.py b/tests/test_performance.py index 766ff31..810b743 100644 --- a/tests/test_performance.py +++ b/tests/test_performance.py @@ -106,172 +106,6 @@ def test_large_file_discovery_performance(self, app, performance_monitor): print(f" Memory: +{results['memory_increase_mb']:.2f}MB") print(f" CPU: {results['avg_cpu_percent']:.1f}%") - def test_bulk_insert_performance(self, app, db, performance_monitor): - """Test performance of bulk database inserts""" - from pixelprobe.models import ScanResult - from pixelprobe.services.scan_executor import BatchProcessor - - num_records = 1000 # Reduced for CI/CD performance - batch_size = 100 - - # Generate test data - test_data = [ - { - 'file_path': f'/test/bulk_{i}.mp4', - 'scan_status': 'pending', - 'is_corrupted': None, - 'file_size': random.randint(1000000, 100000000) - } - for i in range(num_records) - ] - - performance_monitor.start() - - # Measure bulk insert performance - start = time.time() - - def insert_batch(batch): - records = [ScanResult(**item) for item in batch] - db.session.bulk_save_objects(records) - db.session.commit() - return len(batch) - - # Process in batches - results = BatchProcessor.process_in_chunks( - test_data, - batch_size, - insert_batch - ) - - insert_time = time.time() - start - - # Sample final performance - performance_monitor.sample() - perf_results = performance_monitor.get_results() - - # Performance assertions (relaxed for CI/CD) - assert insert_time < 30.0, f"Bulk insert took {insert_time:.2f}s, expected < 30s" - assert sum(results) == num_records, f"Not all records inserted" - - # Report performance - print(f"\nBulk Insert Performance:") - print(f" Records: {num_records}") - print(f" Time: {insert_time:.2f}s") - print(f" Rate: {num_records/insert_time:.0f} records/second") - print(f" Memory: +{perf_results['memory_increase_mb']:.2f}MB") - - def test_scan_memory_usage_over_time(self, app, db, performance_monitor): - """Test memory usage doesn't grow unbounded during long scans""" - from pixelprobe.services.scan_executor import ScanExecutor - - # Create test files in database - from pixelprobe.models import ScanResult - - num_files = 1000 - for i in range(num_files): - file = ScanResult( - file_path=f'/test/memory_{i}.mp4', - scan_status='pending', - is_corrupted=None - ) - db.session.add(file) - - db.session.commit() - - performance_monitor.start() - memory_samples = [] - - # Mock file checking function - def mock_check_file(file_path): - time.sleep(0.001) # Simulate processing - return {'corrupted': False} - - # Create executor - executor = ScanExecutor('test', batch_size=50) - - # Set up progress callback to monitor memory - def progress_callback(data): - current_memory = performance_monitor.process.memory_info().rss / 1024 / 1024 - memory_samples.append(current_memory) - performance_monitor.sample() - - executor.set_progress_callback(progress_callback) - - # Get files to scan - files = ScanResult.query.filter_by(scan_status='pending').all() - file_paths = [f.file_path for f in files] - - # Execute scan with memory monitoring - with patch('pixelprobe.media_checker.PixelProbe.scan_file', mock_check_file): - stats = executor.execute(file_paths, mock_check_file, parallel=True) - - results = performance_monitor.get_results() - - # Calculate memory growth - if memory_samples: - memory_growth = max(memory_samples) - min(memory_samples) - - # Assert memory doesn't grow excessively - assert memory_growth < 50, \ - f"Memory grew by {memory_growth:.2f}MB during scan, expected < 50MB" - - # Report results - print(f"\nMemory Usage During Scan:") - print(f" Files processed: {stats['processed_items']}") - print(f" Duration: {results['duration']:.2f}s") - print(f" Memory growth: {memory_growth:.2f}MB") - print(f" Peak memory: {results['peak_memory_mb']:.2f}MB") - - def test_parallel_vs_sequential_performance(self, app, performance_monitor): - """Compare performance of parallel vs sequential processing""" - from pixelprobe.services.scan_executor import ScanExecutor - - num_items = 100 # Reduced for CI/CD - test_items = list(range(num_items)) - - # Mock processing function - def process_item(item): - # Simulate CPU-bound work - result = sum(i ** 2 for i in range(1000)) - return result - - # Test sequential processing - executor_seq = ScanExecutor('sequential', batch_size=50) - performance_monitor.start() - - start_seq = time.time() - stats_seq = executor_seq.execute(test_items, process_item, parallel=False) - time_seq = time.time() - start_seq - - perf_seq = performance_monitor.get_results() - - # Test parallel processing - executor_par = ScanExecutor('parallel', batch_size=50) - performance_monitor.start() - - start_par = time.time() - stats_par = executor_par.execute(test_items, process_item, parallel=True) - time_par = time.time() - start_par - - perf_par = performance_monitor.get_results() - - # Calculate speedup - speedup = time_seq / time_par if time_par > 0 else 0 - - # Report comparison - print(f"\nParallel vs Sequential Performance:") - print(f" Items: {num_items}") - print(f" Sequential: {time_seq:.2f}s") - print(f" Parallel: {time_par:.2f}s") - print(f" Speedup: {speedup:.2f}x") - print(f" CPU (seq): {perf_seq['avg_cpu_percent']:.1f}%") - print(f" CPU (par): {perf_par['avg_cpu_percent']:.1f}%") - - # Assert parallel is reasonably fast (may not always be faster for small datasets) - # Allow some variance as parallel overhead can make it slower on small datasets - # Lowered threshold to 0.75x to account for timing variance in CI environments - assert speedup > 0.75, f"Expected speedup > 0.75x, got {speedup:.2f}x" - def test_api_response_time(self, app, db): """Test API endpoint response times""" from pixelprobe.models import ScanResult @@ -335,50 +169,6 @@ def test_api_response_time(self, app, db): for endpoint, avg_time in response_times.items(): print(f" {endpoint}: {avg_time*1000:.1f}ms") - def test_scan_throughput(self, app): - """Test maximum scan throughput (files/second)""" - from pixelprobe.services.scan_executor import BatchProcessor - - # Create mock file checker - files_checked = [] - - def mock_check(file_path): - files_checked.append(file_path) - # Simulate minimal processing - time.sleep(0.0001) - return {'corrupted': False} - - # Test with different batch sizes - batch_sizes = [10, 50, 100, 500] - throughputs = {} - - for batch_size in batch_sizes: - files_checked.clear() - test_files = [f'/test/throughput_{i}.mp4' for i in range(1000)] - - start = time.time() - - # Process files - BatchProcessor.process_in_chunks( - test_files, - batch_size, - lambda batch: [mock_check(f) for f in batch] - ) - - elapsed = time.time() - start - throughput = len(files_checked) / elapsed if elapsed > 0 else 0 - throughputs[batch_size] = throughput - - # Report results - print("\nScan Throughput by Batch Size:") - for size, throughput in throughputs.items(): - print(f" Batch {size}: {throughput:.0f} files/second") - - # Assert minimum throughput - max_throughput = max(throughputs.values()) - assert max_throughput > 1000, \ - f"Maximum throughput {max_throughput:.0f} files/s, expected > 1000" - @pytest.mark.slow def test_sustained_load(self, authenticated_client, db, performance_monitor): """Test system behavior under sustained load""" diff --git a/tests/test_postgres_concurrency.py b/tests/test_postgres_concurrency.py new file mode 100644 index 0000000..05c6b80 --- /dev/null +++ b/tests/test_postgres_concurrency.py @@ -0,0 +1,57 @@ +""" +Concurrency regression: parallel scan workers sharing one PixelProbe instance +must persist results safely against a real PostgreSQL backend. StaticPool +previously shared a single raw psycopg2 connection across worker threads. + +Requires PIXELPROBE_TEST_POSTGRES_URI (provided by CI's postgres service); +skipped otherwise. +""" + +import os +from concurrent.futures import ThreadPoolExecutor + +import pytest +from PIL import Image +from sqlalchemy import create_engine, text + +from pixelprobe.media_checker import PixelProbe +from pixelprobe.models import db + +POSTGRES_URI = os.environ.get('PIXELPROBE_TEST_POSTGRES_URI') + + +@pytest.mark.postgres +@pytest.mark.timeout(300) +@pytest.mark.skipif(not POSTGRES_URI, reason='PIXELPROBE_TEST_POSTGRES_URI not set') +def test_parallel_scan_file_saves_are_connection_safe(tmp_path): + engine = create_engine(POSTGRES_URI) + db.metadata.create_all(engine) + + files = [] + for i in range(50): + p = tmp_path / f'img_{i}.png' + Image.new('RGB', (32, 32), (i * 5 % 255, 100, 150)).save(str(p)) + files.append(str(p)) + + checker = PixelProbe(database_path=POSTGRES_URI, max_workers=4) + try: + with ThreadPoolExecutor(max_workers=4) as ex: + results = list(ex.map(lambda f: checker.scan_file(f, force_rescan=True), files)) + + assert len(results) == 50 + assert all(r is not None for r in results) + assert checker.failed_saves == 0, f'{checker.failed_saves} saves failed under concurrency' + + with engine.connect() as conn: + count = conn.execute( + text('SELECT count(*) FROM scan_results WHERE file_path LIKE :p'), + {'p': f'{tmp_path}%'} + ).scalar() + assert count == 50, f'expected 50 persisted rows, got {count}' + finally: + with engine.begin() as conn: + conn.execute( + text('DELETE FROM scan_results WHERE file_path LIKE :p'), + {'p': f'{tmp_path}%'} + ) + engine.dispose() diff --git a/tests/test_real_media_samples.py b/tests/test_real_media_samples.py index 5559d6e..7449398 100644 --- a/tests/test_real_media_samples.py +++ b/tests/test_real_media_samples.py @@ -2,16 +2,26 @@ Tests using real media samples from FFmpeg """ -import pytest import os +import subprocess +from unittest.mock import patch + +import pytest +from PIL import Image + +from pixelprobe.media_checker import PixelProbe from pixelprobe.models import ScanResult +# Fixture scanning is session-cached; the first test to touch it pays the +# full corpus scan (minutes on a loaded CI runner), so every test in this +# module shares one generous timeout +pytestmark = pytest.mark.timeout(900) + @pytest.mark.real_media class TestRealMediaSamples: """Test PixelProbe with real media files""" - @pytest.mark.timeout(120) # 2 minute timeout def test_valid_files_not_corrupted(self, real_scan_results): """Test that valid files are detected as not corrupted""" valid_results = [r for r in real_scan_results if 'valid' in r.file_path] @@ -26,7 +36,6 @@ def test_valid_files_not_corrupted(self, real_scan_results): if filename in ['valid.mp4', 'valid.jpg', 'valid.png', 'valid.wav']: assert not result.is_corrupted, f"{filename} should not be corrupted" - @pytest.mark.timeout(120) # 2 minute timeout def test_corrupted_files_detected(self, real_scan_results): """Test that corrupted files are properly detected""" corrupted_results = [r for r in real_scan_results if 'corrupted' in r.file_path] @@ -51,14 +60,12 @@ def test_corrupted_files_detected(self, real_scan_results): detection_rate = detected / total if total > 0 else 0 assert detection_rate >= 0.20, f"Only {detected}/{total} corrupted files detected ({detection_rate*100:.1f}%)" - @pytest.mark.timeout(60) # 1 minute timeout def test_scan_output_captured(self, real_scan_results): """Test that scan output is properly captured""" # Most scans should have some output results_with_output = [r for r in real_scan_results if r.scan_output] assert len(results_with_output) > 0, "No scan output captured" - @pytest.mark.timeout(60) # 1 minute timeout def test_file_types_detected(self, real_scan_results): """Test that file types are properly detected""" for result in real_scan_results: @@ -73,14 +80,12 @@ def test_file_types_detected(self, real_scan_results): elif filename.endswith('.jpg'): assert 'image' in result.file_type.lower() or 'jpeg' in result.file_type.lower() - @pytest.mark.timeout(60) # 1 minute timeout def test_file_hashes_generated(self, real_scan_results): """Test that file hashes are generated""" for result in real_scan_results: assert result.file_hash is not None assert len(result.file_hash) == 64 # SHA256 hash length - @pytest.mark.timeout(60) # 1 minute timeout def test_hevc_detection(self, real_scan_results): """Test HEVC file detection and warnings""" hevc_results = [r for r in real_scan_results if 'hevc' in r.file_path.lower()] @@ -92,7 +97,6 @@ def test_hevc_detection(self, real_scan_results): assert any('hevc' in str(line).lower() or 'h.265' in str(line).lower() for line in (result.scan_output if isinstance(result.scan_output, list) else [result.scan_output])) - @pytest.mark.timeout(60) # 1 minute timeout def test_scan_tool_recorded(self, real_scan_results): """Test that scan tool is recorded""" for result in real_scan_results: @@ -104,26 +108,20 @@ def test_scan_tool_recorded(self, real_scan_results): class TestCorruptionDetails: """Test specific corruption detection capabilities""" - @pytest.mark.timeout(60) # 1 minute timeout def test_mp3_broken_frame(self, real_scan_results): - """Test MP3 with broken first frame is detected""" + """Test truncated MP3 is detected""" mp3_results = [r for r in real_scan_results if 'corrupted.mp3' in r.file_path] if mp3_results: result = mp3_results[0] - # Should detect issues with broken MP3 - # Note: Some MP3s with minor corruption may still play - assert result.is_corrupted or result.error_message or result.warning_details or result.scan_status == 'completed' + assert result.is_corrupted or result.error_message or result.warning_details - @pytest.mark.timeout(60) # 1 minute timeout def test_invalid_aiff(self, real_scan_results): - """Test invalid AIFF without common chunk""" + """Test AIFF with damaged header is detected""" aiff_results = [r for r in real_scan_results if 'corrupted.aiff' in r.file_path] if aiff_results: result = aiff_results[0] - # Invalid AIFF should be detected - but some may still parse - assert result.is_corrupted or result.error_message or result.warning_details or result.scan_status == 'completed' + assert result.is_corrupted or result.error_message or result.warning_details - @pytest.mark.timeout(60) # 1 minute timeout def test_corrupted_images(self, real_scan_results): """Test corrupted image detection""" image_extensions = ['.jpg', '.png', '.gif', '.bmp'] @@ -134,7 +132,58 @@ def test_corrupted_images(self, real_scan_results): if corrupted_images: # At least one corrupted image of each type should be detected - # Note: Some corrupted images may still have valid headers - detected = any(r.is_corrupted or r.error_message or r.warning_details or r.scan_status == 'completed' + detected = any(r.is_corrupted or r.error_message or r.warning_details for r in corrupted_images) - assert detected, f"No corrupted {ext} files were processed" \ No newline at end of file + assert detected, f"No corrupted {ext} files were processed" + +@pytest.mark.real_media +class TestOversizedImages: + """Regression: images beyond tool pixel/resource limits are warnings, not corruption""" + + def _make_oversized(self, path, mode, size, **save_kwargs): + old_limit = Image.MAX_IMAGE_PIXELS + Image.MAX_IMAGE_PIXELS = None + try: + Image.new(mode, size, 128 if mode != '1' else 0).save(str(path), **save_kwargs) + finally: + Image.MAX_IMAGE_PIXELS = old_limit + + def test_oversized_valid_png_is_warning_not_corrupted(self, tmp_path): + # 400MP exceeds both Pillow's DecompressionBombError threshold + # (2x MAX_IMAGE_PIXELS, ~358MP) and ImageMagick's default 256MP area policy + big = tmp_path / 'big_valid.png' + self._make_oversized(big, 'L', (20000, 20000), compress_level=1) + + checker = PixelProbe(database_path=None) + is_corrupted, corruption_details, scan_tool, scan_output, warning_details = \ + checker._check_image_corruption(str(big)) + + assert is_corrupted is False, f"valid oversized PNG flagged corrupted: {corruption_details}" + assert warning_details, "expected a tool-limit warning" + assert any('limit' in w.lower() for w in warning_details) + + def test_oversized_image_imagemagick_timeout_is_warning(self, tmp_path): + # A huge valid image can outlast the size-scaled ImageMagick timeout + # before hitting its cache limit; with PIL's bomb guard also fired, + # that combination must stay a warning, not corruption + big = tmp_path / 'big_timeout.png' + self._make_oversized(big, 'L', (20000, 20000), compress_level=1) + + checker = PixelProbe(database_path=None) + with patch('pixelprobe.media_checker.safe_subprocess_run', + side_effect=subprocess.TimeoutExpired(cmd='magick', timeout=60)): + is_corrupted, corruption_details, _, _, warning_details = \ + checker._check_image_corruption(str(big)) + assert is_corrupted is False, corruption_details + assert warning_details + + def test_decompression_bomb_scale_image_not_corrupted(self, tmp_path): + # 625MP 1-bit image (just past both guards), tiny on disk: verdict + # stays sane without allocating multiple GB on a CI runner + bomb = tmp_path / 'bomb.png' + self._make_oversized(bomb, '1', (25000, 25000), compress_level=9) + + checker = PixelProbe(database_path=None) + is_corrupted, _, _, _, warning_details = checker._check_image_corruption(str(bomb)) + assert is_corrupted is False + assert warning_details diff --git a/tests/test_synthetic_corruption.py b/tests/test_synthetic_corruption.py new file mode 100644 index 0000000..f2ac94c --- /dev/null +++ b/tests/test_synthetic_corruption.py @@ -0,0 +1,214 @@ +""" +Synthetic corruption matrix: truncations, mid-stream damage, format confusion, +and formats without committed fixtures (SVG, PSD, progressive JPEG, animated +WebP). Fixtures are generated at test time from the committed media samples, +reproducing the real failure modes this tool exists to catch (partial writes, +disk errors, incomplete transfers) rather than codec-specific re-encodes. + +Verdict expectations were verified against the installed ffmpeg/ImageMagick/ +Pillow. Known detection limits, asserted as such rather than papered over: +- Mid-stream damage needs to hit multiple regions; a single overwritten span + in an MP4 can land entirely in tolerated data. +- Animated WebP content damage is undetectable (VP8 decodes garbage without + erroring); only structural truncation is caught. +""" + +import importlib.util +import os +import subprocess +from shutil import which + +import pytest +from PIL import Image + +from pixelprobe.media_checker import IMAGEMAGICK_BINARY, PixelProbe + +SAMPLES_DIR = os.path.join(os.path.dirname(__file__), 'fixtures', 'media_samples') + +# media_samples is not a package; load the committed fixture generator so the +# test-time damage recipe cannot drift from the committed-fixture recipe +_spec = importlib.util.spec_from_file_location( + 'generate_corrupted_fixtures', + os.path.join(SAMPLES_DIR, 'generate_corrupted_fixtures.py')) +_fixture_gen = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_fixture_gen) +scattered_bytes = _fixture_gen.scattered_bytes + +VALID_SVG = (b'' + b'') + + +def _read(name): + with open(os.path.join(SAMPLES_DIR, name), 'rb') as f: + return f.read() + + +@pytest.fixture(scope='module') +def synthetic_dir(tmp_path_factory): + """Generate every synthetic case once for the module.""" + dest = tmp_path_factory.mktemp('synthetic_media') + + def write(name, data): + path = dest / name + path.write_bytes(data) + return path + + mp4 = _read('valid.mp4') + mkv = _read('valid.mkv') + jpg = _read('valid.jpg') + png = _read('valid.png') + + paths = { + 'trunc_header_jpg': write('trunc_header.jpg', jpg[:64]), + 'trunc_header_png': write('trunc_header.png', png[:64]), + 'trunc_header_mp4': write('trunc_header.mp4', mp4[:64]), + 'trunc_header_mkv': write('trunc_header.mkv', mkv[:64]), + 'trunc_half_mp4': write('trunc_half.mp4', mp4[:len(mp4) // 2]), + 'trunc_tail_mkv': write('trunc_tail.mkv', mkv[:-4096]), + 'empty_mp4': write('empty.mp4', b''), + 'empty_jpg': write('empty.jpg', b''), + 'png_as_jpg': write('actually_png.jpg', png), + 'mdat_damaged_mp4': write('mdat_damaged.mp4', scattered_bytes(mp4, stride=8192, span=1024, seed=7)), + 'valid_svg': write('valid.svg', VALID_SVG), + 'corrupted_svg': write('corrupted.svg', VALID_SVG[:40]), + } + + prog = dest / 'progressive.jpg' + Image.new('RGB', (200, 200), (200, 50, 50)).save(str(prog), progressive=True, quality=85) + prog_bytes = prog.read_bytes() + paths['valid_progressive_jpg'] = prog + paths['trunc_progressive_jpg'] = write('trunc_progressive.jpg', + prog_bytes[:int(len(prog_bytes) * 0.7)]) + + if which(IMAGEMAGICK_BINARY): + psd = dest / 'valid.psd' + subprocess.run([IMAGEMAGICK_BINARY, os.path.join(SAMPLES_DIR, 'valid.png'), str(psd)], + check=True, capture_output=True) + paths['valid_psd'] = psd + paths['corrupted_psd'] = write('corrupted.psd', psd.read_bytes()[:200]) + + if which('ffmpeg'): + anim = dest / 'valid_anim.webp' + subprocess.run(['ffmpeg', '-v', 'error', '-f', 'lavfi', '-i', + 'testsrc=duration=2:size=64x64:rate=10', '-y', str(anim)], + check=True, capture_output=True) + anim_bytes = anim.read_bytes() + paths['valid_anim_webp'] = anim + paths['trunc_anim_webp'] = write('trunc_anim.webp', + anim_bytes[:int(len(anim_bytes) * 0.6)]) + + return paths + + +@pytest.fixture(scope='module') +def checker(): + return PixelProbe() + + +def _scan(checker, paths, key): + if key not in paths: + pytest.skip(f'{key} fixture unavailable (missing tool)') + result = checker.scan_file(str(paths[key])) + assert result is not None + return result + + +@pytest.mark.real_media +class TestTruncation: + @pytest.mark.timeout(120) + @pytest.mark.parametrize('key', ['trunc_header_jpg', 'trunc_header_png', + 'trunc_header_mp4', 'trunc_header_mkv']) + def test_header_truncation_detected(self, checker, synthetic_dir, key): + result = _scan(checker, synthetic_dir, key) + assert result['is_corrupted'], f'{key} not flagged: {result.get("corruption_details")}' + + @pytest.mark.timeout(120) + @pytest.mark.parametrize('key', ['trunc_half_mp4', 'trunc_tail_mkv']) + def test_body_truncation_flagged(self, checker, synthetic_dir, key): + # Decoders conceal clean-cut truncation of these containers; the + # frame-count-vs-metadata warning is the honest signal + result = _scan(checker, synthetic_dir, key) + assert result['is_corrupted'] or result.get('warning_details'), \ + f'{key} produced neither corruption nor warning' + + @pytest.mark.timeout(60) + @pytest.mark.parametrize('key', ['empty_mp4', 'empty_jpg']) + def test_zero_byte_files_flagged(self, checker, synthetic_dir, key): + result = _scan(checker, synthetic_dir, key) + assert result['is_corrupted'] + + +@pytest.mark.real_media +class TestMidStreamDamage: + @pytest.mark.timeout(120) + def test_scattered_mdat_damage_detected(self, checker, synthetic_dir): + # Validates the deep-decode path catches mid-stream damage, not just + # header damage (header/moov region is left intact) + result = _scan(checker, synthetic_dir, 'mdat_damaged_mp4') + assert result['is_corrupted'] + + +@pytest.mark.real_media +class TestFormatConfusion: + @pytest.mark.timeout(60) + def test_png_renamed_to_jpg_is_not_corrupted(self, checker, synthetic_dir): + # Content is a valid PNG; a wrong extension must not produce a + # corruption verdict or a crash + result = _scan(checker, synthetic_dir, 'png_as_jpg') + assert not result['is_corrupted'], result.get('corruption_details') + + +@pytest.mark.real_media +class TestUncoveredImageFormats: + @pytest.mark.timeout(60) + def test_valid_svg_clean(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'valid_svg') + assert not result['is_corrupted'], result.get('corruption_details') + + @pytest.mark.timeout(60) + def test_malformed_svg_detected(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'corrupted_svg') + assert result['is_corrupted'] + + @pytest.mark.timeout(60) + def test_valid_psd_clean(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'valid_psd') + assert not result['is_corrupted'], result.get('corruption_details') + + @pytest.mark.timeout(60) + def test_truncated_psd_detected(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'corrupted_psd') + assert result['is_corrupted'] + + @pytest.mark.timeout(60) + def test_valid_progressive_jpeg_clean(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'valid_progressive_jpg') + assert not result['is_corrupted'], result.get('corruption_details') + + @pytest.mark.timeout(60) + def test_truncated_progressive_jpeg_detected(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'trunc_progressive_jpg') + assert result['is_corrupted'] + + @pytest.mark.timeout(120) + def test_valid_animated_webp_clean(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'valid_anim_webp') + assert not result['is_corrupted'], result.get('corruption_details') + + @pytest.mark.timeout(120) + def test_truncated_animated_webp_detected(self, checker, synthetic_dir): + result = _scan(checker, synthetic_dir, 'trunc_anim_webp') + assert result['is_corrupted'] + + +@pytest.mark.real_media +class TestDiscovery: + @pytest.mark.timeout(60) + def test_symlink_loop_discovery_terminates(self, checker, tmp_path): + loop_dir = tmp_path / 'loop' / 'a' + loop_dir.mkdir(parents=True) + (loop_dir / 'self').symlink_to(tmp_path / 'loop' / 'a') + (loop_dir / 'clip.mp4').write_bytes(_read('valid.mp4')) + + files = checker.discover_media_files([str(tmp_path / 'loop')]) + assert any(f.endswith('clip.mp4') for f in files)