Implement configuration management, drag-and-drop functionality, and logging utilities for WebDrop Bridge
This commit is contained in:
parent
04ef84cf9a
commit
6bef2f6119
9 changed files with 1154 additions and 0 deletions
143
src/webdrop_bridge/config.py
Normal file
143
src/webdrop_bridge/config.py
Normal file
|
|
@ -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})"
|
||||
)
|
||||
106
src/webdrop_bridge/core/drag_interceptor.py
Normal file
106
src/webdrop_bridge/core/drag_interceptor.py
Normal file
|
|
@ -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
|
||||
97
src/webdrop_bridge/core/validator.py
Normal file
97
src/webdrop_bridge/core/validator.py
Normal file
|
|
@ -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
|
||||
67
src/webdrop_bridge/main.py
Normal file
67
src/webdrop_bridge/main.py
Normal file
|
|
@ -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())
|
||||
158
src/webdrop_bridge/ui/main_window.py
Normal file
158
src/webdrop_bridge/ui/main_window.py
Normal file
|
|
@ -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"<html><body><h1>Error</h1>"
|
||||
f"<p>Web application file not found: {file_path}</p>"
|
||||
f"</body></html>"
|
||||
)
|
||||
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"<html><body><h1>Error</h1>"
|
||||
f"<p>Failed to load web application: {e}</p>"
|
||||
f"</body></html>"
|
||||
)
|
||||
|
||||
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)
|
||||
100
src/webdrop_bridge/utils/logging.py
Normal file
100
src/webdrop_bridge/utils/logging.py
Normal file
|
|
@ -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)
|
||||
148
tests/unit/test_config.py
Normal file
148
tests/unit/test_config.py
Normal file
|
|
@ -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()
|
||||
154
tests/unit/test_logging.py
Normal file
154
tests/unit/test_logging.py
Normal file
|
|
@ -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
|
||||
181
tests/unit/test_validator.py
Normal file
181
tests/unit/test_validator.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Add a link
Reference in a new issue