# -*- test-case-name: twisted.python.test.test_util # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.python.util}. """ from __future__ import division, absolute_import import os.path, sys import shutil, errno, warnings try: import pwd, grp except ImportError: pwd = grp = None from twisted.trial import unittest from twisted.trial.util import suppress as SUPPRESS from twisted.python.compat import _PY3 from twisted.python import util from twisted.python.versions import Version from twisted.internet import reactor from twisted.internet.interfaces import IReactorProcess from twisted.internet.protocol import ProcessProtocol from twisted.internet.defer import Deferred from twisted.internet.error import ProcessDone if _PY3: MockOS = None from twisted.python.deprecate import _fullyQualifiedName as fullyQualifiedName else: from twisted.test.test_process import MockOS from twisted.python.reflect import fullyQualifiedName class UtilTestCase(unittest.TestCase): def testUniq(self): l = ["a", 1, "ab", "a", 3, 4, 1, 2, 2, 4, 6] self.assertEqual(util.uniquify(l), ["a", 1, "ab", 3, 4, 2, 6]) def testRaises(self): self.failUnless(util.raises(ZeroDivisionError, divmod, 1, 0)) self.failIf(util.raises(ZeroDivisionError, divmod, 0, 1)) try: util.raises(TypeError, divmod, 1, 0) except ZeroDivisionError: pass else: raise unittest.FailTest("util.raises didn't raise when it should have") def test_uidFromNumericString(self): """ When L{uidFromString} is called with a base-ten string representation of an integer, it returns the integer. """ self.assertEqual(util.uidFromString("100"), 100) def test_uidFromUsernameString(self): """ When L{uidFromString} is called with a base-ten string representation of an integer, it returns the integer. """ pwent = pwd.getpwuid(os.getuid()) self.assertEqual(util.uidFromString(pwent.pw_name), pwent.pw_uid) if pwd is None: test_uidFromUsernameString.skip = ( "Username/UID conversion requires the pwd module.") def test_gidFromNumericString(self): """ When L{gidFromString} is called with a base-ten string representation of an integer, it returns the integer. """ self.assertEqual(util.gidFromString("100"), 100) def test_gidFromGroupnameString(self): """ When L{gidFromString} is called with a base-ten string representation of an integer, it returns the integer. """ grent = grp.getgrgid(os.getgid()) self.assertEqual(util.gidFromString(grent.gr_name), grent.gr_gid) if grp is None: test_gidFromGroupnameString.skip = ( "Group Name/GID conversion requires the grp module.") class NameToLabelTests(unittest.TestCase): """ Tests for L{nameToLabel}. """ def test_nameToLabel(self): """ Test the various kinds of inputs L{nameToLabel} supports. """ nameData = [ ('f', 'F'), ('fo', 'Fo'), ('foo', 'Foo'), ('fooBar', 'Foo Bar'), ('fooBarBaz', 'Foo Bar Baz'), ] for inp, out in nameData: got = util.nameToLabel(inp) self.assertEqual( got, out, "nameToLabel(%r) == %r != %r" % (inp, got, out)) class UntilConcludesTests(unittest.TestCase): """ Tests for L{untilConcludes}, an C{EINTR} helper. """ def test_uninterruptably(self): """ L{untilConcludes} calls the function passed to it until the function does not raise either L{OSError} or L{IOError} with C{errno} of C{EINTR}. It otherwise completes with the same result as the function passed to it. """ def f(a, b): self.calls += 1 exc = self.exceptions.pop() if exc is not None: raise exc(errno.EINTR, "Interrupted system call!") return a + b self.exceptions = [None] self.calls = 0 self.assertEqual(util.untilConcludes(f, 1, 2), 3) self.assertEqual(self.calls, 1) self.exceptions = [None, OSError, IOError] self.calls = 0 self.assertEqual(util.untilConcludes(f, 2, 3), 5) self.assertEqual(self.calls, 3) class SwitchUIDTest(unittest.TestCase): """ Tests for L{util.switchUID}. """ if getattr(os, "getuid", None) is None: skip = "getuid/setuid not available" def setUp(self): self.mockos = MockOS() self.patch(util, "os", self.mockos) self.patch(util, "initgroups", self.initgroups) self.initgroupsCalls = [] def initgroups(self, uid, gid): """ Save L{util.initgroups} calls in C{self.initgroupsCalls}. """ self.initgroupsCalls.append((uid, gid)) def test_uid(self): """ L{util.switchUID} calls L{util.initgroups} and then C{os.setuid} with the given uid. """ util.switchUID(12000, None) self.assertEqual(self.initgroupsCalls, [(12000, None)]) self.assertEqual(self.mockos.actions, [("setuid", 12000)]) def test_euid(self): """ L{util.switchUID} calls L{util.initgroups} and then C{os.seteuid} with the given uid if the C{euid} parameter is set to C{True}. """ util.switchUID(12000, None, True) self.assertEqual(self.initgroupsCalls, [(12000, None)]) self.assertEqual(self.mockos.seteuidCalls, [12000]) def test_currentUID(self): """ If the current uid is the same as the uid passed to L{util.switchUID}, then initgroups does not get called, but a warning is issued. """ uid = self.mockos.getuid() util.switchUID(uid, None) self.assertEqual(self.initgroupsCalls, []) self.assertEqual(self.mockos.actions, []) warnings = self.flushWarnings([util.switchUID]) self.assertEqual(len(warnings), 1) self.assertIn('tried to drop privileges and setuid %i' % uid, warnings[0]['message']) self.assertIn('but uid is already %i' % uid, warnings[0]['message']) def test_currentEUID(self): """ If the current euid is the same as the euid passed to L{util.switchUID}, then initgroups does not get called, but a warning is issued. """ euid = self.mockos.geteuid() util.switchUID(euid, None, True) self.assertEqual(self.initgroupsCalls, []) self.assertEqual(self.mockos.seteuidCalls, []) warnings = self.flushWarnings([util.switchUID]) self.assertEqual(len(warnings), 1) self.assertIn('tried to drop privileges and seteuid %i' % euid, warnings[0]['message']) self.assertIn('but euid is already %i' % euid, warnings[0]['message']) class TestMergeFunctionMetadata(unittest.TestCase): """ Tests for L{mergeFunctionMetadata}. """ def test_mergedFunctionBehavesLikeMergeTarget(self): """ After merging C{foo}'s data into C{bar}, the returned function behaves as if it is C{bar}. """ foo_object = object() bar_object = object() def foo(): return foo_object def bar(x, y, ab, c=10, *d, **e): (a, b) = ab return bar_object baz = util.mergeFunctionMetadata(foo, bar) self.assertIdentical(baz(1, 2, (3, 4), quux=10), bar_object) def test_moduleIsMerged(self): """ Merging C{foo} into C{bar} returns a function with C{foo}'s C{__module__}. """ def foo(): pass def bar(): pass bar.__module__ = 'somewhere.else' baz = util.mergeFunctionMetadata(foo, bar) self.assertEqual(baz.__module__, foo.__module__) def test_docstringIsMerged(self): """ Merging C{foo} into C{bar} returns a function with C{foo}'s docstring. """ def foo(): """ This is foo. """ def bar(): """ This is bar. """ baz = util.mergeFunctionMetadata(foo, bar) self.assertEqual(baz.__doc__, foo.__doc__) def test_nameIsMerged(self): """ Merging C{foo} into C{bar} returns a function with C{foo}'s name. """ def foo(): pass def bar(): pass baz = util.mergeFunctionMetadata(foo, bar) self.assertEqual(baz.__name__, foo.__name__) def test_instanceDictionaryIsMerged(self): """ Merging C{foo} into C{bar} returns a function with C{bar}'s dictionary, updated by C{foo}'s. """ def foo(): pass foo.a = 1 foo.b = 2 def bar(): pass bar.b = 3 bar.c = 4 baz = util.mergeFunctionMetadata(foo, bar) self.assertEqual(foo.a, baz.a) self.assertEqual(foo.b, baz.b) self.assertEqual(bar.c, baz.c) class OrderedDictTest(unittest.TestCase): def testOrderedDict(self): d = util.OrderedDict() d['a'] = 'b' d['b'] = 'a' d[3] = 12 d[1234] = 4321 self.assertEqual(repr(d), "{'a': 'b', 'b': 'a', 3: 12, 1234: 4321}") self.assertEqual(d.values(), ['b', 'a', 12, 4321]) del d[3] self.assertEqual(repr(d), "{'a': 'b', 'b': 'a', 1234: 4321}") self.assertEqual(d, {'a': 'b', 'b': 'a', 1234:4321}) self.assertEqual(d.keys(), ['a', 'b', 1234]) self.assertEqual(list(d.iteritems()), [('a', 'b'), ('b','a'), (1234, 4321)]) item = d.popitem() self.assertEqual(item, (1234, 4321)) def testInitialization(self): d = util.OrderedDict({'monkey': 'ook', 'apple': 'red'}) self.failUnless(d._order) d = util.OrderedDict(((1,1),(3,3),(2,2),(0,0))) self.assertEqual(repr(d), "{1: 1, 3: 3, 2: 2, 0: 0}") class InsensitiveDictTest(unittest.TestCase): """ Tests for L{util.InsensitiveDict}. """ def test_preserve(self): """ L{util.InsensitiveDict} preserves the case of keys if constructed with C{preserve=True}. """ dct = util.InsensitiveDict({'Foo':'bar', 1:2, 'fnz':{1:2}}, preserve=1) self.assertEqual(dct['fnz'], {1:2}) self.assertEqual(dct['foo'], 'bar') self.assertEqual(dct.copy(), dct) self.assertEqual(dct['foo'], dct.get('Foo')) self.assertIn(1, dct) self.assertIn('foo', dct) # Make eval() work, urrrrgh: InsensitiveDict = util.InsensitiveDict self.assertEqual(eval(repr(dct)), dct) keys=['Foo', 'fnz', 1] for x in keys: self.assertIn(x, dct.keys()) self.assertIn((x, dct[x]), dct.items()) self.assertEqual(len(keys), len(dct)) del dct[1] del dct['foo'] self.assertEqual(dct.keys(), ['fnz']) def test_noPreserve(self): """ L{util.InsensitiveDict} does not preserves the case of keys if constructed with C{preserve=False}. """ dct = util.InsensitiveDict({'Foo':'bar', 1:2, 'fnz':{1:2}}, preserve=0) keys=['foo', 'fnz', 1] for x in keys: self.assertIn(x, dct.keys()) self.assertIn((x, dct[x]), dct.items()) self.assertEqual(len(keys), len(dct)) del dct[1] del dct['foo'] self.assertEqual(dct.keys(), ['fnz']) def test_unicode(self): """ Unicode keys are case insensitive. """ d = util.InsensitiveDict(preserve=False) d[u"Foo"] = 1 self.assertEqual(d[u"FOO"], 1) self.assertEqual(d.keys(), [u"foo"]) def test_bytes(self): """ Bytes keys are case insensitive. """ d = util.InsensitiveDict(preserve=False) d[b"Foo"] = 1 self.assertEqual(d[b"FOO"], 1) self.assertEqual(d.keys(), [b"foo"]) class PasswordTestingProcessProtocol(ProcessProtocol): """ Write the string C{"secret\n"} to a subprocess and then collect all of its output and fire a Deferred with it when the process ends. """ def connectionMade(self): self.output = [] self.transport.write('secret\n') def childDataReceived(self, fd, output): self.output.append((fd, output)) def processEnded(self, reason): self.finished.callback((reason, self.output)) class GetPasswordTest(unittest.TestCase): if not IReactorProcess.providedBy(reactor): skip = "Process support required to test getPassword" def test_stdin(self): """ Making sure getPassword accepts a password from standard input by running a child process which uses getPassword to read in a string which it then writes it out again. Write a string to the child process and then read one and make sure it is the right string. """ p = PasswordTestingProcessProtocol() p.finished = Deferred() reactor.spawnProcess( p, sys.executable, [sys.executable, '-c', ('import sys\n' 'from twisted.python.util import getPassword\n' 'sys.stdout.write(getPassword())\n' 'sys.stdout.flush()\n')], env={'PYTHONPATH': os.pathsep.join(sys.path)}) def processFinished(result): (reason, output) = result reason.trap(ProcessDone) self.assertIn((1, 'secret'), output) return p.finished.addCallback(processFinished) class SearchUpwardsTest(unittest.TestCase): def testSearchupwards(self): os.makedirs('searchupwards/a/b/c') file('searchupwards/foo.txt', 'w').close() file('searchupwards/a/foo.txt', 'w').close() file('searchupwards/a/b/c/foo.txt', 'w').close() os.mkdir('searchupwards/bar') os.mkdir('searchupwards/bam') os.mkdir('searchupwards/a/bar') os.mkdir('searchupwards/a/b/bam') actual=util.searchupwards('searchupwards/a/b/c', files=['foo.txt'], dirs=['bar', 'bam']) expected=os.path.abspath('searchupwards') + os.sep self.assertEqual(actual, expected) shutil.rmtree('searchupwards') actual=util.searchupwards('searchupwards/a/b/c', files=['foo.txt'], dirs=['bar', 'bam']) expected=None self.assertEqual(actual, expected) class IntervalDifferentialTestCase(unittest.TestCase): def testDefault(self): d = iter(util.IntervalDifferential([], 10)) for i in range(100): self.assertEqual(d.next(), (10, None)) def testSingle(self): d = iter(util.IntervalDifferential([5], 10)) for i in range(100): self.assertEqual(d.next(), (5, 0)) def testPair(self): d = iter(util.IntervalDifferential([5, 7], 10)) for i in range(100): self.assertEqual(d.next(), (5, 0)) self.assertEqual(d.next(), (2, 1)) self.assertEqual(d.next(), (3, 0)) self.assertEqual(d.next(), (4, 1)) self.assertEqual(d.next(), (1, 0)) self.assertEqual(d.next(), (5, 0)) self.assertEqual(d.next(), (1, 1)) self.assertEqual(d.next(), (4, 0)) self.assertEqual(d.next(), (3, 1)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (5, 0)) self.assertEqual(d.next(), (0, 1)) def testTriple(self): d = iter(util.IntervalDifferential([2, 4, 5], 10)) for i in range(100): self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (0, 1)) self.assertEqual(d.next(), (1, 2)) self.assertEqual(d.next(), (1, 0)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (0, 1)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (0, 2)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (0, 1)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (1, 2)) self.assertEqual(d.next(), (1, 0)) self.assertEqual(d.next(), (0, 1)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (2, 0)) self.assertEqual(d.next(), (0, 1)) self.assertEqual(d.next(), (0, 2)) def testInsert(self): d = iter(util.IntervalDifferential([], 10)) self.assertEqual(d.next(), (10, None)) d.addInterval(3) self.assertEqual(d.next(), (3, 0)) self.assertEqual(d.next(), (3, 0)) d.addInterval(6) self.assertEqual(d.next(), (3, 0)) self.assertEqual(d.next(), (3, 0)) self.assertEqual(d.next(), (0, 1)) self.assertEqual(d.next(), (3, 0)) self.assertEqual(d.next(), (3, 0)) self.assertEqual(d.next(), (0, 1)) def testRemove(self): d = iter(util.IntervalDifferential([3, 5], 10)) self.assertEqual(d.next(), (3, 0)) self.assertEqual(d.next(), (2, 1)) self.assertEqual(d.next(), (1, 0)) d.removeInterval(3) self.assertEqual(d.next(), (4, 0)) self.assertEqual(d.next(), (5, 0)) d.removeInterval(5) self.assertEqual(d.next(), (10, None)) self.assertRaises(ValueError, d.removeInterval, 10) class Record(util.FancyEqMixin): """ Trivial user of L{FancyEqMixin} used by tests. """ compareAttributes = ('a', 'b') def __init__(self, a, b): self.a = a self.b = b class DifferentRecord(util.FancyEqMixin): """ Trivial user of L{FancyEqMixin} which is not related to L{Record}. """ compareAttributes = ('a', 'b') def __init__(self, a, b): self.a = a self.b = b class DerivedRecord(Record): """ A class with an inheritance relationship to L{Record}. """ class EqualToEverything(object): """ A class the instances of which consider themselves equal to everything. """ def __eq__(self, other): return True def __ne__(self, other): return False class EqualToNothing(object): """ A class the instances of which consider themselves equal to nothing. """ def __eq__(self, other): return False def __ne__(self, other): return True class EqualityTests(unittest.TestCase): """ Tests for L{FancyEqMixin}. """ def test_identity(self): """ Instances of a class which mixes in L{FancyEqMixin} but which defines no comparison attributes compare by identity. """ class Empty(util.FancyEqMixin): pass self.assertFalse(Empty() == Empty()) self.assertTrue(Empty() != Empty()) empty = Empty() self.assertTrue(empty == empty) self.assertFalse(empty != empty) def test_equality(self): """ Instances of a class which mixes in L{FancyEqMixin} should compare equal if all of their attributes compare equal. They should not compare equal if any of their attributes do not compare equal. """ self.assertTrue(Record(1, 2) == Record(1, 2)) self.assertFalse(Record(1, 2) == Record(1, 3)) self.assertFalse(Record(1, 2) == Record(2, 2)) self.assertFalse(Record(1, 2) == Record(3, 4)) def test_unequality(self): """ Unequality between instances of a particular L{record} should be defined as the negation of equality. """ self.assertFalse(Record(1, 2) != Record(1, 2)) self.assertTrue(Record(1, 2) != Record(1, 3)) self.assertTrue(Record(1, 2) != Record(2, 2)) self.assertTrue(Record(1, 2) != Record(3, 4)) def test_differentClassesEquality(self): """ Instances of different classes which mix in L{FancyEqMixin} should not compare equal. """ self.assertFalse(Record(1, 2) == DifferentRecord(1, 2)) def test_differentClassesInequality(self): """ Instances of different classes which mix in L{FancyEqMixin} should compare unequal. """ self.assertTrue(Record(1, 2) != DifferentRecord(1, 2)) def test_inheritedClassesEquality(self): """ An instance of a class which derives from a class which mixes in L{FancyEqMixin} should compare equal to an instance of the base class if and only if all of their attributes compare equal. """ self.assertTrue(Record(1, 2) == DerivedRecord(1, 2)) self.assertFalse(Record(1, 2) == DerivedRecord(1, 3)) self.assertFalse(Record(1, 2) == DerivedRecord(2, 2)) self.assertFalse(Record(1, 2) == DerivedRecord(3, 4)) def test_inheritedClassesInequality(self): """ An instance of a class which derives from a class which mixes in L{FancyEqMixin} should compare unequal to an instance of the base class if any of their attributes compare unequal. """ self.assertFalse(Record(1, 2) != DerivedRecord(1, 2)) self.assertTrue(Record(1, 2) != DerivedRecord(1, 3)) self.assertTrue(Record(1, 2) != DerivedRecord(2, 2)) self.assertTrue(Record(1, 2) != DerivedRecord(3, 4)) def test_rightHandArgumentImplementsEquality(self): """ The right-hand argument to the equality operator is given a chance to determine the result of the operation if it is of a type unrelated to the L{FancyEqMixin}-based instance on the left-hand side. """ self.assertTrue(Record(1, 2) == EqualToEverything()) self.assertFalse(Record(1, 2) == EqualToNothing()) def test_rightHandArgumentImplementsUnequality(self): """ The right-hand argument to the non-equality operator is given a chance to determine the result of the operation if it is of a type unrelated to the L{FancyEqMixin}-based instance on the left-hand side. """ self.assertFalse(Record(1, 2) != EqualToEverything()) self.assertTrue(Record(1, 2) != EqualToNothing()) class RunAsEffectiveUserTests(unittest.TestCase): """ Test for the L{util.runAsEffectiveUser} function. """ if getattr(os, "geteuid", None) is None: skip = "geteuid/seteuid not available" def setUp(self): self.mockos = MockOS() self.patch(os, "geteuid", self.mockos.geteuid) self.patch(os, "getegid", self.mockos.getegid) self.patch(os, "seteuid", self.mockos.seteuid) self.patch(os, "setegid", self.mockos.setegid) def _securedFunction(self, startUID, startGID, wantUID, wantGID): """ Check if wanted UID/GID matched start or saved ones. """ self.assertTrue(wantUID == startUID or wantUID == self.mockos.seteuidCalls[-1]) self.assertTrue(wantGID == startGID or wantGID == self.mockos.setegidCalls[-1]) def test_forwardResult(self): """ L{util.runAsEffectiveUser} forwards the result obtained by calling the given function """ result = util.runAsEffectiveUser(0, 0, lambda: 1) self.assertEqual(result, 1) def test_takeParameters(self): """ L{util.runAsEffectiveUser} pass the given parameters to the given function. """ result = util.runAsEffectiveUser(0, 0, lambda x: 2*x, 3) self.assertEqual(result, 6) def test_takesKeyworkArguments(self): """ L{util.runAsEffectiveUser} pass the keyword parameters to the given function. """ result = util.runAsEffectiveUser(0, 0, lambda x, y=1, z=1: x*y*z, 2, z=3) self.assertEqual(result, 6) def _testUIDGIDSwitch(self, startUID, startGID, wantUID, wantGID, expectedUIDSwitches, expectedGIDSwitches): """ Helper method checking the calls to C{os.seteuid} and C{os.setegid} made by L{util.runAsEffectiveUser}, when switching from startUID to wantUID and from startGID to wantGID. """ self.mockos.euid = startUID self.mockos.egid = startGID util.runAsEffectiveUser( wantUID, wantGID, self._securedFunction, startUID, startGID, wantUID, wantGID) self.assertEqual(self.mockos.seteuidCalls, expectedUIDSwitches) self.assertEqual(self.mockos.setegidCalls, expectedGIDSwitches) self.mockos.seteuidCalls = [] self.mockos.setegidCalls = [] def test_root(self): """ Check UID/GID switches when current effective UID is root. """ self._testUIDGIDSwitch(0, 0, 0, 0, [], []) self._testUIDGIDSwitch(0, 0, 1, 0, [1, 0], []) self._testUIDGIDSwitch(0, 0, 0, 1, [], [1, 0]) self._testUIDGIDSwitch(0, 0, 1, 1, [1, 0], [1, 0]) def test_UID(self): """ Check UID/GID switches when current effective UID is non-root. """ self._testUIDGIDSwitch(1, 0, 0, 0, [0, 1], []) self._testUIDGIDSwitch(1, 0, 1, 0, [], []) self._testUIDGIDSwitch(1, 0, 1, 1, [0, 1, 0, 1], [1, 0]) self._testUIDGIDSwitch(1, 0, 2, 1, [0, 2, 0, 1], [1, 0]) def test_GID(self): """ Check UID/GID switches when current effective GID is non-root. """ self._testUIDGIDSwitch(0, 1, 0, 0, [], [0, 1]) self._testUIDGIDSwitch(0, 1, 0, 1, [], []) self._testUIDGIDSwitch(0, 1, 1, 1, [1, 0], []) self._testUIDGIDSwitch(0, 1, 1, 2, [1, 0], [2, 1]) def test_UIDGID(self): """ Check UID/GID switches when current effective UID/GID is non-root. """ self._testUIDGIDSwitch(1, 1, 0, 0, [0, 1], [0, 1]) self._testUIDGIDSwitch(1, 1, 0, 1, [0, 1], []) self._testUIDGIDSwitch(1, 1, 1, 0, [0, 1, 0, 1], [0, 1]) self._testUIDGIDSwitch(1, 1, 1, 1, [], []) self._testUIDGIDSwitch(1, 1, 2, 1, [0, 2, 0, 1], []) self._testUIDGIDSwitch(1, 1, 1, 2, [0, 1, 0, 1], [2, 1]) self._testUIDGIDSwitch(1, 1, 2, 2, [0, 2, 0, 1], [2, 1]) def _getDeprecationSuppression(f): """ Returns a tuple of arguments needed to suppress deprecation warnings from a specified function. @param f: function to suppress dperecation warnings for @type f: L{callable} @return: tuple to add to C{suppress} attribute """ return SUPPRESS( category=DeprecationWarning, message='%s was deprecated' % (fullyQualifiedName(f),)) class UnsignedIDTests(unittest.TestCase): """ Tests for L{util.unsignedID} and L{util.setIDFunction}. """ suppress = [ _getDeprecationSuppression(util.unsignedID), _getDeprecationSuppression(util.setIDFunction), ] def setUp(self): """ Save the value of L{util._idFunction} and arrange for it to be restored after the test runs. """ self.addCleanup(setattr, util, '_idFunction', util._idFunction) def test_setIDFunction(self): """ L{util.setIDFunction} returns the last value passed to it. """ value = object() previous = util.setIDFunction(value) result = util.setIDFunction(previous) self.assertIdentical(value, result) def test_setIDFunctionDeprecated(self): """ L{util.setIDFunction} is deprecated. """ self.callDeprecated( (Version("Twisted", 13, 0, 0)), util.setIDFunction, UnsignedIDTests) test_setIDFunctionDeprecated.suppress = [] def test_unsignedID(self): """ L{util.unsignedID} uses the function passed to L{util.setIDFunction} to determine the unique integer id of an object and then adjusts it to be positive if necessary. """ foo = object() bar = object() # A fake object identity mapping objects = {foo: 17, bar: -73} def fakeId(obj): return objects[obj] util.setIDFunction(fakeId) self.assertEqual(util.unsignedID(foo), 17) self.assertEqual(util.unsignedID(bar), (sys.maxsize + 1) * 2 - 73) def test_unsignedIDDeprecated(self): """ L{util.unsignedID} is deprecated, use the builtin C{id} instead. """ self.callDeprecated( (Version("Twisted", 13, 0, 0), "builtin id"), util.unsignedID, UnsignedIDTests) test_unsignedIDDeprecated.suppress = [] def test_defaultIDFunction(self): """ L{util.unsignedID} uses the built in L{id} by default. """ obj = object() idValue = id(obj) if idValue < 0: idValue += (sys.maxsize + 1) * 2 self.assertEqual(util.unsignedID(obj), idValue) class InitGroupsTests(unittest.TestCase): """ Tests for L{util.initgroups}. """ if pwd is None: skip = "pwd not available" def setUp(self): self.addCleanup(setattr, util, "_c_initgroups", util._c_initgroups) self.addCleanup(setattr, util, "setgroups", util.setgroups) def test_initgroupsForceC(self): """ If we fake the presence of the C extension, it's called instead of the Python implementation. """ calls = [] util._c_initgroups = lambda x, y: calls.append((x, y)) setgroupsCalls = [] util.setgroups = calls.append util.initgroups(os.getuid(), 4) self.assertEqual(calls, [(pwd.getpwuid(os.getuid())[0], 4)]) self.assertFalse(setgroupsCalls) def test_initgroupsForcePython(self): """ If we fake the absence of the C extension, the Python implementation is called instead, calling C{os.setgroups}. """ util._c_initgroups = None calls = [] util.setgroups = calls.append util.initgroups(os.getuid(), os.getgid()) # Something should be in the calls, we don't really care what self.assertTrue(calls) def test_initgroupsInC(self): """ If the C extension is present, it's called instead of the Python version. We check that by making sure C{os.setgroups} is not called. """ calls = [] util.setgroups = calls.append try: util.initgroups(os.getuid(), os.getgid()) except OSError: pass self.assertFalse(calls) if util._c_initgroups is None: test_initgroupsInC.skip = "C initgroups not available" class DeprecationTests(unittest.TestCase): """ Tests for deprecations in C{twisted.python.util}. """ def test_getPluginDirs(self): """ L{util.getPluginDirs} is deprecated. """ util.getPluginDirs() warnings = self.flushWarnings(offendingFunctions=[ self.test_getPluginDirs]) self.assertEqual( warnings[0]['message'], "twisted.python.util.getPluginDirs is deprecated since Twisted " "12.2.") self.assertEqual(warnings[0]['category'], DeprecationWarning) self.assertEqual(len(warnings), 1) def test_addPluginDir(self): """ L{util.addPluginDir} is deprecated. """ util.addPluginDir() warnings = self.flushWarnings(offendingFunctions=[ self.test_addPluginDir]) self.assertEqual( warnings[0]['message'], "twisted.python.util.addPluginDir is deprecated since Twisted " "12.2.") self.assertEqual(warnings[0]['category'], DeprecationWarning) self.assertEqual(len(warnings), 1) test_addPluginDir.suppress = [ SUPPRESS(category=DeprecationWarning, message="twisted.python.util.getPluginDirs is deprecated") ] class SuppressedWarningsTests(unittest.TestCase): """ Tests for L{util.runWithWarningsSuppressed}. """ runWithWarningsSuppressed = staticmethod(util.runWithWarningsSuppressed) def test_runWithWarningsSuppressedFiltered(self): """ Warnings from the function called by C{runWithWarningsSuppressed} are suppressed if they match the passed in filter. """ filters = [(("ignore", ".*foo.*"), {}), (("ignore", ".*bar.*"), {})] self.runWithWarningsSuppressed(filters, warnings.warn, "ignore foo") self.runWithWarningsSuppressed(filters, warnings.warn, "ignore bar") self.assertEqual([], self.flushWarnings()) def test_runWithWarningsSuppressedUnfiltered(self): """ Warnings from the function called by C{runWithWarningsSuppressed} are not suppressed if they do not match the passed in filter. """ filters = [(("ignore", ".*foo.*"), {}), (("ignore", ".*bar.*"), {})] self.runWithWarningsSuppressed(filters, warnings.warn, "don't ignore") self.assertEqual( ["don't ignore"], [w['message'] for w in self.flushWarnings()]) def test_passThrough(self): """ C{runWithWarningsSuppressed} returns the result of the function it called. """ self.assertEqual(self.runWithWarningsSuppressed([], lambda: 4), 4) def test_noSideEffects(self): """ Once C{runWithWarningsSuppressed} has returned, it no longer suppresses warnings. """ filters = [(("ignore", ".*foo.*"), {}), (("ignore", ".*bar.*"), {})] self.runWithWarningsSuppressed(filters, lambda: None) warnings.warn("ignore foo") self.assertEqual( ["ignore foo"], [w['message'] for w in self.flushWarnings()]) class FancyStrMixinTests(unittest.TestCase): """ Tests for L{util.FancyStrMixin}. """ def test_sequenceOfStrings(self): """ If C{showAttributes} is set to a sequence of strings, C{__str__} renders using those by looking them up as attributes on the object. """ class Foo(util.FancyStrMixin): showAttributes = ("first", "second") first = 1 second = "hello" self.assertEqual(str(Foo()), "") def test_formatter(self): """ If C{showAttributes} has an item that is a 2-tuple, C{__str__} renders the first item in the tuple as a key and the result of calling the second item with the value of the attribute named by the first item as the value. """ class Foo(util.FancyStrMixin): showAttributes = ( "first", ("second", lambda value: repr(value[::-1]))) first = "hello" second = "world" self.assertEqual("", str(Foo())) def test_override(self): """ If C{showAttributes} has an item that is a 3-tuple, C{__str__} renders the second item in the tuple as a key, and the contents of the attribute named in the first item are rendered as the value. The value is formatted using the third item in the tuple. """ class Foo(util.FancyStrMixin): showAttributes = ("first", ("second", "2nd", "%.1f")) first = 1 second = 2.111 self.assertEqual(str(Foo()), "") def test_fancybasename(self): """ If C{fancybasename} is present, C{__str__} uses it instead of the class name. """ class Foo(util.FancyStrMixin): fancybasename = "Bar" self.assertEqual(str(Foo()), "") def test_repr(self): """ C{__repr__} outputs the same content as C{__str__}. """ class Foo(util.FancyStrMixin): showAttributes = ("first", "second") first = 1 second = "hello" obj = Foo() self.assertEqual(str(obj), repr(obj)) if _PY3: del (SwitchUIDTest, SearchUpwardsTest, RunAsEffectiveUserTests, OrderedDictTest, IntervalDifferentialTestCase, UtilTestCase, TestMergeFunctionMetadata, DeprecationTests, InitGroupsTests, GetPasswordTest, )