Skip to content

Commit

Permalink
Fix setup_now_tt in Docker environment
Browse files Browse the repository at this point in the history
  • Loading branch information
matti-lamppu committed Feb 5, 2025
1 parent 716fe9e commit 82aff9d
Show file tree
Hide file tree
Showing 5 changed files with 63 additions and 60 deletions.
1 change: 1 addition & 0 deletions backend/.dockerignore
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ sonar-project.properties

# tests
**/tests*
!tests/factories*

# celery
broker/*
Expand Down
61 changes: 2 additions & 59 deletions backend/tests/plugins.py
Original file line number Diff line number Diff line change
@@ -1,70 +1,13 @@
from __future__ import annotations

import contextlib
import datetime
from typing import Any
from unittest.mock import patch

import pytest


@pytest.hookimpl(tryfirst=True)
def pytest_load_initial_conftests(early_config: pytest.Config, parser: pytest.Parser, args: list[str]) -> None:
setup_now_tt()


def setup_now_tt():
"""
Setup NowTT for tests so that when freezegun is used,
it will also setup the offset in the database for NowTT.
"""
from freezegun.api import _freeze_time # noqa: PLC2701

freezes: dict[int, int] = {}

class MockFreezeTime(_freeze_time):
def start(self) -> Any:
"""Called when 'freeze_time' is started."""
from utils.db import NowTT

# Calculate offset for time travel.
delta = self.time_to_freeze - datetime.datetime.now()
offset = int(delta.total_seconds())

# Set offset for to database, but ignore errors if test doesn't have database access.
with contextlib.suppress(RuntimeError):
NowTT.set_offset(seconds=offset)
from utils.utils import setup_now_tt

# Save offset in case we make multiple calls to 'freeze_time'.
freezes[id(self)] = offset

return super().start()

def stop(self) -> None:
"""Called when 'freeze_time' is stopped."""
from utils.db import NowTT

# Remove the saved offset.
del freezes[id(self)]

# If there are no more freezes, reset the offset to 0.
# Otherwise, use the last set offset.
if freezes:
key, value = freezes.popitem()
freezes[key] = value
else:
value = 0

# Set offset for to database, but ignore errors if test doesn't have database access.
with contextlib.suppress(RuntimeError):
NowTT.set_offset(seconds=value)

return super().stop()

# Just apply the patch for the whole duration of the test run.
mocked_freeze_factory = patch("freezegun.api._freeze_time", MockFreezeTime)
mocked_freeze_factory.start()
return mocked_freeze_factory
setup_now_tt()


@pytest.hookimpl()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

from freezegun import freeze_time

from tests.plugins import setup_now_tt
from utils.utils import setup_now_tt

if TYPE_CHECKING:
import datetime
Expand Down
57 changes: 57 additions & 0 deletions backend/utils/utils.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

import base64
import contextlib
import datetime
import hashlib
import hmac
Expand All @@ -9,6 +10,7 @@
import re
import urllib.parse
from typing import TYPE_CHECKING, Any, Generic, TypeVar
from unittest.mock import patch

from django.conf import settings
from django.core.cache import cache
Expand All @@ -28,6 +30,7 @@
__all__ = [
"comma_sep_str",
"get_text_search_language",
"setup_now_tt",
"update_query_params",
"with_indices",
]
Expand Down Expand Up @@ -262,3 +265,57 @@ def get_jwt_payload(json_web_token: str) -> dict[str, Any]:
payload_part += "=" * divmod(len(payload_part), 4)[1] # Add padding to the payload if needed
payload: str = base64.urlsafe_b64decode(payload_part).decode() # Decode the payload
return json.loads(payload) # Return the payload as a dict


def setup_now_tt() -> Any:
"""
Setup NowTT for tests so that when freezegun is used,
it will also setup the offset in the database for NowTT.
"""
from freezegun.api import _freeze_time # noqa: PLC2701

freezes: dict[int, int] = {}

class MockFreezeTime(_freeze_time):
def start(self) -> Any:
"""Called when 'freeze_time' is started."""
from utils.db import NowTT

# Calculate offset for time travel.
delta = self.time_to_freeze - local_datetime()
offset = int(delta.total_seconds())

# Set offset for to database, but ignore errors if test doesn't have database access.
with contextlib.suppress(RuntimeError):
NowTT.set_offset(seconds=offset)

# Save offset in case we make multiple calls to 'freeze_time'.
freezes[id(self)] = offset

return super().start()

def stop(self) -> None:
"""Called when 'freeze_time' is stopped."""
from utils.db import NowTT

# Remove the saved offset.
del freezes[id(self)]

# If there are no more freezes, reset the offset to 0.
# Otherwise, use the last set offset.
if freezes:
key, value = freezes.popitem()
freezes[key] = value
else:
value = 0

# Set offset for to database, but ignore errors if test doesn't have database access.
with contextlib.suppress(RuntimeError):
NowTT.set_offset(seconds=value)

return super().stop()

# Just apply the patch for the whole duration of the test run.
mocked_freeze_factory = patch("freezegun.api._freeze_time", MockFreezeTime)
mocked_freeze_factory.start()
return mocked_freeze_factory
2 changes: 2 additions & 0 deletions docker-compose.yml
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ services:
build:
context: backend/
dockerfile: docker/Dockerfile
args:
DEPS: main,admin,celery,test,lint
env_file:
- backend/.env
environment:
Expand Down

0 comments on commit 82aff9d

Please sign in to comment.