| Server IP : 74.208.127.88 / Your IP : 216.73.216.157 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /proc/self/root/usr/lib/python3/dist-packages/twisted/protocols/haproxy/test/ |
Upload File : |
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for L{twisted.protocols.haproxy._parser}.
"""
from twisted.trial.unittest import SynchronousTestCase as TestCase
from twisted.test.proto_helpers import MemoryReactor
from twisted.internet.endpoints import (
_WrapperServerEndpoint, TCP4ServerEndpoint, TCP6ServerEndpoint,
UNIXServerEndpoint, serverFromString, _parse as parseEndpoint
)
from .._wrapper import HAProxyWrappingFactory
from .._parser import unparseEndpoint
class UnparseEndpointTests(TestCase):
"""
Tests to ensure that un-parsing an endpoint string round trips through
escaping properly.
"""
def check(self, input):
"""
Check that the input unparses into the output, raising an assertion
error if it doesn't.
@param input: an input in endpoint-string-description format. (To
ensure determinism, keyword arguments should be in alphabetical
order.)
@type input: native L{str}
"""
self.assertEqual(unparseEndpoint(*parseEndpoint(input)), input)
def test_basicUnparse(self):
"""
An individual word.
"""
self.check("word")
def test_multipleArguments(self):
"""
Multiple arguments.
"""
self.check("one:two")
def test_keywords(self):
"""
Keyword arguments.
"""
self.check("aleph=one:bet=two")
def test_colonInArgument(self):
"""
Escaped ":".
"""
self.check("hello\\:colon\\:world")
def test_colonInKeywordValue(self):
"""
Escaped ":" in keyword value.
"""
self.check("hello=\\:")
def test_colonInKeywordName(self):
"""
Escaped ":" in keyword name.
"""
self.check("\\:=hello")
class HAProxyServerParserTests(TestCase):
"""
Tests that the parser generates the correct endpoints.
"""
def onePrefix(self, description, expectedClass):
"""
Test the C{haproxy} enpdoint prefix against one sub-endpoint type.
@param description: A string endpoint description beginning with
C{haproxy}.
@type description: native L{str}
@param expectedClass: the expected sub-endpoint class given the
description.
@type expectedClass: L{type}
@return: the parsed endpoint
@rtype: L{IStreamServerEndpoint}
@raise twisted.trial.unittest.Failtest: if the parsed endpoint doesn't
match expectations.
"""
reactor = MemoryReactor()
endpoint = serverFromString(reactor, description)
self.assertIsInstance(endpoint, _WrapperServerEndpoint)
self.assertIsInstance(endpoint._wrappedEndpoint, expectedClass)
self.assertIs(endpoint._wrapperFactory, HAProxyWrappingFactory)
return endpoint
def test_tcp4(self):
"""
Test if the parser generates a wrapped TCP4 endpoint.
"""
self.onePrefix('haproxy:tcp:8080', TCP4ServerEndpoint)
def test_tcp6(self):
"""
Test if the parser generates a wrapped TCP6 endpoint.
"""
self.onePrefix('haproxy:tcp6:8080', TCP6ServerEndpoint)
def test_unix(self):
"""
Test if the parser generates a wrapped UNIX endpoint.
"""
self.onePrefix('haproxy:unix:address=/tmp/socket', UNIXServerEndpoint)