exabgp.application.validate

Complexity

Total Complexity

14

Size/Duplication

Total Lines	85
Duplicated Lines	0 %

Importance

Changes

0

3 Functions

Rating	Name	Duplication	Size	Complexity
D	cmdline()	0	44	12
A	main()	0	5	1
A	args()	0	7	1

# encoding: utf-8

"""exabgp configuration validation"""


import sys
import syslog
import argparse

from exabgp.environment import getenv
from exabgp.environment import getconf

from exabgp.debug import trace_interceptor
from exabgp.logger import log

from exabgp.reactor.loop import Reactor
from exabgp.configuration.check import check_generation


def args(sub):
    # fmt:off
    sub.add_argument('-n', '--neighbor', help='check the parsing of the neighbors', action='store_true')
    sub.add_argument('-r', '--route', help='check the parsing of the routes', action='store_true')
    sub.add_argument('-v', '--verbose', help='be verbose in the display', action='store_true')
    sub.add_argument('-p', '--pdb', help='fire the debugger on critical logging, SIGTERM, and exceptions (shortcut for exabgp.pdb.enable=true)', action='store_true')
    sub.add_argument('configuration', help='configuration file(s)', nargs='+', type=str)
    # fmt:on


def cmdline(cmdarg):
    env = getenv()

    # Must be done before setting the logger as it modify its behaviour
    if cmdarg.verbose:
        env.log.all = True
        env.log.level = syslog.LOG_DEBUG

    log.init()

    if cmdarg.pdb:
        env.debug.pdb = True

    if cmdarg.verbose:
        env.log.parser = True

    for configuration in cmdarg.configuration:
        log.notice(f'loading {configuration}', 'configuration')
        location = getconf(configuration)
        if not location:
            log.critical(f'{configuration} is not an exabgp config file', 'configuration')
            sys.exit(1)

        config = Reactor([location]).configuration

        if not config.reload():
            log.critical(f'{configuration} is not a valid config file', 'configuration')
            sys.exit(1)
        log.info(f'\u2713 loading', 'configuration')

        if cmdarg.neighbor:
            log.notice(f'checking neighbors', 'configuration')
            for name, neighbor in config.neighbors.items():
                reparsed = neighbor.string()
                for line in reparsed.split('\n'):
                    log.debug(line, configuration)
                log.info(f'\u2713 neighbor  {name.split()[1]}', 'configuration')

        if cmdarg.route:
            log.notice(f'checking routes', 'configuration')
            if not check_generation(config.neighbors):
                log.critical(f'{configuration} has an invalid route', 'configuration')
                sys.exit(1)
            log.info(f'\u2713 routes', 'configuration')


def main():
    parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
    args(parser)
    trace_interceptor()
    cmdline(parser, parser.parse_args())


if __name__ == '__main__':
    main()