Cache calls to frequently called functions
This commit is contained in:
arabcoders
parent
d48faed4f7
commit
215a023da8
11 changed files with 812 additions and 171 deletions
1
.vscode/settings.json
vendored
1
.vscode/settings.json
vendored
|
|
@ -40,6 +40,7 @@
|
|||
"copyts",
|
||||
"creationflags",
|
||||
"cronsim",
|
||||
"currsize",
|
||||
"dailymotion",
|
||||
"datas",
|
||||
"dateparser",
|
||||
|
|
|
|||
|
|
@ -1251,6 +1251,9 @@ class DownloadQueue(metaclass=Singleton):
|
|||
if task.cancelled():
|
||||
return
|
||||
|
||||
if exc := task.exception():
|
||||
task_name: str = task.get_name() if task.get_name() else "unknown_task"
|
||||
LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {traceback.format_exc()}")
|
||||
if not (exc := task.exception()):
|
||||
return
|
||||
|
||||
task_name: str = task.get_name() if task.get_name() else "unknown_task"
|
||||
tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
|
||||
LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {tb}")
|
||||
|
|
|
|||
|
|
@ -1,13 +1,14 @@
|
|||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
from email.utils import formatdate
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from app.library.Utils import archive_add, archive_delete, archive_read, clean_item, get_archive_id
|
||||
from app.library.encoder import Encoder
|
||||
from app.library.Utils import archive_add, archive_delete, archive_read, clean_item, get_archive_id, get_file
|
||||
from app.library.YTDLPOpts import YTDLPOpts
|
||||
|
||||
LOG: logging.Logger = logging.getLogger("ItemDTO")
|
||||
|
|
@ -47,12 +48,37 @@ class Item:
|
|||
"""If the item should be started automatically."""
|
||||
|
||||
def serialize(self) -> dict:
|
||||
"""
|
||||
Serialize the item to a dictionary.
|
||||
|
||||
Returns:
|
||||
dict: The serialized item.
|
||||
|
||||
"""
|
||||
return self.__dict__.copy()
|
||||
|
||||
def json(self) -> str:
|
||||
return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4)
|
||||
"""
|
||||
Convert the item to a JSON string.
|
||||
|
||||
Returns:
|
||||
str: The JSON string representation of the item.
|
||||
|
||||
"""
|
||||
return Encoder(sort_keys=True, indent=4).encode(self.serialize())
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
"""
|
||||
Get a value from the item by key.
|
||||
|
||||
Args:
|
||||
key (str): The key to get the value for.
|
||||
default (Any): The default value to return if the key is not found.
|
||||
|
||||
Returns:
|
||||
Any: The value for the key, or the default value if the key is not found
|
||||
|
||||
"""
|
||||
return self.__dict__.get(key, default)
|
||||
|
||||
def has_extras(self) -> bool:
|
||||
|
|
@ -77,6 +103,13 @@ class Item:
|
|||
|
||||
@staticmethod
|
||||
def _default_preset() -> str:
|
||||
"""
|
||||
Get the default preset from the configuration.
|
||||
|
||||
Returns:
|
||||
str: The default preset name.
|
||||
|
||||
"""
|
||||
from .config import Config
|
||||
|
||||
return Config.get_instance().default_preset
|
||||
|
|
@ -167,12 +200,33 @@ class Item:
|
|||
return Item(**data)
|
||||
|
||||
def get_archive_id(self) -> str | None:
|
||||
"""
|
||||
Get the archive ID for the download URL.
|
||||
|
||||
Returns:
|
||||
str | None: The archive ID if available, None otherwise.
|
||||
|
||||
"""
|
||||
return get_archive_id(self.url).get("archive_id") if self.url else None
|
||||
|
||||
def get_extractor(self) -> str | None:
|
||||
"""
|
||||
Get the extractor key for the download URL.
|
||||
|
||||
Returns:
|
||||
str | None: The extractor key if available, None otherwise.
|
||||
|
||||
"""
|
||||
return get_archive_id(self.url).get("ie_key") if self.url else None
|
||||
|
||||
def get_ytdlp_opts(self) -> YTDLPOpts:
|
||||
"""
|
||||
Get the yt-dlp options for the item.
|
||||
|
||||
Returns:
|
||||
YTDLPOpts: The yt-dlp options for the item.
|
||||
|
||||
"""
|
||||
params: YTDLPOpts = YTDLPOpts.get_instance()
|
||||
|
||||
if self.preset:
|
||||
|
|
@ -184,14 +238,35 @@ class Item:
|
|||
return params
|
||||
|
||||
def get_archive_file(self) -> str | None:
|
||||
"""
|
||||
Get the archive file path from the yt-dlp options.
|
||||
|
||||
Returns:
|
||||
str | None: The archive file path if available, None otherwise.
|
||||
|
||||
"""
|
||||
return self.get_ytdlp_opts().get_all().get("download_archive")
|
||||
|
||||
def is_archived(self) -> bool:
|
||||
"""
|
||||
Check if the item has been archived.
|
||||
|
||||
Returns:
|
||||
bool: True if the item has been archived, False otherwise.
|
||||
|
||||
"""
|
||||
archive_id: str | None = self.get_archive_id()
|
||||
archive_file: str | None = self.get_archive_file()
|
||||
return len(archive_read(archive_file, [archive_id])) > 0 if archive_file and archive_id else False
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""
|
||||
Get a short string representation of the item.
|
||||
|
||||
Returns:
|
||||
str: A short string representation of the item.
|
||||
|
||||
"""
|
||||
from .config import Config
|
||||
from .Utils import calc_download_path, strip_newline
|
||||
|
||||
|
|
@ -288,6 +363,13 @@ class ItemDTO:
|
|||
_archive_file: str | None = None
|
||||
|
||||
def serialize(self) -> dict:
|
||||
"""
|
||||
Serialize the item to a dictionary.
|
||||
|
||||
Returns:
|
||||
dict: The serialized item.
|
||||
|
||||
"""
|
||||
if "finished" == self.status and not self._recomputed:
|
||||
self.archive_status()
|
||||
|
||||
|
|
@ -295,17 +377,84 @@ class ItemDTO:
|
|||
return item
|
||||
|
||||
def json(self) -> str:
|
||||
return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4)
|
||||
"""
|
||||
Convert the item to a JSON string.
|
||||
|
||||
Returns:
|
||||
str: The JSON string representation of the item.
|
||||
|
||||
"""
|
||||
return Encoder(sort_keys=True, indent=4).encode(self.serialize())
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
"""
|
||||
Get a value from the item by key.
|
||||
|
||||
Args:
|
||||
key (str): The key to get the value for.
|
||||
default (Any): The default value to return if the key is not found.
|
||||
|
||||
Returns:
|
||||
Any: The value for the key, or the default value if the key is not found
|
||||
|
||||
"""
|
||||
return self.__dict__.get(key, default)
|
||||
|
||||
def get_id(self) -> str:
|
||||
"""
|
||||
Get the unique identifier for the item.
|
||||
|
||||
Returns:
|
||||
str: The unique identifier for the item.
|
||||
|
||||
"""
|
||||
return self._id
|
||||
|
||||
def name(self) -> str:
|
||||
"""
|
||||
Get a short string representation of the item.
|
||||
|
||||
Returns:
|
||||
str: A short string representation of the item.
|
||||
|
||||
"""
|
||||
return f'id="{self.id}", title="{self.title}"'
|
||||
|
||||
def get_file(self, download_path: Path | None = None) -> Path | None:
|
||||
"""
|
||||
Get the file path of the downloaded item.
|
||||
|
||||
Args:
|
||||
download_path (Path | None): The base download path. If None, it will be taken from the config.
|
||||
|
||||
Returns:
|
||||
Path | None: The file path of the downloaded item, or None if not available.
|
||||
|
||||
"""
|
||||
if not self.filename:
|
||||
return None
|
||||
|
||||
if not download_path:
|
||||
from .config import Config
|
||||
|
||||
base_path = Path(Config.get_instance().download_path)
|
||||
else:
|
||||
base_path = download_path if isinstance(download_path, Path) else Path(download_path)
|
||||
|
||||
filename = base_path
|
||||
|
||||
if self.folder:
|
||||
filename: Path = filename / self.folder
|
||||
|
||||
filename = filename / self.filename
|
||||
|
||||
try:
|
||||
realFile, status = get_file(download_path=base_path, file=str(filename.relative_to(base_path)))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
return Path(realFile) if status in (200, 302) else None
|
||||
|
||||
def get_archive_id(self) -> str | None:
|
||||
"""
|
||||
Get the archive ID for the download URL.
|
||||
|
|
@ -326,6 +475,13 @@ class ItemDTO:
|
|||
return self.archive_id
|
||||
|
||||
def get_extractor(self) -> str | None:
|
||||
"""
|
||||
Get the extractor key for the download URL.
|
||||
|
||||
Returns:
|
||||
str | None: The extractor key if available, None otherwise.
|
||||
|
||||
"""
|
||||
if self.archive_id:
|
||||
return self.archive_id.split(" ")[0]
|
||||
|
||||
|
|
@ -353,6 +509,13 @@ class ItemDTO:
|
|||
return params
|
||||
|
||||
def archive_status(self, force: bool = False) -> None:
|
||||
"""
|
||||
Recompute the archive status of the item.
|
||||
|
||||
Args:
|
||||
force (bool): If True, force recomputation even if already computed.
|
||||
|
||||
"""
|
||||
if not force and (self._recomputed or not self.archive_id):
|
||||
return
|
||||
|
||||
|
|
@ -415,7 +578,13 @@ class ItemDTO:
|
|||
|
||||
@staticmethod
|
||||
def removed_fields() -> tuple:
|
||||
"""Fields that once existed but are no longer used."""
|
||||
"""
|
||||
Fields that once existed but are no longer used.
|
||||
|
||||
Returns:
|
||||
tuple: A tuple of field names that are no longer used.
|
||||
|
||||
"""
|
||||
return (
|
||||
"thumbnail",
|
||||
"quality",
|
||||
|
|
@ -431,6 +600,9 @@ class ItemDTO:
|
|||
)
|
||||
|
||||
def __post_init__(self):
|
||||
"""
|
||||
Post-initialization to compute archive status if applicable.
|
||||
"""
|
||||
self.get_archive_id()
|
||||
self.get_archive_file()
|
||||
self.archive_status()
|
||||
|
|
|
|||
|
|
@ -1,9 +1,9 @@
|
|||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import traceback
|
||||
from collections.abc import Awaitable
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -464,36 +464,36 @@ class Notification(metaclass=Singleton):
|
|||
try:
|
||||
LOG.info(f"Sending Notification event '{ev.event}: {ev.id}' to '{target.name}'.")
|
||||
|
||||
reqBody: dict[str, Any] = {
|
||||
"method": target.request.method.upper(),
|
||||
"url": target.request.url,
|
||||
"headers": {
|
||||
"User-Agent": f"YTPTube/{self._version}",
|
||||
"X-Event-Id": ev.id,
|
||||
"X-Event": ev.event,
|
||||
"Content-Type": "application/json"
|
||||
if "json" == target.request.type.lower()
|
||||
else "application/x-www-form-urlencoded",
|
||||
},
|
||||
headers: dict[str, str] = {
|
||||
"User-Agent": f"YTPTube/{self._version}",
|
||||
"X-Event-Id": ev.id,
|
||||
"X-Event": ev.event,
|
||||
"Content-Type": "application/json"
|
||||
if "json" == target.request.type.lower()
|
||||
else "application/x-www-form-urlencoded",
|
||||
}
|
||||
|
||||
if len(target.request.headers) > 0:
|
||||
for h in target.request.headers:
|
||||
reqBody["headers"][h.key] = h.value
|
||||
headers.update({h.key: h.value for h in target.request.headers if h.key and h.value})
|
||||
|
||||
body_key: str = "json" if "json" == target.request.type.lower() else "data"
|
||||
reqBody[body_key] = self._deep_unpack(ev.serialize())
|
||||
payload: dict = ev.serialize()
|
||||
|
||||
if "data" != target.request.data_key:
|
||||
reqBody[body_key][target.request.data_key] = reqBody[body_key]["data"]
|
||||
reqBody[body_key].pop("data", None)
|
||||
payload[target.request.data_key] = payload["data"]
|
||||
payload.pop("data", None)
|
||||
|
||||
if "form" == target.request.type.lower():
|
||||
reqBody[body_key][target.request.data_key] = self._encoder.encode(
|
||||
reqBody[body_key][target.request.data_key]
|
||||
)
|
||||
payload[target.request.data_key] = self._encoder.encode(payload[target.request.data_key])
|
||||
else:
|
||||
payload = self._encoder.encode(payload)
|
||||
|
||||
response = await self._client.request(**reqBody)
|
||||
response = await self._client.request(
|
||||
method=target.request.method.upper(),
|
||||
url=target.request.url,
|
||||
headers=headers,
|
||||
data=payload if "form" == target.request.type.lower() else None,
|
||||
content=payload if "json" == target.request.type.lower() else None,
|
||||
)
|
||||
|
||||
respData: dict[str, Any] = {
|
||||
"url": target.request.url,
|
||||
|
|
@ -512,7 +512,9 @@ class Notification(metaclass=Singleton):
|
|||
err_msg = str(e)
|
||||
if not err_msg:
|
||||
err_msg: str = type(e).__name__
|
||||
LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'.")
|
||||
|
||||
tb = "".join(traceback.format_exception(type(e), e, e.__traceback__))
|
||||
LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'. {tb}")
|
||||
return {"url": target.request.url, "status": 500, "text": str(ev)}
|
||||
|
||||
def emit(self, e: Event, _, **__) -> None:
|
||||
|
|
@ -522,18 +524,5 @@ class Notification(metaclass=Singleton):
|
|||
self._offload.submit(self.send, e)
|
||||
return
|
||||
|
||||
def _deep_unpack(self, data: dict) -> dict:
|
||||
for k, v in data.items():
|
||||
if isinstance(v, dict):
|
||||
data[k] = self._deep_unpack(v)
|
||||
if isinstance(v, list):
|
||||
data[k] = [self._deep_unpack(i) for i in v]
|
||||
if isinstance(v, datetime):
|
||||
data[k] = v.isoformat()
|
||||
if isinstance(v, object) and hasattr(v, "serialize"):
|
||||
data[k] = v.serialize()
|
||||
|
||||
return data
|
||||
|
||||
async def noop(self) -> None:
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -8,9 +8,10 @@ import os
|
|||
import re
|
||||
import shlex
|
||||
import socket
|
||||
import time
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from functools import lru_cache
|
||||
from functools import lru_cache, wraps
|
||||
from http.cookiejar import MozillaCookieJar
|
||||
from pathlib import Path
|
||||
from typing import Any, TypeVar
|
||||
|
|
@ -87,8 +88,6 @@ DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}
|
|||
T = TypeVar("T")
|
||||
"Generic type variable."
|
||||
|
||||
ARCHIVE_IDS_CACHE: dict[str, dict] = {}
|
||||
"Cache for archive IDs."
|
||||
|
||||
class StreamingError(Exception):
|
||||
"""Raised when an error occurs during streaming."""
|
||||
|
|
@ -99,6 +98,87 @@ class FileLogFormatter(logging.Formatter):
|
|||
return datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds")
|
||||
|
||||
|
||||
def timed_lru_cache(ttl_seconds: int, max_size: int = 128):
|
||||
"""
|
||||
Decorator that applies an LRU cache with a time-to-live (TTL) to a function.
|
||||
Supports both synchronous and asynchronous functions.
|
||||
|
||||
Args:
|
||||
ttl_seconds (int): Time-to-live in seconds.
|
||||
max_size (int): Maximum size of the cache.
|
||||
|
||||
Returns:
|
||||
Decorated function with caching capabilities.
|
||||
|
||||
"""
|
||||
import inspect
|
||||
|
||||
def decorator(func):
|
||||
is_async = inspect.iscoroutinefunction(func)
|
||||
|
||||
if is_async:
|
||||
# For async functions, we need to cache the actual result, not the coroutine
|
||||
cache = {}
|
||||
cache_expiry = {}
|
||||
|
||||
@wraps(func)
|
||||
async def async_wrapper(*args, **kwargs):
|
||||
key = str(args) + str(sorted(kwargs.items()))
|
||||
current_time = time.monotonic()
|
||||
|
||||
# Check if we have a cached result that hasn't expired
|
||||
if key in cache and key in cache_expiry and current_time < cache_expiry[key]:
|
||||
return cache[key]
|
||||
|
||||
# Call the function and cache the result
|
||||
result = await func(*args, **kwargs)
|
||||
cache[key] = result
|
||||
cache_expiry[key] = current_time + ttl_seconds
|
||||
|
||||
# Limit cache size
|
||||
if len(cache) > max_size:
|
||||
# Remove oldest entries
|
||||
oldest_keys = sorted(cache_expiry.keys(), key=lambda k: cache_expiry[k])[:len(cache) - max_size]
|
||||
for old_key in oldest_keys:
|
||||
cache.pop(old_key, None)
|
||||
cache_expiry.pop(old_key, None)
|
||||
|
||||
return result
|
||||
|
||||
# Expose cache management methods
|
||||
def cache_clear():
|
||||
cache.clear()
|
||||
cache_expiry.clear()
|
||||
|
||||
def cache_info():
|
||||
# Use functools._CacheInfo for compatibility with standard lru_cache
|
||||
from functools import _CacheInfo
|
||||
return _CacheInfo(hits=0, misses=len(cache), maxsize=max_size, currsize=len(cache))
|
||||
|
||||
async_wrapper.cache_clear = cache_clear
|
||||
async_wrapper.cache_info = cache_info
|
||||
return async_wrapper
|
||||
|
||||
# For sync functions, use the original implementation
|
||||
cached = lru_cache(maxsize=max_size)(func)
|
||||
cached_expiry = time.monotonic() + ttl_seconds
|
||||
|
||||
@wraps(func)
|
||||
def sync_wrapper(*args, **kwargs):
|
||||
nonlocal cached_expiry
|
||||
if time.monotonic() >= cached_expiry:
|
||||
cached.cache_clear()
|
||||
cached_expiry = time.monotonic() + ttl_seconds
|
||||
return cached(*args, **kwargs)
|
||||
|
||||
# expose cache_clear, cache_info
|
||||
sync_wrapper.cache_clear = cached.cache_clear
|
||||
sync_wrapper.cache_info = cached.cache_info
|
||||
return sync_wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
def calc_download_path(base_path: str | Path, folder: str | None = None, create_path: bool = True) -> str:
|
||||
"""
|
||||
Calculates download path and prevents folder traversal.
|
||||
|
|
@ -387,7 +467,7 @@ def check_id(file: Path) -> bool | str:
|
|||
return False
|
||||
|
||||
|
||||
@lru_cache(maxsize=512)
|
||||
@timed_lru_cache(ttl_seconds=300, max_size=256)
|
||||
def is_private_address(hostname: str) -> bool:
|
||||
ip: str = socket.gethostbyname(hostname)
|
||||
ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address = ipaddress.ip_address(ip)
|
||||
|
|
@ -545,6 +625,7 @@ def validate_uuid(uuid_str: str, version: int = 4) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
@timed_lru_cache(ttl_seconds=60, max_size=256)
|
||||
def get_file_sidecar(file: Path) -> list[dict]:
|
||||
"""
|
||||
Get sidecar files for the given file.
|
||||
|
|
@ -595,7 +676,6 @@ def get_file_sidecar(file: Path) -> list[dict]:
|
|||
return files
|
||||
|
||||
|
||||
@lru_cache(maxsize=512)
|
||||
def get_possible_images(dir: str) -> list[dict]:
|
||||
images: list = []
|
||||
|
||||
|
|
@ -1093,6 +1173,7 @@ def load_cookies(file: str | Path) -> tuple[bool, MozillaCookieJar]:
|
|||
raise ValueError(msg) from e
|
||||
|
||||
|
||||
@timed_lru_cache(ttl_seconds=300, max_size=256)
|
||||
def get_archive_id(url: str) -> dict[str, str | None]:
|
||||
"""
|
||||
Get the archive ID for a given URL.
|
||||
|
|
@ -1116,9 +1197,6 @@ def get_archive_id(url: str) -> dict[str, str | None]:
|
|||
"archive_id": None,
|
||||
}
|
||||
|
||||
if url in ARCHIVE_IDS_CACHE:
|
||||
return ARCHIVE_IDS_CACHE[url]
|
||||
|
||||
if YTDLP_INFO_CLS is None:
|
||||
YTDLP_INFO_CLS = YTDLP(
|
||||
params={
|
||||
|
|
@ -1151,7 +1229,6 @@ def get_archive_id(url: str) -> dict[str, str | None]:
|
|||
LOG.exception(e)
|
||||
LOG.error(f"Error getting archive ID: {e}")
|
||||
|
||||
ARCHIVE_IDS_CACHE.update({url: idDict})
|
||||
return idDict
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -10,12 +10,10 @@ import operator
|
|||
import os
|
||||
import subprocess # qa: ignore
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import anyio
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from app.library.cache import Cache
|
||||
from app.library.Utils import timed_lru_cache
|
||||
|
||||
LOG: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -238,7 +236,8 @@ class FFProbeResult:
|
|||
}
|
||||
|
||||
|
||||
async def ffprobe(file: str) -> FFProbeResult:
|
||||
@timed_lru_cache(ttl_seconds=300, max_size=128)
|
||||
async def ffprobe(file: Path|str) -> FFProbeResult:
|
||||
"""
|
||||
Run ffprobe on a file and return the parsed data as a dictionary.
|
||||
|
||||
|
|
@ -249,21 +248,12 @@ async def ffprobe(file: str) -> FFProbeResult:
|
|||
dict: A dictionary containing the parsed data.
|
||||
|
||||
"""
|
||||
from app.library.Services import Services
|
||||
|
||||
f = Path(file)
|
||||
f = Path(file) if isinstance(file, str) else file
|
||||
|
||||
if not f.exists():
|
||||
msg = f"No such media file '{file}'."
|
||||
raise OSError(msg)
|
||||
|
||||
cache: Cache | None = Services.get_instance().get("cache")
|
||||
cache_key: str = f"ffprobe:{f!s}:{f.stat().st_size}"
|
||||
|
||||
if cache and (cached := cache.get(cache_key)):
|
||||
LOG.debug(f"ffprobe cache hit for '{cache_key}'")
|
||||
return cached
|
||||
|
||||
try:
|
||||
async with await anyio.open_file(os.devnull, "w") as tempf:
|
||||
await asyncio.create_subprocess_exec(
|
||||
|
|
@ -310,7 +300,4 @@ async def ffprobe(file: str) -> FFProbeResult:
|
|||
elif stream.is_attachment():
|
||||
result.attachment.append(stream)
|
||||
|
||||
if cache:
|
||||
cache.set(cache_key, result, ttl=300)
|
||||
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -1,12 +1,10 @@
|
|||
import asyncio
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from aiohttp import web
|
||||
from aiohttp.web import Request, Response
|
||||
|
||||
from app.library.config import Config
|
||||
from app.library.Download import Download
|
||||
from app.library.DownloadQueue import DownloadQueue
|
||||
from app.library.encoder import Encoder
|
||||
|
|
@ -73,7 +71,7 @@ async def item_delete(request: Request, queue: DownloadQueue, encoder: Encoder)
|
|||
|
||||
|
||||
@route("GET", "api/history/{id}", "item_view")
|
||||
async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, config: Config) -> Response:
|
||||
async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder) -> Response:
|
||||
"""
|
||||
Update an item in the history.
|
||||
|
||||
|
|
@ -82,7 +80,6 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, co
|
|||
queue (DownloadQueue): The download queue instance.
|
||||
encoder (Encoder): The encoder instance.
|
||||
notify (EventBus): The event bus instance.
|
||||
config (Config): The configuration instance.
|
||||
|
||||
Returns:
|
||||
Response: The response object.
|
||||
|
|
@ -104,23 +101,11 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, co
|
|||
"ffprobe": {},
|
||||
}
|
||||
|
||||
if item.info.filename:
|
||||
if "finished" == item.info.status and (filename := item.info.get_file()):
|
||||
from app.library.ffprobe import ffprobe
|
||||
|
||||
try:
|
||||
from app.library.ffprobe import ffprobe
|
||||
from app.library.Utils import get_file
|
||||
|
||||
filename = Path(config.download_path)
|
||||
if item.info.folder:
|
||||
filename: Path = filename / item.info.folder
|
||||
|
||||
filename: Path = filename / item.info.filename
|
||||
|
||||
if filename.exists():
|
||||
realFile, status = get_file(
|
||||
download_path=config.download_path, file=str(filename.relative_to(config.download_path))
|
||||
)
|
||||
if status in (web.HTTPOk.status_code, web.HTTPFound.status_code):
|
||||
info["ffprobe"] = await ffprobe(str(realFile))
|
||||
info["ffprobe"] = await ffprobe(filename)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
|
|
|||
207
app/tests/test_ffprobe.py
Normal file
207
app/tests/test_ffprobe.py
Normal file
|
|
@ -0,0 +1,207 @@
|
|||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestFFProbe:
|
||||
"""Test the ffprobe module functionality."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up test files."""
|
||||
self.temp_dir = tempfile.mkdtemp()
|
||||
self.test_file = Path(self.temp_dir) / "test_video.mp4"
|
||||
self.test_file.touch()
|
||||
|
||||
def teardown_method(self):
|
||||
"""Clean up test files."""
|
||||
import shutil
|
||||
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ffprobe_with_existing_file(self):
|
||||
"""Test ffprobe with an existing file."""
|
||||
from app.library.ffprobe import FFProbeResult, ffprobe
|
||||
|
||||
# Mock subprocess to avoid actual ffprobe execution
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
||||
# Mock the subprocess result
|
||||
mock_process = AsyncMock()
|
||||
mock_process.wait.return_value = 0
|
||||
mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}}', b"")
|
||||
mock_subprocess.return_value = mock_process
|
||||
|
||||
with patch("anyio.open_file") as mock_open_file:
|
||||
mock_file = AsyncMock()
|
||||
mock_open_file.return_value.__aenter__.return_value = mock_file
|
||||
|
||||
result = await ffprobe(str(self.test_file))
|
||||
assert isinstance(result, FFProbeResult)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ffprobe_with_nonexistent_file(self):
|
||||
"""Test ffprobe with a non-existent file."""
|
||||
from app.library.ffprobe import ffprobe
|
||||
|
||||
nonexistent_file = Path(self.temp_dir) / "does_not_exist.mp4"
|
||||
|
||||
with pytest.raises(OSError, match="No such media file"):
|
||||
await ffprobe(str(nonexistent_file))
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ffprobe_caching_behavior(self):
|
||||
"""Test that ffprobe results are cached with enhanced async timed_lru_cache."""
|
||||
from app.library.ffprobe import ffprobe
|
||||
|
||||
# Test that the function has been decorated with caching
|
||||
assert hasattr(ffprobe, "cache_clear"), "ffprobe should have cache_clear method from timed_lru_cache"
|
||||
assert hasattr(ffprobe, "cache_info"), "ffprobe should have cache_info method from timed_lru_cache"
|
||||
|
||||
# Clear cache to start fresh
|
||||
ffprobe.cache_clear()
|
||||
|
||||
# Mock subprocess to avoid actual ffprobe execution
|
||||
call_count = 0
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
||||
def create_mock_process(*_args, **_kwargs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
mock_process = AsyncMock()
|
||||
mock_process.wait.return_value = 0
|
||||
mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}, "streams": []}', b"")
|
||||
return mock_process
|
||||
|
||||
mock_subprocess.side_effect = create_mock_process
|
||||
|
||||
with patch("anyio.open_file") as mock_open_file:
|
||||
mock_file = AsyncMock()
|
||||
mock_open_file.return_value.__aenter__.return_value = mock_file
|
||||
|
||||
# First call should execute the function
|
||||
result1 = await ffprobe(str(self.test_file))
|
||||
assert result1 is not None
|
||||
assert isinstance(result1.metadata, dict)
|
||||
first_call_count = call_count
|
||||
|
||||
# Second call with same argument should use cached result
|
||||
result2 = await ffprobe(str(self.test_file))
|
||||
assert result2 is not None
|
||||
assert isinstance(result2.metadata, dict)
|
||||
|
||||
# The subprocess should not be called again for the actual ffprobe execution
|
||||
# (it may be called for the -h check, but the main execution should be cached)
|
||||
assert call_count == first_call_count, "Second call should use cached result"
|
||||
|
||||
# Results should be equivalent (same data, may not be same object due to async nature)
|
||||
assert result1.metadata == result2.metadata
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ffprobe_with_path_object(self):
|
||||
"""Test ffprobe with Path object input."""
|
||||
from app.library.ffprobe import ffprobe
|
||||
|
||||
# Mock subprocess to avoid actual ffprobe execution
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
||||
mock_process = AsyncMock()
|
||||
mock_process.wait.return_value = 0
|
||||
mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}}', b"")
|
||||
mock_subprocess.return_value = mock_process
|
||||
|
||||
with patch("anyio.open_file") as mock_open_file:
|
||||
mock_file = AsyncMock()
|
||||
mock_open_file.return_value.__aenter__.return_value = mock_file
|
||||
|
||||
# Test with Path object
|
||||
result = await ffprobe(self.test_file)
|
||||
assert result is not None
|
||||
|
||||
def test_ffprobe_result_properties(self):
|
||||
"""Test FFProbeResult object properties."""
|
||||
from app.library.ffprobe import FFProbeResult, FFStream
|
||||
|
||||
result = FFProbeResult()
|
||||
|
||||
# Test empty result
|
||||
assert result.video == []
|
||||
assert result.audio == []
|
||||
assert result.subtitle == []
|
||||
assert result.attachment == []
|
||||
assert result.metadata == {}
|
||||
|
||||
# Test adding streams
|
||||
video_stream = FFStream({
|
||||
"index": 0,
|
||||
"codec_type": "video",
|
||||
"codec_name": "h264",
|
||||
"width": 1920,
|
||||
"height": 1080
|
||||
})
|
||||
|
||||
audio_stream = FFStream({
|
||||
"index": 1,
|
||||
"codec_type": "audio",
|
||||
"codec_name": "aac",
|
||||
"channels": 2
|
||||
})
|
||||
|
||||
result.video.append(video_stream)
|
||||
result.audio.append(audio_stream)
|
||||
|
||||
assert len(result.video) == 1
|
||||
assert len(result.audio) == 1
|
||||
assert result.video[0].is_video()
|
||||
assert result.audio[0].is_audio()
|
||||
|
||||
def test_stream_object_methods(self):
|
||||
"""Test Stream object methods."""
|
||||
from app.library.ffprobe import FFStream
|
||||
|
||||
# Test video stream
|
||||
video_stream = FFStream({
|
||||
"codec_type": "video",
|
||||
"codec_name": "h264",
|
||||
"width": 1920,
|
||||
"height": 1080,
|
||||
"duration": "10.5"
|
||||
})
|
||||
|
||||
assert video_stream.is_video()
|
||||
assert not video_stream.is_audio()
|
||||
assert not video_stream.is_subtitle()
|
||||
assert not video_stream.is_attachment()
|
||||
|
||||
# Test audio stream
|
||||
audio_stream = FFStream({
|
||||
"codec_type": "audio",
|
||||
"codec_name": "aac",
|
||||
"channels": 2,
|
||||
"sample_rate": "44100"
|
||||
})
|
||||
|
||||
assert not audio_stream.is_video()
|
||||
assert audio_stream.is_audio()
|
||||
assert not audio_stream.is_subtitle()
|
||||
assert not audio_stream.is_attachment()
|
||||
|
||||
# Test subtitle stream
|
||||
subtitle_stream = FFStream({
|
||||
"codec_type": "subtitle",
|
||||
"codec_name": "subrip"
|
||||
})
|
||||
|
||||
assert not subtitle_stream.is_video()
|
||||
assert not subtitle_stream.is_audio()
|
||||
assert subtitle_stream.is_subtitle()
|
||||
assert not subtitle_stream.is_attachment()
|
||||
|
||||
# Test attachment stream
|
||||
attachment_stream = FFStream({
|
||||
"codec_type": "attachment",
|
||||
"codec_name": "ttf"
|
||||
})
|
||||
|
||||
assert not attachment_stream.is_video()
|
||||
assert not attachment_stream.is_audio()
|
||||
assert not attachment_stream.is_subtitle()
|
||||
assert attachment_stream.is_attachment()
|
||||
|
|
@ -165,3 +165,76 @@ class TestItemDTO:
|
|||
ok2 = dto.archive_delete()
|
||||
assert ok2 is True
|
||||
m_del.assert_called_once()
|
||||
|
||||
def test_get_file_method(self):
|
||||
"""Test ItemDTO get_file method returns correct path."""
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
# Create ItemDTO with filename but no folder
|
||||
dto = ItemDTO(
|
||||
id="test-id",
|
||||
title="Test Video",
|
||||
url="https://youtube.com/watch?v=test123",
|
||||
folder="",
|
||||
status="finished",
|
||||
filename="test_video.mp4",
|
||||
)
|
||||
|
||||
# Test with no filename returns None
|
||||
dto_no_file = ItemDTO(
|
||||
id="test-id-2",
|
||||
title="Test Video 2",
|
||||
url="https://youtube.com/watch?v=test456",
|
||||
folder="",
|
||||
status="finished",
|
||||
)
|
||||
assert dto_no_file.get_file() is None
|
||||
|
||||
# Mock get_file function to return success (without custom download_path)
|
||||
with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config:
|
||||
mock_get_file.return_value = ("/downloads/test_video.mp4", 200)
|
||||
mock_config.get_instance.return_value.download_path = "/downloads"
|
||||
|
||||
result = dto.get_file()
|
||||
assert result == Path("/downloads/test_video.mp4")
|
||||
|
||||
# Test with folder
|
||||
dto_with_folder = ItemDTO(
|
||||
id="test-id-3",
|
||||
title="Test Video 3",
|
||||
url="https://youtube.com/watch?v=test789",
|
||||
folder="media",
|
||||
status="finished",
|
||||
filename="test_video.mp4",
|
||||
)
|
||||
|
||||
with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config:
|
||||
mock_get_file.return_value = ("/downloads/media/test_video.mp4", 200)
|
||||
mock_config.get_instance.return_value.download_path = "/downloads"
|
||||
|
||||
result = dto_with_folder.get_file()
|
||||
assert result == Path("/downloads/media/test_video.mp4")
|
||||
|
||||
# Test with file not found
|
||||
with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config:
|
||||
mock_get_file.return_value = ("/downloads/test_video.mp4", 404)
|
||||
mock_config.get_instance.return_value.download_path = "/downloads"
|
||||
|
||||
result = dto.get_file()
|
||||
assert result is None
|
||||
|
||||
# Test with exception during file access
|
||||
with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config:
|
||||
mock_get_file.side_effect = ValueError("File path error")
|
||||
mock_config.get_instance.return_value.download_path = "/downloads"
|
||||
|
||||
result = dto.get_file()
|
||||
assert result is None
|
||||
|
||||
# Test with custom download_path parameter (Config not imported in this case)
|
||||
with patch("app.library.ItemDTO.get_file") as mock_get_file:
|
||||
mock_get_file.return_value = ("/custom/test_video.mp4", 200)
|
||||
|
||||
result = dto.get_file(download_path=Path("/custom"))
|
||||
assert result == Path("/custom/test_video.mp4")
|
||||
|
|
|
|||
|
|
@ -1003,87 +1003,6 @@ class TestNotification:
|
|||
assert result is None
|
||||
mock_worker_instance.submit.assert_called_once_with(notification.send, event)
|
||||
|
||||
def test_deep_unpack_simple_dict(self):
|
||||
"""Test _deep_unpack method with simple dictionary."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
data = {"key1": "value1", "key2": 123}
|
||||
result = notification._deep_unpack(data)
|
||||
|
||||
assert result == {"key1": "value1", "key2": 123}
|
||||
|
||||
def test_deep_unpack_nested_dict(self):
|
||||
"""Test _deep_unpack method with nested dictionary."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
data = {
|
||||
"level1": {
|
||||
"level2": {
|
||||
"value": "nested"
|
||||
}
|
||||
}
|
||||
}
|
||||
result = notification._deep_unpack(data)
|
||||
|
||||
assert result["level1"]["level2"]["value"] == "nested"
|
||||
|
||||
def test_deep_unpack_with_datetime(self):
|
||||
"""Test _deep_unpack method with datetime objects."""
|
||||
from datetime import UTC, datetime
|
||||
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
test_datetime = datetime.now(tz=UTC)
|
||||
data = {"timestamp": test_datetime}
|
||||
result = notification._deep_unpack(data)
|
||||
|
||||
assert result["timestamp"] == test_datetime.isoformat()
|
||||
|
||||
def test_deep_unpack_with_serializable_object(self):
|
||||
"""Test _deep_unpack method with object having serialize method."""
|
||||
with patch("app.library.Notifications.Config") as mock_config, \
|
||||
patch("app.library.Notifications.BackgroundWorker") as mock_worker:
|
||||
|
||||
mock_config.get_instance.return_value = Mock(
|
||||
debug=False, config_path="/tmp", app_version="1.0.0"
|
||||
)
|
||||
mock_worker.get_instance.return_value = Mock()
|
||||
|
||||
notification = Notification.get_instance()
|
||||
|
||||
# Mock object with serialize method
|
||||
mock_obj = Mock()
|
||||
mock_obj.serialize.return_value = {"serialized": "data"}
|
||||
|
||||
data = {"object": mock_obj}
|
||||
result = notification._deep_unpack(data)
|
||||
|
||||
assert result["object"] == {"serialized": "data"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_noop(self):
|
||||
"""Test noop method."""
|
||||
|
|
|
|||
|
|
@ -44,6 +44,7 @@ from app.library.Utils import (
|
|||
str_to_dt,
|
||||
strip_newline,
|
||||
tail_log,
|
||||
timed_lru_cache,
|
||||
validate_url,
|
||||
validate_uuid,
|
||||
ytdlp_reject,
|
||||
|
|
@ -65,6 +66,233 @@ class TestStreamingError:
|
|||
assert isinstance(error, Exception)
|
||||
|
||||
|
||||
class TestTimedLruCache:
|
||||
"""Test the timed_lru_cache decorator."""
|
||||
|
||||
def test_timed_lru_cache_basic_functionality(self):
|
||||
"""Test that timed_lru_cache caches function results."""
|
||||
call_count = 0
|
||||
|
||||
@timed_lru_cache(ttl_seconds=60, max_size=10)
|
||||
def test_function(x):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
return x * 2
|
||||
|
||||
# First call should execute the function
|
||||
result1 = test_function(5)
|
||||
assert result1 == 10
|
||||
assert call_count == 1
|
||||
|
||||
# Second call with same argument should return cached result
|
||||
result2 = test_function(5)
|
||||
assert result2 == 10
|
||||
assert call_count == 1 # Should not increment
|
||||
|
||||
# Different argument should execute the function again
|
||||
result3 = test_function(10)
|
||||
assert result3 == 20
|
||||
assert call_count == 2
|
||||
|
||||
def test_timed_lru_cache_expiration(self):
|
||||
"""Test that cache expires after TTL."""
|
||||
import time
|
||||
|
||||
call_count = 0
|
||||
|
||||
@timed_lru_cache(ttl_seconds=1, max_size=10) # 1 second TTL
|
||||
def test_function(x):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
return x * 2
|
||||
|
||||
# First call
|
||||
result1 = test_function(5)
|
||||
assert result1 == 10
|
||||
assert call_count == 1
|
||||
|
||||
# Second call immediately should be cached
|
||||
result2 = test_function(5)
|
||||
assert result2 == 10
|
||||
assert call_count == 1
|
||||
|
||||
# Wait for cache to expire
|
||||
time.sleep(1.1)
|
||||
|
||||
# Call after expiration should execute function again
|
||||
result3 = test_function(5)
|
||||
assert result3 == 10
|
||||
assert call_count == 2
|
||||
|
||||
def test_timed_lru_cache_methods_exposed(self):
|
||||
"""Test that cache_clear and cache_info methods are exposed."""
|
||||
from app.library.Utils import timed_lru_cache
|
||||
|
||||
@timed_lru_cache(ttl_seconds=60, max_size=10)
|
||||
def test_function(x):
|
||||
return x * 2
|
||||
|
||||
# Test that methods exist
|
||||
assert hasattr(test_function, "cache_clear")
|
||||
assert hasattr(test_function, "cache_info")
|
||||
|
||||
# Call function to populate cache
|
||||
test_function(5)
|
||||
|
||||
# Get cache info
|
||||
info = test_function.cache_info()
|
||||
assert info.hits == 0
|
||||
assert info.misses == 1
|
||||
|
||||
# Call again to test hit
|
||||
test_function(5)
|
||||
info = test_function.cache_info()
|
||||
assert info.hits == 1
|
||||
assert info.misses == 1
|
||||
|
||||
# Clear cache
|
||||
test_function.cache_clear()
|
||||
info = test_function.cache_info()
|
||||
assert info.hits == 0
|
||||
assert info.misses == 0
|
||||
|
||||
def test_timed_lru_cache_max_size(self):
|
||||
"""Test that cache respects max_size limit."""
|
||||
from app.library.Utils import timed_lru_cache
|
||||
|
||||
@timed_lru_cache(ttl_seconds=60, max_size=2)
|
||||
def test_function(x):
|
||||
return x * 2
|
||||
|
||||
# Fill cache to max size
|
||||
test_function(1)
|
||||
test_function(2)
|
||||
|
||||
info = test_function.cache_info()
|
||||
assert info.misses == 2
|
||||
|
||||
# Adding another item should not exceed max size
|
||||
test_function(3)
|
||||
info = test_function.cache_info()
|
||||
assert info.misses == 3
|
||||
|
||||
# Test that earlier entries may be evicted
|
||||
test_function(1) # This might be a cache miss due to LRU eviction
|
||||
|
||||
|
||||
class TestAsyncTimedLruCache:
|
||||
"""Test async functionality of timed_lru_cache decorator."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_timed_lru_cache_basic(self):
|
||||
"""Test basic async caching functionality."""
|
||||
from app.library.Utils import timed_lru_cache
|
||||
|
||||
call_count = 0
|
||||
|
||||
@timed_lru_cache(ttl_seconds=300, max_size=128)
|
||||
async def async_test_func(x):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
return x * 2
|
||||
|
||||
# First call should execute the function
|
||||
result1 = await async_test_func(5)
|
||||
assert result1 == 10
|
||||
assert call_count == 1
|
||||
|
||||
# Second call should use cached result
|
||||
result2 = await async_test_func(5)
|
||||
assert result2 == 10
|
||||
assert call_count == 1 # Should not increment
|
||||
|
||||
# Different argument should execute function again
|
||||
result3 = await async_test_func(3)
|
||||
assert result3 == 6
|
||||
assert call_count == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_timed_lru_cache_expiry(self):
|
||||
"""Test that async cache entries expire after TTL."""
|
||||
from app.library.Utils import timed_lru_cache
|
||||
|
||||
call_count = 0
|
||||
|
||||
@timed_lru_cache(ttl_seconds=0.1, max_size=128) # 100ms TTL
|
||||
async def async_expire_func(x):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
return x * 3
|
||||
|
||||
# First call
|
||||
result1 = await async_expire_func(2)
|
||||
assert result1 == 6
|
||||
assert call_count == 1
|
||||
|
||||
# Immediate second call should use cache
|
||||
result2 = await async_expire_func(2)
|
||||
assert result2 == 6
|
||||
assert call_count == 1
|
||||
|
||||
# Wait for cache to expire
|
||||
await asyncio.sleep(0.15)
|
||||
|
||||
# Third call should execute function again due to expiry
|
||||
result3 = await async_expire_func(2)
|
||||
assert result3 == 6
|
||||
assert call_count == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_cache_methods(self):
|
||||
"""Test that async cached functions expose cache management methods."""
|
||||
from app.library.Utils import timed_lru_cache
|
||||
|
||||
@timed_lru_cache(ttl_seconds=300, max_size=128)
|
||||
async def async_method_test(x):
|
||||
return x + 1
|
||||
|
||||
# Test that cache methods exist
|
||||
assert hasattr(async_method_test, "cache_clear")
|
||||
assert hasattr(async_method_test, "cache_info")
|
||||
|
||||
# Test cache_info
|
||||
info = async_method_test.cache_info()
|
||||
assert hasattr(info, "hits")
|
||||
assert hasattr(info, "misses")
|
||||
assert hasattr(info, "maxsize")
|
||||
assert hasattr(info, "currsize")
|
||||
assert info.maxsize == 128
|
||||
|
||||
# Test cache_clear
|
||||
await async_method_test(1)
|
||||
async_method_test.cache_clear()
|
||||
info_after_clear = async_method_test.cache_info()
|
||||
assert info_after_clear.currsize == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_cache_max_size(self):
|
||||
"""Test that async cache respects max_size parameter."""
|
||||
from app.library.Utils import timed_lru_cache
|
||||
|
||||
@timed_lru_cache(ttl_seconds=300, max_size=2)
|
||||
async def async_limited_func(x):
|
||||
return x * 4
|
||||
|
||||
# Fill cache beyond max_size
|
||||
result1 = await async_limited_func(1)
|
||||
result2 = await async_limited_func(2)
|
||||
result3 = await async_limited_func(3) # Should evict oldest entry
|
||||
|
||||
# Verify results
|
||||
assert result1 == 4
|
||||
assert result2 == 8
|
||||
assert result3 == 12
|
||||
|
||||
# Check cache size is limited
|
||||
info = async_limited_func.cache_info()
|
||||
assert info.currsize <= 2
|
||||
|
||||
|
||||
class TestFileLogFormatter:
|
||||
"""Test the FileLogFormatter class."""
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue