quaik8/sdb2-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Utils lint fixes

Browse Source
This commit is contained in:
JSC
2025-07-31 21:56:03 +02:00
parent 8847131f24
commit 01bb48c206
4 changed files with 85 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/utils/audio.py
View File

@@ -34,7 +34,7 @@ def get_audio_duration(file_path: Path) -> int:
probe = ffmpeg.probe(str(file_path)) probe = ffmpeg.probe(str(file_path))
duration = float(probe["format"]["duration"]) duration = float(probe["format"]["duration"])
return int(duration * 1000) # Convert to milliseconds return int(duration * 1000) # Convert to milliseconds
except Exception as e: except (ffmpeg.Error, KeyError, ValueError, TypeError, Exception) as e:
logger.warning("Failed to get duration for %s: %s", file_path, e) logger.warning("Failed to get duration for %s: %s", file_path, e)
return 0 return 0

10
app/utils/cookies.py
View File

@@ -3,14 +3,14 @@
def parse_cookies(cookie_header: str) -> dict[str, str]: def parse_cookies(cookie_header: str) -> dict[str, str]:
"""Parse HTTP cookie header into a dictionary.""" """Parse HTTP cookie header into a dictionary."""
cookies = {} cookies: dict[str, str] = {}
if not cookie_header: if not cookie_header:
return cookies return cookies
for cookie in cookie_header.split(";"): for cookie_part in cookie_header.split(";"):
cookie = cookie.strip() cookie_str = cookie_part.strip()
if "=" in cookie: if "=" in cookie_str:
name, value = cookie.split("=", 1) name, value = cookie_str.split("=", 1)
cookies[name.strip()] = value.strip() cookies[name.strip()] = value.strip()
return cookies return cookies

23
app/utils/credit_decorators.py
View File

