This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2696

cleanup: use stat_result[stat.ST_MTIME] instead of stat_result.st_mtime
ClosedPublic

Authored by durin42 on Mar 5 2018, 4:08 PM.

Details

Reviewers
yuja
Group Reviewers
hg-reviewers
Commits
rHGffa3026d4196: cleanup: use stat_result[stat.ST_MTIME] instead of stat_result.st_mtime
Summary

The latter is floating point by default, and we've been doing
os.stat_float_times(False). Unfortunately, os.stat_float_times was
removed between Python 3.7.0a1 and 3.7.0b2, so we have to stop using
it.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

durin42 created this revision.Mar 5 2018, 4:08 PM
indygreg added a subscriber: indygreg.Mar 5 2018, 7:05 PM
indygreg added inline comments.
mercurial/cext/osutil.c
126

Is this valid for Python 2? The passed argument will likely be a PyInt on Python 2. I thought PyLong_AsLong only operates on int types?

durin42 added inline comments.Mar 5 2018, 7:58 PM
mercurial/cext/osutil.c
126

Seems to (I also could have sworn int and long became the same thing in 2.6), as it passes the entire testsuite.

durin42 updated this revision to Diff 6676.Mar 6 2018, 1:58 PM
durin42 updated this revision to Diff 6679.Mar 6 2018, 2:02 PM
yuja accepted this revision.Mar 8 2018, 7:57 AM
This revision is now accepted and ready to land.Mar 8 2018, 7:57 AM
yuja added a comment.Mar 8 2018, 8:00 AM

Queued, thanks.

Maybe we'll need to update cffi code as a follow-up.

mercurial/chgserver.py
552

Fixed name conflicts and st_mtime oversight.

Closed by commit rHGffa3026d4196: cleanup: use stat_result[stat.ST_MTIME] instead of stat_result.st_mtime (authored by durin42). · Explain WhyMar 8 2018, 8:06 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 6676

hgext/extdiff.py

