Skip to content

Commit

Permalink
Merge pull request #78 from azmeuk/issue-70-timestamp-file
Browse files Browse the repository at this point in the history
timestamp files support
  • Loading branch information
simon-weber authored May 5, 2024
2 parents 8f54687 + aeab305 commit aa13def
Show file tree
Hide file tree
Showing 3 changed files with 59 additions and 6 deletions.
16 changes: 16 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,22 @@ To avoid displaying the informative text when re-executing, you can set the
reexec_if_needed(quiet=True)
```

### timestamp_file

A common time can be shared between several execution contexts by using a file
to store the time to mock, instead of environment variables. This is useful
to control the time of a running process for instance. Here is a schematized
use case:

```python
reexec_if_needed(remove_vars=False)

with fake_time("1970-01-01 00:00:00", timestamp_file="/tmp/timestamp"):
subprocess.run("/some/server/process")

with fake_time("2000-01-01 00:00:00", timestamp_file="/tmp/timestamp"):
assert request_the_server_process_date() == "2000-01-01 00:00:00"

Performance
-----------

Expand Down
34 changes: 28 additions & 6 deletions libfaketime/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,13 +126,18 @@ def end_callback(instance):


class fake_time:
def __init__(self, datetime_spec, only_main_thread=True, tz_offset=None):
def __init__(self, datetime_spec=None, only_main_thread=True, tz_offset=None, timestamp_file=None):
self.only_main_thread = only_main_thread
self.timezone_str = 'UTC'
if tz_offset is not None:
self.timezone_str = 'Etc/GMT{0:+}'.format(-tz_offset)

if not datetime_spec and not timestamp_file:
raise ValueError("Either 'datetime_spec' or 'timestamp_file' must be passed.")

self.time_to_freeze = datetime_spec
self.timestamp_file = timestamp_file

if isinstance(datetime_spec, basestring):
self.time_to_freeze = utc.localize(dateutil.parser.parse(datetime_spec)) \
.astimezone(timezone(self.timezone_str))
Expand Down Expand Up @@ -164,21 +169,31 @@ def _should_patch_uuid(self):
def _format_datetime(self, _datetime):
return _datetime.strftime(_FAKETIME_FMT)

def _update_time(self, time):
if not self.timestamp_file:
os.environ['FAKETIME'] = self._format_datetime(time)
else:
if time:
with open(self.timestamp_file, "w") as fd:
fd.write(self._format_datetime(time))
os.environ['FAKETIME_TIMESTAMP_FILE'] = self.timestamp_file

def tick(self, delta=datetime.timedelta(seconds=1)):
self.time_to_freeze += delta
os.environ['FAKETIME'] = self._format_datetime(self.time_to_freeze)
self._update_time(self.time_to_freeze)

def __enter__(self):
if self._should_fake():
begin_callback(self)
self._prev_spec = os.environ.get('FAKETIME')
self._prev_tz = os.environ.get('TZ')
self._prev_fmt = os.environ.get('FAKETIME_FMT')
self._prev_timestamp_file = os.environ.get('FAKETIME_TIMESTAMP_FILE')

os.environ['TZ'] = self.timezone_str

time.tzset()
os.environ['FAKETIME'] = self._format_datetime(self.time_to_freeze)
self._update_time(self.time_to_freeze)
os.environ['FAKETIME_FMT'] = _FAKETIME_FMT

func_name = self._should_patch_uuid()
Expand All @@ -200,10 +215,17 @@ def __exit__(self, *exc):
del os.environ['TZ']
time.tzset()

if self._prev_spec is not None:
os.environ['FAKETIME'] = self._prev_spec
if self.timestamp_file:
if self._prev_timestamp_file is not None:
os.environ['FAKETIME_TIMESTAMP_FILE'] = self._prev_timestamp_file
elif 'FAKETIME_TIMESTAMP_FILE' in os.environ:
del os.environ['FAKETIME_TIMESTAMP_FILE']

else:
del os.environ['FAKETIME']
if self._prev_spec is not None:
os.environ['FAKETIME'] = self._prev_spec
else:
del os.environ['FAKETIME']

if self._prev_fmt is not None:
os.environ['FAKETIME_FMT'] = self._prev_spec
Expand Down
15 changes: 15 additions & 0 deletions test/test_faketime.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,21 @@ def test_freeze_time_alias(self):
def test_monotonic_not_mocked(self):
assert os.environ['DONT_FAKE_MONOTONIC'] == '1'

def test_timestmap_file(self, tmpdir):
file_path = str(tmpdir / "faketime.rc")

with fake_time('2000-01-01 10:00:05', timestamp_file=file_path) as fake:
assert datetime.datetime(2000, 1, 1, 10, 0, 5) == datetime.datetime.now()
with open(file_path) as fd:
assert fd.read() == "2000-01-01 10:00:05.000000"

fake.tick(delta=datetime.timedelta(hours=1))
assert datetime.datetime(2000, 1, 1, 11, 0, 5) == datetime.datetime.now()
with open(file_path) as fd:
assert fd.read() == "2000-01-01 11:00:05.000000"

with fake_time(timestamp_file=file_path):
assert datetime.datetime(2000, 1, 1, 11, 0, 5) == datetime.datetime.now()

class TestUUID1Deadlock():

Expand Down

0 comments on commit aa13def

Please sign in to comment.