diff --git a/.devcontainer/test_application_qa.py b/.devcontainer/test_application_qa.py index c4044701fc7c..261fc938fd9a 100644 --- a/.devcontainer/test_application_qa.py +++ b/.devcontainer/test_application_qa.py @@ -8,6 +8,7 @@ import unittest from io import BytesIO +from time import monotonic from unittest.mock import Mock, patch from zipfile import ZipFile @@ -18,14 +19,13 @@ class ApplicationQATests(unittest.TestCase): def test_eventual_completion(self) -> None: probe = Mock(side_effect=[None, None, {"completed": True}]) - with patch("application_qa.time.sleep") as sleep: + with patch("application_qa.time", monotonic=monotonic) as clock: self.assertEqual(wait_for("import", probe), {"completed": True}) - self.assertEqual(sleep.call_count, 2) + self.assertEqual(clock.sleep.call_count, 2) def test_deadline(self) -> None: with ( - patch("application_qa.time.monotonic", side_effect=[0, 1, 181]), - patch("application_qa.time.sleep"), + patch("application_qa.time", monotonic=Mock(side_effect=[0, 1, 181])), self.assertRaisesRegex(TimeoutError, "180s: notification"), ): wait_for("notification", lambda: False) diff --git a/.devcontainer/test_doctor.py b/.devcontainer/test_doctor.py index d2424eeb1ded..5c649b873f38 100644 --- a/.devcontainer/test_doctor.py +++ b/.devcontainer/test_doctor.py @@ -8,6 +8,7 @@ import os import unittest +from time import monotonic from unittest.mock import MagicMock, patch from django.db import DatabaseError @@ -23,12 +24,12 @@ def test_database_startup_retry(self) -> None: cache = MagicMock() with ( patch("doctor.connection") as database, - patch("doctor.time.sleep") as sleep, + patch("doctor.time", monotonic=monotonic) as clock, ): database.cursor.side_effect = [DatabaseError("starting"), MagicMock()] wait_for_services(cache) database.close.assert_called_once() - sleep.assert_called_once_with(1) + clock.sleep.assert_called_once_with(1) cache.ping.assert_called_once() def test_valkey_timeout(self) -> None: @@ -36,13 +37,12 @@ def test_valkey_timeout(self) -> None: cache.ping.side_effect = RedisConnectionError("private diagnostic") with ( patch("doctor.connection"), - patch("doctor.time.monotonic", side_effect=[0, 61]), - patch("doctor.time.sleep") as sleep, + patch("doctor.time", monotonic=MagicMock(side_effect=[0, 61])) as clock, self.assertRaisesRegex(SystemExit, "unavailable after 60 seconds") as error, ): wait_for_services(cache) self.assertNotIn("private diagnostic", str(error.exception)) - sleep.assert_not_called() + clock.sleep.assert_not_called() class BrowserTests(unittest.TestCase): diff --git a/.devcontainer/test_environment.py b/.devcontainer/test_environment.py index 591c9a8b7095..7653a33a374a 100644 --- a/.devcontainer/test_environment.py +++ b/.devcontainer/test_environment.py @@ -17,7 +17,8 @@ from io import StringIO from pathlib import Path from tempfile import TemporaryDirectory -from unittest.mock import patch +from time import monotonic +from unittest.mock import Mock, patch from environment import Environment, configuration, initialize, main, project_name @@ -408,10 +409,10 @@ def test_readiness_requires_every_service(self) -> None: "output", side_effect=[json.dumps(entries[:-1]), json.dumps(entries)], ), - patch("environment.time.sleep") as sleep, + patch("environment.time", monotonic=monotonic) as clock, ): self.assertEqual(self.env.wait(), 0) - sleep.assert_called_once() + clock.sleep.assert_called_once() def test_readiness_reports_exit_and_timeout(self) -> None: with ( @@ -424,7 +425,7 @@ def test_readiness_reports_exit_and_timeout(self) -> None: ): self.assertEqual(self.env.wait(), 1) with ( - patch("environment.time.monotonic", side_effect=[0, 1801]), + patch("environment.time", monotonic=Mock(side_effect=[0, 1801])), redirect_stderr(StringIO()), ): self.assertEqual(self.env.wait(), 1) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index aef4f838158b..bff41f8195b4 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -35,7 +35,7 @@ repos: hooks: - id: yamllint - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.17.2 + rev: v0.17.3 hooks: - id: ruff-check args: @@ -48,7 +48,7 @@ repos: - id: doccmd args: [--language, python, --no-pad-file, --command, ruff format] additional_dependencies: - - ruff==0.17.2 + - ruff==0.17.3 language: python - repo: meta hooks: diff --git a/weblate/accounts/tests/test_totp_enrollment.py b/weblate/accounts/tests/test_totp_enrollment.py index 2b1358abbed7..412b62fb552d 100644 --- a/weblate/accounts/tests/test_totp_enrollment.py +++ b/weblate/accounts/tests/test_totp_enrollment.py @@ -36,7 +36,12 @@ class TOTPEnrollmentTest(FixtureTestCase): def setUp(self) -> None: super().setUp() self.timestamp = int(now().timestamp()) - self.enterContext(mock.patch("time.time", return_value=self.timestamp)) + self.enterContext( + mock.patch( + "django_otp.plugins.otp_totp.models.time", + time=mock.Mock(return_value=self.timestamp), + ) + ) self.enterContext( mock.patch("django_otp.oath.time", return_value=self.timestamp) ) @@ -86,7 +91,10 @@ def test_replay_and_fresh_enrollment(self) -> None: TOTPTokenForm(self.user, data={"otp_token": payload["token"]}).is_valid() ) with ( - mock.patch("time.time", return_value=self.timestamp + 30), + mock.patch( + "django_otp.plugins.otp_totp.models.time", + time=mock.Mock(return_value=self.timestamp + 30), + ), mock.patch("django_otp.oath.time", return_value=self.timestamp + 30), mock.patch( "django_otp.models.timezone.now", @@ -333,7 +341,12 @@ def setUp(self) -> None: super().setUp() self.user = User.objects.create_user("concurrent-totp", "totp@example.com") self.timestamp = int(now().timestamp()) - self.enterContext(mock.patch("time.time", return_value=self.timestamp)) + self.enterContext( + mock.patch( + "django_otp.plugins.otp_totp.models.time", + time=mock.Mock(return_value=self.timestamp), + ) + ) self.enterContext( mock.patch("django_otp.oath.time", return_value=self.timestamp) ) diff --git a/weblate/accounts/tests/test_twofactor.py b/weblate/accounts/tests/test_twofactor.py index 9263295962db..894fb2e42a0c 100644 --- a/weblate/accounts/tests/test_twofactor.py +++ b/weblate/accounts/tests/test_twofactor.py @@ -205,7 +205,10 @@ def add_totp( timestamp = now().timestamp() - device.step with ( mock.patch("django_otp.oath.time", return_value=timestamp), - mock.patch("time.time", return_value=timestamp), + mock.patch( + "django_otp.plugins.otp_totp.models.time", + time=mock.Mock(return_value=timestamp), + ), ): totp_response = f"{totp(device.bin_key):06d}" response = self.post_with_callbacks( @@ -801,12 +804,12 @@ def test_team_enforced_2fa(self) -> None: def test_password_requires_recent_second_factor(self) -> None: self.add_totp() - with mock.patch("weblate.accounts.views.time.time") as mocked_time: - mocked_time.return_value = 2_000_000_000 + with mock.patch("weblate.accounts.utils.time") as clock: + clock.time.return_value = 2_000_000_000 stale_session = self.client.session stale_session[SESSION_SECOND_FACTOR_TIMESTAMP] = int( - mocked_time.return_value + clock.time.return_value ) - (SECOND_FACTOR_VERIFY_SECONDS + 1) stale_session.save() @@ -824,12 +827,12 @@ def test_password_requires_recent_second_factor(self) -> None: def test_password_allows_recent_second_factor(self) -> None: self.add_totp() - with mock.patch("weblate.accounts.views.time.time") as mocked_time: - mocked_time.return_value = 2_000_000_000 + with mock.patch("weblate.accounts.utils.time") as clock: + clock.time.return_value = 2_000_000_000 session = self.client.session session[SESSION_SECOND_FACTOR_TIMESTAMP] = ( - int(mocked_time.return_value) - SECOND_FACTOR_VERIFY_SECONDS + int(clock.time.return_value) - SECOND_FACTOR_VERIFY_SECONDS ) session.save() diff --git a/weblate/accounts/tests/test_utils.py b/weblate/accounts/tests/test_utils.py index 7d730ef47d16..1268fbcac31e 100644 --- a/weblate/accounts/tests/test_utils.py +++ b/weblate/accounts/tests/test_utils.py @@ -111,7 +111,9 @@ def get_request(self, data: dict[str, str] | None = None): def test_authenticated_session_expiry_is_throttled(self) -> None: request = self.get_request() - with mock.patch("weblate.accounts.utils.time.time", return_value=1000): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1000) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) self.assertEqual( @@ -122,11 +124,15 @@ def test_authenticated_session_expiry_is_throttled(self) -> None: self.assertEqual(request.session.get_expiry_age(), 1200) request.session.modified = False - with mock.patch("weblate.accounts.utils.time.time", return_value=1599): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1599) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) self.assertFalse(request.session.modified) - with mock.patch("weblate.accounts.utils.time.time", return_value=1600): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1600) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) self.assertTrue(request.session.modified) self.assertEqual(request.session[SESSION_EXPIRY_REFRESHED], 1600) @@ -138,15 +144,21 @@ def test_short_authenticated_session_expiry_refreshes_before_lifetime(self) -> N self.assertEqual(get_session_expiry_refresh_seconds(1), 0) self.assertLess(get_session_expiry_refresh_seconds(90), 90) - with mock.patch("weblate.accounts.utils.time.time", return_value=1000): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1000) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) request.session.modified = False - with mock.patch("weblate.accounts.utils.time.time", return_value=1029): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1029) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) self.assertFalse(request.session.modified) - with mock.patch("weblate.accounts.utils.time.time", return_value=1030): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1030) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) self.assertTrue(request.session.modified) self.assertEqual(request.session[SESSION_EXPIRY_REFRESHED], 1030) @@ -159,7 +171,9 @@ def test_short_authenticated_session_expiry_refreshes_before_lifetime(self) -> N def test_session_expiry_scope_changes(self) -> None: request = self.get_request() - with mock.patch("weblate.accounts.utils.time.time", return_value=1000): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1000) + ): adjust_session_expiry(request=request, user=self.user) self.assertEqual( request.session[SESSION_EXPIRY_SCOPE], SESSION_EXPIRY_SCOPE_LOGIN @@ -167,7 +181,9 @@ def test_session_expiry_scope_changes(self) -> None: self.assertEqual(request.session.get_expiry_age(), 1000) request.session.modified = False - with mock.patch("weblate.accounts.utils.time.time", return_value=1001): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1001) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) self.assertTrue(request.session.modified) self.assertEqual( @@ -182,7 +198,9 @@ def test_second_factor_session_promotes_to_authenticated(self) -> None: request = self.get_request() TOTPDevice.objects.create(user=self.user) - with mock.patch("weblate.accounts.utils.time.time", return_value=1000): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1000) + ): adjust_session_expiry(request=request, user=self.user) self.assertEqual( request.session[SESSION_EXPIRY_SCOPE], SESSION_EXPIRY_SCOPE_2FA @@ -191,7 +209,9 @@ def test_second_factor_session_promotes_to_authenticated(self) -> None: request.session[DEVICE_ID_SESSION_KEY] = "otp_totp.totpdevice/1" request.session.modified = False - with mock.patch("weblate.accounts.utils.time.time", return_value=1001): + with mock.patch( + "weblate.accounts.utils.time", time=mock.Mock(return_value=1001) + ): adjust_session_expiry(request=request, user=self.user, is_login=False) self.assertTrue(request.session.modified) self.assertEqual( @@ -228,7 +248,7 @@ def test_cleanup_social_auth(self) -> None: ) with ( mock.patch("weblate.accounts.tasks.now", return_value=timestamp), - mock.patch("social_django.storage.time.time", return_value=1060), + mock.patch("social_django.storage.time", time=mock.Mock(return_value=1060)), ): cleanup_social_auth() self.assertEqual( diff --git a/weblate/addons/tests.py b/weblate/addons/tests.py index f4f633c67927..977e25a39a74 100644 --- a/weblate/addons/tests.py +++ b/weblate/addons/tests.py @@ -13453,8 +13453,8 @@ def test_broker_tls_handshake_enforces_deadline(self) -> None: with ( patch( - "weblate.addons.fedora_messaging.time.monotonic", - side_effect=[0, 0, 8], + "weblate.addons.fedora_messaging.time", + monotonic=MagicMock(side_effect=[0, 0, 8]), ), self.assertRaisesMessage(TimeoutError, "TLS handshake timed out"), ): diff --git a/weblate/machinery/tests.py b/weblate/machinery/tests.py index d7999aeb1be1..8be92dd5f778 100644 --- a/weblate/machinery/tests.py +++ b/weblate/machinery/tests.py @@ -7,11 +7,13 @@ import json import os import re +from contextlib import contextmanager from copy import copy from datetime import UTC, datetime from functools import partial from io import StringIO from pathlib import Path +from types import SimpleNamespace from typing import TYPE_CHECKING, Any, ClassVar, NoReturn, cast from unittest.mock import AsyncMock, MagicMock, Mock, call, patch from urllib.parse import parse_qs, urlparse @@ -120,6 +122,8 @@ from .types import SourceLanguageChoices if TYPE_CHECKING: + from collections.abc import Generator + from weblate.machinery.base import ( SettingsDict, ) @@ -127,6 +131,19 @@ HTTPError = httpx2.HTTPStatusError +@contextmanager +def mock_machinery_time(timestamp: int) -> Generator[Mock, None, None]: + now = Mock(return_value=timestamp) + clock = SimpleNamespace(time=now) + # Cache timeout creation and expiry checks must advance with retry parsing. + with ( + patch("weblate.machinery.base.time", clock), + patch("django.core.cache.backends.base.time", clock), + patch("django.core.cache.backends.locmem.time", clock), + ): + yield now + + def make_error_response( url: str, status_code: int, @@ -476,13 +493,13 @@ def test_inherited_installation_shares_both_cooldowns(self) -> None: def test_cooldowns_expire(self) -> None: machine = DeepLTranslation({"_installation_id": 1}) - with patch("time.time", return_value=1000): + with mock_machinery_time(1000): machine.set_rate_limit() machine.set_rephrase_rate_limit() - with patch("time.time", return_value=2799): + with mock_machinery_time(2799): self.assertTrue(machine.is_rate_limited()) self.assertTrue(machine.is_rephrase_rate_limited()) - with patch("time.time", return_value=2801): + with mock_machinery_time(2801): self.assertFalse(machine.is_rate_limited()) self.assertFalse(machine.is_rephrase_rate_limited()) @@ -897,7 +914,7 @@ class CustomTranslation(DeepLTranslation): def assert_window_expires(self, machine: DeepLTranslation, window: int) -> None: cache.clear() - with patch("time.time", return_value=10000) as now: + with mock_machinery_time(10000) as now: machine.set_rate_limit() self.assertTrue(machine.is_rate_limited()) self.assertFalse(machine.is_rephrase_rate_limited()) @@ -962,7 +979,7 @@ def test_retry_headers(self) -> None: ({"Retry-After": "1e1000"}, 1800), ({"Retry-After": "1e100"}, 1800), ): - with self.subTest(headers=headers), patch("time.time", return_value=10000): + with self.subTest(headers=headers), mock_machinery_time(10000): response = make_error_response("https://example.com/", 429) response.headers.update(headers) error = HTTPError( @@ -1165,7 +1182,7 @@ def test_sdk_translation_cooldown(self) -> None: ): with ( self.subTest(service=machine.name), - patch("time.time", return_value=10000) as now, + mock_machinery_time(10000) as now, ): cache.clear() machine.cache_translations = False @@ -1201,7 +1218,7 @@ def test_alibaba_translation_cooldown(self) -> None: ("Throttling.Api", 400), ("Unknown", 429), ): - with self.subTest(code=code), patch("time.time", return_value=10000) as now: + with self.subTest(code=code), mock_machinery_time(10000) as now: cache.clear() http_mock.register( "POST", @@ -1241,7 +1258,7 @@ def test_rephrase_retry_delay(self) -> None: for asynchronous in (False, True): with ( self.subTest(asynchronous=asynchronous), - patch("time.time", return_value=10000) as now, + mock_machinery_time(10000) as now, ): cache.clear() machine = DeepLTranslation({}) diff --git a/weblate/oidc/tests.py b/weblate/oidc/tests.py index 53441aedfd58..7d0737acfd6b 100644 --- a/weblate/oidc/tests.py +++ b/weblate/oidc/tests.py @@ -16,7 +16,7 @@ from pathlib import Path from textwrap import dedent from typing import TYPE_CHECKING, Any -from unittest.mock import patch +from unittest.mock import Mock, patch from urllib.parse import parse_qs, urlencode, urlsplit import jwt @@ -847,8 +847,10 @@ def test_invalid_login_attempt(self) -> None: def test_expired_login_attempt(self) -> None: target = self.begin_login() with patch( - "weblate.accounts.oidc.time.time", - return_value=timezone.now().timestamp() + OIDC_ATTEMPT_LIFETIME + 1, + "weblate.accounts.oidc.time", + time=Mock( + return_value=timezone.now().timestamp() + OIDC_ATTEMPT_LIFETIME + 1 + ), ): response = self.password_login(target) self.assertContains(response, "Start signing in again", status_code=400) diff --git a/weblate/trans/tests/test_models.py b/weblate/trans/tests/test_models.py index 32c38e03d299..705dfabc7b10 100644 --- a/weblate/trans/tests/test_models.py +++ b/weblate/trans/tests/test_models.py @@ -595,7 +595,10 @@ def test_glossary_automaton_cache_version_avoids_recreated_key_collision( with ( patch.object(Project, "glossaries", glossaries), - patch("weblate.trans.models.project.time.time_ns", side_effect=[100, 200]), + patch( + "weblate.trans.models.project.time", + time_ns=Mock(side_effect=[100, 200]), + ), patch("weblate.trans.models.component.prefetch_glossary_terms") as prefetch, ): first = get_glossary_automaton(project) diff --git a/weblate/utils/tests/test_bot_protection.py b/weblate/utils/tests/test_bot_protection.py index d80380cc4466..f5c0e6d5f411 100644 --- a/weblate/utils/tests/test_bot_protection.py +++ b/weblate/utils/tests/test_bot_protection.py @@ -184,7 +184,9 @@ def test_challenge_deadlines(self) -> None: with ( self.subTest(age=age), override_settings(BOT_PROTECTION_CHALLENGE_AGE=age), - patch("time.time", return_value=1000), + patch( + "weblate.utils.bot_protection.time", time=Mock(return_value=1000) + ), ): response = self.challenge_response(self.request()) self.assertEqual( @@ -524,12 +526,17 @@ def test_expiry_forgery_and_cookie_tampering(self) -> None: payload["challenge"]["signature"] = "0" * 64 self.assertEqual(self.redeem(payload).status_code, 400) payload = self.solution(self.request()) - with patch("time.time", return_value=time.time() + 1000): + with patch( + "weblate.utils.bot_protection.time", + time=Mock(return_value=time.time() + 1000), + ): self.assertEqual(self.redeem(payload).status_code, 400) response = self.redeem(self.solution(self.request())) request = self.request() request.COOKIES["weblate_pow"] = response.cookies["weblate_pow"].value - with patch("time.time", return_value=time.time() + 4000): + with patch( + "django.core.signing.time", time=Mock(return_value=time.time() + 4000) + ): self.assertFalse(clearance(request)) request.COOKIES["weblate_pow"] += "x" self.assertFalse(clearance(request)) diff --git a/weblate/wladmin/tests.py b/weblate/wladmin/tests.py index a1e4ebce4dee..9eca66e6964b 100644 --- a/weblate/wladmin/tests.py +++ b/weblate/wladmin/tests.py @@ -961,7 +961,7 @@ def test_support_reminder_waits_for_page_navigation(self) -> None: def test_claim_configuration_health_check(self) -> None: with ( - patch("weblate.wladmin.middleware.time.time", return_value=123), + patch("weblate.wladmin.middleware.time", time=Mock(return_value=123)), patch( "weblate.wladmin.middleware.cache.add", return_value=True ) as cache_add, @@ -986,8 +986,8 @@ def test_poll_interval(self) -> None: middleware = ManageMiddleware(Mock()) with ( patch( - "weblate.wladmin.middleware.time.monotonic", - side_effect=[100, 120, 100 + CHECK_POLL_INTERVAL], + "weblate.wladmin.middleware.time", + monotonic=Mock(side_effect=[100, 120, 100 + CHECK_POLL_INTERVAL]), ), patch("weblate.wladmin.middleware.cache.get", return_value=None), patch.object(middleware, "trigger_check") as trigger_check, @@ -1002,8 +1002,11 @@ def test_poll_interval(self) -> None: def test_recent_check(self) -> None: middleware = ManageMiddleware(Mock()) with ( - patch("weblate.wladmin.middleware.time.monotonic", return_value=100), - patch("weblate.wladmin.middleware.time.time", return_value=1000), + patch( + "weblate.wladmin.middleware.time", + monotonic=Mock(return_value=100), + time=Mock(return_value=1000), + ), patch( "weblate.wladmin.middleware.cache.get", return_value=1000 - CHECK_INTERVAL + 1, @@ -1070,7 +1073,7 @@ def test_perform_configuration_health_check(self) -> None: patch.object( ConfigurationError.objects, "configuration_health_check" ) as health_check, - patch("weblate.wladmin.middleware.time.time", return_value=123), + patch("weblate.wladmin.middleware.time", time=Mock(return_value=123)), patch("weblate.wladmin.middleware.cache.set") as cache_set, ): self.assertTrue(perform_configuration_health_check(checks))