mirror of
https://github.com/facebook/sapling.git
synced 2024-10-10 16:57:49 +03:00
c074611dee
Summary: Now that the workers are in Rust, we no longer need the forker version in Python. For now, the Python LFS extension still uses the threaded worker so keep this one for now, when that extension will be removed we can remove the rest of the worker code. In theory, not all repository would benefit from the Rust workers, but these are not supported at FB due to not being remotefilelog based. Reviewed By: DurhamG Differential Revision: D21987295 fbshipit-source-id: d17b9730651671608cf13f7abe6a9bb32251e140
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
# Copyright 2004-present Facebook. All Rights Reserved.
|
|
|
|
from __future__ import absolute_import
|
|
|
|
import errno
|
|
import os
|
|
import time
|
|
import unittest
|
|
|
|
import silenttestrunner
|
|
from edenscm.hgext import extutil
|
|
from edenscm.mercurial import error, vfs
|
|
from hghave import require
|
|
|
|
|
|
locktimeout = 25
|
|
locksuccess = 24
|
|
|
|
|
|
def _exitstatus(code):
|
|
"""convert a posix exit status into the same form returned by
|
|
os.spawnv
|
|
|
|
returns None if the process was stopped instead of exiting"""
|
|
if os.WIFEXITED(code):
|
|
return os.WEXITSTATUS(code)
|
|
elif os.WIFSIGNALED(code):
|
|
return -(os.WTERMSIG(code))
|
|
|
|
|
|
class ExtutilTests(unittest.TestCase):
|
|
def testbgcommandnoblock(self):
|
|
"""runbgcommand() should return without waiting for the process to
|
|
finish."""
|
|
env = os.environ.copy()
|
|
start = time.time()
|
|
extutil.runbgcommand(["sleep", "5"], env)
|
|
end = time.time()
|
|
if end - start >= 1.0:
|
|
self.fail(
|
|
"runbgcommand() took took %s seconds, should have "
|
|
"returned immediately" % (end - start)
|
|
)
|
|
|
|
def testbgcommandfailure1(self):
|
|
"""runbgcommand() should throw if executing the process fails."""
|
|
env = os.environ.copy()
|
|
try:
|
|
extutil.runbgcommand(["no_such_program", "arg1", "arg2"], env)
|
|
self.fail("expected runbgcommand to fail with ENOENT")
|
|
except OSError as ex:
|
|
self.assertEqual(ex.errno, errno.ENOENT)
|
|
|
|
def testbgcommandfailure2(self):
|
|
"""runbgcommand() should throw if executing the process fails."""
|
|
env = os.environ.copy()
|
|
try:
|
|
extutil.runbgcommand([os.devnull, "arg1", "arg2"], env)
|
|
self.fail("expected runbgcommand to fail with EACCES")
|
|
except OSError as ex:
|
|
self.assertEqual(ex.errno, errno.EACCES)
|
|
|
|
def testflock(self):
|
|
testtmp = os.environ["TESTTMP"]
|
|
opener = vfs.vfs(testtmp)
|
|
name = "testlock"
|
|
|
|
with extutil.flock(opener.join(name), "testing a lock", timeout=0):
|
|
otherlock = self.otherprocesslock(opener, name)
|
|
self.assertEqual(
|
|
otherlock, locktimeout, "other process should not have taken the lock"
|
|
)
|
|
|
|
otherlock = self.otherprocesslock(opener, name)
|
|
self.assertEqual(
|
|
otherlock, locksuccess, "other process should have taken the lock"
|
|
)
|
|
|
|
def otherprocesslock(self, opener, name):
|
|
pid = os.fork()
|
|
if pid == 0:
|
|
try:
|
|
with extutil.flock(opener.join(name), "other process lock", timeout=0):
|
|
os._exit(locksuccess)
|
|
except error.LockHeld:
|
|
os._exit(locktimeout)
|
|
else:
|
|
p, st = os.waitpid(pid, 0)
|
|
st = _exitstatus(st) # Convert back to an int
|
|
return st
|
|
|
|
|
|
if __name__ == "__main__":
|
|
silenttestrunner.main(__name__)
|