Browse Source

add docs to parsesockstr, and drop the _ to make it first class. this

allows us to reference it's docs from where it's used...
tags/v0.1.0
John-Mark Gurney 5 years ago
parent
commit
93a5211373
1 changed files with 32 additions and 23 deletions
  1. +32
    -23
      ntunnel.py

+ 32
- 23
ntunnel.py View File

@@ -81,6 +81,7 @@ def _makeunix(path):


return 'unix:%s' % path return 'unix:%s' % path


# Make sure any additions are reflected by tests in test_parsesockstr
_allowedparameters = { _allowedparameters = {
'unix': { 'unix': {
'path': str, 'path': str,
@@ -91,12 +92,33 @@ _allowedparameters = {
}, },
} }


def _parsesockstr(sockstr):
'''Parse a socket string to its parts. If there are no
kwargs (no = after the colon), a dictionary w/ a single
key of default will pass the string after the colon.
def parsesockstr(sockstr):
'''Parse a socket string to its parts.


default is a reserved keyword and MUST NOT be used.'''
The format of sockstr is: 'proto:param=value[,param2=value2]'.
If the proto has a default parameter, the value can be used
directly, like: 'proto:value'. This is only allowed when the
value can unambiguously be determined not to be a param. If
there needs to be an equals '=', then you MUST use the extended
version.

The characters that define 'param' must be all lower case ascii
characters and may contain an underscore. The first character
must not be an underscore.

Supported protocols:
unix:
Default parameter is path.
The path parameter specifies the path to the
unix domain socket. The path MUST start w/ a
slash if it is used as a default parameter.

tcp:
Default parameter is host[:port].
The host parameter specifies the host, and the
port parameter specifies the port of the
connection.
'''


proto, rem = sockstr.split(':', 1) proto, rem = sockstr.split(':', 1)


@@ -126,7 +148,7 @@ def _parsesockstr(sockstr):
async def connectsockstr(sockstr): async def connectsockstr(sockstr):
'''Wrapper for asyncio.open_*_connection.''' '''Wrapper for asyncio.open_*_connection.'''


proto, args = _parsesockstr(sockstr)
proto, args = parsesockstr(sockstr)


if proto == 'unix': if proto == 'unix':
fun = asyncio.open_unix_connection fun = asyncio.open_unix_connection
@@ -140,10 +162,7 @@ async def connectsockstr(sockstr):
async def listensockstr(sockstr, cb): async def listensockstr(sockstr, cb):
'''Wrapper for asyncio.start_x_server. '''Wrapper for asyncio.start_x_server.


The format of sockstr is: 'proto:param=value[,param2=value2]'.
If the proto has a default parameter, the value can be used
directly, like: 'proto:value'. This is only allowed when the
value can unambiguously be determined not to be a param.
For the format of sockstr, please see parsesockstr.


The cb parameter is passed to asyncio's start_server or related The cb parameter is passed to asyncio's start_server or related
calls. Per those docs, the cb parameter is calls or scheduled calls. Per those docs, the cb parameter is calls or scheduled
@@ -151,19 +170,9 @@ async def listensockstr(sockstr, cb):
with two arguments, the reader and writer streams. For more with two arguments, the reader and writer streams. For more
information, see: https://docs.python.org/3/library/asyncio-stream.html#asyncio.start_server information, see: https://docs.python.org/3/library/asyncio-stream.html#asyncio.start_server


The characters that define 'param' must be all lower case ascii
characters and may contain an underscore. The first character
must not be an underscore.

Supported protocols:
unix:
Default parameter is path.
The path parameter specifies the path to the
unix domain socket. The path MUST start w/ a
slash if it is used as a default parameter.
''' '''


proto, args = _parsesockstr(sockstr)
proto, args = parsesockstr(sockstr)


if proto == 'unix': if proto == 'unix':
fun = asyncio.start_unix_server fun = asyncio.start_unix_server
@@ -386,7 +395,7 @@ class Tests_misc(unittest.TestCase):
for i in badstrs: for i in badstrs:
with self.assertRaises(ValueError, with self.assertRaises(ValueError,
msg='Should have failed processing: %s' % repr(i)): msg='Should have failed processing: %s' % repr(i)):
_parsesockstr(i)
parsesockstr(i)


def test_parsesockstr(self): def test_parsesockstr(self):
results = { results = {
@@ -400,7 +409,7 @@ class Tests_misc(unittest.TestCase):
} }


for s, r in results.items(): for s, r in results.items():
self.assertEqual(_parsesockstr(s), r)
self.assertEqual(parsesockstr(s), r)


@async_test @async_test
async def test_listensockstr_bad(self): async def test_listensockstr_bad(self):


Loading…
Cancel
Save