112719b2c5
- Redesigned frontend with Outfit/Figtree typography, coral accent palette, noise texture, glassmorphism header, and staggered animations - Split monolithic index.html into modular JS (app, api, upload, batch, results, page-viewer, utils) and extracted CSS - Fixed worker.py to generate page images for Visual Page Inspector - Added Docker Compose stack (web, worker, redis, postgres) - Added batch upload, HTML report export, rate limiting, and Redis queue - Extended test suite with checker, remediation, worker, and DB tests Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
"""
|
|
Tests for worker.py — all external dependencies mocked.
|
|
"""
|
|
|
|
import pytest
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock, mock_open
|
|
|
|
|
|
class TestProcessJob:
|
|
def test_process_job_success(self, tmp_path):
|
|
import worker
|
|
|
|
mock_checker_instance = MagicMock()
|
|
mock_checker_instance.check_all.return_value = {
|
|
"accessibility_score": 85,
|
|
"grade": "B",
|
|
"issues": [
|
|
{"severity": "WARNING", "category": "Test", "description": "x"},
|
|
{"severity": "ERROR", "category": "Test2", "description": "y"},
|
|
],
|
|
}
|
|
mock_checker_cls = MagicMock(return_value=mock_checker_instance)
|
|
|
|
original_results_dir = worker.RESULTS_DIR
|
|
worker.RESULTS_DIR = tmp_path
|
|
|
|
with patch.object(worker, "set_job_status") as mock_set, \
|
|
patch.object(worker, "update_job_status") as mock_update, \
|
|
patch.object(worker, "log_audit") as mock_audit, \
|
|
patch.dict("sys.modules", {"enterprise_pdf_checker": MagicMock(EnterprisePDFChecker=mock_checker_cls)}):
|
|
|
|
# Need to reload so the `from enterprise_pdf_checker import ...` picks up mock
|
|
import importlib
|
|
importlib.reload(worker)
|
|
worker.RESULTS_DIR = tmp_path
|
|
|
|
worker.process_job({
|
|
"job_id": "pdf_test123",
|
|
"pdf_path": "/uploads/test.pdf",
|
|
"options": {"quick_mode": True},
|
|
})
|
|
|
|
worker.RESULTS_DIR = original_results_dir
|
|
# Result JSON should have been written
|
|
assert (tmp_path / "pdf_test123.result.json").exists()
|
|
|
|
def test_process_job_failure(self, tmp_path):
|
|
import worker
|
|
|
|
mock_checker_cls = MagicMock(side_effect=Exception("PDF corrupted"))
|
|
|
|
original_results_dir = worker.RESULTS_DIR
|
|
worker.RESULTS_DIR = tmp_path
|
|
|
|
with patch.object(worker, "set_job_status") as mock_set, \
|
|
patch.object(worker, "update_job_status") as mock_update, \
|
|
patch.object(worker, "log_audit") as mock_audit, \
|
|
patch.dict("sys.modules", {"enterprise_pdf_checker": MagicMock(EnterprisePDFChecker=mock_checker_cls)}):
|
|
|
|
import importlib
|
|
importlib.reload(worker)
|
|
worker.RESULTS_DIR = tmp_path
|
|
|
|
worker.process_job({
|
|
"job_id": "pdf_fail",
|
|
"pdf_path": "/uploads/bad.pdf",
|
|
"options": {},
|
|
})
|
|
|
|
worker.RESULTS_DIR = original_results_dir
|
|
# Error log should have been written
|
|
assert (tmp_path / "pdf_fail.error.log").exists()
|
|
|
|
|
|
class TestWorkerSignalHandling:
|
|
def test_handle_signal_sets_shutdown(self):
|
|
import worker
|
|
worker.shutdown_requested = False
|
|
worker.handle_signal(15, None) # SIGTERM
|
|
assert worker.shutdown_requested is True
|
|
# Reset
|
|
worker.shutdown_requested = False
|
|
|
|
|
|
class TestWorkerMain:
|
|
@patch("worker.pop_job")
|
|
@patch("worker.process_job")
|
|
def test_main_loop_processes_job(self, mock_process, mock_pop):
|
|
import worker
|
|
|
|
# Return one job then set shutdown
|
|
call_count = [0]
|
|
def side_effect(timeout=5):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
return {"job_id": "pdf_1", "pdf_path": "/test.pdf", "options": {}}
|
|
worker.shutdown_requested = True
|
|
return None
|
|
|
|
mock_pop.side_effect = side_effect
|
|
worker.shutdown_requested = False
|
|
|
|
worker.main()
|
|
|
|
mock_process.assert_called_once()
|
|
# Reset
|
|
worker.shutdown_requested = False
|
|
|
|
@patch("worker.pop_job")
|
|
def test_main_loop_handles_empty_queue(self, mock_pop):
|
|
import worker
|
|
|
|
call_count = [0]
|
|
def side_effect(timeout=5):
|
|
call_count[0] += 1
|
|
if call_count[0] >= 2:
|
|
worker.shutdown_requested = True
|
|
return None
|
|
|
|
mock_pop.side_effect = side_effect
|
|
worker.shutdown_requested = False
|
|
|
|
worker.main()
|
|
|
|
assert call_count[0] >= 2
|
|
worker.shutdown_requested = False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|