Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions src/py_moodle/compat.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
import requests
from bs4 import BeautifulSoup, Tag

from .config import DEFAULT_REQUEST_TIMEOUT


@dataclass(frozen=True)
class MoodleVersion:
Expand Down Expand Up @@ -266,7 +268,7 @@ def detect_moodle_compatibility(
response = session.post(
f"{base_url}/webservice/rest/server.php",
params=request_params,
timeout=30,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
data = response.json()
Expand All @@ -278,7 +280,10 @@ def detect_moodle_compatibility(

if version.major is None:
try:
response = session.get(f"{base_url}/my/")
response = session.get(
f"{base_url}/my/",
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
version = extract_version_from_dashboard(response.text)
except requests.RequestException:
Expand Down
24 changes: 24 additions & 0 deletions src/py_moodle/config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
"""Shared timeout defaults for HTTP operations.

Attributes:
DEFAULT_REQUEST_TIMEOUT: Standard timeout in seconds for routine Moodle
GET and POST requests.
DEFAULT_SCRAPE_TIMEOUT: Shorter timeout in seconds for quick HTML fetches
used as fallback probes.
DEFAULT_UPLOAD_TIMEOUT: Tuple of ``(connect_timeout, read_timeout)`` used
for typical webservice uploads.
DEFAULT_LARGE_UPLOAD_TIMEOUT: Tuple of ``(connect_timeout, read_timeout)``
used for larger package transfers such as SCORM and draft uploads.
"""

DEFAULT_REQUEST_TIMEOUT = 30
DEFAULT_SCRAPE_TIMEOUT = 15
DEFAULT_UPLOAD_TIMEOUT = (30, 3600)
DEFAULT_LARGE_UPLOAD_TIMEOUT = (300, 3600)

__all__ = [
"DEFAULT_LARGE_UPLOAD_TIMEOUT",
"DEFAULT_REQUEST_TIMEOUT",
"DEFAULT_SCRAPE_TIMEOUT",
"DEFAULT_UPLOAD_TIMEOUT",
]
7 changes: 4 additions & 3 deletions src/py_moodle/draftfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

from py_moodle.module import MoodleModuleError

from .config import DEFAULT_LARGE_UPLOAD_TIMEOUT, DEFAULT_REQUEST_TIMEOUT
from .upload import ProgressTracker


Expand All @@ -37,7 +38,7 @@ def get_new_draft_itemid(session: requests.Session, base_url: str, sesskey: str)
]

try:
response = session.post(url, json=payload)
response = session.post(url, json=payload, timeout=DEFAULT_REQUEST_TIMEOUT)
response.raise_for_status()
data = response.json()

Expand Down Expand Up @@ -106,7 +107,7 @@ def detect_upload_repo(session: requests.Session, base_url: str, course_id: int)
"""
page_url = f"{base_url}/course/edit.php?id={course_id}"
try:
response = session.get(page_url)
response = session.get(page_url, timeout=DEFAULT_REQUEST_TIMEOUT)
response.raise_for_status()
except requests.RequestException as e:
raise MoodleDraftFileError(
Expand Down Expand Up @@ -144,7 +145,7 @@ def upload_file_to_draft_area(
file_path: str,
itemid: Optional[int] = None,
savepath: str = "/",
timeout: tuple = (300, 3600),
timeout: tuple = DEFAULT_LARGE_UPLOAD_TIMEOUT,
progress_callback: Optional[Callable[[int], None]] = None,
) -> Tuple[int, str]:
"""
Expand Down
3 changes: 2 additions & 1 deletion src/py_moodle/module.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
from bs4 import BeautifulSoup

from py_moodle.compat import get_session_compatibility
from py_moodle.config import DEFAULT_SCRAPE_TIMEOUT
from py_moodle.course import MoodleCourseError, get_course_with_sections_and_modules

# --- Cache for module IDs ---
Expand Down Expand Up @@ -488,7 +489,7 @@ def _maybe_add_intro(
if modname == "label" and "intro" not in module:
try:
view_url = f"{base_url}/mod/label/view.php?id={cm.get('id')}"
r2 = session.get(view_url, timeout=15)
r2 = session.get(view_url, timeout=DEFAULT_SCRAPE_TIMEOUT)
if r2.status_code == 200:
soup = BeautifulSoup(r2.text, "lxml")
div = soup.select_one("div.contentwithoutlink")
Expand Down
7 changes: 6 additions & 1 deletion src/py_moodle/scorm.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import requests

from .config import DEFAULT_LARGE_UPLOAD_TIMEOUT
from .draftfile import MoodleDraftFileError, upload_file_to_draft_area
from .module import MoodleModuleError, add_generic_module
from .upload import MoodleUploadError, upload_file_webservice
Expand Down Expand Up @@ -63,7 +64,11 @@ def add_scorm(
# 2. Upload the SCORM package using the webservice.
try:
package_draft_itemid = upload_file_webservice(
base_url, token, file_path, (300, 3600), progress_callback
base_url,
token,
file_path,
DEFAULT_LARGE_UPLOAD_TIMEOUT,
progress_callback,
)
except MoodleUploadError as e:
raise MoodleScormError(f"Failed to upload SCORM package via webservice: {e}")
Expand Down
8 changes: 6 additions & 2 deletions src/py_moodle/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

from .auth import LoginError, login
from .compat import DEFAULT_COMPATIBILITY, get_session_compatibility
from .config import DEFAULT_REQUEST_TIMEOUT


class MoodleSessionError(RuntimeError):
Expand Down Expand Up @@ -64,7 +65,10 @@ def _login(self) -> None:

# Fallback extraction if sesskey was not attached by login()
if not self._sesskey:
resp = session.get(f"{self.settings.url}/my/")
resp = session.get(
f"{self.settings.url}/my/",
timeout=DEFAULT_REQUEST_TIMEOUT,
)
self._sesskey = self._compatibility.extract_sesskey(resp.text)

# Validate we have at least one usable token
Expand Down Expand Up @@ -134,7 +138,7 @@ def call(
response = self.session.post(
f"{self.settings.url}/webservice/rest/server.php",
params=request_params,
timeout=30,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
data = response.json()
Expand Down
8 changes: 5 additions & 3 deletions src/py_moodle/upload.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@

import requests

from .config import DEFAULT_UPLOAD_TIMEOUT


class MoodleUploadError(Exception):
"""Exception for webservice upload errors."""
Expand Down Expand Up @@ -52,7 +54,7 @@ def upload_file_webservice(
base_url: str,
token: str,
file_path: str,
timeout: tuple = (30, 3600),
timeout: tuple = DEFAULT_UPLOAD_TIMEOUT,
progress_callback: Optional[Callable[[int], None]] = None,
) -> int:
"""
Expand All @@ -63,8 +65,8 @@ def upload_file_webservice(
base_url: The base URL of the Moodle instance.
token: A valid Moodle webservice token.
file_path: The local path to the file to upload.
timeout: Request timeout in seconds. default (30, 3600)
30s to connect, 1h to upload
timeout: Request timeout in seconds. Defaults to the shared upload
timeout policy (30s to connect, 1h to upload).
progress_callback: Optional function to call with bytes uploaded.

Returns:
Expand Down
131 changes: 131 additions & 0 deletions tests/unit/test_timeouts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
"""Unit tests for shared HTTP timeout defaults."""

from py_moodle.compat import detect_moodle_compatibility
from py_moodle.config import (
DEFAULT_LARGE_UPLOAD_TIMEOUT,
DEFAULT_REQUEST_TIMEOUT,
DEFAULT_UPLOAD_TIMEOUT,
)
from py_moodle.draftfile import upload_file_to_draft_area
from py_moodle.session import MoodleSession
from py_moodle.settings import Settings
from py_moodle.upload import upload_file_webservice


class FakeResponse:
"""Minimal response object for timeout tests."""

def __init__(self, text="", json_data=None, status_code=200):
self.text = text
self.json_data = json_data
self.status_code = status_code

def raise_for_status(self):
"""Raise for failing HTTP statuses."""
if self.status_code >= 400:
raise RuntimeError(f"HTTP {self.status_code}")

def json(self):
"""Return the configured JSON payload."""
return self.json_data


class RecordingSession:
"""Minimal session that records GET and POST calls."""

def __init__(self, get_response=None, post_response=None):
self.get_response = get_response or FakeResponse()
self.post_response = post_response or FakeResponse(json_data={})
self.get_calls = []
self.post_calls = []

def get(self, url, **kwargs):
"""Record a GET request."""
self.get_calls.append({"url": url, "kwargs": kwargs})
return self.get_response

def post(self, url, **kwargs):
"""Record a POST request."""
self.post_calls.append({"url": url, "kwargs": kwargs})
return self.post_response


def test_session_call_uses_shared_default_request_timeout():
"""Webservice calls should use the centralized request timeout."""
settings = Settings(
env_name="local",
url="https://moodle.example.test",
username="user",
password="secret",
use_cas=False,
cas_url=None,
webservice_token=None,
)
session = RecordingSession(post_response=FakeResponse(json_data={"ok": True}))
moodle_session = MoodleSession(settings)
moodle_session._session = session
moodle_session._token = "token"

response = moodle_session.call("core_webservice_get_site_info")

assert response == {"ok": True}
assert session.post_calls[0]["kwargs"]["timeout"] == DEFAULT_REQUEST_TIMEOUT


def test_detect_moodle_compatibility_uses_shared_default_request_timeout():
"""Compatibility detection should use the shared timeout for HTTP probes."""
session = RecordingSession(
get_response=FakeResponse(
text='<meta name="generator" content="Moodle 4.5.2+ (Build: 20241001)">'
),
post_response=FakeResponse(json_data={"exception": "disabled"}),
)

compatibility = detect_moodle_compatibility(
session, "https://moodle.example.test", token="token"
)

assert compatibility.version.major == 4
assert session.post_calls[0]["kwargs"]["timeout"] == DEFAULT_REQUEST_TIMEOUT
assert session.get_calls[0]["kwargs"]["timeout"] == DEFAULT_REQUEST_TIMEOUT


def test_upload_helpers_use_shared_default_timeouts(tmp_path, monkeypatch):
"""Upload helpers should default to the shared upload timeout constants."""
file_path = tmp_path / "demo.txt"
file_path.write_text("demo content", encoding="utf-8")

upload_calls = []

def mock_requests_post(url, **kwargs):
upload_calls.append({"url": url, "kwargs": kwargs})
return FakeResponse(json_data=[{"itemid": 7}])

monkeypatch.setattr("py_moodle.upload.requests.post", mock_requests_post)

itemid = upload_file_webservice(
"https://moodle.example.test", "token", str(file_path)
)

assert itemid == 7
assert upload_calls[0]["kwargs"]["timeout"] == DEFAULT_UPLOAD_TIMEOUT

monkeypatch.setattr("py_moodle.draftfile.detect_upload_repo", lambda *args: 9)
draft_session = RecordingSession(
post_response=FakeResponse(json_data={"id": 99, "filename": "demo.txt"})
)

draft_itemid, filename = upload_file_to_draft_area(
session=draft_session,
base_url="https://moodle.example.test",
sesskey="sesskey",
course_id=12,
course_context_id=34,
file_path=str(file_path),
)

assert draft_itemid == 99
assert filename == "demo.txt"
assert (
draft_session.post_calls[0]["kwargs"]["timeout"] == DEFAULT_LARGE_UPLOAD_TIMEOUT
)
Loading