manyi/venv/Lib/site-packages/twisted/internet/test/test_address.py

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

from __future__ import division, absolute_import

import re
import os
import socket

from twisted.trial import unittest
from twisted.internet.address import IPv4Address, UNIXAddress, IPv6Address
from twisted.internet.address import HostnameAddress
from twisted.python.compat import nativeString
from twisted.python.runtime import platform

if not platform._supportsSymlinks():
    symlinkSkip = "Platform does not support symlinks"
else:
    symlinkSkip = None

try:
    socket.AF_UNIX
except AttributeError:
    unixSkip = "Platform doesn't support UNIX sockets."
else:
    unixSkip = None


class AddressTestCaseMixin(object):
    def test_addressComparison(self):
        """
        Two different address instances, sharing the same properties are
        considered equal by C{==} and not considered not equal by C{!=}.

        Note: When applied via UNIXAddress class, this uses the same
        filename for both objects being compared.
        """
        self.assertTrue(self.buildAddress() == self.buildAddress())
        self.assertFalse(self.buildAddress() != self.buildAddress())


    def _stringRepresentation(self, stringFunction):
        """
        Verify that the string representation of an address object conforms to a
        simple pattern (the usual one for Python object reprs) and contains
        values which accurately reflect the attributes of the address.
        """
        addr = self.buildAddress()
        pattern = "".join([
           "^",
           "([^\(]+Address)", # class name,
           "\(",       # opening bracket,
           "([^)]+)",  # arguments,
           "\)",       # closing bracket,
           "$"
        ])
        stringValue = stringFunction(addr)
        m = re.match(pattern, stringValue)
        self.assertNotEqual(
            None, m,
            "%s does not match the standard __str__ pattern "
            "ClassName(arg1, arg2, etc)" % (stringValue,))
        self.assertEqual(addr.__class__.__name__, m.group(1))

        args = [x.strip() for x in m.group(2).split(",")]
        self.assertEqual(
            args,
            [argSpec[1] % (getattr(addr, argSpec[0]),)
             for argSpec in self.addressArgSpec])


    def test_str(self):
        """
        C{str} can be used to get a string representation of an address instance
        containing information about that address.
        """
        self._stringRepresentation(str)


    def test_repr(self):
        """
        C{repr} can be used to get a string representation of an address
        instance containing information about that address.
        """
        self._stringRepresentation(repr)


    def test_hash(self):
        """
        C{__hash__} can be used to get a hash of an address, allowing
        addresses to be used as keys in dictionaries, for instance.
        """
        addr = self.buildAddress()
        d = {addr: True}
        self.assertTrue(d[self.buildAddress()])


    def test_differentNamesComparison(self):
        """
        Check that comparison operators work correctly on address objects
        when a different name is passed in
        """
        self.assertFalse(self.buildAddress() == self.buildDifferentAddress())
        self.assertFalse(self.buildDifferentAddress() == self.buildAddress())

        self.assertTrue(self.buildAddress() != self.buildDifferentAddress())
        self.assertTrue(self.buildDifferentAddress() != self.buildAddress())


    def assertDeprecations(self, testMethod, message):
        """
        Assert that the a DeprecationWarning with the given message was
        emitted against the given method.
        """
        warnings = self.flushWarnings([testMethod])
        self.assertEqual(warnings[0]['category'], DeprecationWarning)
        self.assertEqual(warnings[0]['message'], message)
        self.assertEqual(len(warnings), 1)



class IPv4AddressTestCaseMixin(AddressTestCaseMixin):
    addressArgSpec = (("type", "%s"), ("host", "%r"), ("port", "%d"))



class HostnameAddressTests(unittest.TestCase, AddressTestCaseMixin):
    """
    Test case for L{HostnameAddress}.
    """
    addressArgSpec = (("hostname", "%s"), ("port", "%d"))

    def buildAddress(self):
        """
        Create an arbitrary new L{HostnameAddress} instance.

        @return: A L{HostnameAddress} instance.
        """
        return HostnameAddress(b"example.com", 0)


    def buildDifferentAddress(self):
        """
        Like L{buildAddress}, but with a different hostname.

        @return: A L{HostnameAddress} instance.
        """
        return HostnameAddress(b"example.net", 0)



class IPv4AddressTCPTests(unittest.SynchronousTestCase,
                          IPv4AddressTestCaseMixin):
    def buildAddress(self):
        """
        Create an arbitrary new L{IPv4Address} instance with a C{"TCP"}
        type.  A new instance is created for each call, but always for the
        same address.
        """
        return IPv4Address("TCP", "127.0.0.1", 0)


    def buildDifferentAddress(self):
        """
        Like L{buildAddress}, but with a different fixed address.
        """
        return IPv4Address("TCP", "127.0.0.2", 0)


    def test_bwHackDeprecation(self):
        """
        If a value is passed for the C{_bwHack} parameter to L{IPv4Address},
        a deprecation warning is emitted.
        """
        # Construct this for warning side-effects, disregard the actual object.
        IPv4Address("TCP", "127.0.0.3", 0, _bwHack="TCP")

        message = (
            "twisted.internet.address.IPv4Address._bwHack is deprecated "
            "since Twisted 11.0")
        return self.assertDeprecations(self.test_bwHackDeprecation, message)



