remove the footer saying that vim is the best editor -- anyone knows anyway

[pingcheck] / test / connd_state.py

#!/usr/bin/env python3

""" Representation for connd state as returned by tell-connd --status

Christian Herdtweck, Intra2net, January 2015
(c) Intra2net AG 2015
"""

# Version History
# 16/01/15 Christian Herdtweck: started creation

import subprocess
from re import match as regexp
from os import EX_OK

# constants
default_tell_connd_binary = '/usr/intranator/bin/tell-connd'
timeout = 1

ONLINE_STATE_ALWAYS_ONLINE = 'always online'
ONLINE_STATE_ALWAYS_OFFLINE = 'always offline'
ONLINE_STATE_DIAL_ON_COMMAND = 'dial on command'
ONLINE_STATE_DIAL_ON_DEMAND = 'dial on demand'

SUBSYS_DNS       = 'dns'
SUBSYS_DYNDNS    = 'dyndns'
SUBSYS_MAIL      = 'mail'
SUBSYS_NTP       = 'ntp'
SUBSYS_SOCKS     = 'socks'
SUBSYS_VPN       = 'vpn'
SUBSYS_WEBPROXY  = 'webproxy'
SUBSYS_PINGCHECK = 'pingcheck'
ALL_SUBSYS = (SUBSYS_DNS, SUBSYS_DYNDNS, SUBSYS_MAIL, SUBSYS_NTP, \
              SUBSYS_SOCKS, SUBSYS_VPN, SUBSYS_WEBPROXY, SUBSYS_PINGCHECK)

