blob: 2ff7b35273e15658d35dbcbda9815ef47f7c6506 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
# Copyright 2011 Gentoo Foundation
# Distributed under the terms of the GNU General Public License v2
import tempfile
import traceback
import portage
from portage import os
from portage import shutil
from portage.tests import TestCase
class LockNonblockTestCase(TestCase):
def _testLockNonblock(self):
tempdir = tempfile.mkdtemp()
try:
path = os.path.join(tempdir, 'lock_me')
lock1 = portage.locks.lockfile(path)
pid = os.fork()
if pid == 0:
portage.locks._close_fds()
# Disable close_fds since we don't exec
# (see _setup_pipes docstring).
portage.process._setup_pipes({0:0, 1:1, 2:2}, close_fds=False)
rval = 2
try:
try:
lock2 = portage.locks.lockfile(path, flags=os.O_NONBLOCK)
except portage.exception.TryAgain:
rval = os.EX_OK
else:
rval = 1
portage.locks.unlockfile(lock2)
except SystemExit:
raise
except:
traceback.print_exc()
finally:
os._exit(rval)
self.assertEqual(pid > 0, True)
pid, status = os.waitpid(pid, 0)
self.assertEqual(os.WIFEXITED(status), True)
self.assertEqual(os.WEXITSTATUS(status), os.EX_OK)
portage.locks.unlockfile(lock1)
finally:
shutil.rmtree(tempdir)
def testLockNonblock(self):
self._testLockNonblock()
def testLockNonblockHardlink(self):
prev_state = os.environ.pop("__PORTAGE_TEST_HARDLINK_LOCKS", None)
os.environ["__PORTAGE_TEST_HARDLINK_LOCKS"] = "1"
try:
self._testLockNonblock()
finally:
os.environ.pop("__PORTAGE_TEST_HARDLINK_LOCKS", None)
if prev_state is not None:
os.environ["__PORTAGE_TEST_HARDLINK_LOCKS"] = prev_state
|
