Gecko [ theprintave.ca ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2024-11-28 06:59
__init__.py	475 B	-rw-r--r--	2022-02-07 13:12
cert.pem.no_trailing_newline	1.38 KB	-rw-r--r--	2022-02-07 13:12
crash_test_dummy.py	549 B	-rw-r--r--	2022-02-07 13:12
ignore_test_failure.py	30.72 KB	-rw-r--r--	2024-11-22 17:49
iosim.py	17.74 KB	-rw-r--r--	2022-02-07 13:12
key.pem.no_trailing_newline	1.67 KB	-rw-r--r--	2022-02-07 13:12
mock_win32process.py	1.27 KB	-rw-r--r--	2022-02-07 13:12
myrebuilder1.py	151 B	-rw-r--r--	2022-02-07 13:12
myrebuilder2.py	151 B	-rw-r--r--	2022-02-07 13:12
plugin_basic.py	901 B	-rw-r--r--	2022-02-07 13:12
plugin_extra1.py	392 B	-rw-r--r--	2022-02-07 13:12
plugin_extra2.py	550 B	-rw-r--r--	2022-02-07 13:12
process_cmdline.py	123 B	-rw-r--r--	2022-02-07 13:12
process_echoer.py	214 B	-rw-r--r--	2022-02-07 13:12
process_fds.py	984 B	-rw-r--r--	2022-02-07 13:12
process_getargv.py	234 B	-rw-r--r--	2022-02-07 13:12
process_getenv.py	268 B	-rw-r--r--	2022-02-07 13:12
process_linger.py	297 B	-rw-r--r--	2022-02-07 13:12
process_reader.py	178 B	-rw-r--r--	2022-02-07 13:12
process_signal.py	220 B	-rw-r--r--	2022-02-07 13:12
process_stdinreader.py	739 B	-rw-r--r--	2022-02-07 13:12
process_tester.py	779 B	-rw-r--r--	2022-02-07 13:12
process_tty.py	130 B	-rw-r--r--	2022-02-07 13:12
process_twisted.py	1.15 KB	-rw-r--r--	2022-02-07 13:12
proto_helpers.py	1.34 KB	-rw-r--r--	2022-02-07 13:12
reflect_helper_IE.py	60 B	-rw-r--r--	2022-02-07 13:12
reflect_helper_VE.py	81 B	-rw-r--r--	2022-02-07 13:12
reflect_helper_ZDE.py	48 B	-rw-r--r--	2022-02-07 13:12
server.pem	4.34 KB	-rw-r--r--	2022-02-07 13:12
ssl_helpers.py	1.61 KB	-rw-r--r--	2022-02-07 13:12
stdio_test_consumer.py	1.14 KB	-rw-r--r--	2022-02-07 13:12
stdio_test_halfclose.py	2 KB	-rw-r--r--	2022-02-07 13:12
stdio_test_hostpeer.py	1.06 KB	-rw-r--r--	2022-02-07 13:12
stdio_test_lastwrite.py	1.13 KB	-rw-r--r--	2022-02-07 13:12
stdio_test_loseconn.py	1.55 KB	-rw-r--r--	2022-02-07 13:12
stdio_test_producer.py	1.45 KB	-rw-r--r--	2022-02-07 13:12
stdio_test_write.py	902 B	-rw-r--r--	2022-02-07 13:12
stdio_test_writeseq.py	894 B	-rw-r--r--	2022-02-07 13:12
test_abstract.py	3.35 KB	-rw-r--r--	2022-02-07 13:12
test_adbapi.py	25.56 KB	-rw-r--r--	2022-02-07 13:12
test_amp.py	107.92 KB	-rw-r--r--	2022-02-07 13:12
test_application.py	33.34 KB	-rw-r--r--	2022-02-07 13:12
test_compat.py	17.75 KB	-rw-r--r--	2022-02-07 13:12
test_context.py	1.43 KB	-rw-r--r--	2022-02-07 13:12
test_cooperator.py	20.84 KB	-rw-r--r--	2022-02-07 13:12
test_defer.py	121.86 KB	-rw-r--r--	2022-02-07 13:12
test_defgen.py	10.38 KB	-rw-r--r--	2022-02-07 13:12
test_dict.py	1.49 KB	-rw-r--r--	2022-02-07 13:12
test_dirdbm.py	6.69 KB	-rw-r--r--	2022-02-07 13:12
test_error.py	8.72 KB	-rw-r--r--	2022-02-07 13:12
test_factories.py	4.46 KB	-rw-r--r--	2022-02-07 13:12
test_fdesc.py	7.28 KB	-rw-r--r--	2022-02-07 13:12
test_finger.py	1.85 KB	-rw-r--r--	2022-02-07 13:12
test_formmethod.py	3.87 KB	-rw-r--r--	2022-02-07 13:12
test_ftp.py	126.96 KB	-rw-r--r--	2022-02-07 13:12
test_ftp_options.py	2.63 KB	-rw-r--r--	2022-02-07 13:12
test_htb.py	3.08 KB	-rw-r--r--	2022-02-07 13:12
test_ident.py	6.56 KB	-rw-r--r--	2022-02-07 13:12
test_internet.py	45.38 KB	-rw-r--r--	2022-02-07 13:12
test_iosim.py	8.7 KB	-rw-r--r--	2022-02-07 13:12
test_iutils.py	13.31 KB	-rw-r--r--	2022-02-07 13:12
test_lockfile.py	14.98 KB	-rw-r--r--	2022-02-07 13:12
test_log.py	35.23 KB	-rw-r--r--	2022-02-07 13:12
test_logfile.py	17.47 KB	-rw-r--r--	2022-02-07 13:12
test_loopback.py	13.99 KB	-rw-r--r--	2022-02-07 13:12
test_main.py	2.12 KB	-rw-r--r--	2022-02-07 13:12
test_memcache.py	24.69 KB	-rw-r--r--	2022-02-07 13:12
test_modules.py	17.02 KB	-rw-r--r--	2024-11-22 17:49
test_monkey.py	5.39 KB	-rw-r--r--	2022-02-07 13:12
test_paths.py	69.31 KB	-rw-r--r--	2022-02-07 13:12
test_pcp.py	12.23 KB	-rw-r--r--	2022-02-07 13:12
test_persisted.py	13.09 KB	-rw-r--r--	2022-02-07 13:12
test_plugin.py	24.79 KB	-rw-r--r--	2022-02-07 13:12
test_policies.py	32.29 KB	-rw-r--r--	2022-02-07 13:12
test_postfix.py	4.32 KB	-rw-r--r--	2022-02-07 13:12
test_process.py	85.16 KB	-rw-r--r--	2022-02-07 13:12
test_protocols.py	7.16 KB	-rw-r--r--	2022-02-07 13:12
test_randbytes.py	3.19 KB	-rw-r--r--	2022-02-07 13:12
test_rebuild.py	7.15 KB	-rw-r--r--	2022-02-07 13:12
test_reflect.py	23.9 KB	-rw-r--r--	2022-02-07 13:12
test_roots.py	1.6 KB	-rw-r--r--	2022-02-07 13:12
test_shortcut.py	1.92 KB	-rw-r--r--	2022-02-07 13:12
test_sip.py	24.9 KB	-rw-r--r--	2022-02-07 13:12
test_sob.py	5.53 KB	-rw-r--r--	2022-02-07 13:12
test_socks.py	17.09 KB	-rw-r--r--	2022-02-07 13:12
test_ssl.py	22.75 KB	-rw-r--r--	2022-02-07 13:12
test_sslverify.py	113.89 KB	-rw-r--r--	2024-11-22 17:49
test_stateful.py	1.97 KB	-rw-r--r--	2022-02-07 13:12
test_stdio.py	12.43 KB	-rw-r--r--	2022-02-07 13:12
test_strerror.py	5.12 KB	-rw-r--r--	2022-02-07 13:12
test_strports.py	1.67 KB	-rw-r--r--	2022-02-07 13:12
test_task.py	45.26 KB	-rw-r--r--	2022-02-07 13:12
test_tcp.py	64.27 KB	-rw-r--r--	2022-02-07 13:12
test_tcp_internals.py	12.73 KB	-rw-r--r--	2022-02-07 13:12
test_text.py	6.22 KB	-rw-r--r--	2022-02-07 13:12
test_threadable.py	3.26 KB	-rw-r--r--	2022-02-07 13:12
test_threadpool.py	21.44 KB	-rw-r--r--	2022-02-07 13:12
test_threads.py	12.9 KB	-rw-r--r--	2022-02-07 13:12
test_tpfile.py	1.51 KB	-rw-r--r--	2022-02-07 13:12
test_twistd.py	72.04 KB	-rw-r--r--	2022-02-07 13:12
test_twisted.py	6.13 KB	-rw-r--r--	2022-02-07 13:12
test_udp.py	24.43 KB	-rw-r--r--	2022-02-07 13:12
test_unix.py	13.26 KB	-rw-r--r--	2022-02-07 13:12
test_usage.py	22.76 KB	-rw-r--r--	2022-02-07 13:12
testutils.py	5.06 KB	-rw-r--r--	2022-02-07 13:12

Rename

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

# System Imports
import copyreg
import io
import pickle
import sys

# Twisted Imports
from twisted.persisted import aot, crefutil, styles
from twisted.trial import unittest
from twisted.trial.unittest import TestCase

class VersionTests(TestCase):
    def test_nullVersionUpgrade(self):
        global NullVersioned

class NullVersioned:
            def __init__(self):
                self.ok = 0

pkcl = pickle.dumps(NullVersioned())

class NullVersioned(styles.Versioned):
            persistenceVersion = 1

def upgradeToVersion1(self):
                self.ok = 1

mnv = pickle.loads(pkcl)
        styles.doUpgrade()
        assert mnv.ok, "initial upgrade not run!"

def test_versionUpgrade(self):
        global MyVersioned

class MyVersioned(styles.Versioned):
            persistenceVersion = 2
            persistenceForgets = ["garbagedata"]
            v3 = 0
            v4 = 0

def __init__(self):
                self.somedata = "xxx"
                self.garbagedata = lambda q: "cant persist"

def upgradeToVersion3(self):
                self.v3 += 1

def upgradeToVersion4(self):
                self.v4 += 1

mv = MyVersioned()
        assert not (mv.v3 or mv.v4), "hasn't been upgraded yet"
        pickl = pickle.dumps(mv)
        MyVersioned.persistenceVersion = 4
        obj = pickle.loads(pickl)
        styles.doUpgrade()
        assert obj.v3, "didn't do version 3 upgrade"
        assert obj.v4, "didn't do version 4 upgrade"
        pickl = pickle.dumps(obj)
        obj = pickle.loads(pickl)
        styles.doUpgrade()
        assert obj.v3 == 1, "upgraded unnecessarily"
        assert obj.v4 == 1, "upgraded unnecessarily"

def test_nonIdentityHash(self):
        global ClassWithCustomHash

class ClassWithCustomHash(styles.Versioned):
            def __init__(self, unique, hash):
                self.unique = unique
                self.hash = hash

def __hash__(self):
                return self.hash

v1 = ClassWithCustomHash("v1", 0)
        v2 = ClassWithCustomHash("v2", 0)

pkl = pickle.dumps((v1, v2))
        del v1, v2
        ClassWithCustomHash.persistenceVersion = 1
        ClassWithCustomHash.upgradeToVersion1 = lambda self: setattr(
            self, "upgraded", True
        )
        v1, v2 = pickle.loads(pkl)
        styles.doUpgrade()
        self.assertEqual(v1.unique, "v1")
        self.assertEqual(v2.unique, "v2")
        self.assertTrue(v1.upgraded)
        self.assertTrue(v2.upgraded)

def test_upgradeDeserializesObjectsRequiringUpgrade(self):
        global ToyClassA, ToyClassB

class ToyClassA(styles.Versioned):
            pass

class ToyClassB(styles.Versioned):
            pass

x = ToyClassA()
        y = ToyClassB()
        pklA, pklB = pickle.dumps(x), pickle.dumps(y)
        del x, y
        ToyClassA.persistenceVersion = 1

def upgradeToVersion1(self):
            self.y = pickle.loads(pklB)
            styles.doUpgrade()

ToyClassA.upgradeToVersion1 = upgradeToVersion1
        ToyClassB.persistenceVersion = 1

def setUpgraded(self):
            setattr(self, "upgraded", True)

ToyClassB.upgradeToVersion1 = setUpgraded

x = pickle.loads(pklA)
        styles.doUpgrade()
        self.assertTrue(x.y.upgraded)

class VersionedSubClass(styles.Versioned):
    pass

class SecondVersionedSubClass(styles.Versioned):
    pass

class VersionedSubSubClass(VersionedSubClass):
    pass

class VersionedDiamondSubClass(VersionedSubSubClass, SecondVersionedSubClass):
    pass

class AybabtuTests(TestCase):
    """
    L{styles._aybabtu} gets all of classes in the inheritance hierarchy of its
    argument that are strictly between L{Versioned} and the class itself.
    """

def test_aybabtuStrictEmpty(self):
        """
        L{styles._aybabtu} of L{Versioned} itself is an empty list.
        """
        self.assertEqual(styles._aybabtu(styles.Versioned), [])

def test_aybabtuStrictSubclass(self):
        """
        There are no classes I{between} L{VersionedSubClass} and L{Versioned},
        so L{styles._aybabtu} returns an empty list.
        """
        self.assertEqual(styles._aybabtu(VersionedSubClass), [])

def test_aybabtuSubsubclass(self):
        """
        With a sub-sub-class of L{Versioned}, L{styles._aybabtu} returns a list
        containing the intervening subclass.
        """
        self.assertEqual(styles._aybabtu(VersionedSubSubClass), [VersionedSubClass])

def test_aybabtuStrict(self):
        """
        For a diamond-shaped inheritance graph, L{styles._aybabtu} returns a
        list containing I{both} intermediate subclasses.
        """
        self.assertEqual(
            styles._aybabtu(VersionedDiamondSubClass),
            [VersionedSubSubClass, VersionedSubClass, SecondVersionedSubClass],
        )

class MyEphemeral(styles.Ephemeral):
    def __init__(self, x):
        self.x = x

class EphemeralTests(TestCase):
    def test_ephemeral(self):
        o = MyEphemeral(3)
        self.assertEqual(o.__class__, MyEphemeral)
        self.assertEqual(o.x, 3)

pickl = pickle.dumps(o)
        o = pickle.loads(pickl)

self.assertEqual(o.__class__, styles.Ephemeral)
        self.assertFalse(hasattr(o, "x"))

class Pickleable:
    def __init__(self, x):
        self.x = x

def getX(self):
        return self.x

class NotPickleable:
    """
    A class that is not pickleable.
    """

def __reduce__(self):
        """
        Raise an exception instead of pickling.
        """
        raise TypeError("Not serializable.")

class CopyRegistered:
    """
    A class that is pickleable only because it is registered with the
    C{copyreg} module.
    """

def __init__(self):
        """
        Ensure that this object is normally not pickleable.
        """
        self.notPickleable = NotPickleable()

class CopyRegisteredLoaded:
    """
    L{CopyRegistered} after unserialization.
    """

def reduceCopyRegistered(cr):
    """
    Externally implement C{__reduce__} for L{CopyRegistered}.

@param cr: The L{CopyRegistered} instance.

@return: a 2-tuple of callable and argument list, in this case
        L{CopyRegisteredLoaded} and no arguments.
    """
    return CopyRegisteredLoaded, ()

copyreg.pickle(CopyRegistered, reduceCopyRegistered)

class A:
    """
    dummy class
    """

def amethod(self):
        pass

class B:
    """
    dummy class
    """

def bmethod(self):
        pass

def funktion():
    pass

class PicklingTests(TestCase):
    """Test pickling of extra object types."""

def test_module(self):
        pickl = pickle.dumps(styles)
        o = pickle.loads(pickl)
        self.assertEqual(o, styles)

def test_instanceMethod(self):
        obj = Pickleable(4)
        pickl = pickle.dumps(obj.getX)
        o = pickle.loads(pickl)
        self.assertEqual(o(), 4)
        self.assertEqual(type(o), type(obj.getX))

class StringIOTransitionTests(TestCase):
    """
    When pickling a cStringIO in Python 2, it should unpickle as a BytesIO or a
    StringIO in Python 3, depending on the type of its contents.
    """

def test_unpickleBytesIO(self):
        """
        A cStringIO pickled with bytes in it will yield an L{io.BytesIO} on
        python 3.
        """
        pickledStringIWithText = (
            b"ctwisted.persisted.styles\nunpickleStringI\np0\n"
            b"(S'test'\np1\nI0\ntp2\nRp3\n."
        )
        loaded = pickle.loads(pickledStringIWithText)
        self.assertIsInstance(loaded, io.StringIO)
        self.assertEqual(loaded.getvalue(), "test")

class EvilSourceror:
    def __init__(self, x):
        self.a = self
        self.a.b = self
        self.a.b.c = x

class NonDictState:
    def __getstate__(self):
        return self.state

def __setstate__(self, state):
        self.state = state

class AOTTests(TestCase):
    def test_simpleTypes(self):
        obj = (
            1,
            2.0,
            3j,
            True,
            slice(1, 2, 3),
            "hello",
            "world",
            sys.maxsize + 1,
            None,
            Ellipsis,
        )
        rtObj = aot.unjellyFromSource(aot.jellyToSource(obj))
        self.assertEqual(obj, rtObj)

def test_methodSelfIdentity(self):
        a = A()
        b = B()
        a.bmethod = b.bmethod
        b.a = a
        im_ = aot.unjellyFromSource(aot.jellyToSource(b)).a.bmethod
        self.assertEqual(aot._selfOfMethod(im_).__class__, aot._classOfMethod(im_))

def test_methodNotSelfIdentity(self):
        """
        If a class change after an instance has been created,
        L{aot.unjellyFromSource} shoud raise a C{TypeError} when trying to
        unjelly the instance.
        """
        a = A()
        b = B()
        a.bmethod = b.bmethod
        b.a = a
        savedbmethod = B.bmethod
        del B.bmethod
        try:
            self.assertRaises(TypeError, aot.unjellyFromSource, aot.jellyToSource(b))
        finally:
            B.bmethod = savedbmethod

def test_unsupportedType(self):
        """
        L{aot.jellyToSource} should raise a C{TypeError} when trying to jelly
        an unknown type without a C{__dict__} property or C{__getstate__}
        method.
        """

class UnknownType:
            @property
            def __dict__(self):
                raise AttributeError()

self.assertRaises(TypeError, aot.jellyToSource, UnknownType())

def test_basicIdentity(self):
        # Anyone wanting to make this datastructure more complex, and thus this
        # test more comprehensive, is welcome to do so.
        aj = aot.AOTJellier().jellyToAO
        d = {"hello": "world", "method": aj}
        l = [
            1,
            2,
            3,
            'he\tllo\n\n"x world!',
            "goodbye \n\t\u1010 world!",
            1,
            1.0,
            100 ** 100,
            unittest,
            aot.AOTJellier,
            d,
            funktion,
        ]
        t = tuple(l)
        l.append(l)
        l.append(t)
        l.append(t)
        uj = aot.unjellyFromSource(aot.jellyToSource([l, l]))
        assert uj[0] is uj[1]
        assert uj[1][0:5] == l[0:5]

def test_nonDictState(self):
        a = NonDictState()
        a.state = "meringue!"
        assert aot.unjellyFromSource(aot.jellyToSource(a)).state == a.state

def test_copyReg(self):
        """
        L{aot.jellyToSource} and L{aot.unjellyFromSource} honor functions
        registered in the pickle copy registry.
        """
        uj = aot.unjellyFromSource(aot.jellyToSource(CopyRegistered()))
        self.assertIsInstance(uj, CopyRegisteredLoaded)

def test_funkyReferences(self):
        o = EvilSourceror(EvilSourceror([]))
        j1 = aot.jellyToAOT(o)
        oj = aot.unjellyFromAOT(j1)

assert oj.a is oj
        assert oj.a.b is oj.b
        assert oj.c is not oj.c.c

def test_circularTuple(self):
        """
        L{aot.jellyToAOT} can persist circular references through tuples.
        """
        l = []
        t = (l, 4321)
        l.append(t)
        j1 = aot.jellyToAOT(l)
        oj = aot.unjellyFromAOT(j1)
        self.assertIsInstance(oj[0], tuple)
        self.assertIs(oj[0][0], oj)
        self.assertEqual(oj[0][1], 4321)

class CrefUtilTests(TestCase):
    """
    Tests for L{crefutil}.
    """

def test_dictUnknownKey(self):
        """
        L{crefutil._DictKeyAndValue} only support keys C{0} and C{1}.
        """
        d = crefutil._DictKeyAndValue({})
        self.assertRaises(RuntimeError, d.__setitem__, 2, 3)

def test_deferSetMultipleTimes(self):
        """
        L{crefutil._Defer} can be assigned a key only one time.
        """
        d = crefutil._Defer()
        d[0] = 1
        self.assertRaises(RuntimeError, d.__setitem__, 0, 1)

def test_containerWhereAllElementsAreKnown(self):
        """
        A L{crefutil._Container} where all of its elements are known at
        construction time is nonsensical and will result in errors in any call
        to addDependant.
        """
        container = crefutil._Container([1, 2, 3], list)
        self.assertRaises(AssertionError, container.addDependant, {}, "ignore-me")

def test_dontPutCircularReferencesInDictionaryKeys(self):
        """
        If a dictionary key contains a circular reference (which is probably a
        bad practice anyway) it will be resolved by a
        L{crefutil._DictKeyAndValue}, not by placing a L{crefutil.NotKnown}
        into a dictionary key.
        """
        self.assertRaises(
            AssertionError, dict().__setitem__, crefutil.NotKnown(), "value"
        )

def test_dontCallInstanceMethodsThatArentReady(self):
        """
        L{crefutil._InstanceMethod} raises L{AssertionError} to indicate it
        should not be called.  This should not be possible with any of its API
        clients, but is provided for helping to debug.
        """
        self.assertRaises(
            AssertionError,
            crefutil._InstanceMethod("no_name", crefutil.NotKnown(), type),
        )

testCases = [VersionTests, EphemeralTests, PicklingTests]