diff --git a/src/webdrop_bridge/config.py b/src/webdrop_bridge/config.py new file mode 100644 index 0000000..ac8bacb --- /dev/null +++ b/src/webdrop_bridge/config.py @@ -0,0 +1,143 @@ +"""Configuration management for WebDrop Bridge application.""" + +import logging +import os +from dataclasses import dataclass +from pathlib import Path +from typing import List + +from dotenv import load_dotenv + +logger = logging.getLogger(__name__) + + +class ConfigurationError(Exception): + """Raised when configuration is invalid.""" + + pass + + +@dataclass +class Config: + """Application configuration loaded from environment variables. + + Attributes: + app_name: Application display name + app_version: Application version (semantic versioning) + log_level: Logging level (DEBUG, INFO, WARNING, ERROR) + log_file: Optional log file path + allowed_roots: List of whitelisted root directories for file access + webapp_url: URL to load in embedded web application + window_width: Initial window width in pixels + window_height: Initial window height in pixels + enable_logging: Whether to write logs to file + + Raises: + ConfigurationError: If configuration values are invalid + """ + + app_name: str + app_version: str + log_level: str + log_file: Path | None + allowed_roots: List[Path] + webapp_url: str + window_width: int + window_height: int + enable_logging: bool + + @classmethod + def from_env(cls, env_file: str | None = None) -> "Config": + """Load configuration from environment variables. + + Looks for a .env file in the current working directory and loads + environment variables from it. All unset variables use sensible defaults. + + Args: + env_file: Optional path to .env file to load. If None, loads default .env + + Returns: + Config: Configured instance with environment values or defaults + + Raises: + ConfigurationError: If any configured value is invalid + """ + # Load .env file if it exists + load_dotenv(env_file) + + # Extract and validate configuration values + app_name = os.getenv("APP_NAME", "WebDrop Bridge") + app_version = os.getenv("APP_VERSION", "1.0.0") + log_level = os.getenv("LOG_LEVEL", "INFO").upper() + log_file_str = os.getenv("LOG_FILE", "logs/webdrop_bridge.log") + allowed_roots_str = os.getenv("ALLOWED_ROOTS", "Z:/,C:/Users/Public") + webapp_url = os.getenv("WEBAPP_URL", "file:///./webapp/index.html") + window_width = int(os.getenv("WINDOW_WIDTH", "1024")) + window_height = int(os.getenv("WINDOW_HEIGHT", "768")) + enable_logging = os.getenv("ENABLE_LOGGING", "true").lower() == "true" + + # Validate log level + valid_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} + if log_level not in valid_levels: + raise ConfigurationError( + f"Invalid LOG_LEVEL: {log_level}. " + f"Must be one of: {', '.join(valid_levels)}" + ) + + # Validate and parse allowed roots + try: + allowed_roots = [] + for p in allowed_roots_str.split(","): + root_path = Path(p.strip()).resolve() + if not root_path.exists(): + raise ConfigurationError( + f"Allowed root '{p.strip()}' does not exist" + ) + if not root_path.is_dir(): + raise ConfigurationError( + f"Allowed root '{p.strip()}' is not a directory" + ) + allowed_roots.append(root_path) + except ConfigurationError: + raise + except (ValueError, OSError) as e: + raise ConfigurationError( + f"Invalid ALLOWED_ROOTS: {allowed_roots_str}. Error: {e}" + ) from e + + # Validate window dimensions + if window_width <= 0 or window_height <= 0: + raise ConfigurationError( + f"Window dimensions must be positive: " + f"{window_width}x{window_height}" + ) + + # Create log file path if logging enabled + log_file = None + if enable_logging: + log_file = Path(log_file_str).resolve() + + # Validate webapp URL is not empty + if not webapp_url: + raise ConfigurationError("WEBAPP_URL cannot be empty") + + return cls( + app_name=app_name, + app_version=app_version, + log_level=log_level, + log_file=log_file, + allowed_roots=allowed_roots, + webapp_url=webapp_url, + window_width=window_width, + window_height=window_height, + enable_logging=enable_logging, + ) + + def __repr__(self) -> str: + """Return developer-friendly representation.""" + return ( + f"Config(app={self.app_name} v{self.app_version}, " + f"log_level={self.log_level}, " + f"allowed_roots={len(self.allowed_roots)} dirs, " + f"window={self.window_width}x{self.window_height})" + ) diff --git a/src/webdrop_bridge/core/drag_interceptor.py b/src/webdrop_bridge/core/drag_interceptor.py new file mode 100644 index 0000000..d3c1dc0 --- /dev/null +++ b/src/webdrop_bridge/core/drag_interceptor.py @@ -0,0 +1,106 @@ +"""Qt widget for intercepting drag events and initiating native drag operations.""" + +from pathlib import Path +from typing import List, Optional + +from PySide6.QtCore import Qt, Signal +from PySide6.QtGui import QDrag, QMimeData +from PySide6.QtWidgets import QWidget + +from webdrop_bridge.core.validator import PathValidator, ValidationError + + +class DragInterceptor(QWidget): + """Widget that handles drag initiation for file paths. + + Intercepts drag events from web content and initiates native Qt drag + operations, allowing files to be dragged from web content to native + applications. + + Signals: + drag_started: Emitted when a drag operation begins successfully + drag_failed: Emitted when drag initiation fails + """ + + # Signals with string parameters (file paths that were dragged) + drag_started = Signal(list) # List[str] - list of file paths + drag_failed = Signal(str) # str - error message + + def __init__(self, parent: Optional[QWidget] = None): + """Initialize the drag interceptor. + + Args: + parent: Parent widget + """ + super().__init__(parent) + self._validator: Optional[PathValidator] = None + + def set_validator(self, validator: PathValidator) -> None: + """Set the path validator for this interceptor. + + Args: + validator: PathValidator instance to use for validation + """ + self._validator = validator + + def initiate_drag(self, file_paths: List[str]) -> bool: + """Initiate a native drag operation for the given files. + + Args: + file_paths: List of file paths to drag + + Returns: + True if drag was successfully initiated, False otherwise + """ + if not file_paths: + self.drag_failed.emit("No files to drag") + return False + + if not self._validator: + self.drag_failed.emit("Validator not configured") + return False + + # Validate all paths first + validated_paths = [] + for path_str in file_paths: + try: + path = Path(path_str) + if self._validator.validate(path): + validated_paths.append(path) + except ValidationError as e: + self.drag_failed.emit(f"Validation failed for {path_str}: {e}") + return False + + if not validated_paths: + self.drag_failed.emit("No valid files after validation") + return False + + # Create MIME data with file URLs + mime_data = QMimeData() + file_urls = [ + path.as_uri() for path in validated_paths + ] + mime_data.setUrls([QUrl(url) for url in file_urls]) + + # Create and execute drag operation + drag = QDrag(self) + drag.setMimeData(mime_data) + # Use default drag pixmap (small icon) + drag.setPixmap(self.grab(self.rect()).scaled( + 64, 64, Qt.AspectRatioMode.KeepAspectRatio + )) + + # Execute drag operation (blocking call) + drop_action = drag.exec(Qt.DropAction.CopyAction) + + # Check result + if drop_action == Qt.DropAction.CopyAction: + self.drag_started.emit(validated_paths) + return True + else: + self.drag_failed.emit("Drag operation cancelled or failed") + return False + + +# Import QUrl here to avoid circular import at module level +from PySide6.QtCore import QUrl # noqa: E402, F401 diff --git a/src/webdrop_bridge/core/validator.py b/src/webdrop_bridge/core/validator.py new file mode 100644 index 0000000..218fe2d --- /dev/null +++ b/src/webdrop_bridge/core/validator.py @@ -0,0 +1,97 @@ +"""Path validation for secure file operations.""" + +from pathlib import Path +from typing import List + + +class ValidationError(Exception): + """Raised when path validation fails.""" + + pass + + +class PathValidator: + """Validates file paths against security whitelist. + + Ensures that only files within allowed root directories can be accessed. + All paths are resolved to absolute form before validation to prevent + directory traversal attacks. + """ + + def __init__(self, allowed_roots: List[Path]): + """Initialize validator with allowed root directories. + + Args: + allowed_roots: List of Path objects representing allowed root dirs + + Raises: + ValidationError: If any root doesn't exist or isn't a directory + """ + self.allowed_roots = [] + + for root in allowed_roots: + root_path = Path(root).resolve() + if not root_path.exists(): + raise ValidationError( + f"Allowed root '{root}' does not exist" + ) + if not root_path.is_dir(): + raise ValidationError( + f"Allowed root '{root}' is not a directory" + ) + self.allowed_roots.append(root_path) + + def validate(self, path: Path) -> bool: + """Validate that path is within an allowed root directory. + + Args: + path: File path to validate + + Returns: + True if path is valid and accessible + + Raises: + ValidationError: If path fails validation + """ + try: + # Resolve to absolute path (handles symlinks, .., etc) + file_path = Path(path).resolve() + except (OSError, ValueError) as e: + raise ValidationError(f"Cannot resolve path '{path}': {e}") from e + + # Check file exists + if not file_path.exists(): + raise ValidationError(f"File does not exist: {path}") + + # Check it's a regular file (not directory, symlink to dir, etc) + if not file_path.is_file(): + raise ValidationError(f"Path is not a regular file: {path}") + + # Check path is within an allowed root + for allowed_root in self.allowed_roots: + try: + # This raises ValueError if file_path is not relative to root + file_path.relative_to(allowed_root) + return True + except ValueError: + continue + + # Not in any allowed root + raise ValidationError( + f"Path '{file_path}' is not within allowed roots: " + f"{self.allowed_roots}" + ) + + def is_valid(self, path: Path) -> bool: + """Check if path is valid without raising exception. + + Args: + path: File path to check + + Returns: + True if valid, False otherwise + """ + try: + return self.validate(path) + except ValidationError: + return False diff --git a/src/webdrop_bridge/main.py b/src/webdrop_bridge/main.py new file mode 100644 index 0000000..e2941c2 --- /dev/null +++ b/src/webdrop_bridge/main.py @@ -0,0 +1,67 @@ +"""WebDrop Bridge - Application entry point.""" + +import sys +from pathlib import Path + +from PySide6.QtWidgets import QApplication + +from webdrop_bridge.config import Config, ConfigurationError +from webdrop_bridge.ui.main_window import MainWindow +from webdrop_bridge.utils.logging import get_logger, setup_logging + + +def main() -> int: + """Main application entry point. + + Initializes configuration, logging, and creates the main window. + Returns appropriate exit code. + + Returns: + int: Exit code (0 for success, non-zero for error) + """ + try: + # Load configuration from environment + config = Config.from_env() + + # Set up logging + log_file = None + if config.log_file_path: + log_file = Path(config.log_file_path) + + setup_logging( + name="webdrop_bridge", + level=config.log_level, + log_file=log_file, + ) + + logger = get_logger(__name__) + logger.info(f"Starting {config.app_name} v{config.app_version}") + logger.debug(f"Configuration: {config}") + + except ConfigurationError as e: + print(f"Configuration error: {e}", file=sys.stderr) + return 1 + except Exception as e: + print(f"Failed to initialize logging: {e}", file=sys.stderr) + return 1 + + try: + # Create Qt application + app = QApplication(sys.argv) + + # Create and show main window + window = MainWindow(config) + window.show() + + logger.info("Main window opened successfully") + + # Run event loop + return app.exec() + + except Exception as e: + logger.exception(f"Unhandled exception: {e}") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/webdrop_bridge/ui/main_window.py b/src/webdrop_bridge/ui/main_window.py new file mode 100644 index 0000000..a23fe59 --- /dev/null +++ b/src/webdrop_bridge/ui/main_window.py @@ -0,0 +1,158 @@ +"""Main application window with web engine integration.""" + +from pathlib import Path +from typing import Optional + +from PySide6.QtCore import Qt, QUrl +from PySide6.QtWebEngineWidgets import QWebEngineView +from PySide6.QtWidgets import QMainWindow, QVBoxLayout, QWidget + +from webdrop_bridge.config import Config +from webdrop_bridge.core.drag_interceptor import DragInterceptor +from webdrop_bridge.core.validator import PathValidator + + +class MainWindow(QMainWindow): + """Main application window for WebDrop Bridge. + + Displays web content in a QWebEngineView and provides drag-and-drop + integration with the native filesystem. + """ + + def __init__( + self, + config: Config, + parent: Optional[QWidget] = None, + ): + """Initialize the main window. + + Args: + config: Application configuration + parent: Parent widget + """ + super().__init__(parent) + self.config = config + + # Set window properties + self.setWindowTitle(f"{config.app_name} v{config.app_version}") + self.setGeometry( + 100, + 100, + config.window_width, + config.window_height, + ) + + # Create web engine view + self.web_view = QWebEngineView() + + # Create drag interceptor + self.drag_interceptor = DragInterceptor() + + # Set up path validator + validator = PathValidator(config.allowed_roots) + self.drag_interceptor.set_validator(validator) + + # Connect drag interceptor signals + self.drag_interceptor.drag_started.connect(self._on_drag_started) + self.drag_interceptor.drag_failed.connect(self._on_drag_failed) + + # Set up central widget with layout + central_widget = QWidget() + layout = QVBoxLayout() + layout.addWidget(self.web_view) + layout.setContentsMargins(0, 0, 0, 0) + central_widget.setLayout(layout) + self.setCentralWidget(central_widget) + + # Load web application + self._load_webapp() + + # Apply styling if available + self._apply_stylesheet() + + def _load_webapp(self) -> None: + """Load the web application. + + Loads HTML from the configured webapp URL or from local file. + """ + webapp_url = self.config.webapp_url + + if webapp_url.startswith("http://") or webapp_url.startswith("https://"): + # Remote URL + self.web_view.load(QUrl(webapp_url)) + else: + # Local file path + try: + file_path = Path(webapp_url).resolve() + if not file_path.exists(): + self.web_view.setHtml( + f"

