/usr/lib/python2.7/dist-packages/twext/enterprise/test/test_locking.py is in python-twext 0.1.b2.dev15059-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | ##
# Copyright (c) 2012-2015 Apple Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##
"""
Tests for mutual exclusion locks.
"""
from twisted.internet.defer import inlineCallbacks
from twisted.trial.unittest import TestCase
from twext.enterprise.fixtures import buildConnectionPool
from twext.enterprise.locking import NamedLock, LockTimeout
from twext.enterprise.dal.syntax import Select
from twext.enterprise.locking import LockSchema
schemaText = """
create table NAMED_LOCK (LOCK_NAME varchar(255) unique primary key);
"""
class TestLocking(TestCase):
"""
Test locking and unlocking a database row.
"""
def setUp(self):
"""
Build a connection pool for the tests to use.
"""
self.pool = buildConnectionPool(self, schemaText)
@inlineCallbacks
def test_acquire(self):
"""
Acquiring a lock adds a row in that transaction.
"""
txn = self.pool.connection()
yield NamedLock.acquire(txn, u"a test lock")
rows = yield Select(From=LockSchema.NAMED_LOCK).on(txn)
self.assertEquals(rows, [tuple([u"a test lock"])])
@inlineCallbacks
def test_release(self):
"""
Releasing an acquired lock removes the row.
"""
txn = self.pool.connection()
lck = yield NamedLock.acquire(txn, u"a test lock")
yield lck.release()
rows = yield Select(From=LockSchema.NAMED_LOCK).on(txn)
self.assertEquals(rows, [])
@inlineCallbacks
def test_autoRelease(self):
"""
Committing a transaction automatically releases all of its locks.
"""
txn = self.pool.connection()
yield NamedLock.acquire(txn, u"something")
yield txn.commit()
txn2 = self.pool.connection()
rows = yield Select(From=LockSchema.NAMED_LOCK).on(txn2)
self.assertEquals(rows, [])
@inlineCallbacks
def test_timeout(self):
"""
Trying to acquire second lock times out.
"""
txn1 = self.pool.connection()
yield NamedLock.acquire(txn1, u"a test lock")
txn2 = self.pool.connection()
yield self.assertFailure(
NamedLock.acquire(txn2, u"a test lock"), LockTimeout
)
yield txn2.abort()
self.flushLoggedErrors()
|