class ConndState:
    """ representation of connd's status as returned by tell-connd --status """

    online_mode = None
    default_provider = None
    subsys_online = None
    subsys_offline = None
    subsys_disabled = None
    connections = None
    actions = None
    online_ips = None
    connected_vpns = None

    def __str__(self):
        return '[ConndState: {0} (default {1}), {2} conn\'s, {3} ips, {4} vpns ]'.format(\
                self.online_mode, self.default_provider, len(self.connections), \
                len(self.online_ips), len(self.connected_vpns))

    def complete_str(self):
        # general
        parts = ['ConndState: online mode = "{0}" (default provider: {1})\n'.format(\
                  self.online_mode, self.default_provider), ]

        # subsys
        #            '  connctns:
        parts.append('    subsys: online: ')
        if self.subsys_online:
            for subsys in self.subsys_online:
                parts.append(subsys + ' ')
        else:
            parts.append('None ')
        parts.append(';  offline: ')
        if self.subsys_offline:
            for subsys in self.subsys_offline:
                parts.append(subsys + ' ')
        else:
            parts.append('None ')
        parts.append(';  disabled: ')
        if self.subsys_disabled:
            for subsys in self.subsys_disabled:
                parts.append(subsys + ' ')
        else:
            parts.append('None')
        parts.append('\n')

        # connections
        parts.append('     conns: ')
        if self.connections:
            name, info, actions = self.connections[0]
            parts.append('{0}: {1}, {2}\n'.format(name, info, actions))
        else:
            parts.append('None\n')
        for name, info, actions in self.connections[1:]:
            #            '  connctns:
            parts.append('            {0}: {1}, {2}\n'.format(name, info, actions))

        # actions
        #            '  connctns:
        parts.append('   actions: ')
        if self.actions:
            parts.append(self.actions[0] + '\n')
        else:
            parts.append('None\n')
        for action in self.actions[1:]:
            #            '  connctns:
            parts.append('            {0}\n'.format(action))

        # online IPs
        #            '  connctns:
        parts.append('       IPs: ')
        if self.online_ips:
            parts.append(self.online_ips[0])
            for ip in self.online_ips[1:]:
                parts.append(', {0}'.format(ip))
        else:
            parts.append('None')
        parts.append('\n')

        # VPNs
        #            '  connctns:
        parts.append('      VPNs: ')
        if self.connected_vpns:
            parts.append(self.connected_vpns[0])
            for vpn in self.connected_vpns[1:]:
                parts.append(', {0}'.format(vpn))
        else:
            parts.append('None')
        parts.append('\n')


        return ''.join(parts)
    #end: ConndState.complete_str


    @staticmethod
    def run_tell_connd(tell_connd_binary=default_tell_connd_binary):
        """ run tell-connd --status, return output iterator and return code

        catches all it can, so should usually return (output, return_code)
          where output = [line1, line2, ...]
        if return_code != 0, output's first line(s) is error message
        """
        try:
            output = subprocess.check_output([tell_connd_binary, '--status'], \
                    stderr=subprocess.STDOUT, universal_newlines=True, shell=False, timeout=timeout)
            return EX_OK, output.splitlines()
        except subprocess.CalledProcessError as cpe: # non-zero return status
            output = ['tell-connd exited with status {0}'.format(cpe.returncode), ]
            output.extend( cpe.output.splitlines() )
            return cpe.returncode, output
        except subprocess.TimeoutExpired as te:
            output = ['tell-connd timed out after {0}s. Returning -1'.format(te.timeout), ]
            output.extend( te.output.splitlines() )
            return -1, output
        except Exception as e:
            output = [str(e),]
            return -1, output
    #end: ConndState.run_tell_connd


    @staticmethod
    def get_state(tell_connd_binary=default_tell_connd_binary):
        """ get actual state from tell-connd --status

        returns (err_code, output_lines) if something goes wrong running binary;
        raises assertion if output from tell-connd does not match expected format
        """

        state = ConndState()

        err_code, all_lines = ConndState.run_tell_connd(tell_connd_binary)
        if err_code != EX_OK:
            return err_code, all_lines

        output = iter(all_lines)

        # first section
        line = next(output).strip()
        state.online_mode = regexp('online mode\s*:\s*(.+)$', line).groups()[0]
        assert( state.online_mode in (ONLINE_STATE_DIAL_ON_DEMAND, ONLINE_STATE_DIAL_ON_COMMAND, \
                                      ONLINE_STATE_ALWAYS_OFFLINE, ONLINE_STATE_ALWAYS_ONLINE) )
        line = next(output).strip()
        state.default_provider = regexp('default provider\s*:\s*(.*)$', \
                                        line).groups()[0]
        if len(state.default_provider) == 0:
            state.default_provider = None
        line = next(output).strip()
        assert( len(line) == 0 )

        # subsys
        line = next(output).strip()
        assert( line == 'subsys' )
        line = next(output).strip()
        state.subsys_online   = regexp(  'online\s*:\s*(.*)$', line).groups()[0].split()
        for subsys in state.subsys_online:
            assert(subsys in ALL_SUBSYS)
        line = next(output).strip()
        state.subsys_offline  = regexp( 'offline\s*:\s*(.*)$', line).groups()[0].split()
        for subsys in state.subsys_offline:
            assert(subsys in ALL_SUBSYS)
        line = next(output).strip()
        state.subsys_disabled = regexp('disabled\s*:\s*(.*)$', line).groups()[0].split()
        for subsys in state.subsys_disabled:
            assert(subsys in ALL_SUBSYS)
        line = next(output).strip()
        assert( len(line) == 0 )

        # connection map
        state.connections = []
        line = next(output).strip()
        assert( line == 'connection map:' )
        expect_new = True
        for line in output:
            line = line.strip()
            if len(line) == 0:
                continue
            if expect_new:
                if line == 'end of connection map':
                    break
                conn_name, conn_info = regexp('\[\s*(.+)\s*\]\s*:\s*\(\s*(.*)\s*\)', line).groups()
                expect_new = False
            else:
                conn_actions = regexp('actions\s*:\s*\[\s*(.+)\s*\]', line).groups()
                state.connections.append( (conn_name, conn_info, conn_actions) )
                expect_new = True
        #end: for lines
        assert( expect_new )
        line = next(output).strip()
        assert( len(line) == 0 )

        # actions
        line = next(output).strip()
        state.actions = regexp('actions\s*:\s*(.*)', line).groups()[0].split()
        if len(state.actions) == 1 and state.actions[0].strip() == '-':
            state.actions = []
        line = next(output).strip()
        assert( len(line) == 0 )

        # online IPs
        line = next(output).strip()
        state.online_ips = regexp('list of online ips\s*:\s*(.*)', \
                                  line).groups()[0].split()
        if len(state.online_ips) == 1 and state.online_ips[0].strip() == 'NONE':
            state.online_ips = []
        line = next(output).strip()
        assert( len(line) == 0 )

        # VPNs
        state.connected_vpns = []
        line = next(output).strip()
        assert( line == 'vpns connected:' )
        for line in output:
            line = line.strip()
            if len(line) == 0:
                continue
            elif line == 'end of list of connected vpns':
                break
            else:
                state.connected_vpns.append(line)
        #end: for lines

        # done
        line = next(output).strip()
        assert( len(line) == 0 )
        line = next(output).strip()
        assert( line == 'Done.' )

        return state
    #end: ConndState.get_state

#end: class ConndState


def test():
    state = ConndState.get_state()
    print(state)
    print(state.complete_str())

def main():
    """ Main function, called when running file as script; runs test()
    """
    test()
#end: function main


if __name__ == '__main__':
    main()