Error

" + f"

Web application file not found: {file_path}

" + f"" + ) + return + + # Load local file as file:// URL + file_url = file_path.as_uri() + self.web_view.load(QUrl(file_url)) + + except (OSError, ValueError) as e: + self.web_view.setHtml( + f"

Error

" + f"

Failed to load web application: {e}

" + f"" + ) + + def _apply_stylesheet(self) -> None: + """Apply application stylesheet if available.""" + stylesheet_path = Path(__file__).parent.parent.parent.parent / \ + "resources" / "stylesheets" / "default.qss" + + if stylesheet_path.exists(): + try: + with open(stylesheet_path, "r") as f: + stylesheet = f.read() + self.setStyleSheet(stylesheet) + except (OSError, IOError): + # Silently fail if stylesheet can't be read + pass + + def _on_drag_started(self, paths: list) -> None: + """Handle successful drag initiation. + + Args: + paths: List of paths that were dragged + """ + # Can be extended with logging or status bar updates + pass + + def _on_drag_failed(self, error: str) -> None: + """Handle drag operation failure. + + Args: + error: Error message + """ + # Can be extended with logging or user notification + pass + + def closeEvent(self, event) -> None: + """Handle window close event. + + Args: + event: Close event + """ + # Can be extended with save operations or cleanup + event.accept() + + def initiate_drag(self, file_paths: list) -> bool: + """Initiate a drag operation for the given files. + + Called from web content via JavaScript bridge. + + Args: + file_paths: List of file paths to drag + + Returns: + True if drag was initiated successfully + """ + return self.drag_interceptor.initiate_drag(file_paths) diff --git a/src/webdrop_bridge/utils/logging.py b/src/webdrop_bridge/utils/logging.py new file mode 100644 index 0000000..aaafadb --- /dev/null +++ b/src/webdrop_bridge/utils/logging.py @@ -0,0 +1,100 @@ +"""Logging configuration and utilities for WebDrop Bridge.""" + +import logging +import logging.handlers +from pathlib import Path +from typing import Optional + + +def setup_logging( + name: str = "webdrop_bridge", + level: str = "INFO", + log_file: Optional[Path] = None, + fmt: Optional[str] = None, +) -> logging.Logger: + """Configure application-wide logging. + + Sets up both console and file logging (if enabled). All loggers in the + application should use this configured root logger. + + Args: + name: Logger name (typically module name or app name) + level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) + log_file: Optional path to log file. If provided, logs will be written + to this file in addition to console + fmt: Optional custom format string. If None, uses default format. + Default: "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + + Returns: + logging.Logger: Configured logger instance + + Raises: + ValueError: If log_file path is invalid or can't be created + KeyError: If level is not a valid logging level + """ + # Validate logging level + try: + numeric_level = getattr(logging, level.upper()) + except AttributeError as e: + raise KeyError(f"Invalid logging level: {level}") from e + + # Use default format if not provided + if fmt is None: + fmt = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + + # Create formatter + formatter = logging.Formatter(fmt) + + # Get or create logger + logger = logging.getLogger(name) + logger.setLevel(numeric_level) + + # Remove existing handlers to avoid duplicates + logger.handlers.clear() + + # Add console handler + console_handler = logging.StreamHandler() + console_handler.setLevel(numeric_level) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + # Add file handler if log file specified + if log_file: + try: + # Create parent directories if needed + log_file.parent.mkdir(parents=True, exist_ok=True) + + # Use rotating file handler to manage log file size + # Max 10 MB per file, keep 5 backups + file_handler = logging.handlers.RotatingFileHandler( + log_file, + maxBytes=10 * 1024 * 1024, # 10 MB + backupCount=5, + encoding="utf-8", + ) + file_handler.setLevel(numeric_level) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + except (OSError, IOError) as e: + raise ValueError(f"Cannot write to log file {log_file}: {e}") from e + + return logger + + +def get_logger(name: str = __name__) -> logging.Logger: + """Get a logger instance for a module. + + Convenience function to get a logger for a specific module. + Use this in your modules to get properly named loggers. + + Example: + logger = get_logger(__name__) + + Args: + name: Logger name, typically __name__ of the calling module + + Returns: + logging.Logger: Logger instance for the given name + """ + return logging.getLogger(name) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..a9447a2 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,148 @@ +"""Unit tests for configuration system.""" + +import os +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +import pytest + +from webdrop_bridge.config import Config, ConfigurationError + + +@pytest.fixture(autouse=True) +def clear_env(): + """Clear environment variables before each test to avoid persistence.""" + # Save current env + saved_env = os.environ.copy() + + # Clear relevant variables + for key in list(os.environ.keys()): + if key.startswith(('APP_', 'LOG_', 'ALLOWED_', 'WEBAPP_', 'WINDOW_', 'ENABLE_')): + del os.environ[key] + + yield + + # Restore env (cleanup) + os.environ.clear() + os.environ.update(saved_env) + + +class TestConfigFromEnv: + """Test Config.from_env() loading from environment.""" + + def test_from_env_with_all_values(self, tmp_path): + """Test loading config with all environment variables set.""" + # Create .env file + env_file = tmp_path / ".env" + root1 = tmp_path / "root1" + root2 = tmp_path / "root2" + root1.mkdir() + root2.mkdir() + + env_file.write_text( + f"APP_NAME=TestApp\n" + f"APP_VERSION=2.0.0\n" + f"LOG_LEVEL=DEBUG\n" + f"LOG_FILE={tmp_path / 'test.log'}\n" + f"ALLOWED_ROOTS={root1},{root2}\n" + f"WEBAPP_URL=http://localhost:8000\n" + f"WINDOW_WIDTH=1200\n" + f"WINDOW_HEIGHT=800\n" + ) + + # Load config (env vars from file, not system) + config = Config.from_env(str(env_file)) + + assert config.app_name == "TestApp" + assert config.app_version == "2.0.0" + assert config.log_level == "DEBUG" + assert config.allowed_roots == [root1.resolve(), root2.resolve()] + assert config.webapp_url == "http://localhost:8000" + assert config.window_width == 1200 + assert config.window_height == 800 + + def test_from_env_with_defaults(self, tmp_path): + """Test loading config uses defaults when env vars not set.""" + # Create empty .env file + env_file = tmp_path / ".env" + env_file.write_text("") + + config = Config.from_env(str(env_file)) + + assert config.app_name == "WebDrop Bridge" + assert config.app_version == "1.0.0" + assert config.log_level == "INFO" + assert config.window_width == 1024 + assert config.window_height == 768 + + def test_from_env_invalid_log_level(self, tmp_path): + """Test that invalid log level raises ConfigurationError.""" + env_file = tmp_path / ".env" + root1 = tmp_path / "root1" + root1.mkdir() + env_file.write_text(f"LOG_LEVEL=INVALID\nALLOWED_ROOTS={root1}\n") + + with pytest.raises(ConfigurationError, match="Invalid LOG_LEVEL"): + Config.from_env(str(env_file)) + + def test_from_env_invalid_window_dimension(self, tmp_path): + """Test that negative window dimensions raise ConfigurationError.""" + env_file = tmp_path / ".env" + root1 = tmp_path / "root1" + root1.mkdir() + env_file.write_text(f"WINDOW_WIDTH=-100\nALLOWED_ROOTS={root1}\n") + + with pytest.raises(ConfigurationError, match="Window dimensions"): + Config.from_env(str(env_file)) + + def test_from_env_invalid_root_path(self, tmp_path): + """Test that non-existent root paths raise ConfigurationError.""" + env_file = tmp_path / ".env" + env_file.write_text("ALLOWED_ROOTS=/nonexistent/path/that/does/not/exist\n") + + with pytest.raises(ConfigurationError, match="does not exist"): + Config.from_env(str(env_file)) + + def test_from_env_empty_webapp_url(self, tmp_path): + """Test that empty webapp URL raises ConfigurationError.""" + env_file = tmp_path / ".env" + root1 = tmp_path / "root1" + root1.mkdir() + env_file.write_text(f"WEBAPP_URL=\nALLOWED_ROOTS={root1}\n") + + with pytest.raises(ConfigurationError, match="WEBAPP_URL"): + Config.from_env(str(env_file)) + + +class TestConfigValidation: + """Test Config field validation.""" + + def test_root_path_resolution(self, tmp_path): + """Test that root paths are resolved to absolute paths.""" + env_file = tmp_path / ".env" + root_dir = tmp_path / "allowed" + root_dir.mkdir() + + env_file.write_text(f"ALLOWED_ROOTS={root_dir}\n") + + config = Config.from_env(str(env_file)) + + # Should be resolved to absolute path + assert config.allowed_roots[0].is_absolute() + + def test_multiple_root_paths(self, tmp_path): + """Test loading multiple root paths.""" + dir1 = tmp_path / "dir1" + dir2 = tmp_path / "dir2" + dir1.mkdir() + dir2.mkdir() + + env_file = tmp_path / ".env" + env_file.write_text(f"ALLOWED_ROOTS={dir1},{dir2}\n") + + config = Config.from_env(str(env_file)) + + assert len(config.allowed_roots) == 2 + assert config.allowed_roots[0] == dir1.resolve() + assert config.allowed_roots[1] == dir2.resolve() diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py new file mode 100644 index 0000000..d2c7d52 --- /dev/null +++ b/tests/unit/test_logging.py @@ -0,0 +1,154 @@ +"""Unit tests for logging module.""" + +import logging +from pathlib import Path +from tempfile import TemporaryDirectory + +import pytest + +from webdrop_bridge.utils.logging import get_logger, setup_logging + + +class TestSetupLogging: + """Test logging configuration.""" + + def test_setup_logging_console_only(self): + """Test console-only logging setup.""" + logger = setup_logging(name="test_console", level="DEBUG") + + assert logger.name == "test_console" + assert logger.level == logging.DEBUG + assert len(logger.handlers) == 1 + assert isinstance(logger.handlers[0], logging.StreamHandler) + + def test_setup_logging_with_file(self, tmp_path): + """Test logging setup with file handler.""" + log_file = tmp_path / "test.log" + + logger = setup_logging( + name="test_file", + level="INFO", + log_file=log_file, + ) + + assert logger.name == "test_file" + assert len(logger.handlers) == 2 + + # Find file handler + file_handler = None + for handler in logger.handlers: + if isinstance(handler, logging.handlers.RotatingFileHandler): + file_handler = handler + break + + assert file_handler is not None + assert log_file.exists() + + def test_setup_logging_invalid_level(self): + """Test that invalid log level raises KeyError.""" + with pytest.raises(KeyError, match="Invalid logging level"): + setup_logging(level="INVALID") + + def test_setup_logging_invalid_file_path(self): + """Test that inaccessible file path raises ValueError.""" + # Use a path that can't be created (on Windows, CON is reserved) + if Path.cwd().drive: # Windows + invalid_path = Path("CON") / "invalid.log" + else: # Unix - use root-only directory + invalid_path = Path("/root") / "invalid.log" + + with pytest.raises(ValueError, match="Cannot write to log file"): + setup_logging(log_file=invalid_path) + + def test_setup_logging_custom_format(self, tmp_path): + """Test logging with custom format string.""" + log_file = tmp_path / "test.log" + custom_fmt = "%(levelname)s:%(name)s:%(message)s" + + logger = setup_logging( + name="test_format", + level="INFO", + log_file=log_file, + fmt=custom_fmt, + ) + + # Log a test message + logger.info("test message") + + # Check format in log file + content = log_file.read_text() + assert "INFO:test_format:test message" in content + + def test_setup_logging_creates_parent_dirs(self, tmp_path): + """Test that setup_logging creates parent directories.""" + log_file = tmp_path / "nested" / "dir" / "test.log" + + logger = setup_logging( + name="test_nested", + level="INFO", + log_file=log_file, + ) + + logger.info("test") + + assert log_file.exists() + assert log_file.parent.exists() + + def test_setup_logging_removes_duplicates(self): + """Test that multiple setup calls don't duplicate handlers.""" + logger_name = "test_duplicates" + + # First setup + logger1 = setup_logging(name=logger_name, level="DEBUG") + initial_handler_count = len(logger1.handlers) + + # Second setup (should clear old handlers) + logger2 = setup_logging(name=logger_name, level="INFO") + + assert logger2 is logger1 # Same logger object + assert len(logger2.handlers) == initial_handler_count + + +class TestGetLogger: + """Test get_logger convenience function.""" + + def test_get_logger(self): + """Test getting a logger instance.""" + logger = get_logger("test.module") + + assert isinstance(logger, logging.Logger) + assert logger.name == "test.module" + + def test_get_logger_default_name(self): + """Test get_logger uses __name__ by default.""" + logger = get_logger() + + # Should return a logger (when no name provided, uses logging module __name__) + assert isinstance(logger, logging.Logger) + assert logger.name == "webdrop_bridge.utils.logging" + + +class TestLogRotation: + """Test log file rotation.""" + + def test_rotating_file_handler_configured(self, tmp_path): + """Test that file handler is configured for rotation.""" + log_file = tmp_path / "test.log" + + logger = setup_logging( + name="test_rotation", + level="INFO", + log_file=log_file, + ) + + # Find rotating file handler + rotating_handler = None + for handler in logger.handlers: + if isinstance(handler, logging.handlers.RotatingFileHandler): + rotating_handler = handler + break + + assert rotating_handler is not None + # Default: 10 MB max, 5 backups + assert rotating_handler.maxBytes == 10 * 1024 * 1024 + assert rotating_handler.backupCount == 5 diff --git a/tests/unit/test_validator.py b/tests/unit/test_validator.py new file mode 100644 index 0000000..525d46a --- /dev/null +++ b/tests/unit/test_validator.py @@ -0,0 +1,181 @@ +"""Unit tests for path validator.""" + +from pathlib import Path +from tempfile import TemporaryDirectory + +import pytest + +from webdrop_bridge.core.validator import PathValidator, ValidationError + + +class TestPathValidator: + """Test path validation.""" + + def test_validator_initialization(self, tmp_path): + """Test creating a validator with valid roots.""" + dir1 = tmp_path / "dir1" + dir2 = tmp_path / "dir2" + dir1.mkdir() + dir2.mkdir() + + validator = PathValidator([dir1, dir2]) + + assert len(validator.allowed_roots) == 2 + + def test_validator_nonexistent_root(self, tmp_path): + """Test that nonexistent root raises ValidationError.""" + nonexistent = tmp_path / "nonexistent" + + with pytest.raises(ValidationError, match="does not exist"): + PathValidator([nonexistent]) + + def test_validator_non_directory_root(self, tmp_path): + """Test that non-directory root raises ValidationError.""" + file_path = tmp_path / "file.txt" + file_path.write_text("test") + + with pytest.raises(ValidationError, match="not a directory"): + PathValidator([file_path]) + + def test_validate_valid_file(self, tmp_path): + """Test validating a file within allowed root.""" + file_path = tmp_path / "test.txt" + file_path.write_text("test content") + + validator = PathValidator([tmp_path]) + + assert validator.validate(file_path) is True + assert validator.is_valid(file_path) is True + + def test_validate_nonexistent_file(self, tmp_path): + """Test that nonexistent file raises ValidationError.""" + validator = PathValidator([tmp_path]) + nonexistent = tmp_path / "nonexistent.txt" + + with pytest.raises(ValidationError, match="does not exist"): + validator.validate(nonexistent) + + def test_validate_directory_path(self, tmp_path): + """Test that directory path raises ValidationError.""" + subdir = tmp_path / "subdir" + subdir.mkdir() + + validator = PathValidator([tmp_path]) + + with pytest.raises(ValidationError, match="not a regular file"): + validator.validate(subdir) + + def test_validate_file_outside_roots(self, tmp_path): + """Test that file outside allowed roots raises ValidationError.""" + allowed_dir = tmp_path / "allowed" + other_dir = tmp_path / "other" + allowed_dir.mkdir() + other_dir.mkdir() + + file_in_other = other_dir / "test.txt" + file_in_other.write_text("test") + + validator = PathValidator([allowed_dir]) + + with pytest.raises(ValidationError, match="not within allowed roots"): + validator.validate(file_in_other) + + def test_validate_multiple_roots(self, tmp_path): + """Test validating files in multiple allowed roots.""" + dir1 = tmp_path / "dir1" + dir2 = tmp_path / "dir2" + dir1.mkdir() + dir2.mkdir() + + file1 = dir1 / "file1.txt" + file2 = dir2 / "file2.txt" + file1.write_text("content1") + file2.write_text("content2") + + validator = PathValidator([dir1, dir2]) + + assert validator.validate(file1) is True + assert validator.validate(file2) is True + + def test_validate_with_relative_path(self, tmp_path): + """Test validating with relative path (gets resolved).""" + import os + + file_path = tmp_path / "test.txt" + file_path.write_text("test") + + validator = PathValidator([tmp_path]) + + # Change to tmp_path directory + original_cwd = os.getcwd() + try: + os.chdir(tmp_path) + # Use relative path + assert validator.validate(Path("test.txt")) is True + finally: + os.chdir(original_cwd) + + def test_validate_with_path_traversal(self, tmp_path): + """Test that path traversal attacks are blocked.""" + allowed_dir = tmp_path / "allowed" + other_dir = tmp_path / "other" + allowed_dir.mkdir() + other_dir.mkdir() + + file_in_other = other_dir / "secret.txt" + file_in_other.write_text("secret") + + validator = PathValidator([allowed_dir]) + + # Try to access file outside root using .. + traversal_path = allowed_dir / ".." / "other" / "secret.txt" + + with pytest.raises(ValidationError, match="not within allowed roots"): + validator.validate(traversal_path) + + def test_is_valid_doesnt_raise(self, tmp_path): + """Test that is_valid() never raises exceptions.""" + validator = PathValidator([tmp_path]) + + # These should all return False, not raise + assert validator.is_valid(Path("/nonexistent")) is False + assert validator.is_valid(tmp_path) is False # Directory, not file + + # Valid file should return True + file_path = tmp_path / "test.txt" + file_path.write_text("test") + assert validator.is_valid(file_path) is True + + +class TestPathValidatorEdgeCases: + """Test edge cases in path validation.""" + + def test_symlink_to_valid_file(self, tmp_path): + """Test validating a symlink to a valid file.""" + # Skip on Windows if symlink creation fails + actual_file = tmp_path / "actual.txt" + actual_file.write_text("content") + + try: + symlink = tmp_path / "link.txt" + symlink.symlink_to(actual_file) + + validator = PathValidator([tmp_path]) + # Symlinks resolve to their target, should validate + assert validator.validate(symlink) is True + + except (OSError, NotImplementedError): + # Skip if symlinks not supported + pytest.skip("Symlinks not supported on this platform") + + def test_nested_files_in_allowed_root(self, tmp_path): + """Test validating files in nested subdirectories.""" + nested_dir = tmp_path / "a" / "b" / "c" + nested_dir.mkdir(parents=True) + + nested_file = nested_dir / "file.txt" + nested_file.write_text("content") + + validator = PathValidator([tmp_path]) + + assert validator.validate(nested_file) is True