util/jsoncomm: support PathLike

Add support for `util.types.PathLike` paths for socket addresses,
instead of just plain strings. Test it by using pathlib.Path to
create the address in the corresponding test.
This commit is contained in:
Christian Kellner 2020-07-24 19:33:36 +02:00 committed by Tom Gundersen
parent 0aa44c23bb
commit 28947f3bae
2 changed files with 7 additions and 5 deletions

View file

@ -14,6 +14,7 @@ import os
import socket
from typing import Any
from typing import Optional
from .types import PathLike
class FdSet():
@ -137,7 +138,7 @@ class Socket(contextlib.AbstractContextManager):
self._unlink = None
@classmethod
def new_client(cls, connect_to: Optional[str] = None):
def new_client(cls, connect_to: Optional[PathLike] = None):
"""Create Client
Create a new client socket.
@ -163,7 +164,7 @@ class Socket(contextlib.AbstractContextManager):
# Connect the socket. This has no effect other than specifying the
# default destination for send operations.
if connect_to is not None:
sock.connect(connect_to)
sock.connect(os.fspath(connect_to))
except:
if sock is not None:
sock.close()
@ -172,7 +173,7 @@ class Socket(contextlib.AbstractContextManager):
return cls(sock, None)
@classmethod
def new_server(cls, bind_to: str):
def new_server(cls, bind_to: PathLike):
"""Create Server
Create a new listener socket.
@ -199,7 +200,7 @@ class Socket(contextlib.AbstractContextManager):
# as well. We do not guarantee atomicity, so you better make sure
# you do not rely on it.
sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
sock.bind(bind_to)
sock.bind(os.fspath(bind_to))
unlink = os.open(os.path.join(".", path[0]), os.O_CLOEXEC | os.O_PATH)
except:
if unlink is not None:

View file

@ -4,6 +4,7 @@
import asyncio
import os
import pathlib
import tempfile
import unittest
@ -13,7 +14,7 @@ from osbuild.util import jsoncomm
class TestUtilJsonComm(unittest.TestCase):
def setUp(self):
self.dir = tempfile.TemporaryDirectory()
self.address = os.path.join(self.dir.name, "listener")
self.address = pathlib.Path(self.dir.name, "listener")
self.server = jsoncomm.Socket.new_server(self.address)
self.client = jsoncomm.Socket.new_client(self.address)