pretty fast (at least faster than having to compare the entire tree). pretty fast (at least faster than having to compare the entire tree).
''' '''
from __future__ import absolute_import from __future__ import absolute_import
import os import os
import re import re
import shutil import shutil
import stat
import tempfile import tempfile
from mercurial.i18n import _ from mercurial.i18n import _
from mercurial.node import ( from mercurial.node import (
nullid, nullid,
short, short,
) )
from mercurial import ( from mercurial import (
archival, archival,
cpstat = os.lstat(copy_fn) cpstat = os.lstat(copy_fn)
# Some tools copy the file and attributes, so mtime may not detect # Some tools copy the file and attributes, so mtime may not detect
# all changes. A size check will detect more cases, but not all. # all changes. A size check will detect more cases, but not all.
# The only certain way to detect every case is to diff all files, # The only certain way to detect every case is to diff all files,
# which could be expensive. # which could be expensive.
# copyfile() carries over the permission, so the mode check could # copyfile() carries over the permission, so the mode check could
# be in an 'elif' branch, but for the case where the file has # be in an 'elif' branch, but for the case where the file has
# changed without affecting mtime or size. # changed without affecting mtime or size.
if (cpstat.st_mtime != st.st_mtime or cpstat.st_size != st.st_size if (cpstat[stat.ST_MTIME] != st[stat.ST_MTIME]
or cpstat.st_size != st.st_size
or (cpstat.st_mode & 0o100) != (st.st_mode & 0o100)): or (cpstat.st_mode & 0o100) != (st.st_mode & 0o100)):
ui.debug('file changed while diffing. ' ui.debug('file changed while diffing. '
'Overwriting: %s (src: %s)\n' % (working_fn, copy_fn)) 'Overwriting: %s (src: %s)\n' % (working_fn, copy_fn))
util.copyfile(copy_fn, working_fn) util.copyfile(copy_fn, working_fn)
return 1 return 1
finally: finally:
ui.note(_('cleaning up temp directory\n')) ui.note(_('cleaning up temp directory\n'))

hgext/shelve.py

shelved change has a distinct name. For details, see the help for "hg shelved change has a distinct name. For details, see the help for "hg
shelve". shelve".
""" """
from __future__ import absolute_import from __future__ import absolute_import
import collections import collections
import errno import errno
import itertools import itertools
import stat
from mercurial.i18n import _ from mercurial.i18n import _
from mercurial import ( from mercurial import (
bookmarks, bookmarks,
bundle2, bundle2,
bundlerepo, bundlerepo,
changegroup, changegroup,
cmdutil, cmdutil,
def clear(cls, repo): def clear(cls, repo):
repo.vfs.unlinkpath(cls._filename, ignoremissing=True) repo.vfs.unlinkpath(cls._filename, ignoremissing=True)
def cleanupoldbackups(repo): def cleanupoldbackups(repo):
vfs = vfsmod.vfs(repo.vfs.join(backupdir)) vfs = vfsmod.vfs(repo.vfs.join(backupdir))
maxbackups = repo.ui.configint('shelve', 'maxbackups') maxbackups = repo.ui.configint('shelve', 'maxbackups')
hgfiles = [f for f in vfs.listdir() hgfiles = [f for f in vfs.listdir()
if f.endswith('.' + patchextension)] if f.endswith('.' + patchextension)]
hgfiles = sorted([(vfs.stat(f).st_mtime, f) for f in hgfiles]) hgfiles = sorted([(vfs.stat(f)[stat.ST_MTIME], f) for f in hgfiles])
if 0 < maxbackups and maxbackups < len(hgfiles): if 0 < maxbackups and maxbackups < len(hgfiles):
bordermtime = hgfiles[-maxbackups][0] bordermtime = hgfiles[-maxbackups][0]
else: else:
bordermtime = None bordermtime = None
for mtime, f in hgfiles[:len(hgfiles) - maxbackups]: for mtime, f in hgfiles[:len(hgfiles) - maxbackups]:
if mtime == bordermtime: if mtime == bordermtime:
# keep it, because timestamp can't decide exact order of backups # keep it, because timestamp can't decide exact order of backups
continue continue
raise raise
return [] return []
info = [] info = []
for (name, _type) in names: for (name, _type) in names:
pfx, sfx = name.rsplit('.', 1) pfx, sfx = name.rsplit('.', 1)
if not pfx or sfx != patchextension: if not pfx or sfx != patchextension:
continue continue
st = shelvedfile(repo, name).stat() st = shelvedfile(repo, name).stat()
info.append((st.st_mtime, shelvedfile(repo, pfx).filename())) info.append((st[stat.ST_MTIME], shelvedfile(repo, pfx).filename()))
return sorted(info, reverse=True) return sorted(info, reverse=True)
def listcmd(ui, repo, pats, opts): def listcmd(ui, repo, pats, opts):
"""subcommand that displays the list of shelves""" """subcommand that displays the list of shelves"""
pats = set(pats) pats = set(pats)
width = 80 width = 80
if not ui.plain(): if not ui.plain():
width = ui.termwidth() width = ui.termwidth()

mercurial/cext/osutil.c

static void listdir_stat_dealloc(PyObject *o) static void listdir_stat_dealloc(PyObject *o)
{ {
o->ob_type->tp_free(o); o->ob_type->tp_free(o);
} }
static PyObject *listdir_stat_getitem(PyObject *self, PyObject *key) static PyObject *listdir_stat_getitem(PyObject *self, PyObject *key)
{ {
long index = PyInt_AsLong(key); long index = PyLong_AsLong(key);
indygregUnsubmitted
Not Done

Is this valid for Python 2? The passed argument will likely be a PyInt on Python 2. I thought PyLong_AsLong only operates on int types?

indygreg: Is this valid for Python 2? The passed argument will likely be a PyInt on Python 2. I thought…
durin42AuthorUnsubmitted
Not Done

Seems to (I also could have sworn int and long became the same thing in 2.6), as it passes the entire testsuite.

durin42: Seems to (I also could have sworn int and long became the same thing in 2.6), as it passes the…
if (index == -1 && PyErr_Occurred()) { if (index == -1 && PyErr_Occurred()) {
return NULL; return NULL;
} }
if (index != 8) { if (index != 8) {
PyErr_Format(PyExc_IndexError, "osutil.stat objects only " PyErr_Format(PyExc_IndexError, "osutil.stat objects only "
"support stat.ST_MTIME in " "support stat.ST_MTIME in "
"__getitem__"); "__getitem__");
return NULL; return NULL;

mercurial/chgserver.py

from __future__ import absolute_import from __future__ import absolute_import
import hashlib import hashlib
import inspect import inspect
import os import os
import re import re
import socket import socket
import stat
import struct import struct
import time import time
from .i18n import _ from .i18n import _
from . import ( from . import (
commandserver, commandserver,
encoding, encoding,
mtimehash is not included in confighash because we only know the paths of mtimehash is not included in confighash because we only know the paths of
extensions after importing them (there is imp.find_module but that faces extensions after importing them (there is imp.find_module but that faces
race conditions). We need to calculate confighash without importing. race conditions). We need to calculate confighash without importing.
""" """
def trystat(path): def trystat(path):
try: try:
st = os.stat(path) st = os.stat(path)
return (st.st_mtime, st.st_size) return (st[stat.ST_MTIME], st.st_size)
except OSError: except OSError:
# could be ENOENT, EPERM etc. not fatal in any case # could be ENOENT, EPERM etc. not fatal in any case
pass pass
return _hashlist(map(trystat, paths))[:12] return _hashlist(map(trystat, paths))[:12]
class hashstate(object): class hashstate(object):
"""a structure storing confighash, mtimehash, paths used for mtimehash""" """a structure storing confighash, mtimehash, paths used for mtimehash"""
def __init__(self, confighash, mtimehash, mtimepaths): def __init__(self, confighash, mtimehash, mtimepaths):
tempaddress = _tempaddress(self._baseaddress) tempaddress = _tempaddress(self._baseaddress)
os.symlink(os.path.basename(self._realaddress), tempaddress) os.symlink(os.path.basename(self._realaddress), tempaddress)
util.rename(tempaddress, self._baseaddress) util.rename(tempaddress, self._baseaddress)
def _issocketowner(self): def _issocketowner(self):
try: try:
stat = os.stat(self._realaddress) stat = os.stat(self._realaddress)
return (stat.st_ino == self._socketstat.st_ino and return (stat.st_ino == self._socketstat.st_ino and
stat.st_mtime == self._socketstat.st_mtime) stat.st_mtime == self._socketstat[stat.ST_MTIME])
yujaUnsubmitted
Not Done

Fixed name conflicts and st_mtime oversight.

yuja: Fixed name conflicts and st_mtime oversight.
except OSError: except OSError:
return False return False
def unlinksocket(self, address): def unlinksocket(self, address):
if not self._issocketowner(): if not self._issocketowner():
return return
# it is possible to have a race condition here that we may # it is possible to have a race condition here that we may
# remove another server's socket file. but that's okay # remove another server's socket file. but that's okay

mercurial/context.py

return None return None
return rp, self._changectx._parents[0]._manifest.get(rp, nullid) return rp, self._changectx._parents[0]._manifest.get(rp, nullid)
def size(self): def size(self):
return self._repo.wvfs.lstat(self._path).st_size return self._repo.wvfs.lstat(self._path).st_size
def date(self): def date(self):
t, tz = self._changectx.date() t, tz = self._changectx.date()
try: try:
return (self._repo.wvfs.lstat(self._path).st_mtime, tz) return (self._repo.wvfs.lstat(self._path)[stat.ST_MTIME], tz)
except OSError as err: except OSError as err:
if err.errno != errno.ENOENT: if err.errno != errno.ENOENT:
raise raise
return (t, tz) return (t, tz)
def exists(self): def exists(self):
return self._repo.wvfs.exists(self._path) return self._repo.wvfs.exists(self._path)

mercurial/debugcommands.py

import collections import collections
import difflib import difflib
import errno import errno
import operator import operator
import os import os
import random import random
import socket import socket
import ssl import ssl
import stat
import string import string
import subprocess import subprocess
import sys import sys
import tempfile import tempfile
import time import time
from .i18n import _ from .i18n import _
from .node import ( from .node import (
l = method(False) l = method(False)
except error.LockHeld: except error.LockHeld:
l = None l = None
if l: if l:
l.release() l.release()
else: else:
try: try:
stat = vfs.lstat(name) st = vfs.lstat(name)
age = now - stat.st_mtime age = now - st[stat.ST_MTIME]
user = util.username(stat.st_uid) user = util.username(st.st_uid)
locker = vfs.readlock(name) locker = vfs.readlock(name)
if ":" in locker: if ":" in locker:
host, pid = locker.split(':') host, pid = locker.split(':')
if host == socket.gethostname(): if host == socket.gethostname():
locker = 'user %s, process %s' % (user, pid) locker = 'user %s, process %s' % (user, pid)
else: else:
locker = 'user %s, process %s, host %s' \ locker = 'user %s, process %s, host %s' \
% (user, pid, host) % (user, pid, host)

mercurial/dirstate.py

"""filecache for files in the repository root""" """filecache for files in the repository root"""
def join(self, obj, fname): def join(self, obj, fname):
return obj._join(fname) return obj._join(fname)
def _getfsnow(vfs): def _getfsnow(vfs):
'''Get "now" timestamp on filesystem''' '''Get "now" timestamp on filesystem'''
tmpfd, tmpname = vfs.mkstemp() tmpfd, tmpname = vfs.mkstemp()
try: try:
return os.fstat(tmpfd).st_mtime return os.fstat(tmpfd)[stat.ST_MTIME]
finally: finally:
os.close(tmpfd) os.close(tmpfd)
vfs.unlink(tmpname) vfs.unlink(tmpname)
class dirstate(object): class dirstate(object):
def __init__(self, opener, ui, root, validate, sparsematchfn): def __init__(self, opener, ui, root, validate, sparsematchfn):
'''Create a new dirstate object. '''Create a new dirstate object.
_('file %r in dirstate clashes with %r') % (d, f)) _('file %r in dirstate clashes with %r') % (d, f))
self._dirty = True self._dirty = True
self._updatedfiles.add(f) self._updatedfiles.add(f)
self._map.addfile(f, oldstate, state, mode, size, mtime) self._map.addfile(f, oldstate, state, mode, size, mtime)
def normal(self, f): def normal(self, f):
'''Mark a file normal and clean.''' '''Mark a file normal and clean.'''
s = os.lstat(self._join(f)) s = os.lstat(self._join(f))
mtime = s.st_mtime mtime = s[stat.ST_MTIME]
self._addpath(f, 'n', s.st_mode, self._addpath(f, 'n', s.st_mode,
s.st_size & _rangemask, mtime & _rangemask) s.st_size & _rangemask, mtime & _rangemask)
self._map.copymap.pop(f, None) self._map.copymap.pop(f, None)
if f in self._map.nonnormalset: if f in self._map.nonnormalset:
self._map.nonnormalset.remove(f) self._map.nonnormalset.remove(f)
if mtime > self._lastnormaltime: if mtime > self._lastnormaltime:
# Remember the most recent modification timeslot for status(), # Remember the most recent modification timeslot for status(),
# to make sure we won't miss future size-preserving file content # to make sure we won't miss future size-preserving file content
def _writedirstate(self, st): def _writedirstate(self, st):
# notify callbacks about parents change # notify callbacks about parents change
if self._origpl is not None and self._origpl != self._pl: if self._origpl is not None and self._origpl != self._pl:
for c, callback in sorted(self._plchangecallbacks.iteritems()): for c, callback in sorted(self._plchangecallbacks.iteritems()):
callback(self, self._origpl, self._pl) callback(self, self._origpl, self._pl)
self._origpl = None self._origpl = None
# use the modification time of the newly created temporary file as the # use the modification time of the newly created temporary file as the
# filesystem's notion of 'now' # filesystem's notion of 'now'
now = util.fstat(st).st_mtime & _rangemask now = util.fstat(st)[stat.ST_MTIME] & _rangemask
# enough 'delaywrite' prevents 'pack_dirstate' from dropping # enough 'delaywrite' prevents 'pack_dirstate' from dropping
# timestamp of each entries in dirstate, because of 'now > mtime' # timestamp of each entries in dirstate, because of 'now > mtime'
delaywrite = self._ui.configint('debug', 'dirstate.delaywrite') delaywrite = self._ui.configint('debug', 'dirstate.delaywrite')
if delaywrite > 0: if delaywrite > 0:
# do we have any files to delay for? # do we have any files to delay for?
for f, e in self._map.iteritems(): for f, e in self._map.iteritems():
if e[0] == 'n' and e[3] == now: if e[0] == 'n' and e[3] == now:
dadd(fn) dadd(fn)
elif state == 'n': elif state == 'n':
if (size >= 0 and if (size >= 0 and
((size != st.st_size and size != st.st_size & _rangemask) ((size != st.st_size and size != st.st_size & _rangemask)
or ((mode ^ st.st_mode) & 0o100 and checkexec)) or ((mode ^ st.st_mode) & 0o100 and checkexec))
or size == -2 # other parent or size == -2 # other parent
or fn in copymap): or fn in copymap):
madd(fn) madd(fn)
elif time != st.st_mtime and time != st.st_mtime & _rangemask: elif (time != st[stat.ST_MTIME]
and time != st[stat.ST_MTIME] & _rangemask):
ladd(fn) ladd(fn)
elif st.st_mtime == lastnormaltime: elif st[stat.ST_MTIME] == lastnormaltime:
# fn may have just been marked as normal and it may have # fn may have just been marked as normal and it may have
# changed in the same second without changing its size. # changed in the same second without changing its size.
# This can happen if we quickly do multiple commits. # This can happen if we quickly do multiple commits.
# Force lookup, so we don't miss such a racy file change. # Force lookup, so we don't miss such a racy file change.
ladd(fn) ladd(fn)
elif listclean: elif listclean:
cadd(fn) cadd(fn)
elif state == 'm': elif state == 'm':

mercurial/hg.py

# hg.py - repository classes for mercurial # hg.py - repository classes for mercurial
# #
# Copyright 2005-2007 Matt Mackall <mpm@selenic.com> # Copyright 2005-2007 Matt Mackall <mpm@selenic.com>
# Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.com> # Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import errno import errno
import hashlib import hashlib
import os import os
import shutil import shutil
import stat
from .i18n import _ from .i18n import _
from .node import ( from .node import (
nullid, nullid,
) )
from . import ( from . import (
bookmarks, bookmarks,
maxmtime = -1 maxmtime = -1
for attr, fname in foi: for attr, fname in foi:
prefix = getattr(self._repo, attr) prefix = getattr(self._repo, attr)
p = os.path.join(prefix, fname) p = os.path.join(prefix, fname)
try: try:
st = os.stat(p) st = os.stat(p)
except OSError: except OSError:
st = os.stat(prefix) st = os.stat(prefix)
state.append((st.st_mtime, st.st_size)) state.append((st[stat.ST_MTIME], st.st_size))
maxmtime = max(maxmtime, st.st_mtime) maxmtime = max(maxmtime, st[stat.ST_MTIME])
return tuple(state), maxmtime return tuple(state), maxmtime
def copy(self): def copy(self):
"""Obtain a copy of this class instance. """Obtain a copy of this class instance.
A new localrepository instance is obtained. The new instance should be A new localrepository instance is obtained. The new instance should be
completely independent of the original. completely independent of the original.

mercurial/hgweb/common.py

# hgweb/common.py - Utility functions needed by hgweb_mod and hgwebdir_mod # hgweb/common.py - Utility functions needed by hgweb_mod and hgwebdir_mod
# #
# Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net> # Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net>
# Copyright 2005, 2006 Matt Mackall <mpm@selenic.com> # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import base64 import base64
import errno import errno
import mimetypes import mimetypes
import os import os
import stat
from .. import ( from .. import (
encoding, encoding,
pycompat, pycompat,
util, util,
) )
httpserver = util.httpserver httpserver = util.httpserver
"""stat fn if it exists, spath otherwise""" """stat fn if it exists, spath otherwise"""
cl_path = os.path.join(spath, fn) cl_path = os.path.join(spath, fn)
if os.path.exists(cl_path): if os.path.exists(cl_path):
return os.stat(cl_path) return os.stat(cl_path)
else: else:
return os.stat(spath) return os.stat(spath)
def get_mtime(spath): def get_mtime(spath):
return get_stat(spath, "00changelog.i").st_mtime return get_stat(spath, "00changelog.i")[stat.ST_MTIME]
def ispathsafe(path): def ispathsafe(path):
"""Determine if a path is safe to use for filesystem access.""" """Determine if a path is safe to use for filesystem access."""
parts = path.split('/') parts = path.split('/')
for part in parts: for part in parts:
if (part in ('', pycompat.oscurdir, pycompat.ospardir) or if (part in ('', pycompat.oscurdir, pycompat.ospardir) or
pycompat.ossep in part or pycompat.ossep in part or
pycompat.osaltsep is not None and pycompat.osaltsep in part): pycompat.osaltsep is not None and pycompat.osaltsep in part):

mercurial/posix.py

# changes. # changes.
return (self.stat.st_mode == other.stat.st_mode and return (self.stat.st_mode == other.stat.st_mode and
self.stat.st_ino == other.stat.st_ino and self.stat.st_ino == other.stat.st_ino and
self.stat.st_dev == other.stat.st_dev and self.stat.st_dev == other.stat.st_dev and
self.stat.st_nlink == other.stat.st_nlink and self.stat.st_nlink == other.stat.st_nlink and
self.stat.st_uid == other.stat.st_uid and self.stat.st_uid == other.stat.st_uid and
self.stat.st_gid == other.stat.st_gid and self.stat.st_gid == other.stat.st_gid and
self.stat.st_size == other.stat.st_size and self.stat.st_size == other.stat.st_size and
self.stat.st_mtime == other.stat.st_mtime and self.stat[stat.ST_MTIME] == other.stat[stat.ST_MTIME] and
self.stat.st_ctime == other.stat.st_ctime) self.stat[stat.ST_CTIME] == other.stat[stat.ST_CTIME])
except AttributeError: except AttributeError:
return False return False
def __ne__(self, other): def __ne__(self, other):
return not self == other return not self == other
def executablepath(): def executablepath():
return None # available on Windows only return None # available on Windows only

mercurial/util.py

# copystat also copies mode # copystat also copies mode
shutil.copystat(src, dest) shutil.copystat(src, dest)
else: else:
shutil.copymode(src, dest) shutil.copymode(src, dest)
if oldstat and oldstat.stat: if oldstat and oldstat.stat:
newstat = filestat.frompath(dest) newstat = filestat.frompath(dest)
if newstat.isambig(oldstat): if newstat.isambig(oldstat):
# stat of copied file is ambiguous to original one # stat of copied file is ambiguous to original one
advanced = (oldstat.stat.st_mtime + 1) & 0x7fffffff advanced = (
oldstat.stat[stat.ST_MTIME] + 1) & 0x7fffffff
os.utime(dest, (advanced, advanced)) os.utime(dest, (advanced, advanced))
except shutil.Error as inst: except shutil.Error as inst:
raise Abort(str(inst)) raise Abort(str(inst))
def copyfiles(src, dst, hardlink=None, progress=lambda t, pos: None): def copyfiles(src, dst, hardlink=None, progress=lambda t, pos: None):
"""Copy a directory tree using hardlinks if possible.""" """Copy a directory tree using hardlinks if possible."""
num = 0 num = 0
__hash__ = object.__hash__ __hash__ = object.__hash__
def __eq__(self, old): def __eq__(self, old):
try: try:
# if ambiguity between stat of new and old file is # if ambiguity between stat of new and old file is
# avoided, comparison of size, ctime and mtime is enough # avoided, comparison of size, ctime and mtime is enough
# to exactly detect change of a file regardless of platform # to exactly detect change of a file regardless of platform
return (self.stat.st_size == old.stat.st_size and return (self.stat.st_size == old.stat.st_size and
self.stat.st_ctime == old.stat.st_ctime and self.stat[stat.ST_CTIME] == old.stat[stat.ST_CTIME] and
self.stat.st_mtime == old.stat.st_mtime) self.stat[stat.ST_MTIME] == old.stat[stat.ST_MTIME])
except AttributeError: except AttributeError:
pass pass
try: try:
return self.stat is None and old.stat is None return self.stat is None and old.stat is None
except AttributeError: except AttributeError:
return False return False
def isambig(self, old): def isambig(self, old):
Therefore, all "S[n-1].ctime == S[n].ctime" cases should be Therefore, all "S[n-1].ctime == S[n].ctime" cases should be
treated as ambiguous regardless of mtime, to avoid overlooking treated as ambiguous regardless of mtime, to avoid overlooking
by confliction between such mtime. by confliction between such mtime.
Advancing mtime "if isambig(oldstat)" ensures "S[n-1].mtime != Advancing mtime "if isambig(oldstat)" ensures "S[n-1].mtime !=
S[n].mtime", even if size of a file isn't changed. S[n].mtime", even if size of a file isn't changed.
""" """
try: try:
return (self.stat.st_ctime == old.stat.st_ctime) return (self.stat[stat.ST_CTIME] == old.stat[stat.ST_CTIME])
except AttributeError: except AttributeError:
return False return False
def avoidambig(self, path, old): def avoidambig(self, path, old):
"""Change file stat of specified path to avoid ambiguity """Change file stat of specified path to avoid ambiguity
'old' should be previous filestat of 'path'. 'old' should be previous filestat of 'path'.
This skips avoiding ambiguity, if a process doesn't have This skips avoiding ambiguity, if a process doesn't have
appropriate privileges for 'path'. This returns False in this appropriate privileges for 'path'. This returns False in this
case. case.
Otherwise, this returns True, as "ambiguity is avoided". Otherwise, this returns True, as "ambiguity is avoided".
""" """
advanced = (old.stat.st_mtime + 1) & 0x7fffffff advanced = (old.stat[stat.ST_MTIME] + 1) & 0x7fffffff
try: try:
os.utime(path, (advanced, advanced)) os.utime(path, (advanced, advanced))
except OSError as inst: except OSError as inst:
if inst.errno == errno.EPERM: if inst.errno == errno.EPERM:
# utime() on the file created by another user causes EPERM, # utime() on the file created by another user causes EPERM,
# if a process doesn't have appropriate privileges # if a process doesn't have appropriate privileges
return False return False
raise raise
self._fp.close() self._fp.close()
filename = localpath(self.__name) filename = localpath(self.__name)
oldstat = self._checkambig and filestat.frompath(filename) oldstat = self._checkambig and filestat.frompath(filename)
if oldstat and oldstat.stat: if oldstat and oldstat.stat:
rename(self._tempname, filename) rename(self._tempname, filename)
newstat = filestat.frompath(filename) newstat = filestat.frompath(filename)
if newstat.isambig(oldstat): if newstat.isambig(oldstat):
# stat of changed file is ambiguous to original one # stat of changed file is ambiguous to original one
advanced = (oldstat.stat.st_mtime + 1) & 0x7fffffff advanced = (oldstat.stat[stat.ST_MTIME] + 1) & 0x7fffffff
os.utime(filename, (advanced, advanced)) os.utime(filename, (advanced, advanced))
else: else:
rename(self._tempname, filename) rename(self._tempname, filename)
def discard(self): def discard(self):
if not self._fp.closed: if not self._fp.closed:
try: try:
os.unlink(self._tempname) os.unlink(self._tempname)

tests/svn-safe-append.py

#!/usr/bin/env python #!/usr/bin/env python
from __future__ import absolute_import from __future__ import absolute_import
__doc__ = """Same as `echo a >> b`, but ensures a changed mtime of b. __doc__ = """Same as `echo a >> b`, but ensures a changed mtime of b.
Without this svn will not detect workspace changes.""" Without this svn will not detect workspace changes."""
import os import os
import stat
import sys import sys
text = sys.argv[1] text = sys.argv[1]
fname = sys.argv[2] fname = sys.argv[2]
f = open(fname, "ab") f = open(fname, "ab")
try: try:
before = os.fstat(f.fileno()).st_mtime before = os.fstat(f.fileno())[stat.ST_MTIME]
f.write(text) f.write(text)
f.write("\n") f.write("\n")
finally: finally:
f.close() f.close()
inc = 1 inc = 1
now = os.stat(fname).st_mtime now = os.stat(fname)[stat.ST_MTIME]
while now == before: while now == before:
t = now + inc t = now + inc
inc += 1 inc += 1
os.utime(fname, (t, t)) os.utime(fname, (t, t))
now = os.stat(fname).st_mtime now = os.stat(fname)[stat.ST_MTIME]

tests/test-atomictempfile.py

from __future__ import absolute_import from __future__ import absolute_import
import glob import glob
import os import os
import shutil import shutil
import stat
import tempfile import tempfile
import unittest import unittest
from mercurial import ( from mercurial import (
pycompat, pycompat,
util, util,
) )
atomictempfile = util.atomictempfile atomictempfile = util.atomictempfile
f.write(b'FOO') f.write(b'FOO')
f.close() f.close()
# try some times, because reproduction of ambiguity depends on # try some times, because reproduction of ambiguity depends on
# "filesystem time" # "filesystem time"
for i in xrange(5): for i in xrange(5):
atomicwrite(False) atomicwrite(False)
oldstat = os.stat(self._filename) oldstat = os.stat(self._filename)
if oldstat.st_ctime != oldstat.st_mtime: if oldstat[stat.ST_CTIME] != oldstat[stat.ST_MTIME]:
# subsequent changing never causes ambiguity # subsequent changing never causes ambiguity
continue continue
repetition = 3 repetition = 3
# repeat atomic write with checkambig=True, to examine # repeat atomic write with checkambig=True, to examine
# whether st_mtime is advanced multiple times as expected # whether st_mtime is advanced multiple times as expected
for j in xrange(repetition): for j in xrange(repetition):
atomicwrite(True) atomicwrite(True)
newstat = os.stat(self._filename) newstat = os.stat(self._filename)
if oldstat.st_ctime != newstat.st_ctime: if oldstat[stat.ST_CTIME] != newstat[stat.ST_CTIME]:
# timestamp ambiguity was naturally avoided while repetition # timestamp ambiguity was naturally avoided while repetition
continue continue
# st_mtime should be advanced "repetition" times, because # st_mtime should be advanced "repetition" times, because
# all atomicwrite() occurred at same time (in sec) # all atomicwrite() occurred at same time (in sec)
self.assertTrue(newstat.st_mtime == oldtime = (oldstat[stat.ST_MTIME] + repetition) & 0x7fffffff
((oldstat.st_mtime + repetition) & 0x7fffffff)) self.assertTrue(newstat[stat.ST_MTIME] == oldtime)
# no more examination is needed, if assumption above is true # no more examination is needed, if assumption above is true
break break
else: else:
# This platform seems too slow to examine anti-ambiguity # This platform seems too slow to examine anti-ambiguity
# of file timestamp (or test happened to be executed at # of file timestamp (or test happened to be executed at
# bad timing). Exit silently in this case, because running # bad timing). Exit silently in this case, because running
# on other faster platforms can detect problems # on other faster platforms can detect problems
pass pass

tests/test-context.py

from __future__ import absolute_import, print_function from __future__ import absolute_import, print_function
import os import os
import stat
from mercurial.node import hex from mercurial.node import hex
from mercurial import ( from mercurial import (
context, context,
encoding, encoding,
hg, hg,
scmutil, scmutil,
ui as uimod, ui as uimod,
) )
date=(0, 0)) date=(0, 0))
ctx.p1().manifest() # side effect: cache manifestctx ctx.p1().manifest() # side effect: cache manifestctx
n = repo.commitctx(ctx) n = repo.commitctx(ctx)
print('commit %s: %s' % (i, hex(n))) print('commit %s: %s' % (i, hex(n)))
# touch 00manifest.i mtime so storecache could expire. # touch 00manifest.i mtime so storecache could expire.
# repo.__dict__['manifestlog'] is deleted by transaction releasefn. # repo.__dict__['manifestlog'] is deleted by transaction releasefn.
st = repo.svfs.stat('00manifest.i') st = repo.svfs.stat('00manifest.i')
repo.svfs.utime('00manifest.i', (st.st_mtime + 1, st.st_mtime + 1)) repo.svfs.utime('00manifest.i',
(st[stat.ST_MTIME] + 1, st[stat.ST_MTIME] + 1))
# read the file just committed # read the file just committed
try: try:
if repo[n][i].data() != i: if repo[n][i].data() != i:
print('data mismatch') print('data mismatch')
except Exception as ex: except Exception as ex:
print('cannot read data: %r' % ex) print('cannot read data: %r' % ex)

tests/test-filecache.py

from __future__ import absolute_import, print_function from __future__ import absolute_import, print_function
import os import os
import stat
import subprocess import subprocess
import sys import sys
if subprocess.call(['python', '%s/hghave' % os.environ['TESTDIR'], if subprocess.call(['python', '%s/hghave' % os.environ['TESTDIR'],
'cacheable']): 'cacheable']):
sys.exit(80) sys.exit(80)
from mercurial import ( from mercurial import (
# try some times, because reproduction of ambiguity depends on # try some times, because reproduction of ambiguity depends on
# "filesystem time" # "filesystem time"
for i in xrange(5): for i in xrange(5):
fp = open(filename, 'w') fp = open(filename, 'w')
fp.write('FOO') fp.write('FOO')
fp.close() fp.close()
oldstat = os.stat(filename) oldstat = os.stat(filename)
if oldstat.st_ctime != oldstat.st_mtime: if oldstat[stat.ST_CTIME] != oldstat[stat.ST_MTIME]:
# subsequent changing never causes ambiguity # subsequent changing never causes ambiguity
continue continue
repetition = 3 repetition = 3
# repeat changing via checkambigatclosing, to examine whether # repeat changing via checkambigatclosing, to examine whether
# st_mtime is advanced multiple times as expected # st_mtime is advanced multiple times as expected
for i in xrange(repetition): for i in xrange(repetition):
# explicit closing # explicit closing
fp = vfsmod.checkambigatclosing(open(filename, 'a')) fp = vfsmod.checkambigatclosing(open(filename, 'a'))
fp.write('FOO') fp.write('FOO')
fp.close() fp.close()
# implicit closing by "with" statement # implicit closing by "with" statement
with vfsmod.checkambigatclosing(open(filename, 'a')) as fp: with vfsmod.checkambigatclosing(open(filename, 'a')) as fp:
fp.write('BAR') fp.write('BAR')
newstat = os.stat(filename) newstat = os.stat(filename)
if oldstat.st_ctime != newstat.st_ctime: if oldstat[stat.ST_CTIME] != newstat[stat.ST_CTIME]:
# timestamp ambiguity was naturally avoided while repetition # timestamp ambiguity was naturally avoided while repetition
continue continue
# st_mtime should be advanced "repetition * 2" times, because # st_mtime should be advanced "repetition * 2" times, because
# all changes occurred at same time (in sec) # all changes occurred at same time (in sec)
expected = (oldstat.st_mtime + repetition * 2) & 0x7fffffff expected = (oldstat[stat.ST_MTIME] + repetition * 2) & 0x7fffffff
if newstat.st_mtime != expected: if newstat[stat.ST_MTIME] != expected:
print("'newstat.st_mtime %s is not %s (as %s + %s * 2)" % print("'newstat[stat.ST_MTIME] %s is not %s (as %s + %s * 2)" %
(newstat.st_mtime, expected, oldstat.st_mtime, repetition)) (newstat[stat.ST_MTIME], expected,
oldstat[stat.ST_MTIME], repetition))
# no more examination is needed regardless of result # no more examination is needed regardless of result
break break
else: else:
# This platform seems too slow to examine anti-ambiguity # This platform seems too slow to examine anti-ambiguity
# of file timestamp (or test happened to be executed at # of file timestamp (or test happened to be executed at
# bad timing). Exit silently in this case, because running # bad timing). Exit silently in this case, because running
# on other faster platforms can detect problems # on other faster platforms can detect problems