# Path: src/foolscap/util.py
# class AsyncAND(defer.Deferred):
#     def __init__(self, deferredList):
#     def _cbDeferred(self, result, succeeded):
# def get_local_ip_for(target='A.ROOT-SERVERS.NET'):
# def format_time(when, mode):
# def move_into_place(source, dest):
# def isSubstring(small, big):
# def allocate_tcp_port():
# def _make_socket():
# def ensure_tuple_str(input_tuple):
# def ensure_dict_binary(d):
# def ensure_dict_str(d):
# def ensure_dict_str_keys(d):
# FORMAT_TIME_MODES = ["short-local", "long-local", "utc", "epoch"]


import gc
import re
import sys
    from twisted.python import log
from twisted.python import log
from twisted.trial import unittest
from twisted.internet.main import CONNECTION_LOST, CONNECTION_DONE
from twisted.python.failure import Failure
from twisted.application import service
from foolscap.tokens import Violation
from foolscap.eventual import flushEventualQueue
from foolscap.test.common import HelperTarget, TargetMixin, ShouldFailMixin
from foolscap.test.common import RIMyTarget, Target, TargetWithoutInterfaces, \
     BrokenTarget, MakeTubsMixin
from foolscap.api import RemoteException, DeadReferenceError
from foolscap.call import CopiedFailure
from foolscap.logging import log as flog
from ..util import long_type
        # call through a manually-defined RemoteMethodSchema
        rr, target = self.setupTarget(Target(), True)
        d = rr.callRemote("add", 3, 4, _methodConstraint=RIMyTarget['add1'])
        d.addCallback(lambda res: self.assertEqual(res, 7))
        return d
    testCall4.timeout = 2

    def testChoiceOf(self):
        # this is a really small test case to check specific bugs. We
        # definitely need more here.

        # in bug (#13), the ChoiceOf constraint did not override the
        # checkToken() call to its children, which had the consequence of not
        # propagating the maxLength= attribute of the StringConstraint to the
        # children (using the default of 1000 bytes instead).
        rr, target = self.setupTarget(HelperTarget())
        d = rr.callRemote("choice1", 4)
        d.addCallback(lambda res: self.assertEqual(res, None))
        d.addCallback(lambda res: rr.callRemote("choice1", b"a"*2000))
        d.addCallback(lambda res: self.assertEqual(res, None))
        # False does not conform
        d.addCallback(lambda res:
                      self.shouldFail(Violation, "testChoiceOf", None,
                                      rr.callRemote, "choice1", False))
        return d

    def testMegaSchema(self):
        # try to exercise all our constraints at once
        rr, target = self.setupTarget(HelperTarget())
        t = (set([1, 2, 3]),

# Path: src/foolscap/util.py
# class AsyncAND(defer.Deferred):
#     def __init__(self, deferredList):
#     def _cbDeferred(self, result, succeeded):
# def get_local_ip_for(target='A.ROOT-SERVERS.NET'):
# def format_time(when, mode):
# def move_into_place(source, dest):
# def isSubstring(small, big):
# def allocate_tcp_port():
# def _make_socket():
# def ensure_tuple_str(input_tuple):
# def ensure_dict_binary(d):
# def ensure_dict_str(d):
# def ensure_dict_str_keys(d):
# FORMAT_TIME_MODES = ["short-local", "long-local", "utc", "epoch"]


# Path: src/foolscap/test/test_call.py
import gc
import re
import sys
    from twisted.python import log
from twisted.python import log
from twisted.trial import unittest
from twisted.internet.main import CONNECTION_LOST, CONNECTION_DONE
from twisted.python.failure import Failure
from twisted.application import service
from foolscap.tokens import Violation
from foolscap.eventual import flushEventualQueue
from foolscap.test.common import HelperTarget, TargetMixin, ShouldFailMixin
from foolscap.test.common import RIMyTarget, Target, TargetWithoutInterfaces, \
     BrokenTarget, MakeTubsMixin
from foolscap.api import RemoteException, DeadReferenceError
from foolscap.call import CopiedFailure
from foolscap.logging import log as flog
from ..util import long_type

        # call through a manually-defined RemoteMethodSchema
        rr, target = self.setupTarget(Target(), True)
        d = rr.callRemote("add", 3, 4, _methodConstraint=RIMyTarget['add1'])
        d.addCallback(lambda res: self.assertEqual(res, 7))
        return d
    testCall4.timeout = 2

    def testChoiceOf(self):
        # this is a really small test case to check specific bugs. We
        # definitely need more here.

        # in bug (#13), the ChoiceOf constraint did not override the
        # checkToken() call to its children, which had the consequence of not
        # propagating the maxLength= attribute of the StringConstraint to the
        # children (using the default of 1000 bytes instead).
        rr, target = self.setupTarget(HelperTarget())
        d = rr.callRemote("choice1", 4)
        d.addCallback(lambda res: self.assertEqual(res, None))
        d.addCallback(lambda res: rr.callRemote("choice1", b"a"*2000))
        d.addCallback(lambda res: self.assertEqual(res, None))
        # False does not conform
        d.addCallback(lambda res:
                      self.shouldFail(Violation, "testChoiceOf", None,
                                      rr.callRemote, "choice1", False))
        return d

    def testMegaSchema(self):
        # try to exercise all our constraints at once
        rr, target = self.setupTarget(HelperTarget())
        t = (set([1, 2, 3]),

             b"str", True, 12, long_type(12), 19.3, None,