class IPv4AddressUDPTests(unittest.SynchronousTestCase,
                          IPv4AddressTestCaseMixin):
    def buildAddress(self):
        """
        Create an arbitrary new L{IPv4Address} instance with a C{"UDP"}
        type.  A new instance is created for each call, but always for the
        same address.
        """
        return IPv4Address("UDP", "127.0.0.1", 0)


    def buildDifferentAddress(self):
        """
        Like L{buildAddress}, but with a different fixed address.
        """
        return IPv4Address("UDP", "127.0.0.2", 0)


    def test_bwHackDeprecation(self):
        """
        If a value is passed for the C{_bwHack} parameter to L{IPv4Address},
        a deprecation warning is emitted.
        """
        # Construct this for warning side-effects, disregard the actual object.
        IPv4Address("UDP", "127.0.0.3", 0, _bwHack="UDP")

        message = (
            "twisted.internet.address.IPv4Address._bwHack is deprecated "
            "since Twisted 11.0")
        return self.assertDeprecations(self.test_bwHackDeprecation, message)



class IPv6AddressTests(unittest.SynchronousTestCase, AddressTestCaseMixin):
    addressArgSpec = (("type", "%s"), ("host", "%r"), ("port", "%d"))

    def buildAddress(self):
        """
        Create an arbitrary new L{IPv6Address} instance with a C{"TCP"}
        type.  A new instance is created for each call, but always for the
        same address.
        """
        return IPv6Address("TCP", "::1", 0)


    def buildDifferentAddress(self):
        """
        Like L{buildAddress}, but with a different fixed address.
        """
        return IPv6Address("TCP", "::2", 0)



class UNIXAddressTests(unittest.SynchronousTestCase):
    skip = unixSkip
    addressArgSpec = (("name", "%r"),)

    def setUp(self):
        self._socketAddress = self.mktemp()
        self._otherAddress = self.mktemp()


    def buildAddress(self):
        """
        Create an arbitrary new L{UNIXAddress} instance.  A new instance is
        created for each call, but always for the same address.
        """
        return UNIXAddress(self._socketAddress)


    def buildDifferentAddress(self):
        """
        Like L{buildAddress}, but with a different fixed address.
        """
        return UNIXAddress(self._otherAddress)


    def test_repr(self):
        """
        The repr of L{UNIXAddress} returns with the filename that the
        L{UNIXAddress} is for.
        """
        self.assertEqual(repr(self.buildAddress()), "UNIXAddress('%s')" % (
            nativeString(self._socketAddress)))


    def test_comparisonOfLinkedFiles(self):
        """
        UNIXAddress objects compare as equal if they link to the same file.
        """
        linkName = self.mktemp()
        with open(self._socketAddress, 'w') as self.fd:
            os.symlink(os.path.abspath(self._socketAddress), linkName)
            self.assertEqual(UNIXAddress(self._socketAddress),
                             UNIXAddress(linkName))
            self.assertEqual(UNIXAddress(linkName),
                             UNIXAddress(self._socketAddress))
    if not unixSkip:
        test_comparisonOfLinkedFiles.skip = symlinkSkip


    def test_hashOfLinkedFiles(self):
        """
        UNIXAddress Objects that compare as equal have the same hash value.
        """
        linkName = self.mktemp()
        with open(self._socketAddress, 'w') as self.fd:
            os.symlink(os.path.abspath(self._socketAddress), linkName)
            self.assertEqual(hash(UNIXAddress(self._socketAddress)),
                            hash(UNIXAddress(linkName)))
    if not unixSkip:
        test_hashOfLinkedFiles.skip = symlinkSkip



class EmptyUNIXAddressTests(unittest.SynchronousTestCase,
                            AddressTestCaseMixin):
    """
    Tests for L{UNIXAddress} operations involving a L{None} address.
    """
    skip = unixSkip
    addressArgSpec = (("name", "%r"),)

    def setUp(self):
        self._socketAddress = self.mktemp()


    def buildAddress(self):
        """
        Create an arbitrary new L{UNIXAddress} instance.  A new instance is
        created for each call, but always for the same address. This builds it
        with a fixed address of L{None}.
        """
        return UNIXAddress(None)


    def buildDifferentAddress(self):
        """
        Like L{buildAddress}, but with a random temporary directory.
        """
        return UNIXAddress(self._socketAddress)


    def test_comparisonOfLinkedFiles(self):
        """
        A UNIXAddress referring to a L{None} address does not compare equal to a
        UNIXAddress referring to a symlink.
        """
        linkName = self.mktemp()
        with open(self._socketAddress, 'w') as self.fd:
            os.symlink(os.path.abspath(self._socketAddress), linkName)
            self.assertNotEqual(UNIXAddress(self._socketAddress),
                                UNIXAddress(None))
            self.assertNotEqual(UNIXAddress(None),
                                UNIXAddress(self._socketAddress))
    if not unixSkip:
        test_comparisonOfLinkedFiles.skip = symlinkSkip


    def test_emptyHash(self):
        """
        C{__hash__} can be used to get a hash of an address, even one referring
        to L{None} rather than a real path.
        """
        addr = self.buildAddress()
        d = {addr: True}
        self.assertTrue(d[self.buildAddress()])


    def test_bwHackDeprecation(self):
        """
        If a value is passed for the C{_bwHack} parameter to L{UNIXAddress},
        a deprecation warning is emitted.
        """
        # Construct this for warning side-effects, disregard the actual object.
        UNIXAddress(None, _bwHack='UNIX')

        message = (
            "twisted.internet.address.UNIXAddress._bwHack is deprecated "
            "since Twisted 11.0")
        return self.assertDeprecations(self.test_bwHackDeprecation, message)