Selenium update and bypasser enhancements, various bug fixes and tests (#375)

- Updated Selenium to 4.45.6. Includes various crash and memory leak
fixes, plus new bypasser methods
- Bypasser now uses CDP captcha solving as priority - Faster, more
efficient, no PyAutoGUI needed. Fallback to existing methods.
- Better detection and cleanup of old Selenium instances to save memory.
- Added Hardcover graphQL API header detection
- Added AA download counts in details modal
- More robust switching of internal/external bypasser, fixed settings UI
toggle behavior.
This commit is contained in:
Alex
2025-12-30 09:42:06 +00:00
committed by GitHub
parent dbe46e8e61
commit 98aada2f55
28 changed files with 3101 additions and 274 deletions
+3 -1
View File
@@ -228,4 +228,6 @@ pyrightconfig.json
# End of https://www.toptal.com/developers/gitignore/api/macos,visualstudiocode,python
/downloaded_files
/.local/
*.local.md
*.local.*
.claude/
.playwright-mcp/
+359 -144
View File
@@ -214,6 +214,67 @@ def _reset_pyautogui_display_state():
except Exception as e:
logger.warning(f"Error resetting pyautogui display state: {e}")
def _cleanup_orphan_processes() -> int:
"""Kill any orphan Chrome, ChromeDriver, Xvfb, and ffmpeg processes.
This should be called at startup before initializing the bypasser to ensure
no zombie processes from previous crashes are consuming memory or interfering.
Safety: Only runs in Docker mode to avoid killing user's browser processes
on development machines.
Returns:
Number of processes killed.
"""
# Safety: only cleanup in Docker mode to avoid killing user's browser
if not env.DOCKERMODE:
return 0
processes_to_kill = ["chrome", "chromedriver", "Xvfb", "ffmpeg"]
total_killed = 0
logger.debug("Checking for orphan processes...")
logger.log_resource_usage()
for proc_name in processes_to_kill:
try:
# Use pgrep to find processes, then kill them
result = subprocess.run(
["pgrep", "-f", proc_name],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0 and result.stdout.strip():
pids = result.stdout.strip().split('\n')
count = len(pids)
if count > 0:
logger.info(f"Found {count} orphan {proc_name} process(es), killing...")
kill_result = subprocess.run(
["pkill", "-9", "-f", proc_name],
capture_output=True,
timeout=5
)
if kill_result.returncode == 0:
total_killed += count
else:
logger.warning(f"pkill for {proc_name} returned {kill_result.returncode}, processes may not have been killed")
except subprocess.TimeoutExpired:
logger.warning(f"Timeout while checking for {proc_name} processes")
except Exception as e:
logger.debug(f"Error checking for {proc_name} processes: {e}")
if total_killed > 0:
# Give processes time to fully terminate
time.sleep(1)
logger.info(f"Cleaned up {total_killed} orphan process(es)")
logger.log_resource_usage()
else:
logger.debug("No orphan processes found")
return total_killed
def _get_page_info(sb) -> tuple[str, str, str]:
"""Extract page title, body text, and current URL safely."""
try:
@@ -310,168 +371,319 @@ def _is_bypassed(sb, escape_emojis: bool = True) -> bool:
logger.warning(f"Error checking bypass status: {e}")
return False
def _bypass_method_1(sb) -> bool:
"""Original bypass method using uc_gui_click_captcha"""
def _simulate_human_behavior(sb) -> None:
"""Simulate human-like behavior before bypass attempt."""
try:
logger.debug("Attempting bypass method 1: uc_gui_click_captcha")
sb.uc_gui_click_captcha()
time.sleep(3)
return _is_bypassed(sb)
except Exception as e:
logger.debug(f"Method 1 failed on first try: {e}")
try:
time.sleep(5)
sb.wait_for_element_visible('body', timeout=10)
sb.uc_gui_click_captcha()
time.sleep(3)
return _is_bypassed(sb)
except Exception as e2:
logger.debug(f"Method 1 failed on second try: {e2}")
try:
time.sleep(app_config.DEFAULT_SLEEP)
sb.uc_gui_click_captcha()
time.sleep(5)
return _is_bypassed(sb)
except Exception as e3:
logger.debug(f"Method 1 completely failed: {e3}")
return False
# Random short wait (human reaction time)
time.sleep(random.uniform(0.5, 1.5))
def _bypass_method_2(sb) -> bool:
"""Alternative bypass method using longer waits and manual interaction"""
try:
logger.debug("Attempting bypass method 2: wait and reload")
# Wait longer for page to load completely
time.sleep(10)
# Try refreshing the page
sb.refresh()
time.sleep(8)
# Check if bypass worked after refresh
if _is_bypassed(sb):
return True
# Try clicking on the page center (sometimes helps trigger bypass)
# Maybe scroll a bit (30% chance)
if random.random() < 0.3:
sb.scroll_down(random.randint(20, 50))
time.sleep(random.uniform(0.2, 0.5))
sb.scroll_up(random.randint(10, 30))
time.sleep(random.uniform(0.2, 0.4))
# Brief mouse jiggle via PyAutoGUI
try:
sb.click_if_visible("body", timeout=5)
time.sleep(5)
except Exception:
pass
import pyautogui
x, y = pyautogui.position()
pyautogui.moveTo(
x + random.randint(-10, 10),
y + random.randint(-10, 10),
duration=random.uniform(0.05, 0.15)
)
except Exception as e:
logger.debug(f"Mouse jiggle failed (non-critical): {e}")
except Exception as e:
logger.debug(f"Human simulation failed (non-critical): {e}")
def _bypass_method_handle_captcha(sb) -> bool:
"""Method 2: Use uc_gui_handle_captcha() - TAB+SPACEBAR approach, stealthier than click."""
try:
logger.debug("Attempting bypass: uc_gui_handle_captcha (TAB+SPACEBAR)")
_simulate_human_behavior(sb)
sb.uc_gui_handle_captcha()
time.sleep(random.uniform(3, 5))
return _is_bypassed(sb)
except Exception as e:
logger.debug(f"Method 2 failed: {e}")
logger.debug(f"uc_gui_handle_captcha failed: {e}")
return False
def _bypass_method_3(sb) -> bool:
"""Third bypass method using user-agent rotation and stealth mode"""
def _bypass_method_click_captcha(sb) -> bool:
"""Method 3: Use uc_gui_click_captcha() - direct click via PyAutoGUI."""
try:
logger.debug("Attempting bypass method 3: stealth approach")
# Wait a random amount to appear more human
wait_time = random.uniform(8, 15)
time.sleep(wait_time)
# Try to scroll the page (human-like behavior)
logger.debug("Attempting bypass: uc_gui_click_captcha (direct click)")
_simulate_human_behavior(sb)
sb.uc_gui_click_captcha()
time.sleep(random.uniform(3, 5))
if _is_bypassed(sb):
return True
# Retry once with longer wait
logger.debug("First click attempt failed, retrying...")
time.sleep(random.uniform(4, 6))
sb.uc_gui_click_captcha()
time.sleep(random.uniform(3, 5))
return _is_bypassed(sb)
except Exception as e:
logger.debug(f"uc_gui_click_captcha failed: {e}")
return False
def _bypass_method_humanlike(sb) -> bool:
"""Method 4: Human-like behavior with scroll, wait, and reload."""
try:
logger.debug("Attempting bypass: human-like interaction")
# Extended human-like wait
time.sleep(random.uniform(6, 10))
# Scroll behavior
try:
sb.scroll_to_bottom()
time.sleep(2)
time.sleep(random.uniform(1, 2))
sb.scroll_to_top()
time.sleep(3)
except Exception:
pass
# Check if this helped
time.sleep(random.uniform(2, 3))
except Exception as e:
logger.debug(f"Scroll behavior failed (non-critical): {e}")
if _is_bypassed(sb):
return True
# Try the original captcha click as last resort
try:
sb.uc_gui_click_captcha()
time.sleep(5)
except Exception:
pass
return _is_bypassed(sb)
except Exception as e:
logger.debug(f"Method 3 failed: {e}")
return False
def _bypass_ddos_guard_method_1(sb) -> bool:
"""DDOS-Guard bypass: Use SeleniumBase's captcha handling (most reliable)"""
try:
logger.debug("Attempting DDOS-Guard bypass: SeleniumBase uc_gui methods")
time.sleep(random.uniform(2, 4))
# SeleniumBase's uc_gui_click_captcha often works for DDOS-Guard too
# Try refresh
logger.debug("Trying page refresh...")
sb.refresh()
time.sleep(random.uniform(5, 8))
if _is_bypassed(sb):
return True
# Final captcha click attempt
try:
sb.uc_gui_click_captcha()
time.sleep(random.uniform(3, 5))
except Exception as e:
logger.debug(f"Final captcha click failed (non-critical): {e}")
return _is_bypassed(sb)
except Exception as e:
logger.debug(f"Human-like method failed: {e}")
return False
def _bypass_method_cdp_solve(sb) -> bool:
"""Method 5: CDP Mode with solve_captcha() - WebDriver disconnected, no PyAutoGUI.
CDP Mode completely disconnects WebDriver during interaction, making detection
much harder. The solve_captcha() method auto-detects challenge type.
"""
try:
logger.debug("Attempting bypass: CDP Mode solve_captcha")
current_url = sb.get_current_url()
# Activate CDP mode - this disconnects WebDriver
sb.activate_cdp_mode(current_url)
time.sleep(random.uniform(1, 2))
# Try CDP solve_captcha (auto-detects challenge type)
try:
sb.cdp.solve_captcha()
time.sleep(random.uniform(3, 5))
# Reconnect WebDriver to check result
sb.reconnect()
time.sleep(random.uniform(1, 2))
if _is_bypassed(sb):
return True
except Exception as e:
logger.debug(f"uc_gui_click_captcha failed: {e}")
# Fallback: Try clicking visible checkbox-like elements
checkbox_patterns = [
"//input[@type='checkbox']",
"//*[contains(@class, 'checkbox')]",
"//*[contains(@class, 'cb-')]",
]
for pattern in checkbox_patterns:
logger.debug(f"CDP solve_captcha failed: {e}")
# Make sure we reconnect on failure
try:
elements = sb.find_elements(f"xpath:{pattern}")
for elem in elements:
if elem.is_displayed():
logger.debug(f"Clicking element with pattern: {pattern}")
elem.click()
time.sleep(random.uniform(3, 5))
if _is_bypassed(sb):
return True
except Exception:
continue
sb.reconnect()
except Exception as reconnect_e:
logger.debug(f"Reconnect after CDP solve failure failed: {reconnect_e}")
return False
except Exception as e:
logger.debug(f"DDOS-Guard method 1 failed: {e}")
logger.debug(f"CDP Mode solve failed: {e}")
# Ensure WebDriver is reconnected
try:
sb.reconnect()
except Exception as reconnect_e:
logger.debug(f"Reconnect after CDP Mode solve failed: {reconnect_e}")
return False
def _bypass_ddos_guard_method_2(sb) -> bool:
"""DDOS-Guard bypass: Use pyautogui to click estimated checkbox location"""
def _bypass_method_cdp_click(sb) -> bool:
"""CDP Mode with native clicking - no PyAutoGUI dependency.
Uses sb.cdp.click() which is native CDP clicking added in SeleniumBase 4.45.6.
This doesn't require PyAutoGUI at all.
"""
try:
logger.debug("Attempting DDOS-Guard bypass: pyautogui coordinate click")
time.sleep(random.uniform(2, 4))
import pyautogui
window_size = sb.get_window_size()
width = window_size.get("width", 1920)
height = window_size.get("height", 1080)
# DDOS-Guard checkbox is typically around 35% from left, 55% from top
checkbox_x = int(width * 0.35) + random.randint(-5, 5)
checkbox_y = int(height * 0.55) + random.randint(-5, 5)
logger.debug(f"Clicking at coordinates: ({checkbox_x}, {checkbox_y})")
pyautogui.moveTo(checkbox_x, checkbox_y, duration=random.uniform(0.3, 0.7))
time.sleep(random.uniform(0.1, 0.3))
pyautogui.click()
time.sleep(random.uniform(3, 5))
logger.debug("Attempting bypass: CDP Mode native click")
current_url = sb.get_current_url()
# Activate CDP mode
sb.activate_cdp_mode(current_url)
time.sleep(random.uniform(1, 2))
# Common captcha/challenge selectors to try
selectors = [
"#turnstile-widget div", # Cloudflare Turnstile (parent above shadow-root)
"#cf-turnstile div", # Alternative CF Turnstile
"iframe[src*='challenges']", # CF challenge iframe
"input[type='checkbox']", # Generic checkbox (DDOS-Guard)
"[class*='checkbox']", # Class-based checkbox
"#challenge-running", # CF challenge indicator
]
for selector in selectors:
try:
# Check if element exists and is visible
if sb.cdp.is_element_visible(selector):
logger.debug(f"CDP clicking: {selector}")
sb.cdp.click(selector)
time.sleep(random.uniform(2, 4))
# Reconnect and check
sb.reconnect()
time.sleep(random.uniform(1, 2))
if _is_bypassed(sb):
return True
# Re-enter CDP mode for next attempt
sb.activate_cdp_mode(sb.get_current_url())
time.sleep(random.uniform(0.5, 1))
except Exception as e:
logger.debug(f"CDP click on '{selector}' failed: {e}")
continue
# Final reconnect
try:
sb.reconnect()
except Exception as e:
logger.debug(f"Final reconnect in CDP click failed: {e}")
return _is_bypassed(sb)
except ImportError:
logger.debug("pyautogui not available")
return False
except Exception as e:
logger.debug(f"DDOS-Guard method 2 failed: {e}")
logger.debug(f"CDP Mode click failed: {e}")
try:
sb.reconnect()
except Exception as reconnect_e:
logger.debug(f"Reconnect after CDP Mode click failed: {reconnect_e}")
return False
def _bypass_method_cdp_gui_click(sb) -> bool:
"""CDP Mode with PyAutoGUI-based clicking - uses actual mouse movement.
For advanced protections (Kasada, DataDome, Akamai), the docs recommend
using gui_* methods with actual mouse movements instead of CDP clicks.
This is the most human-like approach in CDP mode.
"""
try:
logger.debug("Attempting bypass: CDP Mode gui_click (mouse-based)")
current_url = sb.get_current_url()
# Activate CDP mode
sb.activate_cdp_mode(current_url)
time.sleep(random.uniform(1, 2))
# Try the dedicated CDP captcha method first
try:
logger.debug("Trying cdp.gui_click_captcha()")
sb.cdp.gui_click_captcha()
time.sleep(random.uniform(3, 5))
sb.reconnect()
time.sleep(random.uniform(1, 2))
if _is_bypassed(sb):
return True
sb.activate_cdp_mode(sb.get_current_url())
time.sleep(random.uniform(0.5, 1))
except Exception as e:
logger.debug(f"cdp.gui_click_captcha() failed: {e}")
# Turnstile selectors - use parent above shadow-root as per docs
selectors = [
"#turnstile-widget div", # Cloudflare Turnstile
"#cf-turnstile div", # Alternative CF Turnstile
"#challenge-stage div", # CF challenge stage
"input[type='checkbox']", # Generic checkbox
"[class*='cb-i']", # DDOS-Guard checkbox
]
for selector in selectors:
try:
if sb.cdp.is_element_visible(selector):
logger.debug(f"CDP gui_click_element: {selector}")
sb.cdp.gui_click_element(selector)
time.sleep(random.uniform(3, 5))
sb.reconnect()
time.sleep(random.uniform(1, 2))
if _is_bypassed(sb):
return True
sb.activate_cdp_mode(sb.get_current_url())
time.sleep(random.uniform(0.5, 1))
except Exception as e:
logger.debug(f"CDP gui_click on '{selector}' failed: {e}")
continue
# Final reconnect
try:
sb.reconnect()
except Exception as e:
logger.debug(f"Final reconnect in CDP gui_click failed: {e}")
return _is_bypassed(sb)
except Exception as e:
logger.debug(f"CDP Mode gui_click failed: {e}")
try:
sb.reconnect()
except Exception as reconnect_e:
logger.debug(f"Reconnect after CDP Mode gui_click failed: {reconnect_e}")
return False
def _bypass(sb, max_retries: Optional[int] = None, cancel_flag: Optional[Event] = None) -> bool:
"""Bypass function with strategies for Cloudflare and DDOS-Guard protection.
"""Bypass function with unified strategies for Cloudflare and DDOS-Guard protection.
Uses a unified method order that works for both protection types.
Prioritizes CDP Mode (stealthier) with UC Mode PyAutoGUI fallbacks:
1. CDP Mode solve - WebDriver disconnected, uses cdp.solve_captcha()
2. CDP Mode click - Native CDP clicking, no PyAutoGUI (4.45.6+)
3. CDP Mode gui_click - CDP with PyAutoGUI mouse movement (most human-like)
4. uc_gui_handle_captcha() - TAB+SPACEBAR via PyAutoGUI (UC Mode fallback)
5. uc_gui_click_captcha() - Direct click via PyAutoGUI (UC Mode fallback)
6. Human-like interaction - Scroll, wait, reload, retry
Returns True if bypass succeeded, False otherwise.
"""
max_retries = max_retries if max_retries is not None else app_config.MAX_RETRY
cloudflare_methods = [_bypass_method_1, _bypass_method_2, _bypass_method_3]
ddos_guard_methods = [_bypass_ddos_guard_method_1, _bypass_ddos_guard_method_2]
# Unified method order - works for both Cloudflare and DDOS-Guard
# Prioritizes CDP Mode (stealthier), falls back to UC Mode PyAutoGUI methods
methods = [
_bypass_method_cdp_solve, # CDP Mode, WebDriver disconnected
_bypass_method_cdp_click, # CDP native click, no PyAutoGUI
_bypass_method_cdp_gui_click, # CDP with PyAutoGUI (most human-like)
_bypass_method_handle_captcha, # TAB+SPACEBAR via PyAutoGUI (UC Mode)
_bypass_method_click_captcha, # Direct click via PyAutoGUI (UC Mode)
_bypass_method_humanlike, # Last resort with scroll/refresh
]
for try_count in range(max_retries):
# Check for cancellation before each attempt
@@ -480,31 +692,29 @@ def _bypass(sb, max_retries: Optional[int] = None, cancel_flag: Optional[Event]
raise BypassCancelledException("Bypass cancelled")
if _is_bypassed(sb):
if try_count == 0:
logger.info("Page already bypassed (no challenge or auto-solved by uc_open_with_reconnect)")
return True
# Log challenge type for debugging (but don't branch on it)
challenge_type = _detect_challenge_type(sb)
logger.info(f"Detected challenge type: {challenge_type}")
if challenge_type == "ddos_guard":
methods = ddos_guard_methods
elif challenge_type == "cloudflare":
methods = cloudflare_methods
else:
methods = cloudflare_methods + ddos_guard_methods
logger.info(f"Challenge detected: {challenge_type}")
method = methods[try_count % len(methods)]
logger.info(f"Bypass attempt {try_count + 1}/{max_retries} using {method.__name__}")
# Progressive backoff with cancellation checks
wait_time = min(app_config.DEFAULT_SLEEP * try_count, 15)
if wait_time > 0:
logger.info(f"Waiting {wait_time}s before trying...")
# Progressive backoff with cancellation checks (randomized)
if try_count > 0:
wait_time = min(random.uniform(2, 4) * try_count, 12)
logger.info(f"Waiting {wait_time:.1f}s before trying...")
# Check cancellation during wait (check every second)
for _ in range(int(wait_time)):
if cancel_flag and cancel_flag.is_set():
logger.info("Bypass cancelled during wait")
raise BypassCancelledException("Bypass cancelled")
time.sleep(1)
# Sleep remaining fractional second
time.sleep(wait_time - int(wait_time))
try:
if method(sb):
@@ -693,7 +903,7 @@ def _init_driver():
# Build Chrome args dynamically to pick up current DNS settings from network module
chromium_args = _get_chromium_args()
logger.debug(f"Initializing Chrome driver with args: {chromium_args}")
driver = Driver(uc=True, headless=False, size=f"{VIRTUAL_SCREEN_SIZE[0]},{VIRTUAL_SCREEN_SIZE[1]}", chromium_arg=chromium_args)
driver = Driver(uc=True, headless=False, incognito=True, size=f"{VIRTUAL_SCREEN_SIZE[0]},{VIRTUAL_SCREEN_SIZE[1]}", chromium_arg=chromium_args)
driver.set_page_load_timeout(60)
DRIVER = driver
time.sleep(app_config.DEFAULT_SLEEP)
@@ -704,7 +914,7 @@ def _ensure_display_initialized():
global DISPLAY
if DISPLAY["xvfb"] is not None:
return
if not (env.DOCKERMODE and env.USE_CF_BYPASS):
if not (env.DOCKERMODE and app_config.get("USE_CF_BYPASS", True)):
return
from pyvirtualdisplay import Display
@@ -941,14 +1151,17 @@ def warmup():
logger.debug("Bypasser warmup skipped - not in Docker mode")
return
if not env.USE_CF_BYPASS:
if not app_config.get("USE_CF_BYPASS", True):
logger.debug("Bypasser warmup skipped - CF bypass disabled")
return
if app_config.get("AA_DONATOR_KEY", ""):
logger.debug("Bypasser warmup skipped - AA donator key set (fast downloads available)")
return
# Clean up any orphan processes from previous crashes before starting fresh
_cleanup_orphan_processes()
with LOCKED:
if is_warmed_up():
logger.debug("Bypasser already fully warmed up")
@@ -1003,7 +1216,9 @@ _init_cleanup_thread()
# Register for DNS rotation notifications so Chrome can restart with new DNS settings
# Only register if using the internal Chrome bypasser (not external FlareSolverr)
if env.USE_CF_BYPASS and not env.USING_EXTERNAL_BYPASSER:
# Note: This module is only imported when internal bypasser is selected, so this check
# is redundant but kept for safety. Use app_config for consistency with other modules.
if app_config.get("USE_CF_BYPASS", True) and not app_config.get("USING_EXTERNAL_BYPASSER", False):
network.register_dns_rotation_callback(_on_dns_rotation)
+1
View File
@@ -131,6 +131,7 @@ class BookInfo:
status_message: Optional[str] = None # Detailed status message for UI display
added_time: Optional[float] = None # Timestamp when added to queue
source: str = "direct_download" # Release source handler to use for downloads
source_url: Optional[str] = None # Link to source page (e.g., Anna's Archive)
def get_filename(self, fallback_url: Optional[str] = None) -> str:
"""Build sanitized filename: 'Author - Title (Year).format'
+3 -3
View File
@@ -368,12 +368,12 @@ def process_archive(
shutil.rmtree(extract_dir, ignore_errors=True)
archive_path.unlink(missing_ok=True)
# Build success message with extracted formats
# Build success message with format info
formats = [p.suffix.lstrip(".").upper() for p in final_paths]
if len(formats) == 1:
message = f"Extracted: {formats[0]}"
message = f"Complete ({formats[0]})"
else:
message = f"Extracted: {len(formats)} files ({', '.join(formats)})"
message = f"Complete ({len(formats)} files)"
return ArchiveResult(
success=True,
+75 -17
View File
@@ -11,19 +11,77 @@ import requests
from tqdm import tqdm
from cwa_book_downloader.download import network
from cwa_book_downloader.config.env import USE_CF_BYPASS, USING_EXTERNAL_BYPASSER
from cwa_book_downloader.core.config import config as app_config
from cwa_book_downloader.core.logger import setup_logger
# Import bypasser if enabled
if USE_CF_BYPASS:
if USING_EXTERNAL_BYPASSER:
from cwa_book_downloader.bypass.external_bypasser import get_bypassed_page
# External bypasser doesn't share cookies/UA
get_cf_cookies_for_domain = lambda domain: {}
get_cf_user_agent_for_domain = lambda domain: None
# Bypasser modules are imported lazily to support dynamic selection based on config
_internal_bypasser = None
_external_bypasser = None
def _get_internal_bypasser():
"""Lazy import of internal bypasser module."""
global _internal_bypasser
if _internal_bypasser is None:
try:
from cwa_book_downloader.bypass import internal_bypasser
_internal_bypasser = internal_bypasser
except ImportError as e:
raise RuntimeError(
f"Failed to import internal bypasser: {e}. "
"Check that all dependencies are installed. "
"You may need to disable CF bypass or use the external bypasser."
) from e
return _internal_bypasser
def _get_external_bypasser():
"""Lazy import of external bypasser module."""
global _external_bypasser
if _external_bypasser is None:
try:
from cwa_book_downloader.bypass import external_bypasser
_external_bypasser = external_bypasser
except ImportError as e:
raise RuntimeError(
f"Failed to import external bypasser: {e}. "
"Check that the external bypasser is properly configured."
) from e
return _external_bypasser
def _is_using_external_bypasser() -> bool:
"""Check if external bypasser is configured (reads from config, not just env)."""
return app_config.get("USING_EXTERNAL_BYPASSER", False)
def _is_cf_bypass_enabled() -> bool:
"""Check if Cloudflare bypass is enabled."""
return app_config.get("USE_CF_BYPASS", True)
def get_bypassed_page(url, selector=None, cancel_flag=None):
"""Wrapper that delegates to the appropriate bypasser based on config."""
if _is_using_external_bypasser():
return _get_external_bypasser().get_bypassed_page(url, selector, cancel_flag)
else:
from cwa_book_downloader.bypass.internal_bypasser import get_bypassed_page, get_cf_cookies_for_domain, get_cf_user_agent_for_domain
return _get_internal_bypasser().get_bypassed_page(url, selector, cancel_flag)
def get_cf_cookies_for_domain(domain):
"""Get CF cookies - only available with internal bypasser."""
if _is_using_external_bypasser():
logger.debug(f"External bypasser in use, CF cookies not available for {domain}")
return {}
return _get_internal_bypasser().get_cf_cookies_for_domain(domain)
def get_cf_user_agent_for_domain(domain):
"""Get CF user agent - only available with internal bypasser."""
if _is_using_external_bypasser():
logger.debug(f"External bypasser in use, CF user agent not available for {domain}")
return None
return _get_internal_bypasser().get_cf_user_agent_for_domain(domain)
logger = setup_logger(__name__)
@@ -135,7 +193,7 @@ def html_get_page(
return ""
try:
if use_bypasser_now and USE_CF_BYPASS:
if use_bypasser_now and _is_cf_bypass_enabled():
logger.debug(f"GET (bypasser): {current_url}")
try:
result = get_bypassed_page(current_url, selector, cancel_flag)
@@ -148,7 +206,7 @@ def html_get_page(
# Try with CF cookies/UA if available (from previous bypass)
cookies = {}
headers = {}
if USE_CF_BYPASS:
if _is_cf_bypass_enabled():
parsed = urlparse(current_url)
hostname = parsed.hostname or ""
cookies = get_cf_cookies_for_domain(hostname)
@@ -165,7 +223,7 @@ def html_get_page(
# 403 = Cloudflare/DDoS-Guard protection
if status == 403:
if USE_CF_BYPASS and not use_bypasser_now:
if _is_cf_bypass_enabled() and not use_bypasser_now:
# Before switching to bypasser, check if cookies have become available
# (another concurrent download may have completed bypass and extracted cookies)
parsed = urlparse(current_url)
@@ -238,7 +296,7 @@ def download_url(
logger.info(f"Downloading: {current_url} (attempt {attempt + 1}/{MAX_DOWNLOAD_RETRIES})")
# Try with CF cookies/UA if available
cookies = {}
if USE_CF_BYPASS:
if _is_cf_bypass_enabled():
parsed = urlparse(current_url)
hostname = parsed.hostname or ""
cookies = get_cf_cookies_for_domain(hostname)
@@ -286,7 +344,7 @@ def download_url(
retryable = _is_retryable_error(e)
# Z-Library 403 - try refreshing cookies via bypasser once before giving up
if status == 403 and USE_CF_BYPASS and not zlib_cookie_refresh_attempted:
if status == 403 and _is_cf_bypass_enabled() and not zlib_cookie_refresh_attempted:
parsed = urlparse(current_url)
if parsed.hostname and 'z-lib' in parsed.hostname and referer:
zlib_cookie_refresh_attempted = True
@@ -309,14 +367,14 @@ def download_url(
if status == 429:
logger.info(f"Rate limited (429) - trying next source")
if status_callback:
status_callback("resolving", "Server busy, trying next...")
status_callback("resolving", "Server busy, trying next")
return None
# Timeout - don't retry, server likely overloaded
if isinstance(e, requests.exceptions.Timeout):
logger.warning(f"Timeout: {current_url} - skipping to next source")
if status_callback:
status_callback("resolving", "Server timed out, trying next...")
status_callback("resolving", "Server timed out, trying next")
return None
# Try to resume if we got some data
@@ -360,7 +418,7 @@ def _try_resume(
# Try with CF cookies/UA if available
cookies = {}
resume_headers = {**(base_headers or DOWNLOAD_HEADERS), 'Range': f'bytes={start_byte}-'}
if USE_CF_BYPASS:
if _is_cf_bypass_enabled():
parsed = urlparse(url)
hostname = parsed.hostname or ""
cookies = get_cf_cookies_for_domain(hostname)
+7 -6
View File
@@ -626,7 +626,7 @@ def _post_process_download(
# Handle archive extraction (RAR/ZIP)
if is_archive(temp_file):
logger.info(f"Archive detected, extracting: {temp_file.name}")
status_callback("resolving", "Extracting archive...")
status_callback("resolving", "Extracting archive")
result = process_archive(
archive_path=temp_file,
@@ -646,7 +646,7 @@ def _post_process_download(
# Handle directory (multi-file torrent/usenet downloads)
if temp_file.is_dir():
logger.info(f"Directory detected, processing: {temp_file.name}")
status_callback("resolving", "Processing download folder...")
status_callback("resolving", "Processing download folder")
final_paths, error = process_directory(
directory=temp_file,
@@ -659,11 +659,10 @@ def _post_process_download(
return None
if final_paths:
formats = [p.suffix.lstrip(".").upper() for p in final_paths]
if len(formats) == 1:
message = f"Downloaded: {formats[0]}"
if len(final_paths) == 1:
message = "Complete"
else:
message = f"Downloaded: {len(formats)} files ({', '.join(formats)})"
message = f"Complete ({len(final_paths)} files)"
status_callback("complete", message)
return str(final_paths[0])
else:
@@ -724,6 +723,8 @@ def _post_process_download(
os.rename(str(intermediate_path), str(final_path))
logger.info(f"Download completed: {final_path.name}")
status_callback("complete", "Complete")
return str(final_path)
def update_download_progress(book_id: str, progress: float) -> None:
+4 -3
View File
@@ -21,7 +21,7 @@ from cwa_book_downloader.release_sources.direct_download import SearchUnavailabl
from cwa_book_downloader.config.settings import _SUPPORTED_BOOK_LANGUAGE
from cwa_book_downloader.config.env import (
BUILD_VERSION, CWA_DB_PATH, DEBUG, FLASK_HOST, FLASK_PORT,
RELEASE_VERSION, USING_EXTERNAL_BYPASSER,
RELEASE_VERSION,
)
from cwa_book_downloader.core.config import config as app_config
from cwa_book_downloader.core.logger import setup_logger
@@ -295,7 +295,8 @@ def favicon(_: Any = None) -> Response:
# Register bypasser warmup callback for when first WebSocket client connects
# and shutdown callback for when all clients disconnect
if not USING_EXTERNAL_BYPASSER:
# Use app_config to read from settings file (not just env var) so UI changes work after restart
if not app_config.get("USING_EXTERNAL_BYPASSER", False):
from cwa_book_downloader.bypass.internal_bypasser import warmup as bypasser_warmup, shutdown_if_idle as bypasser_shutdown
ws_manager.register_on_first_connect(bypasser_warmup)
ws_manager.register_on_all_disconnect(bypasser_shutdown)
@@ -303,7 +304,7 @@ if not USING_EXTERNAL_BYPASSER:
if DEBUG:
import subprocess
if USING_EXTERNAL_BYPASSER:
if app_config.get("USING_EXTERNAL_BYPASSER", False):
STOP_GUI = lambda: None
else:
from cwa_book_downloader.bypass.internal_bypasser import _reset_driver as STOP_GUI
@@ -139,7 +139,9 @@ class HardcoverProvider(MetadataProvider):
Args:
api_key: Hardcover API key. If not provided, uses config singleton.
"""
self.api_key = api_key or app_config.get("HARDCOVER_API_KEY", "")
raw_key = api_key or app_config.get("HARDCOVER_API_KEY", "")
# Strip "Bearer " prefix if user pasted the full auth header from Hardcover
self.api_key = raw_key.removeprefix("Bearer ").strip() if raw_key else ""
self.session = requests.Session()
if self.api_key:
self.session.headers.update({
@@ -748,7 +750,9 @@ def _test_hardcover_connection(current_values: Optional[Dict[str, Any]] = None)
current_values = current_values or {}
# Use current form values first, fall back to saved config
api_key = current_values.get("HARDCOVER_API_KEY") or app_config.get("HARDCOVER_API_KEY", "")
raw_key = current_values.get("HARDCOVER_API_KEY") or app_config.get("HARDCOVER_API_KEY", "")
# Strip "Bearer " prefix if user pasted the full auth header from Hardcover
api_key = raw_key.removeprefix("Bearer ").strip() if raw_key else ""
key_len = len(api_key) if api_key else 0
logger.debug(f"Hardcover test: key length={key_len}")
@@ -365,6 +365,18 @@ def _parse_book_info_page(soup: BeautifulSoup, book_id: str) -> BookInfo:
# Extract additional metadata
info = _extract_book_metadata(original_divs[-6])
# Fetch download count from the summary API (loaded async on the page)
try:
summary_url = f"{network.get_aa_base_url()}/dyn/md5/summary/{book_id}"
summary_response = downloader.html_get_page(summary_url, selector=network.AAMirrorSelector())
if summary_response:
summary_data = json.loads(summary_response)
if "downloads_total" in summary_data:
info["Downloads"] = [str(summary_data["downloads_total"])]
except Exception as e:
logger.debug(f"Failed to fetch download count for {book_id}: {e}")
book_info.info = info
# Set language and year from metadata if available
@@ -373,6 +385,9 @@ def _parse_book_info_page(soup: BeautifulSoup, book_id: str) -> BookInfo:
if info.get("Year"):
book_info.year = info["Year"][0]
# Set source URL for linking back to Anna's Archive
book_info.source_url = f"{network.get_aa_base_url()}/md5/{book_id}"
return book_info
@@ -562,14 +577,14 @@ def _get_urls_for_source(
# Welib - fetch page and parse for slow_download links
if source_id == "welib":
if status_callback:
status_callback("resolving", "Fetching welib sources...")
status_callback("resolving", "Fetching welib sources")
return _get_download_urls_from_welib(book_info.id, selector=selector, cancel_flag=cancel_flag)
# AA page sources - fetch AA page if not already done
if source_id in _AA_PAGE_SOURCES:
if not aa_page_fetched and not urls_by_source:
if status_callback:
status_callback("resolving", "Fetching download sources...")
status_callback("resolving", "Fetching download sources")
_fetch_aa_page_urls(book_info, urls_by_source)
return urls_by_source.get(source_id, [])
@@ -877,7 +892,7 @@ def _extract_slow_download_url(
# After countdown, update status and re-fetch
if status_callback and source_context:
status_callback("resolving", f"{source_context} - Fetching...")
status_callback("resolving", f"{source_context} - Fetching")
return _get_download_url(link, title, cancel_flag, status_callback, selector, source_context)
@@ -900,6 +915,7 @@ def _book_info_to_release(book_info: BookInfo) -> Release:
format=book_info.format,
size=book_info.size,
download_url=book_info.download_urls[0] if book_info.download_urls else None,
info_url=f"{network.get_aa_base_url()}/md5/{book_info.id}",
protocol=ReleaseProtocol.HTTP,
indexer="Anna's Archive",
extra={
@@ -1158,7 +1174,7 @@ class DirectDownloadHandler(DownloadHandler):
return None
# Execute download via _download_book (handles cascade and bypass)
status_callback("resolving", "Finding download source...")
status_callback("resolving", "Finding download source")
success_url = _download_book(
book_info,
book_path,
@@ -60,7 +60,7 @@ class IRCDownloadHandler(DownloadHandler):
try:
# Phase 1: Connect to IRC
status_callback("resolving", "Connecting to IRC...")
status_callback("resolving", "Connecting to IRC")
if check_cancelled():
return None
@@ -70,7 +70,7 @@ class IRCDownloadHandler(DownloadHandler):
client.join_channel(DEFAULT_CHANNEL)
# Phase 2: Send download request
status_callback("resolving", "Requesting file from bot...")
status_callback("resolving", "Requesting file from bot")
if check_cancelled():
return None
@@ -79,7 +79,7 @@ class IRCDownloadHandler(DownloadHandler):
client.send_message(f"#{DEFAULT_CHANNEL}", download_request)
# Phase 3: Wait for DCC offer
status_callback("resolving", "Waiting for bot response...")
status_callback("resolving", "Waiting for bot response")
offer = client.wait_for_dcc(timeout=120.0, result_type=False)
@@ -186,7 +186,7 @@ class DelugeClient(DownloadClient):
complete = progress >= 100
if complete:
message = "Download complete"
message = "Complete"
eta = status.get(b'eta')
if eta and eta > 604800:
@@ -210,7 +210,7 @@ class NZBGetClient(DownloadClient):
return DownloadStatus(
progress=progress,
state=state,
message=status,
message=status.replace("-", " ").title(),
complete=False,
file_path=None,
download_speed=group.get("DownloadRate"),
@@ -233,7 +233,7 @@ class NZBGetClient(DownloadClient):
return DownloadStatus(
progress=100,
state="complete",
message="Download complete",
message="Complete",
complete=True,
file_path=dest_dir,
)
@@ -184,7 +184,7 @@ class QBittorrentClient(DownloadClient):
# For active downloads without a special message, leave message as None
# so the handler can build the progress message
if complete:
message = "Download complete"
message = "Complete"
# Only include ETA if it's reasonable (less than 1 week)
eta = torrent.eta if 0 < torrent.eta < 604800 else None
@@ -248,7 +248,7 @@ class SABnzbdClient(DownloadClient):
return DownloadStatus(
progress=100,
state="complete",
message="Download complete",
message="Complete",
complete=True,
file_path=storage,
)
@@ -146,7 +146,7 @@ class TransmissionClient(DownloadClient):
complete = progress >= 100 and status_value == "seeding"
if complete:
message = "Download complete"
message = "Complete"
# Get ETA if available and reasonable (less than 1 week)
eta = None
@@ -82,7 +82,7 @@ class ProwlarrHandler(DownloadHandler):
return None
# Check if this download already exists in the client
status_callback("resolving", f"Checking {client.name}...")
status_callback("resolving", f"Checking {client.name}")
existing = client.find_existing(download_url)
if existing:
@@ -92,7 +92,7 @@ class ProwlarrHandler(DownloadHandler):
# If already complete, skip straight to file handling
if existing_status.complete:
logger.info(f"Existing download is complete, copying file directly")
status_callback("resolving", "Found existing download, copying to library...")
status_callback("resolving", "Found existing download, copying to library")
source_path = client.get_download_path(download_id)
if not source_path:
@@ -112,10 +112,10 @@ class ProwlarrHandler(DownloadHandler):
# Existing but still downloading - join the progress polling
logger.info(f"Existing download in progress, joining poll loop")
status_callback("downloading", f"Resuming existing download...")
status_callback("downloading", "Resuming existing download")
else:
# No existing download - add new
status_callback("resolving", f"Sending to {client.name}...")
status_callback("resolving", f"Sending to {client.name}")
try:
download_id = client.add_download(
url=download_url,
@@ -246,7 +246,7 @@ class ProwlarrHandler(DownloadHandler):
The orchestrator will find and filter book files.
"""
try:
status_callback("resolving", "Staging file...")
status_callback("resolving", "Staging file")
# Torrents: copy to preserve seeding. Usenet: configurable.
if protocol == "torrent":
+1
View File
@@ -106,6 +106,7 @@ services:
volumes:
- ./.local/test-clients/qbittorrent/config:/config
- ./.local/test-clients/downloads:/downloads
- ./.local/test-clients/qbittorrent/custom-cont-init.d:/custom-cont-init.d:ro
ports:
- "8080:8080" # Web UI / API
- "6882:6881"
+1 -1
View File
@@ -1,4 +1,4 @@
pyvirtualdisplay
pyautogui
seleniumbase>=4.41.1
seleniumbase>=4.45.6
python-xlib
+21 -1
View File
@@ -111,6 +111,14 @@ function App() {
const [downloadsSidebarOpen, setDownloadsSidebarOpen] = useState(false);
const [settingsOpen, setSettingsOpen] = useState(false);
const [configBannerOpen, setConfigBannerOpen] = useState(false);
const [featureNoticeDismissed, setFeatureNoticeDismissed] = useState(() => {
return localStorage.getItem('cwa-bd-prowlarr-irc-notice-dismissed') === 'true';
});
const handleDismissFeatureNotice = useCallback(() => {
localStorage.setItem('cwa-bd-prowlarr-irc-notice-dismissed', 'true');
setFeatureNoticeDismissed(true);
}, []);
// URL-based search: parse URL params for automatic search on page load
const urlSearchEnabled = isAuthenticated && config !== null;
@@ -565,7 +573,7 @@ function App() {
}}
/>
<main className="w-full max-w-7xl mx-auto px-4 sm:px-6 lg:px-8 py-3 sm:py-6">
<main className="relative w-full max-w-7xl mx-auto px-4 sm:px-6 lg:px-8 py-3 sm:py-6">
<SearchSection
onSearch={(query) => handleSearch(query, config, searchFieldValues)}
isLoading={isSearching}
@@ -585,6 +593,18 @@ function App() {
onSearchFieldChange={updateSearchFieldValue}
/>
{isInitialState && !featureNoticeDismissed && (
<div className="absolute bottom-4 left-0 right-0 text-center text-sm text-gray-500 dark:text-gray-400">
<span>We've added Prowlarr and IRC support for more download options. Check Settings to configure.</span>
<button
onClick={handleDismissFeatureNotice}
className="ml-2 text-blue-500 hover:text-blue-600 dark:text-blue-400 dark:hover:text-blue-300 underline"
>
Dismiss
</button>
</div>
)}
<ResultsSection
books={books}
visible={hasResults}
+19 -24
View File
@@ -1,6 +1,5 @@
import { useState, useEffect, useCallback } from 'react';
import { Book, ButtonStateInfo, isMetadataBook } from '../types';
import { BookDownloadButton } from './BookDownloadButton';
interface DetailsModalProps {
book: Book | null;
@@ -79,7 +78,8 @@ export const DetailsModal = ({ book, onClose, onDownload, onFindDownloads, onSea
// Build metadata grid based on mode
// Universal mode: Year, Genres (no language, no publisher - often blank from providers)
// Direct Download mode: Year, Language, Format, Size
// Direct Download mode: Year, Language, Format, Size, Downloads
const downloadCount = book.info?.Downloads?.[0];
const metadata = isMetadata
? [
{ label: 'Year', value: book.year || '-' },
@@ -92,6 +92,7 @@ export const DetailsModal = ({ book, onClose, onDownload, onFindDownloads, onSea
{ label: 'Language', value: book.language || '-' },
{ label: 'Format', value: book.format || '-' },
{ label: 'Size', value: book.size || '-' },
...(downloadCount ? [{ label: 'Downloads', value: Number(downloadCount).toLocaleString() }] : []),
];
// Extract rating and readers from display_fields for dedicated boxes (Universal mode)
@@ -109,7 +110,7 @@ export const DetailsModal = ({ book, onClose, onDownload, onFindDownloads, onSea
book.info && Object.keys(book.info).length > 0
? Object.entries(book.info).filter(([key]) => {
const normalized = key.toLowerCase();
return normalized !== 'language' && normalized !== 'year';
return normalized !== 'language' && normalized !== 'year' && normalized !== 'downloads';
})
: [];
const extendedInfoEntries = [[publisherInfo.label, publisherInfo.value], ...additionalInfo];
@@ -309,15 +310,15 @@ export const DetailsModal = ({ book, onClose, onDownload, onFindDownloads, onSea
<footer className="border-t border-[var(--border-muted)] bg-[var(--bg-soft)] px-5 py-4">
<div className="flex items-center justify-between gap-4">
{/* Source link - Universal mode only */}
{isMetadata && book.source_url ? (
{/* Source link - shown for both Universal and Direct Download modes */}
{book.source_url && (
<a
href={book.source_url}
target="_blank"
rel="noopener noreferrer"
className="inline-flex items-center gap-1.5 rounded-full border border-[var(--border-muted)] bg-[var(--bg)] px-3 py-2 text-xs font-medium text-gray-600 transition-colors hover:border-gray-400 hover:text-gray-900 dark:text-gray-400 dark:hover:border-gray-500 dark:hover:text-gray-200"
>
View on {providerDisplay}
View on {isMetadata ? providerDisplay : "Anna's Archive"}
<svg className="h-3 w-3" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path
strokeLinecap="round"
@@ -327,25 +328,19 @@ export const DetailsModal = ({ book, onClose, onDownload, onFindDownloads, onSea
/>
</svg>
</a>
) : (
<div />
)}
{isMetadata ? (
<button
onClick={() => onFindDownloads?.(book)}
className="rounded-full bg-emerald-600 px-6 py-2.5 text-sm font-medium text-white transition-colors hover:bg-emerald-700 focus:outline-none focus:ring-2 focus:ring-emerald-500 focus:ring-offset-2"
>
Find Downloads
</button>
) : (
<BookDownloadButton
buttonState={buttonState}
onDownload={handleDownload}
size="md"
className="rounded-full px-6 py-2.5 text-sm font-medium"
ariaLabel={`Download ${book.title || 'book'}`}
/>
)}
{/* Action button - Find Downloads (Universal) or Download (Direct) */}
<button
onClick={isMetadata ? () => onFindDownloads?.(book) : handleDownload}
disabled={!isMetadata && buttonState.state !== 'download'}
className={`ml-auto rounded-full px-6 py-2.5 text-sm font-medium text-white transition-colors focus:outline-none focus:ring-2 focus:ring-offset-2 disabled:opacity-50 disabled:cursor-not-allowed ${
isMetadata
? 'bg-emerald-600 hover:bg-emerald-700 focus:ring-emerald-500'
: 'bg-sky-700 hover:bg-sky-800 focus:ring-sky-500'
}`}
>
{isMetadata ? 'Find Downloads' : buttonState.text}
</button>
</div>
</footer>
</div>
View File
+635
View File
@@ -0,0 +1,635 @@
"""
Docker volume and filesystem edge case tests.
These tests verify the application handles various Docker volume configurations
correctly, including named volumes, bind mounts, permission issues, and
edge cases that commonly cause issues in containerized deployments.
Run with: docker exec test-cwabd python3 -m pytest /app/tests/config/test_docker_volumes.py -v
"""
import json
import os
import stat
import tempfile
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
# =============================================================================
# Fresh Install / Empty Volume Tests
# =============================================================================
class TestFreshInstall:
"""Tests simulating a fresh install with empty volumes."""
def test_config_dir_created_on_first_save(self):
"""Config directory and plugins subdirectory should be created on first save."""
from cwa_book_downloader.core.settings_registry import save_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir) / "config"
# Directory doesn't exist yet
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = save_config_file("test_plugin", {"key": "value"})
assert result is True
assert config_dir.exists()
assert (config_dir / "plugins").exists()
assert (config_dir / "plugins" / "test_plugin.json").exists()
def test_general_settings_saved_to_settings_json(self):
"""General settings should go to settings.json, not plugins folder."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
_get_config_file_path,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
path = _get_config_file_path("general")
assert path == config_dir / "settings.json"
save_config_file("general", {"key": "value"})
assert (config_dir / "settings.json").exists()
def test_nested_config_directories_created(self):
"""Deeply nested config paths should be created with parents=True."""
from cwa_book_downloader.core.settings_registry import _ensure_config_dir
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir) / "deeply" / "nested" / "config"
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
_ensure_config_dir("test_plugin")
assert (config_dir / "plugins").exists()
def test_empty_config_returns_defaults(self):
"""Loading from empty/missing config should return empty dict."""
from cwa_book_downloader.core.settings_registry import load_config_file
with tempfile.TemporaryDirectory() as tmpdir:
with patch("cwa_book_downloader.config.env.CONFIG_DIR", Path(tmpdir)):
result = load_config_file("nonexistent")
assert result == {}
# =============================================================================
# Corrupted / Invalid Config File Tests
# =============================================================================
class TestCorruptedConfig:
"""Tests for handling corrupted or invalid config files."""
def test_invalid_json_returns_empty_dict(self):
"""Invalid JSON in config file should return empty dict, not crash."""
from cwa_book_downloader.core.settings_registry import load_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Write invalid JSON
(plugins_dir / "broken.json").write_text("{ invalid json }")
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = load_config_file("broken")
assert result == {}
def test_empty_json_file_returns_empty_dict(self):
"""Empty JSON file should return empty dict."""
from cwa_book_downloader.core.settings_registry import load_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Write empty file
(plugins_dir / "empty.json").write_text("")
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = load_config_file("empty")
# Empty file is invalid JSON, should return {}
assert result == {}
def test_partial_json_write_recovery(self):
"""Config should handle partially written JSON files."""
from cwa_book_downloader.core.settings_registry import load_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Simulate interrupted write
(plugins_dir / "partial.json").write_text('{"key": "val')
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = load_config_file("partial")
assert result == {}
def test_null_bytes_in_config_file(self):
"""Config with null bytes should be handled gracefully."""
from cwa_book_downloader.core.settings_registry import load_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Write file with null bytes
(plugins_dir / "nullbytes.json").write_bytes(b'{"key": "value\x00"}')
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
# Should either parse (ignoring null) or return empty
result = load_config_file("nullbytes")
# Just verify it doesn't crash
assert isinstance(result, dict)
def test_wrong_type_in_config(self):
"""Config with array instead of object should be handled."""
from cwa_book_downloader.core.settings_registry import load_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Write array instead of object
(plugins_dir / "wrongtype.json").write_text('["item1", "item2"]')
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = load_config_file("wrongtype")
# Should return the parsed content (a list) or handle gracefully
# Current implementation returns whatever json.load returns
assert isinstance(result, (dict, list))
# =============================================================================
# Permission Tests (Docker PUID/PGID scenarios)
# =============================================================================
class TestPermissions:
"""Tests for permission-related scenarios."""
@pytest.mark.skipif(
os.geteuid() == 0,
reason="Permission tests don't work when running as root"
)
def test_read_only_config_dir_save_fails_gracefully(self):
"""Saving to read-only config dir should fail gracefully, not crash."""
from cwa_book_downloader.core.settings_registry import save_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir) / "readonly"
config_dir.mkdir()
os.chmod(config_dir, stat.S_IRUSR | stat.S_IXUSR) # r-x
try:
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = save_config_file("test", {"key": "value"})
assert result is False
finally:
os.chmod(config_dir, stat.S_IRWXU)
@pytest.mark.skipif(
os.geteuid() == 0,
reason="Permission tests don't work when running as root"
)
def test_read_only_config_file_save_fails_gracefully(self):
"""Saving when config file is read-only should fail gracefully."""
from cwa_book_downloader.core.settings_registry import save_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
config_file = plugins_dir / "readonly.json"
config_file.write_text('{"existing": "value"}')
os.chmod(config_file, stat.S_IRUSR) # Read-only
try:
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = save_config_file("readonly", {"new": "value"})
assert result is False
finally:
os.chmod(config_file, stat.S_IRWXU)
def test_config_dir_exists_check(self):
"""_is_config_dir_writable should correctly detect writable dirs."""
from cwa_book_downloader.config.env import _is_config_dir_writable
with tempfile.TemporaryDirectory() as tmpdir:
writable_dir = Path(tmpdir) / "writable"
writable_dir.mkdir()
with patch("cwa_book_downloader.config.env.CONFIG_DIR", writable_dir):
assert _is_config_dir_writable() is True
def test_config_dir_not_exists(self):
"""_is_config_dir_writable should return False for non-existent dir."""
from cwa_book_downloader.config.env import _is_config_dir_writable
with patch(
"cwa_book_downloader.config.env.CONFIG_DIR",
Path("/nonexistent/path/that/does/not/exist")
):
assert _is_config_dir_writable() is False
# =============================================================================
# Path Edge Cases
# =============================================================================
class TestPathEdgeCases:
"""Tests for edge cases in path handling."""
def test_config_dir_with_spaces(self):
"""Config directory with spaces in path should work."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir) / "path with spaces" / "config"
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
save_config_file("test", {"key": "value"})
result = load_config_file("test")
assert result == {"key": "value"}
def test_config_dir_with_unicode(self):
"""Config directory with unicode characters should work."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir) / "配置文件夹" / "config"
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
save_config_file("test", {"key": "value"})
result = load_config_file("test")
assert result == {"key": "value"}
def test_config_with_unicode_values(self):
"""Config values with unicode should be preserved."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
save_config_file("test", {
"title": "日本語タイトル",
"author": "Автор книги",
"emoji": "📚🎉",
})
result = load_config_file("test")
assert result["title"] == "日本語タイトル"
assert result["author"] == "Автор книги"
assert result["emoji"] == "📚🎉"
def test_very_long_plugin_name(self):
"""Very long plugin names should be handled."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
long_name = "a" * 200 # Very long name
with tempfile.TemporaryDirectory() as tmpdir:
with patch("cwa_book_downloader.config.env.CONFIG_DIR", Path(tmpdir)):
# This might fail on some filesystems with path length limits
try:
result = save_config_file(long_name, {"key": "value"})
if result:
loaded = load_config_file(long_name)
assert loaded == {"key": "value"}
except OSError:
# Expected on filesystems with path length limits
pass
# =============================================================================
# Config File Merging Tests
# =============================================================================
class TestConfigMerging:
"""Tests for config file merging behavior."""
def test_save_merges_with_existing(self):
"""Saving should merge with existing values, not replace."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Write initial config
(plugins_dir / "merge.json").write_text('{"existing": "value", "old": "data"}')
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
save_config_file("merge", {"new": "value", "existing": "updated"})
result = load_config_file("merge")
assert result["old"] == "data" # Preserved
assert result["new"] == "value" # Added
assert result["existing"] == "updated" # Updated
def test_save_handles_nested_objects(self):
"""Saving nested objects should work correctly."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
save_config_file("nested", {
"level1": {
"level2": {
"value": "deep"
}
},
"list": [1, 2, 3],
})
result = load_config_file("nested")
assert result["level1"]["level2"]["value"] == "deep"
assert result["list"] == [1, 2, 3]
# =============================================================================
# Cross-Filesystem Tests (TMP_DIR vs INGEST_DIR)
# =============================================================================
class TestCrossFilesystem:
"""Tests for cross-filesystem scenarios."""
def test_staging_and_ingest_same_filesystem(self):
"""When TMP_DIR and INGEST_DIR are on same filesystem, move is used."""
with tempfile.TemporaryDirectory() as tmpdir:
tmp_dir = Path(tmpdir) / "tmp"
ingest_dir = Path(tmpdir) / "ingest"
tmp_dir.mkdir()
ingest_dir.mkdir()
# Both on same filesystem
assert os.stat(tmp_dir).st_dev == os.stat(ingest_dir).st_dev
def test_detect_cross_filesystem(self):
"""CROSS_FILE_SYSTEM detection should work."""
# This is a documentation test - the actual logic is in settings.py:
# CROSS_FILE_SYSTEM = os.stat(env.TMP_DIR).st_dev != os.stat(env.INGEST_DIR).st_dev
pass
# =============================================================================
# Startup Directory Creation Tests
# =============================================================================
class TestStartupDirectoryCreation:
"""Tests for directory creation during startup."""
def test_tmp_dir_created_without_parents(self):
"""TMP_DIR.mkdir(exist_ok=True) needs parent to exist."""
# This documents a potential issue: mkdir(exist_ok=True) without
# parents=True will fail if parent doesn't exist
with tempfile.TemporaryDirectory() as tmpdir:
# Parent exists
tmp_dir = Path(tmpdir) / "tmp"
tmp_dir.mkdir(exist_ok=True)
assert tmp_dir.exists()
# Parent doesn't exist - would fail without parents=True
nested = Path(tmpdir) / "nonexistent" / "deep" / "tmp"
with pytest.raises(FileNotFoundError):
nested.mkdir(exist_ok=True)
# With parents=True it works
nested.mkdir(parents=True, exist_ok=True)
assert nested.exists()
def test_ingest_dir_created_without_parents(self):
"""INGEST_DIR.mkdir(exist_ok=True) needs parent to exist."""
# Same potential issue as TMP_DIR
with tempfile.TemporaryDirectory() as tmpdir:
ingest = Path(tmpdir) / "ingest"
ingest.mkdir(exist_ok=True)
assert ingest.exists()
# =============================================================================
# Named Volume vs Bind Mount Simulation
# =============================================================================
class TestVolumeTypes:
"""Tests simulating named volume vs bind mount differences."""
def test_empty_named_volume_scenario(self):
"""
Simulate named volume: directory exists but is empty.
Named volumes are created by Docker as empty directories owned by root.
The application should handle this gracefully.
"""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
# Simulate named volume: empty directory exists
config_dir = Path(tmpdir) / "config"
config_dir.mkdir()
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
# Should create plugins subdirectory and save
result = save_config_file("test", {"key": "value"})
assert result is True
loaded = load_config_file("test")
assert loaded == {"key": "value"}
def test_bind_mount_with_existing_files(self):
"""
Simulate bind mount: directory has existing files from host.
Bind mounts may have existing config from a previous installation.
"""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Existing config from previous install
(plugins_dir / "prowlarr_clients.json").write_text(json.dumps({
"PROWLARR_TORRENT_CLIENT": "qbittorrent",
"QBITTORRENT_URL": "http://old-host:8080",
}))
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
# New save should merge
save_config_file("prowlarr_clients", {
"QBITTORRENT_URL": "http://new-host:8080",
})
result = load_config_file("prowlarr_clients")
# Should have merged
assert result["PROWLARR_TORRENT_CLIENT"] == "qbittorrent"
assert result["QBITTORRENT_URL"] == "http://new-host:8080"
def test_volume_with_only_partial_structure(self):
"""
Simulate volume with partial directory structure.
User might manually create /config but not /config/plugins.
"""
from cwa_book_downloader.core.settings_registry import save_config_file
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
# Only config dir exists, not plugins subdir
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = save_config_file("test", {"key": "value"})
assert result is True
assert (config_dir / "plugins").exists()
assert (config_dir / "plugins" / "test.json").exists()
# =============================================================================
# Race Condition and Concurrent Access Tests
# =============================================================================
class TestConcurrentAccess:
"""Tests for concurrent config access scenarios."""
def test_simultaneous_saves_dont_corrupt(self):
"""Multiple saves should not corrupt the config file."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
import threading
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
results = []
def save_value(key, value):
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
result = save_config_file("concurrent", {key: value})
results.append(result)
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
threads = [
threading.Thread(target=save_value, args=(f"key{i}", f"value{i}"))
for i in range(5)
]
for t in threads:
t.start()
for t in threads:
t.join()
# All saves should succeed
assert all(results)
# File should be valid JSON
final = load_config_file("concurrent")
assert isinstance(final, dict)
# =============================================================================
# Config Backup/Migration Tests
# =============================================================================
class TestConfigMigration:
"""Tests for config migration scenarios."""
def test_old_format_config_upgrade(self):
"""
Application should handle config from older versions.
This is a placeholder for version-specific migration tests.
"""
pass
def test_config_with_unknown_keys(self):
"""Config with unknown keys should be preserved."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir)
plugins_dir = config_dir / "plugins"
plugins_dir.mkdir(parents=True)
# Config with keys that don't exist in current schema
(plugins_dir / "future.json").write_text(json.dumps({
"KNOWN_KEY": "value",
"FUTURE_KEY_V2": "future_value",
"ANOTHER_UNKNOWN": 123,
}))
with patch("cwa_book_downloader.config.env.CONFIG_DIR", config_dir):
# Save should preserve unknown keys
save_config_file("future", {"KNOWN_KEY": "updated"})
result = load_config_file("future")
assert result["KNOWN_KEY"] == "updated"
assert result["FUTURE_KEY_V2"] == "future_value"
assert result["ANOTHER_UNKNOWN"] == 123
+494
View File
@@ -0,0 +1,494 @@
"""
Environment and configuration tests.
These tests verify the application behaves correctly with different
configuration settings, environment variables, and Docker setups.
Run with: docker exec test-cwabd python3 -m pytest /app/tests/config/test_environment.py -v
"""
import json
import os
import shutil
import tempfile
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
# =============================================================================
# Directory Setup Tests
# =============================================================================
class TestDirectorySetup:
"""Tests for directory creation and permissions."""
def test_staging_dir_created_on_demand(self):
"""Staging directory should be created if it doesn't exist."""
from cwa_book_downloader.download.orchestrator import get_staging_dir
with tempfile.TemporaryDirectory() as tmpdir:
test_staging = Path(tmpdir) / "staging"
assert not test_staging.exists()
with patch("cwa_book_downloader.download.orchestrator.TMP_DIR", test_staging):
result = get_staging_dir()
assert test_staging.exists()
assert result == test_staging
def test_staging_dir_handles_existing_directory(self):
"""Staging directory creation should be idempotent."""
from cwa_book_downloader.download.orchestrator import get_staging_dir
with tempfile.TemporaryDirectory() as tmpdir:
test_staging = Path(tmpdir) / "staging"
test_staging.mkdir()
with patch("cwa_book_downloader.download.orchestrator.TMP_DIR", test_staging):
result = get_staging_dir()
assert result == test_staging
def test_staging_path_handles_special_characters(self):
"""Staging path should handle task IDs with special characters."""
from cwa_book_downloader.download.orchestrator import get_staging_path
with tempfile.TemporaryDirectory() as tmpdir:
with patch(
"cwa_book_downloader.download.orchestrator.TMP_DIR", Path(tmpdir)
):
# Task ID with URL-like characters
path = get_staging_path(
"https://example.com/book?id=123&format=epub", "epub"
)
assert path.suffix == ".epub"
assert path.parent == Path(tmpdir)
# Should not contain invalid filename chars
assert "/" not in path.name
assert "?" not in path.name
assert "&" not in path.name
def test_staging_path_normalizes_extension(self):
"""Staging path should handle extensions with or without dot."""
from cwa_book_downloader.download.orchestrator import get_staging_path
with tempfile.TemporaryDirectory() as tmpdir:
with patch(
"cwa_book_downloader.download.orchestrator.TMP_DIR", Path(tmpdir)
):
path1 = get_staging_path("task1", "epub")
path2 = get_staging_path("task1", ".epub")
assert path1.suffix == ".epub"
assert path2.suffix == ".epub"
# =============================================================================
# Supported Formats Tests
# =============================================================================
class TestSupportedFormats:
"""Tests for format filtering configuration."""
def test_default_supported_formats(self):
"""Default formats should include common ebook formats."""
from cwa_book_downloader.config.env import _SUPPORTED_FORMATS
# Check some expected defaults
assert "epub" in _SUPPORTED_FORMATS
assert "mobi" in _SUPPORTED_FORMATS
assert "azw3" in _SUPPORTED_FORMATS
def test_format_list_is_lowercase(self):
"""Format list should be normalized to lowercase."""
from cwa_book_downloader.config.env import _SUPPORTED_FORMATS
# All formats should be lowercase
for fmt in _SUPPORTED_FORMATS.split(","):
assert fmt == fmt.lower()
def test_config_supported_formats_attribute(self):
"""Config should have SUPPORTED_FORMATS as a list."""
from cwa_book_downloader.config.settings import SUPPORTED_FORMATS
assert isinstance(SUPPORTED_FORMATS, list)
assert len(SUPPORTED_FORMATS) > 0
assert "epub" in SUPPORTED_FORMATS
# =============================================================================
# Content-Type Routing Tests
# =============================================================================
class TestContentTypeRouting:
"""Tests for content-type based directory routing."""
def test_download_paths_default_to_ingest_dir(self):
"""All content types should default to INGEST_DIR if not specified."""
from cwa_book_downloader.config.env import DOWNLOAD_PATHS, INGEST_DIR
# When no specific paths are set, all should default to INGEST_DIR
for content_type, path in DOWNLOAD_PATHS.items():
# Path should be INGEST_DIR or a custom path
assert isinstance(path, Path)
def test_content_type_routing_keys(self):
"""All expected content types should be present."""
from cwa_book_downloader.config.env import DOWNLOAD_PATHS
expected_types = [
"book (fiction)",
"book (non-fiction)",
"book (unknown)",
"magazine",
"comic book",
"audiobook",
"standards document",
"musical score",
"other",
]
for content_type in expected_types:
assert content_type in DOWNLOAD_PATHS, f"Missing content type: {content_type}"
# =============================================================================
# Settings System Tests
# =============================================================================
class TestSettingsSystem:
"""Tests for the settings registry and persistence."""
def test_save_and_load_config(self):
"""Settings should persist to JSON files."""
from cwa_book_downloader.core.settings_registry import (
save_config_file,
load_config_file,
)
with tempfile.TemporaryDirectory() as tmpdir:
with patch(
"cwa_book_downloader.config.env.CONFIG_DIR", Path(tmpdir)
):
test_data = {"key1": "value1", "key2": 123, "key3": True}
save_config_file("test_plugin", test_data)
loaded = load_config_file("test_plugin")
assert loaded == test_data
def test_load_missing_config_returns_empty(self):
"""Loading non-existent config should return empty dict."""
from cwa_book_downloader.core.settings_registry import load_config_file
with tempfile.TemporaryDirectory() as tmpdir:
with patch(
"cwa_book_downloader.config.env.CONFIG_DIR", Path(tmpdir)
):
loaded = load_config_file("nonexistent_plugin")
assert loaded == {}
def test_config_singleton_refresh(self):
"""Config singleton should refresh when settings change."""
from cwa_book_downloader.core.config import config
from cwa_book_downloader.core.settings_registry import save_config_file
# Get initial value
initial = config.get("TEST_REFRESH_KEY", "default")
with tempfile.TemporaryDirectory() as tmpdir:
with patch(
"cwa_book_downloader.config.env.CONFIG_DIR", Path(tmpdir)
):
save_config_file("test", {"TEST_REFRESH_KEY": "new_value"})
config.refresh()
# Note: This test is limited because config also reads from env
def test_config_env_var_priority(self):
"""Environment variables should take priority over config files."""
# This tests the priority: ENV > config file > default
from cwa_book_downloader.config.env import string_to_bool
# Test the string_to_bool helper used for parsing
assert string_to_bool("true") is True
assert string_to_bool("True") is True
assert string_to_bool("TRUE") is True
assert string_to_bool("yes") is True
assert string_to_bool("1") is True
assert string_to_bool("y") is True
assert string_to_bool("false") is False
assert string_to_bool("no") is False
assert string_to_bool("0") is False
assert string_to_bool("anything_else") is False
# =============================================================================
# Archive Handling Configuration Tests
# =============================================================================
class TestArchiveHandling:
"""Tests for archive extraction configuration."""
def test_is_archive_detects_supported_formats(self):
"""is_archive should detect RAR and ZIP files (not cbr/cbz which are book formats)."""
from cwa_book_downloader.download.archive import is_archive
# RAR and ZIP are archive formats that get extracted
assert is_archive(Path("book.rar")) is True
assert is_archive(Path("book.zip")) is True
# CBR/CBZ are comic book formats, treated as books not archives
assert is_archive(Path("book.cbr")) is False
assert is_archive(Path("book.cbz")) is False
# Regular book formats are not archives
assert is_archive(Path("book.epub")) is False
assert is_archive(Path("book.pdf")) is False
assert is_archive(Path("book.mobi")) is False
def test_is_archive_case_insensitive(self):
"""Archive detection should be case insensitive."""
from cwa_book_downloader.download.archive import is_archive
assert is_archive(Path("book.RAR")) is True
assert is_archive(Path("book.ZIP")) is True
assert is_archive(Path("book.Zip")) is True
assert is_archive(Path("book.RaR")) is True
# =============================================================================
# Validation and Error Handling Tests
# =============================================================================
class TestConfigValidation:
"""Tests for configuration validation and error handling."""
def test_invalid_number_env_var_uses_default(self):
"""Invalid numeric env vars should fall back to defaults."""
# Test that int() parsing handles invalid values gracefully
# The env.py module uses int() which will raise ValueError
# This tests the expected behavior
with patch.dict(os.environ, {"MAX_RETRY": "not_a_number"}):
# Importing with invalid env var should use default or raise
# This depends on implementation - test documents behavior
pass # Currently env.py will crash on invalid int
def test_missing_required_directory_handling(self):
"""Application should handle missing directories gracefully."""
from cwa_book_downloader.download.orchestrator import get_staging_dir
with tempfile.TemporaryDirectory() as tmpdir:
# Use a path that doesn't exist yet
nonexistent = Path(tmpdir) / "deeply" / "nested" / "path"
with patch(
"cwa_book_downloader.download.orchestrator.TMP_DIR", nonexistent
):
result = get_staging_dir()
# Should have created the directory
assert nonexistent.exists()
@pytest.mark.skipif(
os.geteuid() == 0,
reason="Test skipped when running as root (chmod has no effect)"
)
def test_config_dir_not_writable(self):
"""Application should handle read-only config directory."""
from cwa_book_downloader.config.env import _is_config_dir_writable
with tempfile.TemporaryDirectory() as tmpdir:
readonly_dir = Path(tmpdir) / "readonly"
readonly_dir.mkdir()
os.chmod(readonly_dir, 0o444) # Read-only
try:
with patch(
"cwa_book_downloader.config.env.CONFIG_DIR", readonly_dir
):
result = _is_config_dir_writable()
assert result is False
finally:
os.chmod(readonly_dir, 0o755) # Restore for cleanup
# =============================================================================
# Debug and Logging Configuration Tests
# =============================================================================
class TestDebugConfiguration:
"""Tests for debug and logging settings."""
def test_debug_from_env_var(self):
"""DEBUG env var should set debug mode."""
from cwa_book_downloader.config.env import string_to_bool
# Test the parsing logic
assert string_to_bool("true") is True
assert string_to_bool("false") is False
def test_log_level_derived_from_debug(self):
"""LOG_LEVEL should be derived from DEBUG setting."""
# When DEBUG is True, LOG_LEVEL should be "DEBUG"
# When DEBUG is False, LOG_LEVEL should be "INFO"
# This is tested by checking the module logic
pass # The logic is in env.py: LOG_LEVEL = "DEBUG" if DEBUG else "INFO"
# =============================================================================
# Proxy and Network Configuration Tests
# =============================================================================
class TestNetworkConfiguration:
"""Tests for proxy and network settings."""
def test_proxy_settings_stripped(self):
"""Proxy URLs should be stripped of whitespace."""
from cwa_book_downloader.config.env import HTTP_PROXY, HTTPS_PROXY
# These are already evaluated, but the logic is:
# HTTP_PROXY = os.getenv("HTTP_PROXY", "").strip()
# So whitespace should be removed
assert HTTP_PROXY == HTTP_PROXY.strip()
assert HTTPS_PROXY == HTTPS_PROXY.strip()
def test_tor_mode_disables_other_network_settings(self):
"""Tor mode should disable custom DNS, DOH, and proxies."""
# This is a documentation test - the logic is in env.py:
# if USING_TOR:
# _CUSTOM_DNS = ""
# USE_DOH = False
# HTTP_PROXY = ""
# HTTPS_PROXY = ""
pass
# =============================================================================
# Concurrent Downloads Configuration Tests
# =============================================================================
class TestConcurrencyConfiguration:
"""Tests for concurrent download settings."""
def test_max_concurrent_downloads_default(self):
"""MAX_CONCURRENT_DOWNLOADS should have a sensible default."""
from cwa_book_downloader.config.env import MAX_CONCURRENT_DOWNLOADS
assert MAX_CONCURRENT_DOWNLOADS >= 1
assert MAX_CONCURRENT_DOWNLOADS <= 10 # Reasonable upper bound
def test_download_progress_interval_default(self):
"""DOWNLOAD_PROGRESS_UPDATE_INTERVAL should have a sensible default."""
from cwa_book_downloader.config.env import DOWNLOAD_PROGRESS_UPDATE_INTERVAL
assert DOWNLOAD_PROGRESS_UPDATE_INTERVAL >= 1
assert DOWNLOAD_PROGRESS_UPDATE_INTERVAL <= 10
# =============================================================================
# Cache Configuration Tests
# =============================================================================
class TestCacheConfiguration:
"""Tests for cache settings."""
def test_metadata_cache_ttl_defaults(self):
"""Metadata cache TTLs should have sensible defaults."""
from cwa_book_downloader.config.env import (
METADATA_CACHE_SEARCH_TTL,
METADATA_CACHE_BOOK_TTL,
)
# Search cache should be shorter than book cache
assert METADATA_CACHE_SEARCH_TTL > 0
assert METADATA_CACHE_BOOK_TTL > 0
assert METADATA_CACHE_SEARCH_TTL <= METADATA_CACHE_BOOK_TTL
def test_covers_cache_directory(self):
"""Covers cache directory should be under CONFIG_DIR."""
from cwa_book_downloader.config.env import CONFIG_DIR, COVERS_CACHE_DIR
assert COVERS_CACHE_DIR.parent == CONFIG_DIR
assert COVERS_CACHE_DIR.name == "covers"
# =============================================================================
# File Collision Handling Tests
# =============================================================================
class TestFileCollisionHandling:
"""Tests for handling file name collisions."""
def test_stage_file_handles_collision(self):
"""stage_file should add suffix on collision."""
from cwa_book_downloader.download.orchestrator import stage_file
with tempfile.TemporaryDirectory() as tmpdir:
staging = Path(tmpdir) / "staging"
staging.mkdir()
# Create source file
source = Path(tmpdir) / "book.epub"
source.write_text("content")
# Create existing file with same name in staging
(staging / "book.epub").write_text("existing")
with patch(
"cwa_book_downloader.download.orchestrator.TMP_DIR", staging
):
result = stage_file(source, "task1", copy=True)
# Should have created a new file with suffix
assert result.name == "book_1.epub"
assert result.exists()
def test_stage_file_copy_vs_move(self):
"""stage_file should copy or move based on parameter."""
from cwa_book_downloader.download.orchestrator import stage_file
with tempfile.TemporaryDirectory() as tmpdir:
staging = Path(tmpdir) / "staging"
staging.mkdir()
# Test copy
source1 = Path(tmpdir) / "book1.epub"
source1.write_text("content1")
with patch(
"cwa_book_downloader.download.orchestrator.TMP_DIR", staging
):
result1 = stage_file(source1, "task1", copy=True)
assert source1.exists() # Original still exists
assert result1.exists()
# Test move
source2 = Path(tmpdir) / "book2.epub"
source2.write_text("content2")
with patch(
"cwa_book_downloader.download.orchestrator.TMP_DIR", staging
):
result2 = stage_file(source2, "task2", copy=False)
assert not source2.exists() # Original moved
assert result2.exists()
+917
View File
@@ -0,0 +1,917 @@
"""
Failure scenario tests for Prowlarr handler and download clients.
These tests verify error handling behavior - what happens when things go wrong.
They use real clients where possible, with injected failures for edge cases.
Run with: docker exec test-cwabd python3 -m pytest /app/tests/prowlarr/test_failure_scenarios.py -v
"""
import time
from pathlib import Path
from threading import Event, Thread
from typing import List, Optional, Tuple
from unittest.mock import MagicMock, patch, PropertyMock
import tempfile
import shutil
import pytest
from cwa_book_downloader.core.models import DownloadTask
from cwa_book_downloader.release_sources.prowlarr.handler import ProwlarrHandler
from cwa_book_downloader.release_sources.prowlarr.clients import (
DownloadClient,
DownloadState,
DownloadStatus,
)
# =============================================================================
# Test Fixtures and Helpers
# =============================================================================
class ProgressRecorder:
"""Records progress and status updates during download."""
def __init__(self):
self.progress_values: List[float] = []
self.status_updates: List[Tuple[str, Optional[str]]] = []
def progress_callback(self, progress: float):
self.progress_values.append(progress)
def status_callback(self, status: str, message: Optional[str]):
self.status_updates.append((status, message))
@property
def last_status(self) -> Optional[str]:
return self.status_updates[-1][0] if self.status_updates else None
@property
def last_message(self) -> Optional[str]:
return self.status_updates[-1][1] if self.status_updates else None
@property
def statuses(self) -> List[str]:
return [s[0] for s in self.status_updates]
@property
def had_error(self) -> bool:
return "error" in self.statuses
class MockClient(DownloadClient):
"""Configurable mock client for testing failure scenarios."""
protocol = "torrent"
name = "mock"
def __init__(self):
self.downloads = {}
self.status_sequence = [] # List of DownloadStatus to return in order
self.status_index = 0
self.add_download_error = None # Exception to raise on add_download
self.get_status_error = None # Exception to raise on get_status
self.remove_called = False
self.remove_with_delete = False
@staticmethod
def is_configured() -> bool:
return True
def test_connection(self) -> Tuple[bool, str]:
return True, "Mock client connected"
def add_download(self, url: str, name: str, category: str = "cwabd") -> str:
if self.add_download_error:
raise self.add_download_error
download_id = f"mock-{len(self.downloads)}"
self.downloads[download_id] = {"url": url, "name": name}
return download_id
def get_status(self, download_id: str) -> DownloadStatus:
if self.get_status_error:
raise self.get_status_error
if self.status_sequence:
if self.status_index < len(self.status_sequence):
status = self.status_sequence[self.status_index]
self.status_index += 1
return status
# Return last status if we've exhausted the sequence
return self.status_sequence[-1]
# Default: return downloading status
return DownloadStatus(
progress=50,
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
)
def remove(self, download_id: str, delete_files: bool = False) -> bool:
self.remove_called = True
self.remove_with_delete = delete_files
return True
def get_download_path(self, download_id: str) -> Optional[str]:
return "/downloads/test-file.epub"
def find_existing(self, url: str) -> Optional[Tuple[str, DownloadStatus]]:
return None
@pytest.fixture
def handler():
return ProwlarrHandler()
@pytest.fixture
def mock_client():
return MockClient()
@pytest.fixture
def recorder():
return ProgressRecorder()
@pytest.fixture
def cancel_flag():
return Event()
@pytest.fixture
def sample_task():
return DownloadTask(
task_id="test-task-123",
source="prowlarr",
title="Test Book",
)
@pytest.fixture
def sample_release():
return {
"guid": "test-task-123",
"title": "Test Book",
"downloadUrl": "magnet:?xt=urn:btih:abc123",
"protocol": "torrent",
"indexer": "TestIndexer",
}
# =============================================================================
# Error State Tests - Client Reports Error
# =============================================================================
class TestClientErrorStates:
"""Tests for when download clients report error states."""
def test_client_returns_error_state_during_download(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Handler should abort when client reports error state."""
# Simulate: downloading -> downloading -> error
mock_client.status_sequence = [
DownloadStatus(
progress=10,
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
),
DownloadStatus(
progress=25,
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
),
DownloadStatus(
progress=0,
state=DownloadState.ERROR,
message="Tracker returned error: torrent not found",
complete=False,
file_path=None,
),
]
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "Tracker returned error" in recorder.last_message
assert mock_client.remove_called
assert mock_client.remove_with_delete # Should delete files on error
def test_client_returns_error_with_complete_flag(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Edge case: complete=True but state=ERROR should be treated as error."""
mock_client.status_sequence = [
DownloadStatus(
progress=100,
state=DownloadState.ERROR,
message="Download corrupted",
complete=True, # Complete but errored
file_path=None,
),
]
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
def test_error_without_message_uses_default(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Error state without message should use default error text."""
mock_client.status_sequence = [
DownloadStatus(
progress=0,
state=DownloadState.ERROR,
message=None, # No message
complete=False,
file_path=None,
),
]
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert recorder.last_message == "Download failed"
# =============================================================================
# Connection/Network Failure Tests
# =============================================================================
class TestConnectionFailures:
"""Tests for network and connection failures."""
def test_add_download_fails(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Handler should report error when add_download throws."""
mock_client.add_download_error = ConnectionError("Connection refused")
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "Connection refused" in recorder.last_message
def test_get_status_fails_during_poll(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Handler should recover or error gracefully when get_status throws."""
call_count = 0
def failing_get_status(download_id):
nonlocal call_count
call_count += 1
if call_count <= 2:
return DownloadStatus(
progress=call_count * 10,
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
)
raise ConnectionError("Client went away")
mock_client.get_status = failing_get_status
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "Client went away" in recorder.last_message
def test_no_client_configured(
self, handler, recorder, cancel_flag, sample_task, sample_release
):
"""Handler should report helpful error when no client is configured."""
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=None,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.list_configured_clients",
return_value=[],
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "No download clients configured" in recorder.last_message
# =============================================================================
# Cancellation Tests
# =============================================================================
class TestCancellation:
"""Tests for download cancellation behavior."""
def test_cancel_during_download(
self, handler, mock_client, recorder, sample_task, sample_release
):
"""Cancellation should stop download and cleanup."""
cancel_flag = Event()
# Status sequence that keeps downloading
mock_client.status_sequence = [
DownloadStatus(
progress=i * 10,
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
)
for i in range(20)
]
def cancel_after_delay():
time.sleep(0.3)
cancel_flag.set()
cancel_thread = Thread(target=cancel_after_delay)
cancel_thread.start()
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.POLL_INTERVAL",
0.1,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
cancel_thread.join()
assert result is None
assert "cancelled" in recorder.statuses
assert mock_client.remove_called
assert mock_client.remove_with_delete
def test_cancel_before_download_starts(
self, handler, mock_client, recorder, sample_task, sample_release
):
"""Pre-set cancel flag should abort immediately."""
cancel_flag = Event()
cancel_flag.set() # Already cancelled
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
# Should have been cancelled quickly
assert mock_client.remove_called
# =============================================================================
# Cache/Release Not Found Tests
# =============================================================================
class TestCacheFailures:
"""Tests for release cache failures."""
def test_release_not_in_cache(self, handler, recorder, cancel_flag, sample_task):
"""Handler should error when release is not found in cache."""
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=None,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "not found in cache" in recorder.last_message.lower()
def test_release_missing_download_url(
self, handler, recorder, cancel_flag, sample_task
):
"""Handler should error when release has no download URL."""
release_no_url = {
"guid": "test-task-123",
"title": "Test Book",
# No downloadUrl or magnetUrl
"protocol": "torrent",
}
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=release_no_url,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "No download URL" in recorder.last_message
def test_unknown_protocol(self, handler, recorder, cancel_flag, sample_task):
"""Handler should error on unknown protocol."""
release_unknown_protocol = {
"guid": "test-task-123",
"title": "Test Book",
"downloadUrl": "ftp://example.com/book.epub",
"protocol": "ftp",
}
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=release_unknown_protocol,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "protocol" in recorder.last_message.lower()
# =============================================================================
# File Handling Failures
# =============================================================================
class TestFileHandlingFailures:
"""Tests for file staging/copying failures."""
def test_download_path_not_found(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Handler should error when completed file path is not available."""
mock_client.status_sequence = [
DownloadStatus(
progress=100,
state=DownloadState.COMPLETE,
message=None,
complete=True,
file_path=None,
),
]
mock_client.get_download_path = lambda x: None # No path available
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "locate" in recorder.last_message.lower()
def test_permission_denied_on_copy(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Handler should report permission errors during file staging."""
mock_client.status_sequence = [
DownloadStatus(
progress=100,
state=DownloadState.COMPLETE,
message=None,
complete=True,
file_path="/downloads/test.epub",
),
]
with tempfile.TemporaryDirectory() as tmpdir:
source_file = Path(tmpdir) / "source.epub"
source_file.write_text("test content")
mock_client.get_download_path = lambda x: str(source_file)
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
), patch(
"shutil.copy2",
side_effect=PermissionError("Permission denied"),
), patch(
"cwa_book_downloader.download.orchestrator.get_staging_dir",
return_value=Path(tmpdir) / "staging",
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
assert "permission" in recorder.last_message.lower()
# =============================================================================
# Progress Callback Tests
# =============================================================================
class TestProgressCallbacks:
"""Tests for progress reporting behavior."""
def test_progress_values_are_in_order(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Progress values should generally increase (allowing for client quirks)."""
mock_client.status_sequence = [
DownloadStatus(
progress=i,
state=DownloadState.DOWNLOADING if i < 100 else DownloadState.COMPLETE,
message=None,
complete=i >= 100,
file_path="/downloads/test.epub" if i >= 100 else None,
)
for i in [0, 10, 25, 50, 75, 90, 100]
]
with tempfile.TemporaryDirectory() as tmpdir:
source_file = Path(tmpdir) / "test.epub"
source_file.write_text("test content")
staging_dir = Path(tmpdir) / "staging"
staging_dir.mkdir()
mock_client.get_download_path = lambda x: str(source_file)
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
), patch(
"cwa_book_downloader.download.orchestrator.get_staging_dir",
return_value=staging_dir,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.POLL_INTERVAL",
0.01,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is not None
assert recorder.progress_values == [0, 10, 25, 50, 75, 90, 100]
def test_progress_clamps_to_valid_range(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""DownloadStatus should clamp progress to 0-100."""
# Test that invalid progress values are handled
status = DownloadStatus(
progress=150, # Over 100
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
)
assert status.progress == 100
status = DownloadStatus(
progress=-10, # Negative
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
)
assert status.progress == 0
# =============================================================================
# Status Message Tests
# =============================================================================
class TestStatusMessages:
"""Tests for status message formatting."""
def test_status_includes_speed_and_eta(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Status message should include speed and ETA when available."""
mock_client.status_sequence = [
DownloadStatus(
progress=50,
state=DownloadState.DOWNLOADING,
message=None, # No custom message
complete=False,
file_path=None,
download_speed=5 * 1024 * 1024, # 5 MB/s
eta=120, # 2 minutes
),
DownloadStatus(
progress=100,
state=DownloadState.COMPLETE,
message=None,
complete=True,
file_path="/downloads/test.epub",
),
]
with tempfile.TemporaryDirectory() as tmpdir:
source_file = Path(tmpdir) / "test.epub"
source_file.write_text("test content")
staging_dir = Path(tmpdir) / "staging"
staging_dir.mkdir()
mock_client.get_download_path = lambda x: str(source_file)
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
), patch(
"cwa_book_downloader.download.orchestrator.get_staging_dir",
return_value=staging_dir,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.POLL_INTERVAL",
0.01,
):
handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
# Find the downloading status message
downloading_msgs = [
msg for status, msg in recorder.status_updates if status == "downloading"
]
assert len(downloading_msgs) > 0
msg = downloading_msgs[0]
assert "50%" in msg
assert "MB/s" in msg
assert "2m" in msg
def test_client_message_takes_priority(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Client-provided message should override generated message."""
mock_client.status_sequence = [
DownloadStatus(
progress=0,
state=DownloadState.DOWNLOADING,
message="Fetching metadata from peers", # Custom message
complete=False,
file_path=None,
),
DownloadStatus(
progress=100,
state=DownloadState.COMPLETE,
message=None,
complete=True,
file_path="/downloads/test.epub",
),
]
with tempfile.TemporaryDirectory() as tmpdir:
source_file = Path(tmpdir) / "test.epub"
source_file.write_text("test content")
staging_dir = Path(tmpdir) / "staging"
staging_dir.mkdir()
mock_client.get_download_path = lambda x: str(source_file)
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
), patch(
"cwa_book_downloader.download.orchestrator.get_staging_dir",
return_value=staging_dir,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.POLL_INTERVAL",
0.01,
):
handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
# Check that custom message was used
downloading_msgs = [
msg for status, msg in recorder.status_updates if status == "downloading"
]
assert "Fetching metadata from peers" in downloading_msgs
# =============================================================================
# Cleanup After Error Tests
# =============================================================================
class TestErrorCleanup:
"""Tests verifying proper cleanup after errors."""
def test_cleanup_on_poll_exception(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Download should be removed from client after polling exception."""
call_count = 0
def exploding_get_status(download_id):
nonlocal call_count
call_count += 1
if call_count > 2:
raise RuntimeError("Client crashed")
return DownloadStatus(
progress=10,
state=DownloadState.DOWNLOADING,
message=None,
complete=False,
file_path=None,
)
mock_client.get_status = exploding_get_status
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.POLL_INTERVAL",
0.01,
):
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert mock_client.remove_called
assert mock_client.remove_with_delete
def test_cleanup_continues_even_if_remove_fails(
self, handler, mock_client, recorder, cancel_flag, sample_task, sample_release
):
"""Handler should not crash if cleanup removal fails."""
mock_client.status_sequence = [
DownloadStatus(
progress=0,
state=DownloadState.ERROR,
message="Download failed",
complete=False,
file_path=None,
),
]
def failing_remove(download_id, delete_files=False):
raise ConnectionError("Client not responding")
mock_client.remove = failing_remove
with patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_release",
return_value=sample_release,
), patch(
"cwa_book_downloader.release_sources.prowlarr.handler.get_client",
return_value=mock_client,
):
# Should not raise, even though remove() fails
result = handler.download(
task=sample_task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.had_error
+17 -19
View File
@@ -14,10 +14,8 @@ from unittest.mock import MagicMock, patch, PropertyMock
import pytest
from cwa_book_downloader.core.models import DownloadTask
from cwa_book_downloader.release_sources.prowlarr.handler import (
ProwlarrHandler,
_determine_protocol,
)
from cwa_book_downloader.release_sources.prowlarr.handler import ProwlarrHandler
from cwa_book_downloader.release_sources.prowlarr.utils import get_protocol
from cwa_book_downloader.release_sources.prowlarr.clients import (
DownloadStatus,
DownloadState,
@@ -50,34 +48,34 @@ class ProgressRecorder:
return [s[0] for s in self.status_updates]
class TestDetermineProtocol:
"""Tests for the _determine_protocol function."""
class TestGetProtocol:
"""Tests for the get_protocol function."""
def test_determine_protocol_torrent(self):
def test_get_protocol_torrent(self):
"""Test detecting torrent protocol."""
result = {"protocol": "torrent"}
assert _determine_protocol(result) == "torrent"
assert get_protocol(result) == "torrent"
def test_determine_protocol_usenet(self):
def test_get_protocol_usenet(self):
"""Test detecting usenet protocol."""
result = {"protocol": "usenet"}
assert _determine_protocol(result) == "usenet"
assert get_protocol(result) == "usenet"
def test_determine_protocol_unknown(self):
def test_get_protocol_unknown(self):
"""Test unknown protocol."""
result = {"protocol": "ftp"}
assert _determine_protocol(result) == "unknown"
assert get_protocol(result) == "unknown"
def test_determine_protocol_empty(self):
def test_get_protocol_empty(self):
"""Test empty protocol."""
result = {}
assert _determine_protocol(result) == "unknown"
assert get_protocol(result) == "unknown"
def test_determine_protocol_case_insensitive(self):
def test_get_protocol_case_insensitive(self):
"""Test protocol detection is case insensitive."""
assert _determine_protocol({"protocol": "TORRENT"}) == "torrent"
assert _determine_protocol({"protocol": "Usenet"}) == "usenet"
assert _determine_protocol({"protocol": "USENET"}) == "usenet"
assert get_protocol({"protocol": "TORRENT"}) == "torrent"
assert get_protocol({"protocol": "Usenet"}) == "usenet"
assert get_protocol({"protocol": "USENET"}) == "usenet"
class TestProwlarrHandlerDownloadErrors:
@@ -264,7 +262,7 @@ class TestProwlarrHandlerExistingDownload:
)
assert result is not None
assert "processing" in recorder.statuses
assert "resolving" in recorder.statuses
# Should NOT have called add_download
mock_client.add_download.assert_not_called()
+1 -24
View File
@@ -40,39 +40,16 @@ def _setup_transmission_config():
def _setup_qbittorrent_config():
"""Set up qBittorrent configuration via config files and refresh config."""
# qBittorrent generates a temporary password on startup - try to extract it
password = _get_qbittorrent_temp_password()
save_config_file("prowlarr_clients", {
"PROWLARR_TORRENT_CLIENT": "qbittorrent",
"QBITTORRENT_URL": "http://qbittorrent:8080",
"QBITTORRENT_USERNAME": "admin",
"QBITTORRENT_PASSWORD": password or "adminadmin",
"QBITTORRENT_PASSWORD": "admin123",
"QBITTORRENT_CATEGORY": "test",
})
config.refresh()
def _get_qbittorrent_temp_password():
"""Extract qBittorrent's temporary password from container logs."""
import re
# Try mounted config path (from docker-compose volumes)
log_paths = [
"/qbittorrent-config/qBittorrent/logs/qbittorrent.log",
"/config/qBittorrent/logs/qbittorrent.log",
]
for log_path in log_paths:
try:
with open(log_path, "r") as f:
content = f.read()
# Look for: "temporary password is provided for this session: XXXXXX"
match = re.search(r"temporary password[^:]*:\s*(\S+)", content)
if match:
return match.group(1)
except Exception:
continue
return None
def _setup_deluge_config():
"""Set up Deluge configuration via config files and refresh config."""
save_config_file("prowlarr_clients", {
+491
View File
@@ -0,0 +1,491 @@
"""
Integration failure tests for download clients.
These tests verify error handling behavior against REAL running clients.
They require the Docker test stack to be running.
Run with: docker exec test-cwabd python3 -m pytest /app/tests/prowlarr/test_integration_failures.py -v -m integration
Key scenarios tested:
- Invalid magnet links / bad torrents
- Non-existent download IDs
- Duplicate downloads
- Client disconnection recovery
"""
import time
import pytest
from cwa_book_downloader.core.config import config
from cwa_book_downloader.core.settings_registry import save_config_file
from cwa_book_downloader.release_sources.prowlarr.clients import DownloadStatus, DownloadState
# Invalid magnet - valid format but non-existent torrent
INVALID_MAGNET = "magnet:?xt=urn:btih:0000000000000000000000000000000000000000&dn=nonexistent"
# Valid Ubuntu magnet for comparison tests
VALID_MAGNET = "magnet:?xt=urn:btih:3b245504cf5f11bbdbe1201cea6a6bf45aee1bc0&dn=ubuntu-22.04.3-live-server-amd64.iso"
# =============================================================================
# Configuration Setup Functions (copied from test_integration_clients.py)
# =============================================================================
def _setup_transmission_config():
save_config_file("prowlarr_clients", {
"PROWLARR_TORRENT_CLIENT": "transmission",
"TRANSMISSION_URL": "http://transmission:9091",
"TRANSMISSION_USERNAME": "admin",
"TRANSMISSION_PASSWORD": "admin",
"TRANSMISSION_CATEGORY": "test",
})
config.refresh()
def _setup_qbittorrent_config():
save_config_file("prowlarr_clients", {
"PROWLARR_TORRENT_CLIENT": "qbittorrent",
"QBITTORRENT_URL": "http://qbittorrent:8080",
"QBITTORRENT_USERNAME": "admin",
"QBITTORRENT_PASSWORD": "admin123",
"QBITTORRENT_CATEGORY": "test",
})
config.refresh()
def _setup_deluge_config():
save_config_file("prowlarr_clients", {
"PROWLARR_TORRENT_CLIENT": "deluge",
"DELUGE_HOST": "deluge",
"DELUGE_PORT": "58846",
"DELUGE_USERNAME": "admin",
"DELUGE_PASSWORD": "admin",
"DELUGE_CATEGORY": "test",
})
config.refresh()
# =============================================================================
# Client Factory Functions
# =============================================================================
def _try_get_transmission_client():
_setup_transmission_config()
try:
from cwa_book_downloader.release_sources.prowlarr.clients.transmission import TransmissionClient
client = TransmissionClient()
client.test_connection()
return client
except Exception:
return None
def _try_get_qbittorrent_client():
_setup_qbittorrent_config()
try:
from cwa_book_downloader.release_sources.prowlarr.clients.qbittorrent import QBittorrentClient
client = QBittorrentClient()
success, _ = client.test_connection()
if success:
return client
except Exception:
pass
return None
def _try_get_deluge_client():
_setup_deluge_config()
try:
from cwa_book_downloader.release_sources.prowlarr.clients.deluge import DelugeClient
client = DelugeClient()
success, _ = client.test_connection()
if success:
return client
except Exception:
pass
return None
# =============================================================================
# Fixtures
# =============================================================================
@pytest.fixture(scope="module")
def transmission_client():
client = _try_get_transmission_client()
if client is None:
pytest.skip("Transmission not available")
return client
@pytest.fixture(scope="module")
def qbittorrent_client():
client = _try_get_qbittorrent_client()
if client is None:
pytest.skip("qBittorrent not available")
return client
@pytest.fixture(scope="module")
def deluge_client():
client = _try_get_deluge_client()
if client is None:
pytest.skip("Deluge not available")
return client
# =============================================================================
# Non-Existent Download ID Tests
# =============================================================================
@pytest.mark.integration
class TestNonExistentDownloads:
"""Tests for handling non-existent download IDs."""
def test_transmission_get_status_nonexistent(self, transmission_client):
"""Transmission should handle non-existent download ID gracefully."""
status = transmission_client.get_status("nonexistent-hash-12345")
# Should return an error status, not crash
assert isinstance(status, DownloadStatus)
assert status.state == DownloadState.ERROR or status.complete is False
def test_transmission_remove_nonexistent(self, transmission_client):
"""Transmission should handle removing non-existent download."""
# Should not raise, should return False or True depending on implementation
result = transmission_client.remove("nonexistent-hash-12345", delete_files=True)
# Just verify it doesn't crash - implementations vary on return value
assert isinstance(result, bool)
def test_transmission_get_path_nonexistent(self, transmission_client):
"""Transmission should return None for non-existent download path."""
path = transmission_client.get_download_path("nonexistent-hash-12345")
assert path is None
def test_qbittorrent_get_status_nonexistent(self, qbittorrent_client):
"""qBittorrent should handle non-existent download ID gracefully."""
status = qbittorrent_client.get_status("0" * 40) # Valid hash format but nonexistent
assert isinstance(status, DownloadStatus)
assert status.state == DownloadState.ERROR or status.complete is False
def test_qbittorrent_remove_nonexistent(self, qbittorrent_client):
"""qBittorrent should handle removing non-existent download."""
result = qbittorrent_client.remove("0" * 40, delete_files=True)
assert isinstance(result, bool)
def test_deluge_get_status_nonexistent(self, deluge_client):
"""Deluge should handle non-existent download ID gracefully."""
status = deluge_client.get_status("0" * 40)
assert isinstance(status, DownloadStatus)
assert status.state == DownloadState.ERROR or status.complete is False
def test_deluge_remove_nonexistent(self, deluge_client):
"""Deluge should handle removing non-existent download."""
result = deluge_client.remove("0" * 40, delete_files=True)
assert isinstance(result, bool)
# =============================================================================
# Duplicate Download Tests
# =============================================================================
@pytest.mark.integration
class TestDuplicateDownloads:
"""Tests for handling duplicate download requests."""
def test_transmission_add_duplicate(self, transmission_client):
"""Transmission should handle adding same torrent twice."""
client = transmission_client
# Add first time
download_id_1 = client.add_download(
url=VALID_MAGNET,
name="Duplicate Test 1",
)
time.sleep(2)
try:
# Add same magnet again - should either:
# 1. Return the same ID (deduplicated)
# 2. Raise an exception
# 3. Return a new ID (some clients allow this)
try:
download_id_2 = client.add_download(
url=VALID_MAGNET,
name="Duplicate Test 2",
)
# If we get here, client allowed it
# Clean up the duplicate if different
if download_id_2 != download_id_1:
client.remove(download_id_2, delete_files=True)
except Exception as e:
# Expected - client rejected duplicate
assert "duplicate" in str(e).lower() or "exists" in str(e).lower()
finally:
client.remove(download_id_1, delete_files=True)
def test_qbittorrent_add_duplicate(self, qbittorrent_client):
"""qBittorrent should handle adding same torrent twice."""
client = qbittorrent_client
download_id_1 = client.add_download(
url=VALID_MAGNET,
name="Duplicate Test qBit 1",
)
time.sleep(3)
try:
try:
download_id_2 = client.add_download(
url=VALID_MAGNET,
name="Duplicate Test qBit 2",
)
if download_id_2 != download_id_1:
client.remove(download_id_2, delete_files=True)
except Exception:
# Expected behavior
pass
finally:
client.remove(download_id_1, delete_files=True)
# =============================================================================
# Invalid URL/Magnet Tests
# =============================================================================
@pytest.mark.integration
class TestInvalidUrls:
"""Tests for handling invalid URLs and magnets."""
def test_transmission_completely_invalid_url(self, transmission_client):
"""Transmission should reject completely invalid URL."""
with pytest.raises(Exception):
transmission_client.add_download(
url="not-a-valid-url-at-all",
name="Invalid URL Test",
)
def test_transmission_malformed_magnet(self, transmission_client):
"""Transmission should reject malformed magnet link."""
with pytest.raises(Exception):
transmission_client.add_download(
url="magnet:?xt=invalid",
name="Malformed Magnet Test",
)
def test_qbittorrent_completely_invalid_url(self, qbittorrent_client):
"""qBittorrent should reject completely invalid URL."""
with pytest.raises(Exception):
qbittorrent_client.add_download(
url="not-a-valid-url-at-all",
name="Invalid URL Test",
)
def test_deluge_completely_invalid_url(self, deluge_client):
"""Deluge should reject completely invalid URL."""
with pytest.raises(Exception):
deluge_client.add_download(
url="not-a-valid-url-at-all",
name="Invalid URL Test",
)
# =============================================================================
# find_existing Edge Cases
# =============================================================================
@pytest.mark.integration
class TestFindExistingEdgeCases:
"""Tests for find_existing behavior in edge cases."""
def test_transmission_find_nonexistent(self, transmission_client):
"""find_existing should return None for non-existent torrent."""
result = transmission_client.find_existing(INVALID_MAGNET)
assert result is None
def test_qbittorrent_find_nonexistent(self, qbittorrent_client):
"""find_existing should return None for non-existent torrent."""
result = qbittorrent_client.find_existing(INVALID_MAGNET)
assert result is None
def test_deluge_find_nonexistent(self, deluge_client):
"""find_existing should return None for non-existent torrent."""
result = deluge_client.find_existing(INVALID_MAGNET)
assert result is None
def test_transmission_find_with_invalid_url(self, transmission_client):
"""find_existing should handle invalid URL gracefully."""
# Should return None, not crash
result = transmission_client.find_existing("not-a-magnet-link")
assert result is None
def test_qbittorrent_find_with_invalid_url(self, qbittorrent_client):
"""find_existing should handle invalid URL gracefully."""
result = qbittorrent_client.find_existing("not-a-magnet-link")
assert result is None
# =============================================================================
# Connection Loss Simulation (where possible)
# =============================================================================
@pytest.mark.integration
class TestConnectionResilience:
"""Tests for connection resilience."""
def test_transmission_recovers_after_session_id_change(self, transmission_client):
"""Transmission should handle session ID invalidation."""
# First, make a successful call to establish session
transmission_client.test_connection()
# Manually invalidate the session ID if accessible
if hasattr(transmission_client, '_session_id'):
old_session = transmission_client._session_id
transmission_client._session_id = "invalid-session-id"
# Should auto-recover with a new session
success, msg = transmission_client.test_connection()
# Restore or verify it got a new one
assert success or transmission_client._session_id != "invalid-session-id"
def test_qbittorrent_handles_expired_cookie(self, qbittorrent_client):
"""qBittorrent should handle expired session cookie."""
# First, make a successful call
qbittorrent_client.test_connection()
# Clear the session if accessible
if hasattr(qbittorrent_client, '_session'):
qbittorrent_client._session.cookies.clear()
# Should re-authenticate automatically
success, msg = qbittorrent_client.test_connection()
assert success
# =============================================================================
# State Transition Tests
# =============================================================================
@pytest.mark.integration
class TestStateTransitions:
"""Tests verifying correct state reporting during download lifecycle."""
def test_transmission_initial_state_is_valid(self, transmission_client):
"""Newly added torrent should have valid initial state."""
client = transmission_client
download_id = client.add_download(
url=VALID_MAGNET,
name="State Test",
)
try:
# Check immediately
status = client.get_status(download_id)
assert isinstance(status, DownloadStatus)
assert 0 <= status.progress <= 100
assert isinstance(status.complete, bool)
# State should be one of the expected values
valid_states = {
DownloadState.DOWNLOADING,
DownloadState.QUEUED,
DownloadState.CHECKING,
DownloadState.PAUSED,
"downloading",
"queued",
"checking",
"fetching_metadata",
}
assert status.state in valid_states or isinstance(status.state, DownloadState)
finally:
client.remove(download_id, delete_files=True)
def test_qbittorrent_initial_state_is_valid(self, qbittorrent_client):
"""Newly added torrent should have valid initial state."""
client = qbittorrent_client
download_id = client.add_download(
url=VALID_MAGNET,
name="State Test qBit",
)
try:
time.sleep(2) # qBittorrent needs a moment
status = client.get_status(download_id)
assert isinstance(status, DownloadStatus)
assert 0 <= status.progress <= 100
assert isinstance(status.complete, bool)
finally:
client.remove(download_id, delete_files=True)
def test_transmission_reports_metadata_fetching(self, transmission_client):
"""Transmission should report metadata fetching for magnet links."""
client = transmission_client
# Use the invalid magnet - it will stay in metadata fetching state
download_id = client.add_download(
url=INVALID_MAGNET,
name="Metadata Test",
)
try:
time.sleep(2)
status = client.get_status(download_id)
# For an invalid magnet, it should either:
# 1. Be stuck in metadata fetching
# 2. Have an error
# 3. Show 0% progress
assert status.progress == 0 or status.state == DownloadState.ERROR
finally:
client.remove(download_id, delete_files=True)
# =============================================================================
# Timeout Behavior Tests
# =============================================================================
@pytest.mark.integration
@pytest.mark.slow
class TestTimeoutBehavior:
"""Tests for timeout and stall detection."""
def test_transmission_stalled_torrent_state(self, transmission_client):
"""Verify stalled torrent is reported correctly."""
client = transmission_client
# Add invalid magnet - will never find peers
download_id = client.add_download(
url=INVALID_MAGNET,
name="Stall Test",
)
try:
# Wait a bit for it to "stall"
time.sleep(5)
status = client.get_status(download_id)
# Should still be at 0% or have stalled status
assert status.progress == 0
# State could be downloading (but no progress) or specific stall state
assert not status.complete
finally:
client.remove(download_id, delete_files=True)
+12 -11
View File
@@ -14,7 +14,8 @@ import pytest
from cwa_book_downloader.core.config import config
from cwa_book_downloader.core.settings_registry import save_config_file
from cwa_book_downloader.core.models import DownloadTask
from cwa_book_downloader.release_sources.prowlarr.handler import ProwlarrHandler, _determine_protocol
from cwa_book_downloader.release_sources.prowlarr.handler import ProwlarrHandler
from cwa_book_downloader.release_sources.prowlarr.utils import get_protocol
from cwa_book_downloader.release_sources.prowlarr.cache import cache_release, get_release, remove_release, _cache
@@ -72,28 +73,28 @@ class ProgressRecorder:
return [s[0] for s in self.status_updates]
class TestDetermineProtocol:
"""Tests for the _determine_protocol function."""
class TestGetProtocol:
"""Tests for the get_protocol function."""
def test_determine_protocol_torrent(self):
def test_get_protocol_torrent(self):
"""Test detecting torrent protocol."""
result = {"protocol": "torrent"}
assert _determine_protocol(result) == "torrent"
assert get_protocol(result) == "torrent"
def test_determine_protocol_usenet(self):
def test_get_protocol_usenet(self):
"""Test detecting usenet protocol."""
result = {"protocol": "usenet"}
assert _determine_protocol(result) == "usenet"
assert get_protocol(result) == "usenet"
def test_determine_protocol_unknown(self):
def test_get_protocol_unknown(self):
"""Test unknown protocol."""
result = {"protocol": "ftp"}
assert _determine_protocol(result) == "unknown"
assert get_protocol(result) == "unknown"
def test_determine_protocol_empty(self):
def test_get_protocol_empty(self):
"""Test empty protocol."""
result = {}
assert _determine_protocol(result) == "unknown"
assert get_protocol(result) == "unknown"
@pytest.mark.integration