Several test-driven fixes

- Improved test suite exposed several issues with the handling of command line arguments which are now fixed - Fixed a race condition that would cause remote characters to be lost intermittently when running remote commands that finish immediately. - Added automated testing that actually spins up a random listener and initiator in a private Reticulum network and passes data between them, uncovering more issues which are now fixed. - Fixed (hopefully) an issue where `rnsh` doesn't know what version it is.
2024-10-01 01:15:37 -04:00 · 2023-02-14 15:42:55 -06:00 · 2023-02-14 15:42:55 -06:00 · 27664df0b3
commit 27664df0b3
parent d4cb31e220
12 changed files with 508 additions and 85 deletions
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@ -54,3 +54,6 @@ jobs:

      - name: Test with pytest
        run: poetry run pytest -m "not skip_ci" tests
+
+      - name: Vulnerability check
+        run: poetry run safety check
--- a/README.md
+++ b/README.md
@ -17,13 +17,35 @@ out.

 Anyway, there's a lot of room for improvement.

-## Alpha Software
+## Contents
+
+- [Alpha Disclaimer](#reminder--alpha-software)
+- [Recent Changes](#recent-changes)
+- [Quickstart](#quickstart)
+- [Options](#options)
+- [How it works](#how-it-works)
+- [Roadmap](#roadmap)
+- [Active TODO](#todo)
+
+### Reminder: Alpha Software
 These early versions will be buggy. There will sometimes be major
 breaking changes to the command line parameters between releases.
 There will sometimes be breaking changes in the protocol between
 releases. Use at your own peril!

 ## Recent Changes
+### v0.0.8
+- Improved test suite exposed several issues with the handling of
+command line arguments which are now fixed
+- Fixed a race condition that would cause remote characters to be 
+  lost intermittently when running remote commands that finish
+  immediately.
+- Added automated testing that actually spins up a random listener
+  and initiator in a private Reticulum network and passes data
+  between them, uncovering more issues which are now fixed.
+- Fixed (hopefully) an issue where `rnsh` doesn't know what
+  version it is.
+
 ### v0.0.7
 Added `-A` command line option. This listener option causes the
 remote command line to be appended to the arguments list of the
@ -200,7 +222,25 @@ The protocol is build on top of the Reticulum `Request` and
 - [X] ~~Pip package with command-line utility support~~
 - [X] ~~Publish to PyPI~~
 - [X] ~~Improve signal handling~~
- [ ] Protocol improvements (throughput!)
- [ ] Test on several *nixes
 - [X] ~~Make it scriptable (currently requires a tty)~~
- [ ] Documentation improvements
+- [X] ~~Protocol improvements (throughput!)~~
+- [X] ~~Documentation improvements~~
+- [ ] Test on several platforms
+- [ ] Fix issues that come up with testing
+- [ ] Fix issues with running `rnsh` in a binary pipeline, i.e. 
+  piping the output of `tar` over `rsh`.
+- [ ] Beta release
+- [ ] Test and fix more issues
+- [ ] V1.0
+- [ ] Enhancement Ideas
+  - [ ] `authorized_keys` mode similar to SSH
+  - [ ] Git over `rnsh` (git remote helper)
+  - [ ] Sliding window acknowledgements for improved throughput
+
+## Miscellaneous
+
+By piping into/out of `rnsh`, it should be possible to transfer
+files using the same method discussed in 
+[this article](https://cromwell-intl.com/open-source/tar-and-ssh.html).
+I tested it just now and it doesn't work right. There's probably some
+subtle garbling of the data at one end of the stream or the other.
--- a/pyproject.toml
+++ b/pyproject.toml
@ -1,6 +1,6 @@
 [tool.poetry]
 name = "rnsh"
-version = "0.0.7"
+version = "0.0.8"
 description = "Shell over Reticulum"
 authors = ["acehoss <acehoss@acehoss.net>"]
 license = "MIT"
@ -11,6 +11,7 @@ python = "^3.9"
 docopt = "^0.6.2"
 psutil = "^5.9.4"
 rns = "^0.4.8"
+tomli = "^2.0.1"

 [tool.poetry.scripts]
 rnsh = 'rnsh.rnsh:rnsh_cli'
@ -18,8 +19,8 @@ rnsh = 'rnsh.rnsh:rnsh_cli'
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.2.1"
 setuptools = "^67.2.0"
-black = "^23.1.0"
 pytest-asyncio = "^0.20.3"
+safety = "^2.3.5"

 [tool.pytest.ini_options]
 markers = [
--- a/rnsh/init.py
+++ b/rnsh/init.py
@ -19,3 +19,21 @@
 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
+
+import os
+module_abs_filename = os.path.abspath(__file__)
+module_dir = os.path.dirname(module_abs_filename)
+# print(os.path.dirname(module_dir))
+
+def _get_version():
+    try:
+        try:
+            import tomli
+            return tomli.load(open(os.path.join(os.path.dirname(module_dir), "pyproject.toml"), "rb"))["tool"]["poetry"]["version"]
+        except:
+            from importlib.metadata import version
+            return version(__package__)
+    except:
+        return "0.0.0"
+
+__version__ = _get_version()
--- a/rnsh/args.py
+++ b/rnsh/args.py
@ -1,6 +1,6 @@
 from typing import TypeVar
 import RNS
-import importlib.metadata
+import rnsh
 import docopt
 import sys

@ -64,13 +64,16 @@ class Args:
    def __init__(self, argv: [str]):
        global usage
        try:
-            argv, program_args = _split_array_at(argv, "--")
+            self.argv = argv
+            self.program_args = []
+            self.docopts_argv, self.program_args = _split_array_at(self.argv, "--")
            # need to add first arg after -- back onto argv for docopts, but only for listener
-            if len(program_args) > 0 and next(filter(lambda a: a == "-l" or a == "--listen", argv), None) is not None:
-                argv.append(program_args[0])
-                self.program_args = program_args[1:]
+            if next(filter(lambda a: a == "-l" or a == "--listen", self.docopts_argv), None) is not None \
+                    and len(self.program_args) > 0:
+                self.docopts_argv.append(self.program_args[0])
+                self.program_args = self.program_args[1:]
    
-            args = docopt.docopt(usage, argv=argv[1:], version=f"rnsh {importlib.metadata.version('rnsh')}")
+            args = docopt.docopt(usage, argv=self.docopts_argv[1:], version=f"rnsh {rnsh.__version__}")
            # json.dump(args, sys.stdout)
    
            self.service_name = args.get("--service", None) or "default"
@ -87,23 +90,23 @@ class Args:
                    self.announce = int(announce)
            except ValueError:
                print("Invalid value for --announce")
-                self.valid = False
+                sys.exit(1)
            self.no_auth = args.get("--no-auth", None) or False
            self.allowed = args.get("--allowed", None) or []
            self.remote_cmd_as_args = args.get("--remote-command-as-args", None) or False
            self.no_remote_cmd = args.get("--no-remote-command", None) or False
            self.program = args.get("<program>", None)
-            self.program_args = args.get("<arg>", None) or []
-            if self.program is not None:
-                self.program_args.insert(0, self.program)
-            self.program_args.extend(program_args)
+            if len(self.program_args) == 0:
+                self.program_args = args.get("<arg>", None) or []
            self.no_id = args.get("--no-id", None) or False
            self.mirror = args.get("--mirror", None) or False
            self.timeout = args.get("--timeout", None) or RNS.Transport.PATH_REQUEST_TIMEOUT
            self.destination = args.get("<destination_hash>", None)
            self.help = args.get("--help", None) or False
+            self.command_line = [self.program] if self.program else []
+            self.command_line.extend(self.program_args)
        except Exception as e:
-            print("Error parsing arguments: {e}")
+            print(f"Error parsing arguments: {e}")
            print()
            print(usage)
            sys.exit(1)
--- a/rnsh/process.py
+++ b/rnsh/process.py
@ -96,6 +96,8 @@ def tty_read(fd: int) -> bytes:
                        if data is not None and len(data) > 0:
                            result.extend(data)
                            return result
+                        elif len(result) > 0:
+                            return result
                        else:
                            raise EOFError
                    if data is not None and len(data) > 0:
@ -225,7 +227,7 @@ class TTYRestorer(contextlib.AbstractContextManager):
            return copy.deepcopy(termios.tcgetattr(self._fd))
        return None

-    def set_attr(self, attr: [any], when: int = termios.TCSANOW):
+    def set_attr(self, attr: [any], when: int = termios.TCSADRAIN):
        """
        Set termios attributes
        :param attr: attribute list to set
@ -349,6 +351,7 @@ class CallbackSubprocess:
        self._pid: int = None
        self._child_fd: int = None
        self._return_code: int = None
+        self._eof: bool = False

    def terminate(self, kill_delay: float = 1.0):
        """
@ -437,7 +440,7 @@ class CallbackSubprocess:
        return termios.tcgetattr(self._child_fd)

    def ttysetraw(self):
-        tty.setraw(self._child_fd, termios.TCSANOW)
+        tty.setraw(self._child_fd, termios.TCSADRAIN)

    def start(self):
        """
@ -458,6 +461,8 @@ class CallbackSubprocess:
            env[key] = self._env[key]

        program = self._command[0]
+        assert isinstance(program, str)
+
        # match = re.search("^/bin/(.*sh)$", program)
        # if match:
        #     self._command[0] = "-" + match.group(1)
@ -509,15 +514,24 @@ class CallbackSubprocess:
                    if data is not None and len(data) > 0:
                        callback(data)
            except EOFError:
+                self._eof = True
                tty_unset_reader_callbacks(self._child_fd)
-                callback(CTRL_D)
+                callback(bytearray())

        tty_add_reader_callback(self._child_fd, functools.partial(reader, self._child_fd, self._stdout_cb), self._loop)

+    @property
+    def eof(self):
+        return self._eof or not self.running
+
    @property
    def return_code(self) -> int:
        return self._return_code

+    @property
+    def pid(self) -> int:
+        return self._pid
+

 async def main():
    """
--- a/rnsh/rnsh.py
+++ b/rnsh/rnsh.py
@ -248,7 +248,7 @@ def _protocol_response_chars_take(link_mdu: int, version: int) -> int:

 def _protocol_request_chars_take(link_mdu: int, version: int, term: str, cmd: str) -> int:
    if version >= _PROTOCOL_VERSION_2:
-        return link_mdu - 15 * 8 - len(term) - len(cmd) - 20 # TODO: tune
+        return link_mdu - 15 * 8 - len(term) - len(cmd) - 20  # TODO: tune
    else:
        return link_mdu // 2

@ -417,7 +417,7 @@ class Session:
                    if first_term_state is not None:
                        # TODO: use a more specific error
                        with contextlib.suppress(Exception):
-                            self.process.tcsetattr(termios.TCSANOW, term_state[0])
+                            self.process.tcsetattr(termios.TCSADRAIN, term_state[0])
            if stdin is not None and len(stdin) > 0:
                if data[Session.REQUEST_IDX_VERS] < _PROTOCOL_VERSION_2:
                    stdin = base64.b64decode(stdin)
@ -542,7 +542,7 @@ def _subproc_terminated(link: RNS.Link, return_code: int):
    _loop.call_soon_threadsafe(cleanup)


-def _listen_start_proc(link: RNS.Link, remote_identity: str | None, term: str, cmd: str | None,
+def _listen_start_proc(link: RNS.Link, remote_identity: str | None, term: str, cmd: [str],
                       loop: asyncio.AbstractEventLoop) -> Session | None:
    log = _get_logger("_listen_start_proc")
    try:
@ -839,9 +839,6 @@ async def _initiate(configdir: str, identitypath: str, verbosity: int, quietness
    log = _get_logger("_initiate")
    loop = asyncio.get_running_loop()
    _new_data = asyncio.Event()
-    command = command
-    if command is not None and len(command) == 1:
-        command = shlex.split(command[0])

    data_buffer = bytearray(sys.stdin.buffer.read()) if not os.isatty(sys.stdin.fileno()) else bytearray()

@ -870,6 +867,7 @@ async def _initiate(configdir: str, identitypath: str, verbosity: int, quietness
    mdu = 64
    rtt = 5
    first_loop = True
+    cmdline = " ".join(command or [])
    while not await _check_finished():
        try:
            log.debug("top of client loop")
@ -895,7 +893,7 @@ async def _initiate(configdir: str, identitypath: str, verbosity: int, quietness
                mdu = _protocol_request_chars_take(_link.MDU,
                                                   _PROTOCOL_VERSION_DEFAULT,
                                                   os.environ.get("TERM", ""),
-                                                   " ".join(command))
+                                                   cmdline)
                _new_data.set()

            if _link:
@ -916,6 +914,7 @@ async def _initiate(configdir: str, identitypath: str, verbosity: int, quietness
            return 255

        await process.event_wait_any([_new_data, _finished], timeout=min(max(rtt * 50, 5), 120))
+    return 0


 def _loop_set_signal(sig, loop):
@ -942,7 +941,7 @@ async def _rnsh_cli_main():
    if args.listen:
        # log.info("command " + args.command)
        await _listen(configdir=args.config,
-                      command=args.program_args,
+                      command=args.command_line,
                      identitypath=args.identity,
                      service_name=args.service_name,
                      verbosity=args.verbose,
@ -964,7 +963,7 @@ async def _rnsh_cli_main():
            destination=args.destination,
            service_name=args.service_name,
            timeout=args.timeout,
-            command=args.program_args
+            command=args.command_line
        )
        return return_code if args.mirror else 0
    else:
@ -979,7 +978,7 @@ def rnsh_cli():
    with contextlib.suppress(Exception):
        if not os.isatty(sys.stdin.fileno()):
            time.sleep(0.1)  # attempting to deal with an issue with missing input
-            tty.setraw(sys.stdin.fileno(), termios.TCSANOW)
+            tty.setraw(sys.stdin.fileno(), termios.TCSADRAIN)

    with process.TTYRestorer(sys.stdin.fileno()) as _tr, retry.RetryThread() as _retry_timer:
        return_code = asyncio.run(_rnsh_cli_main())
--- a/tests/helpers.py
+++ b/tests/helpers.py
@ -0,0 +1,131 @@
+import logging
+import types
+import typing
+import tempfile
+
+import pytest
+
+import rnsh.rnsh
+import asyncio
+import rnsh.process
+import contextlib
+import threading
+import os
+import pathlib
+import tests
+import shutil
+import random
+
+module_logger = logging.getLogger(__name__)
+
+module_abs_filename = os.path.abspath(tests.__file__)
+module_dir = os.path.dirname(module_abs_filename)
+
+
+class SubprocessReader(contextlib.AbstractContextManager):
+    def __init__(self, argv: [str], env: dict = None, name: str = None):
+        self._log = module_logger.getChild(self.__class__.__name__ + ("" if name is None else f"({name})"))
+        self.name = name or "subproc"
+        self.process: rnsh.process.CallbackSubprocess
+        self.loop = asyncio.get_running_loop()
+        self.env = env or os.environ.copy()
+        self.argv = argv
+        self._lock = threading.RLock()
+        self._stdout = bytearray()
+        self.return_code: int = None
+        self.process = rnsh.process.CallbackSubprocess(argv=self.argv,
+                                                       env=self.env,
+                                                       loop=self.loop,
+                                                       stdout_callback=self._stdout_cb,
+                                                       terminated_callback=self._terminated_cb)
+
+    def _stdout_cb(self, data):
+        self._log.debug(f"_stdout_cb({data})")
+        with self._lock:
+            self._stdout.extend(data)
+
+    def read(self):
+        self._log.debug(f"read()")
+        with self._lock:
+            data = self._stdout.copy()
+            self._stdout.clear()
+        self._log.debug(f"read() returns {data}")
+        return data
+
+    def _terminated_cb(self, rc):
+        self._log.debug(f"_terminated_cb({rc})")
+        self.return_code = rc
+
+    def start(self):
+        self._log.debug(f"start()")
+        self.process.start()
+
+    def cleanup(self):
+        self._log.debug(f"cleanup()")
+        if self.process and self.process.running:
+            self.process.terminate(kill_delay=0.1)
+
+    def __exit__(self, __exc_type: typing.Type[BaseException], __exc_value: BaseException,
+                 __traceback: types.TracebackType) -> bool:
+        self._log.debug(f"__exit__({__exc_type}, {__exc_value}, {__traceback})")
+        self.cleanup()
+        return False
+
+
+def replace_text_in_file(filename: str, text: str, replacement: str):
+    # Read in the file
+    with open(filename, 'r') as file:
+        filedata = file.read()
+
+    # Replace the target string
+    filedata = filedata.replace(text, replacement)
+
+    # Write the file out again
+    with open(filename, 'w') as file:
+        file.write(filedata)
+
+
+class tempdir(object):
+    """Sets the cwd within the context
+
+    Args:
+        path (Path): The path to the cwd
+    """
+    def __init__(self, cd: bool = False):
+        self.cd = cd
+        self.tempdir = tempfile.TemporaryDirectory()
+        self.path = self.tempdir.name
+        self.origin = pathlib.Path().absolute()
+        self.configfile = os.path.join(self.path, "config")
+
+    def setup_files(self):
+        shutil.copy(os.path.join(module_dir, "reticulum_test_config"), self.configfile)
+        port1 = random.randint(30000, 65000)
+        port2 = port1 + 1
+        replace_text_in_file(self.configfile, "22222", str(port1))
+        replace_text_in_file(self.configfile, "22223", str(port2))
+
+
+    def __enter__(self):
+        self.setup_files()
+        if self.cd:
+            os.chdir(self.path)
+
+        return self.path
+
+    def __exit__(self, exc, value, tb):
+        if self.cd:
+            os.chdir(self.origin)
+        self.tempdir.__exit__(exc, value, tb)
+
+
+def test_config_and_cleanup():
+    td = None
+    with tests.helpers.tempdir() as td:
+        assert os.path.isfile(os.path.join(td, "config"))
+        with open(os.path.join(td, "config"), 'r') as file:
+            filedata = file.read()
+        assert filedata.index("acehoss test config") > 0
+        with pytest.raises(ValueError):
+            filedata.index("22222")
+    assert not os.path.exists(os.path.join(td, "config"))
--- a/tests/reticulum_test_config
+++ b/tests/reticulum_test_config
@ -0,0 +1,15 @@
+# acehoss test config
+[reticulum]
+  enable_transport = False
+  share_instance = Yes
+  shared_instance_port = 22222
+  instance_control_port = 22223
+  panic_on_interface_error = No
+
+[logging]
+  loglevel = 7
+
+[interfaces]
+  [[Default Interface]]
+    type = AutoInterface
+    enabled = Yes
--- a/tests/test_args.py
+++ b/tests/test_args.py
@ -0,0 +1,77 @@
+import rnsh.args
+import shlex
+import docopt
+
+def test_program_args():
+    docopt_threw = False
+    try:
+        args = rnsh.args.Args(shlex.split("rnsh -l -n one two three"))
+        assert args.listen
+        assert args.program == "one"
+        assert args.program_args == ["two", "three"]
+        assert args.command_line == ["one", "two", "three"]
+    except docopt.DocoptExit:
+        docopt_threw = True
+    assert not docopt_threw
+
+
+def test_program_args_dash():
+    docopt_threw = False
+    try:
+        args = rnsh.args.Args(shlex.split("rnsh -l -n -- one -l -C"))
+        assert args.listen
+        assert args.program == "one"
+        assert args.program_args == ["-l", "-C"]
+        assert args.command_line == ["one", "-l", "-C"]
+    except docopt.DocoptExit:
+        docopt_threw = True
+    assert not docopt_threw
+
+def test_program_initiate_no_args():
+    docopt_threw = False
+    try:
+        args = rnsh.args.Args(shlex.split("rnsh one"))
+        assert not args.listen
+        assert args.destination == "one"
+        assert args.command_line == []
+    except docopt.DocoptExit:
+        docopt_threw = True
+    assert not docopt_threw
+
+def test_program_initiate_dash_args():
+    docopt_threw = False
+    try:
+        args = rnsh.args.Args(shlex.split("rnsh --config ~/Projects/rnsh/testconfig -s test -vvvvvvv a5f72aefc2cb3cdba648f73f77c4e887 -- -l"))
+        assert not args.listen
+        assert args.config == "~/Projects/rnsh/testconfig"
+        assert args.service_name == "test"
+        assert args.verbose == 7
+        assert args.destination == "a5f72aefc2cb3cdba648f73f77c4e887"
+        assert args.command_line == ["-l"]
+    except docopt.DocoptExit:
+        docopt_threw = True
+    assert not docopt_threw
+
+
+def test_program_listen_config_print():
+    docopt_threw = False
+    try:
+        args = rnsh.args.Args(shlex.split("rnsh -l --config testconfig -p"))
+        assert args.listen
+        assert args.config == "testconfig"
+        assert args.print_identity
+        assert args.command_line == []
+    except docopt.DocoptExit:
+        docopt_threw = True
+    assert not docopt_threw
+
+
+def test_split_at():
+    a, b = rnsh.args._split_array_at(["one", "two", "three"], "two")
+    assert a == ["one"]
+    assert b == ["three"]
+
+def test_split_at_not_found():
+    a, b = rnsh.args._split_array_at(["one", "two", "three"], "four")
+    assert a == ["one", "two", "three"]
+    assert b == []
--- a/tests/test_process.py
+++ b/tests/test_process.py
@ -1,68 +1,20 @@
-import uuid
+import tests.helpers
 import time
 import pytest
 import rnsh.process
-import contextlib
 import asyncio
 import logging
-import os
-import threading
-import types
-import typing
 import multiprocessing.pool
 logging.getLogger().setLevel(logging.DEBUG)


-class State(contextlib.AbstractContextManager):
-    def __init__(self, argv: [str], loop: asyncio.AbstractEventLoop, env: dict = None):
-        self.process: rnsh.process.CallbackSubprocess
-        self.loop = loop
-        self.env = env or os.environ.copy()
-        self.argv = argv
-        self._lock = threading.RLock()
-        self._stdout = bytearray()
-        self.return_code: int = None
-        self.process = rnsh.process.CallbackSubprocess(argv=self.argv,
-                                                       env=self.env,
-                                                       loop=self.loop,
-                                                       stdout_callback=self._stdout_cb,
-                                                       terminated_callback=self._terminated_cb)
-
-    def _stdout_cb(self, data):
-        with self._lock:
-            self._stdout.extend(data)
-
-    def read(self):
-        with self._lock:
-            data = self._stdout.copy()
-            self._stdout.clear()
-            return data
-
-    def _terminated_cb(self, rc):
-        self.return_code = rc
-
-    def start(self):
-        self.process.start()
-
-    def cleanup(self):
-        if self.process and self.process.running:
-            self.process.terminate(kill_delay=0.1)
-
-    def __exit__(self, __exc_type: typing.Type[BaseException], __exc_value: BaseException,
-                 __traceback: types.TracebackType) -> bool:
-        self.cleanup()
-        return False
-
-
@pytest.mark.skip_ci
@pytest.mark.asyncio
 async def test_echo():
    """
    Echoing some text through cat.
    """
-    loop = asyncio.get_running_loop()
-    with State(argv=["/bin/cat"],
-               loop=loop) as state:
+    with tests.helpers.SubprocessReader(argv=["/bin/cat"]) as state:
        state.start()
        assert state.process is not None
        assert state.process.running
@ -84,9 +36,7 @@ async def test_echo_live():
    """
    Check for immediate echo
    """
-    loop = asyncio.get_running_loop()
-    with State(argv=["/bin/cat"],
-               loop=loop) as state:
+    with tests.helpers.SubprocessReader(argv=["/bin/cat"]) as state:
        state.start()
        assert state.process is not None
        assert state.process.running
@ -101,6 +51,38 @@ async def test_echo_live():
        assert decoded == message
        assert not state.process.running

+@pytest.mark.skip_ci
+@pytest.mark.asyncio
+async def test_double_echo_live():
+    """
+    Check for immediate echo
+    """
+    with tests.helpers.SubprocessReader(name="state", argv=["/bin/cat"]) as state:
+        with tests.helpers.SubprocessReader(name="state2", argv=["/bin/cat"]) as state2:
+            state.start()
+            state2.start()
+            assert state.process is not None
+            assert state.process.running
+            assert state2.process is not None
+            assert state2.process.running
+            message = "t"
+            state.process.write(message.encode("utf-8"))
+            state2.process.write(message.encode("utf-8"))
+            await asyncio.sleep(0.1)
+            data = state.read()
+            data2 = state2.read()
+            state.process.write(rnsh.process.CTRL_C)
+            state2.process.write(rnsh.process.CTRL_C)
+            await asyncio.sleep(0.1)
+            assert len(data) > 0
+            assert len(data2) > 0
+            decoded = data.decode("utf-8")
+            decoded2 = data.decode("utf-8")
+            assert decoded == message
+            assert decoded2 == message
+            assert not state.process.running
+            assert not state2.process.running
+

@pytest.mark.asyncio
 async def test_event_wait_any():
--- a/tests/test_rnsh.py
+++ b/tests/test_rnsh.py
@ -1,7 +1,17 @@
 import logging
-import rnsh.rnsh
 logging.getLogger().setLevel(logging.DEBUG)

+import tests.helpers
+import rnsh.rnsh
+import rnsh.process
+import shlex
+import pytest
+import time
+import asyncio
+import re
+import os
+
+

 def test_check_magic():
    magic = rnsh.rnsh._PROTOCOL_VERSION_0
@ -13,3 +23,133 @@ def test_check_magic():
    magic = magic | 0x00ffff0000000000
    # make sure it fails now
    assert not rnsh.rnsh._protocol_check_magic(magic)
+
+
+def test_version():
+    # version = importlib.metadata.version(rnsh.__version__)
+    assert rnsh.__version__ != "0.0.0"
+    assert rnsh.__version__ != "0.0.1"
+
+
+@pytest.mark.skip_ci
+@pytest.mark.asyncio
+async def test_wrapper():
+    with tests.helpers.tempdir() as td:
+        with tests.helpers.SubprocessReader(argv=shlex.split(f"date")) as wrapper:
+            wrapper.start()
+            assert wrapper.process is not None
+            assert wrapper.process.running
+            await asyncio.sleep(1)
+            text = wrapper.read().decode("utf-8")
+            assert len(text) > 5
+            assert not wrapper.process.running
+
+
+
+@pytest.mark.skip_ci
+@pytest.mark.asyncio
+async def test_rnsh_listen_start_stop():
+    with tests.helpers.tempdir() as td:
+        with tests.helpers.SubprocessReader(argv=shlex.split(f"poetry run rnsh -l --config \"{td}\" -n -C -vvvvvv -- /bin/ls")) as wrapper:
+            wrapper.start()
+            await asyncio.sleep(0.1)
+            assert wrapper.process.running
+            # wait for process to start up
+            await asyncio.sleep(3)
+            # read the output
+            text = wrapper.read().decode("utf-8")
+            # listener should have printed "listening
+            assert text.index("listening") is not None
+            # stop process with SIGINT
+            wrapper.process.write(rnsh.process.CTRL_C)
+            # wait for process to wind down
+            start_time = time.time()
+            while wrapper.process.running and time.time() - start_time < 5:
+                await asyncio.sleep(0.1)
+            assert not wrapper.process.running
+
+
+async def get_id_and_dest(td: str) -> tuple[str, str]:
+    with tests.helpers.SubprocessReader(name="getid", argv=shlex.split(f"poetry run -- rnsh -l --config \"{td}\" -p")) as wrapper:
+        wrapper.start()
+        await asyncio.sleep(0.1)
+        assert wrapper.process.running
+        # wait for process to start up
+        await asyncio.sleep(3)
+        # read the output
+        text = wrapper.read().decode("utf-8").replace("\r", "").replace("\n", "")
+        assert text.index("Identity") is not None
+        match = re.search(r"<([a-f0-9]{32})>[^<]+<([a-f0-9]{32})>", text)
+        assert match is not None
+        ih = match.group(1)
+        assert len(ih) == 32
+        dh = match.group(2)
+        assert len(dh) == 32
+        await asyncio.sleep(0.1)
+        assert not wrapper.process.running
+        return ih, dh
+
+
+
+@pytest.mark.skip_ci
+@pytest.mark.asyncio
+async def test_rnsh_get_id_and_dest() -> [int]:
+    with tests.helpers.tempdir() as td:
+        ih, dh = await get_id_and_dest(td)
+        assert len(ih) == 32
+        assert len(dh) == 32
+
+
+async def do_connected_test(listener_args: str, initiator_args: str, test: callable):
+    with tests.helpers.tempdir() as td:
+        ih, dh = await get_id_and_dest(td)
+        assert len(ih) == 32
+        assert len(dh) == 32
+        with tests.helpers.SubprocessReader(name="listener", argv=shlex.split(f"poetry run -- rnsh -l --config \"{td}\" {listener_args}")) as listener, \
+                tests.helpers.SubprocessReader(name="initiator", argv=shlex.split(f"poetry run -- rnsh --config \"{td}\" {dh} {initiator_args}")) as initiator:
+            # listener startup
+            listener.start()
+            await asyncio.sleep(0.1)
+            assert listener.process.running
+            # wait for process to start up
+            await asyncio.sleep(3)
+            # read the output
+            text = listener.read().decode("utf-8")
+            assert text.index(dh) is not None
+
+            # initiator run
+            initiator.start()
+            assert initiator.process.running
+
+            await test(td, ih, dh, listener, initiator)
+
+            # expect test to shut down initiator
+            assert not initiator.process.running
+
+            # stop process with SIGINT
+            listener.process.write(rnsh.process.CTRL_C)
+            # wait for process to wind down
+            start_time = time.time()
+            while listener.process.running and time.time() - start_time < 5:
+                await asyncio.sleep(0.1)
+            assert not listener.process.running
+
+
+@pytest.mark.skip_ci
+@pytest.mark.asyncio
+async def test_rnsh_get_echo_through():
+    cwd = os.getcwd()
+
+    async def test(td: str, ih: str, dh: str, listener: tests.helpers.SubprocessReader,
+                   initiator: tests.helpers.SubprocessReader):
+        start_time = time.time()
+        while initiator.return_code is None and time.time() - start_time < 3:
+            await asyncio.sleep(0.1)
+        text = initiator.read().decode("utf-8").replace("\r", "").replace("\n", "")
+        assert text == cwd
+
+    await do_connected_test("-n -C -- /bin/pwd", "", test)
+
+
+
+