Support connecting to existing pipes (and not touching them)

pipe_io_server/pipe_io_server.py

@@ -5,11 +5,18 @@ from .pipe_writer_server import PipeWriterServer
 
 class PipeIOServer(PipeReaderServer, PipeWriterServer):
     # pylint: disable=abstract-method
-    def __init__(self, in_pipe, out_pipe, permissions=DEFAULT_PERMISSIONS):
+    def __init__(
+            self,
+            in_pipe,
+            out_pipe,
+            permissions=DEFAULT_PERMISSIONS,
+            manage_pipes=True
+    ):
         super().__init__(
             in_pipe=in_pipe,
             out_pipe=out_pipe,
-            permissions=permissions
+            permissions=permissions,
+            manage_pipes=manage_pipes
         )
 
     def _io_threads(self):

pipe_io_server/pipe_reader_server.py

@@ -4,11 +4,19 @@ from .pipe_reader_thread import PipeReaderThread
 
 
 class PipeReaderServer(PipeIOThread):
-    def __init__(self, in_pipe, permissions=DEFAULT_PERMISSIONS, **kwargs):
+    def __init__(
+            self,
+            in_pipe,
+            permissions=DEFAULT_PERMISSIONS,
+            manage_pipes=True,
+            **kwargs
+    ):
         super().__init__(**kwargs)
-        self._in_pipe = in_pipe
-        Pipe(self.in_pipe).recreate(permissions)
         self._reader_thread = None
+        self._manage_pipes = manage_pipes
+        self._in_pipe = in_pipe
+        if self._manage_pipes:
+            Pipe(self.in_pipe).recreate(permissions)
 
     @property
     def in_pipe(self):
@@ -27,4 +35,5 @@ class PipeReaderServer(PipeIOThread):
 
     def stop(self):
         super().stop()
+        if self._manage_pipes:
             Pipe(self.in_pipe).remove()

pipe_io_server/pipe_writer_server.py

@@ -4,11 +4,19 @@ from .pipe_writer_thread import PipeWriterThread
 
 
 class PipeWriterServer(PipeIOThread):
-    def __init__(self, out_pipe, permissions=DEFAULT_PERMISSIONS, **kwargs):
+    def __init__(
+            self,
+            out_pipe,
+            permissions=DEFAULT_PERMISSIONS,
+            manage_pipes=True,
+            **kwargs
+    ):
         super().__init__(**kwargs)
-        self._out_pipe = out_pipe
-        Pipe(self.out_pipe).recreate(permissions)
         self._writer_thread = None
+        self._manage_pipes = manage_pipes
+        self._out_pipe = out_pipe
+        if self._manage_pipes:
+            Pipe(self.out_pipe).recreate(permissions)
 
     @property
     def out_pipe(self):
@@ -26,4 +34,5 @@ class PipeWriterServer(PipeIOThread):
 
     def stop(self):
         super().stop()
+        if self._manage_pipes:
             Pipe(self.out_pipe).remove()

pipe_io_server/test_pipe_io_server.py

@@ -64,10 +64,23 @@ class IOPipes:
         return self.out_pipe.readline().rstrip(b'\n')
 
 
-def test_run_creates_pipes(io_pipes):
-    for path in (io_pipes.in_pipe_path, io_pipes.out_pipe_path):
-        assert exists(path)
-        assert S_ISFIFO(stat(path).st_mode)
+def pipes_exist(*paths):
+    predicate = lambda path: exists(path) and S_ISFIFO(stat(path).st_mode)
+    return all(predicate(path) for path in paths)
+
+
+def test_manage_pipes():
+    pipe_io = PipeIOServer(*get_test_init_params(), manage_pipes=True)
+    assert pipes_exist(pipe_io.in_pipe, pipe_io.out_pipe)
+    pipe_io.stop()
+    assert not pipes_exist(pipe_io.in_pipe, pipe_io.out_pipe)
+
+
+def test_no_manage_pipes():
+    pipe_io = PipeIOServer(*get_test_init_params(), manage_pipes=False)
+    assert not pipes_exist(pipe_io.in_pipe, pipe_io.out_pipe)
+    pipe_io.stop()
+    assert not pipes_exist(pipe_io.in_pipe, pipe_io.out_pipe)
 
 
 def test_stop():