@@ -1,6 +1,8 @@
"""Decorators for credit management and validation.""" """Decorators for credit management and validation."""
import functools import functools
import inspect
import types
from collections.abc import Awaitable, Callable from collections.abc import Awaitable, Callable
from typing import Any, TypeVar from typing import Any, TypeVar
@@ -16,7 +18,7 @@ def requires_credits(
user_id_param: str = "user_id", user_id_param: str = "user_id",
metadata_extractor: Callable[..., dict[str, Any]] | None = None, metadata_extractor: Callable[..., dict[str, Any]] | None = None,
) -> Callable[[F], F]: ) -> Callable[[F], F]:
"""Decorator to enforce credit requirements for actions. """Enforce credit requirements for actions.
Args: Args:
action_type: The type of action that requires credits action_type: The type of action that requires credits
@@ -40,14 +42,13 @@ def requires_credits(
""" """
def decorator(func: F) -> F: def decorator(func: F) -> F:
@functools.wraps(func) @functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any: async def wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: ANN401
# Extract user ID from parameters # Extract user ID from parameters
user_id = None user_id = None
if user_id_param in kwargs: if user_id_param in kwargs:
user_id = kwargs[user_id_param] user_id = kwargs[user_id_param]
else: else:
# Try to find user_id in function signature # Try to find user_id in function signature
import inspect
sig = inspect.signature(func) sig = inspect.signature(func)
param_names = list(sig.parameters.keys()) param_names = list(sig.parameters.keys())
if user_id_param in param_names: if user_id_param in param_names:
@@ -74,14 +75,14 @@ def requires_credits(
# Execute the function # Execute the function
success = False success = False
result = None
try: try:
result = await func(*args, **kwargs) result = await func(*args, **kwargs)
success = bool(result) # Consider function result as success indicator success = bool(result) # Consider function result as success indicator
return result
except Exception: except Exception:
success = False success = False
raise raise
else:
return result
finally: finally:
# Deduct credits based on success # Deduct credits based on success
await credit_service.deduct_credits( await credit_service.deduct_credits(
@@ -97,7 +98,7 @@ def validate_credits_only(
credit_service_factory: Callable[[], CreditService], credit_service_factory: Callable[[], CreditService],
user_id_param: str = "user_id", user_id_param: str = "user_id",
) -> Callable[[F], F]: ) -> Callable[[F], F]:
"""Decorator to only validate credits without deducting them. """Validate credits without deducting them.
Useful for checking if a user can perform an action before actual execution. Useful for checking if a user can perform an action before actual execution.
@@ -112,14 +113,13 @@ def validate_credits_only(
""" """
def decorator(func: F) -> F: def decorator(func: F) -> F:
@functools.wraps(func) @functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any: async def wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: ANN401
# Extract user ID from parameters # Extract user ID from parameters
user_id = None user_id = None
if user_id_param in kwargs: if user_id_param in kwargs:
user_id = kwargs[user_id_param] user_id = kwargs[user_id_param]
else: else:
# Try to find user_id in function signature # Try to find user_id in function signature
import inspect
sig = inspect.signature(func) sig = inspect.signature(func)
param_names = list(sig.parameters.keys()) param_names = list(sig.parameters.keys())
if user_id_param in param_names: if user_id_param in param_names:
@@ -178,7 +178,12 @@ class CreditManager:
self.validated = True self.validated = True
return self return self
async def __aexit__(self, exc_type: type, exc_val: Exception, exc_tb: Any) -> None: async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: types.TracebackType | None,
) -> None:
"""Exit context manager - deduct credits based on success.""" """Exit context manager - deduct credits based on success."""
if self.validated: if self.validated:
# If no exception occurred, consider it successful # If no exception occurred, consider it successful

131
tests/services/test_vlc_player.py
View File

@@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, Mock, patch
import pytest import pytest
from app.models.credit_transaction import CreditTransaction
from app.models.sound import Sound from app.models.sound import Sound
from app.models.user import User from app.models.user import User
from app.services.vlc_player import VLCPlayerService, get_vlc_player_service from app.services.vlc_player import VLCPlayerService, get_vlc_player_service
@@ -323,52 +324,51 @@ class TestVLCPlayerService:
with patch("app.services.vlc_player.SoundRepository", return_value=mock_sound_repo): with patch("app.services.vlc_player.SoundRepository", return_value=mock_sound_repo):
with patch("app.services.vlc_player.UserRepository", return_value=mock_user_repo): with patch("app.services.vlc_player.UserRepository", return_value=mock_user_repo):
with patch("app.services.vlc_player.socket_manager") as mock_socket: with patch("app.services.vlc_player.socket_manager") as mock_socket:
with patch("app.services.vlc_player.select") as mock_select: # Mock the file path utility
# Mock the file path utility with patch("app.services.vlc_player.get_sound_file_path") as mock_get_path:
with patch("app.services.vlc_player.get_sound_file_path") as mock_get_path: mock_path = Mock()
mock_path = Mock() mock_path.exists.return_value = True
mock_path.exists.return_value = True mock_get_path.return_value = mock_path
mock_get_path.return_value = mock_path
# Mock sound repository responses # Mock sound repository responses
updated_sound = Sound( updated_sound = Sound(
id=1, id=1,
type="SDB", type="SDB",
name="Test Sound", name="Test Sound",
filename="test.mp3", filename="test.mp3",
duration=5000, duration=5000,
size=1024, size=1024,
hash="test_hash", hash="test_hash",
play_count=1, # Updated count play_count=1, # Updated count
) )
mock_sound_repo.get_by_id.return_value = sample_sound mock_sound_repo.get_by_id.return_value = sample_sound
mock_sound_repo.update.return_value = updated_sound mock_sound_repo.update.return_value = updated_sound
# Mock admin user # Mock admin user
admin_user = User( admin_user = User(
id=1, id=1,
email="admin@test.com", email="admin@test.com",
name="Admin User", name="Admin User",
role="admin", role="admin",
) )
mock_user_repo.get_by_id.return_value = admin_user mock_user_repo.get_by_id.return_value = admin_user
# Mock socket broadcast # Mock socket broadcast
mock_socket.broadcast_to_all = AsyncMock() mock_socket.broadcast_to_all = AsyncMock()
result = await vlc_service_with_db.play_sound(sample_sound) result = await vlc_service_with_db.play_sound(sample_sound)
# Wait a bit for the async task to complete # Wait a bit for the async task to complete
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
assert result is True assert result is True
# Verify subprocess was called # Verify subprocess was called
mock_subprocess.assert_called_once() mock_subprocess.assert_called_once()
# Note: The async task runs in the background, so we can't easily # Note: The async task runs in the background, so we can't easily
# verify the database operations in this test without more complex # verify the database operations in this test without more complex
# mocking or using a real async test framework setup # mocking or using a real async test framework setup
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_record_play_count_success(self, vlc_service_with_db): async def test_record_play_count_success(self, vlc_service_with_db):
@@ -401,40 +401,39 @@ class TestVLCPlayerService:
with patch("app.services.vlc_player.SoundRepository", return_value=mock_sound_repo): with patch("app.services.vlc_player.SoundRepository", return_value=mock_sound_repo):
with patch("app.services.vlc_player.UserRepository", return_value=mock_user_repo): with patch("app.services.vlc_player.UserRepository", return_value=mock_user_repo):
with patch("app.services.vlc_player.socket_manager") as mock_socket: with patch("app.services.vlc_player.socket_manager") as mock_socket:
with patch("app.services.vlc_player.select") as mock_select: # Setup mocks
# Setup mocks mock_sound_repo.get_by_id.return_value = test_sound
mock_sound_repo.get_by_id.return_value = test_sound mock_user_repo.get_by_id.return_value = admin_user
mock_user_repo.get_by_id.return_value = admin_user
# Mock socket broadcast # Mock socket broadcast
mock_socket.broadcast_to_all = AsyncMock() mock_socket.broadcast_to_all = AsyncMock()
await vlc_service_with_db._record_play_count(1, "Test Sound") await vlc_service_with_db._record_play_count(1, "Test Sound")
# Verify sound repository calls # Verify sound repository calls
mock_sound_repo.get_by_id.assert_called_once_with(1) mock_sound_repo.get_by_id.assert_called_once_with(1)
mock_sound_repo.update.assert_called_once_with( mock_sound_repo.update.assert_called_once_with(
test_sound, {"play_count": 1}, test_sound, {"play_count": 1},
) )
# Verify user repository calls # Verify user repository calls
mock_user_repo.get_by_id.assert_called_once_with(1) mock_user_repo.get_by_id.assert_called_once_with(1)
# Verify session operations # Verify session operations
mock_session.add.assert_called_once() mock_session.add.assert_called_once()
mock_session.commit.assert_called_once() mock_session.commit.assert_called_once()
mock_session.close.assert_called_once() mock_session.close.assert_called_once()
# Verify socket broadcast # Verify socket broadcast
mock_socket.broadcast_to_all.assert_called_once_with( mock_socket.broadcast_to_all.assert_called_once_with(
"sound_played", "sound_played",
{ {
"sound_id": 1, "sound_id": 1,
"sound_name": "Test Sound", "sound_name": "Test Sound",
"user_id": 1, "user_id": 1,
"play_count": 1, "play_count": 1,
}, },
) )
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_record_play_count_no_session_factory(self, vlc_service): async def test_record_play_count_no_session_factory(self, vlc_service):