Encapsulate shell commands via implementing "Command" classes

[buhtz]

Problem

BIT does call a lot of shell commands using system.os() or the subprocess package. Those calls are all over the whole code base and lack of isolation. This makes it hard to track them (e.g. logging) and to mock them in tests.

Solution

Coding is in progress in a hidden branch. I do experiment with some ideas. In short each command will have its own class. And some of them can be combined, e.g. class SshCmd with class RsyncCmd.

Known commands used by BIT

• rsync
• fusemount
• ssh
• ssh-add
• ssh-keygen
• ssh-agent
• sshfs
• encfsctl
• chmod
• diff
• xdpyinfo
• encryptfs-verify
• blkid
• mount
• udevadm
• shutdown
• unity (deprecated?)
• man
• Combine "ps" and "grep"
• (Maybe there are some more.)

Quick n dirty pseudo code

class _ShellCommand():
    """Abstract base class for specific commands."""
    class Result:
        """Data structure holding return code, stderr and several more
        things. See execute().
        """
        pass

    def __init__(self):
        self._cmd = None
        """List with the command and its arguments."""

        self._env = None
        """Environment variables used when execute with subprocess."""

    def as_pretty(self, indent: Union[int, str]) -> str:
        """Return the command as a multi line string in human readable
        form.

        Args:
            indent: Number of blanks or a specific string each line is
                indented with.

        Returns:
            A multi line string.
        """

        # example
        return '''
            rsync
            -a
            --foo bar
            -x=7,2
        '''

    def add(self, items: Union[str, list], before_item: str=None, after_item: str=None):
        """Insert one or multiple items to self._cmd. If before_/after_item is
        specified the location is determined.
        """

    def has(self, items: Union[str, list]) -> bool:
        """True if items exist (in this order)"""

    def execute(self) -> Result:
        """Execute the command using subprocess module. Results are stored
        in an extra structure or dict and returned.
        """
        bli, bla, blub = subprocess.run(self._cmd, self._env)

        # result processing
        result = Result(bli, bla, blub)

        return result

        
class RsyncCmd(_ShellCommand):
    def __init__(self):
        super().__init__()

    @property
    def destination(self) -> str:
        # iterate on self._cmd to determine destination and return it

    @property.setter
    def destination(self, dest: str):
        # check if DEST does exist

        # inject DEST into self._cmd

class SshCmd(_ShellCommand):
    @property
    def jump(self) -> tuple[str, str, Union[int, None]]:
        return ('user', 'host', None)

    @property.setter
    def jump(self, tuple[str, str, Union[int, None]]):
        # (usr, host, port)

    @property.setter
    def jumphost(self, host: str):
        # (usr, host, port)

    @property.setter
    def jumpuser(self, user: str):
        # (usr, host, port)


class SshfsCmd(_ShellCommand):
    def __init__(self):
        super().__init__()

        self._sshcmd = None
        """Used via 'sshfs -o ssh_command='"""

    @property
    def ssh_command(self) -> SshCmd:
        return self._sshcmd

    @property.setter
    def ssh_command(self, cmd: SshCmd):
        self._sshcmd = cmd

    def execute(self):
        if self.ssh_command:
            # combine self._cmd with self._sshcmd parameters to one cmd

        super().execute()
        

class EncfsCmd(_ShellCommand):
    pass

class EncfsConrolCmd(_ShellCommand):
    """encfsctrl"""
    pass

class FUserMountCmd(_ShellCommand):
    """fusermount"""
    pass

class SshAgentCmd(_ShellCommand):
    """ssh-agent"""
    pass

class SshAddCmd(_ShellCommand):
    """ssh-add"""
    pass


# Anything else???