ospd.main.main() - Code Metrics - Inspection of "Cleanup" - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#198)

unknown

created 2020-02-07 07:36 UTC

ospd.main.main() C

↳ Parent: ospd.main

Complexity

Conditions

Size

Total Lines	59
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	38
nop	3
dl	0
loc	59
rs	6.6346
c	0
b	0
f	0

How to fix Long Method

# Copyright (C) 2019 Greenbone Networks GmbH
#
# SPDX-License-Identifier: GPL-2.0-or-later
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

import logging

from logging.handlers import SysLogHandler, WatchedFileHandler

import os
import sys
import atexit
import signal

from functools import partial

from typing import Type, Optional

from ospd.misc import go_to_background, create_pid, remove_pidfile
from ospd.ospd import OSPDaemon
from ospd.parser import create_parser, ParserType
from ospd.server import TlsServer, UnixSocketServer

COPYRIGHT = """Copyright (C) 2014, 2015, 2018, 2019 Greenbone Networks GmbH
License GPLv2+: GNU GPL version 2 or later
This is free software: you are free to change and redistribute it.
There is NO WARRANTY, to the extent permitted by law."""


def print_version(daemon: OSPDaemon, file=sys.stdout):
    """ Prints the server version and license information."""

    scanner_name = daemon.get_scanner_name()
    server_version = daemon.get_server_version()
    protocol_version = daemon.get_protocol_version()
    daemon_name = daemon.get_daemon_name()
    daemon_version = daemon.get_daemon_version()

    print(
        "OSP Server for {0}: {1}".format(scanner_name, server_version),
        file=file,
    )
    print("OSP: {0}".format(protocol_version), file=file)
    print("{0}: {1}".format(daemon_name, daemon_version), file=file)
    print(file=file)
    print(COPYRIGHT, file=file)


def init_logging(
    name: str,
    log_level: int,
    *,
    log_file: Optional[str] = None,
    foreground: Optional[bool] = False
):

    rootlogger = logging.getLogger()
    rootlogger.setLevel(log_level)

    if foreground:
        console = logging.StreamHandler()
        console.setFormatter(
            logging.Formatter(
                '%(asctime)s {}: %(levelname)s: (%(name)s) %(message)s'.format(
                    name
                )
            )
        )
        rootlogger.addHandler(console)
    elif log_file:
        logfile = WatchedFileHandler(log_file)
        logfile.setFormatter(
            logging.Formatter(
                '%(asctime)s {}: %(levelname)s: (%(name)s) %(message)s'.format(
                    name
                )
            )
        )
        rootlogger.addHandler(logfile)
    else:
        syslog = SysLogHandler('/dev/log')
        syslog.setFormatter(
            logging.Formatter(
                '{}: %(levelname)s: (%(name)s) %(message)s'.format(name)
            )
        )
        rootlogger.addHandler(syslog)
        # Duplicate syslog's file descriptor to stout/stderr.
        syslog_fd = syslog.socket.fileno()
        os.dup2(syslog_fd, 1)
        os.dup2(syslog_fd, 2)


def main(
    name: str,
    daemon_class: Type[OSPDaemon],
    parser: Optional[ParserType] = None,
):
    """ OSPD Main function. """

    if not parser:
        parser = create_parser(name)
    args = parser.parse_arguments()

    if args.version:
        args.foreground = True

    init_logging(
        name, args.log_level, log_file=args.log_file, foreground=args.foreground
    )

    if args.port == 0:
        server = UnixSocketServer(
            args.unix_socket, args.socket_mode, args.stream_timeout,
        )
    else:
        server = TlsServer(
            args.address,
            args.port,
            args.cert_file,
            args.key_file,
            args.ca_file,
            args.stream_timeout,
        )

    daemon = daemon_class(**vars(args))

    if args.version:
        print_version(daemon)
        sys.exit()

    if args.list_commands:
        print(daemon.get_help_text())

    if not args.foreground:
        go_to_background()

    if not create_pid(args.pid_file):
        sys.exit()

    # Set signal handler and cleanup
    atexit.register(remove_pidfile, pidfile=args.pid_file)
    signal.signal(signal.SIGTERM, partial(remove_pidfile, args.pid_file))

    if not daemon.check():
        return 1

    daemon.init()

    daemon.run(server)

    return 0


1			# Copyright (C) 2019 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: GPL-2.0-or-later
4			#
5			# This program is free software; you can redistribute it and/or
6			# modify it under the terms of the GNU General Public License
7			# as published by the Free Software Foundation; either version 2
8			# of the License, or (at your option) any later version.
9			#
10			# This program is distributed in the hope that it will be useful,
11			# but WITHOUT ANY WARRANTY; without even the implied warranty of
12			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			# GNU General Public License for more details.
14			#
15			# You should have received a copy of the GNU General Public License
16			# along with this program; if not, write to the Free Software
17			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18
19			import logging
20
21			from logging.handlers import SysLogHandler, WatchedFileHandler
22
23			import os
24			import sys
25			import atexit
26			import signal
27
28			from functools import partial
29
30			from typing import Type, Optional
31
32			from ospd.misc import go_to_background, create_pid, remove_pidfile
33			from ospd.ospd import OSPDaemon
34			from ospd.parser import create_parser, ParserType
35			from ospd.server import TlsServer, UnixSocketServer
36
37			COPYRIGHT = """Copyright (C) 2014, 2015, 2018, 2019 Greenbone Networks GmbH
38			License GPLv2+: GNU GPL version 2 or later
39			This is free software: you are free to change and redistribute it.
40			There is NO WARRANTY, to the extent permitted by law."""
41
42
43			def print_version(daemon: OSPDaemon, file=sys.stdout):
44			""" Prints the server version and license information."""
45
46			scanner_name = daemon.get_scanner_name()
47			server_version = daemon.get_server_version()
48			protocol_version = daemon.get_protocol_version()
49			daemon_name = daemon.get_daemon_name()
50			daemon_version = daemon.get_daemon_version()
51
52			print(
53			"OSP Server for {0}: {1}".format(scanner_name, server_version),
54			file=file,
55			)
56			print("OSP: {0}".format(protocol_version), file=file)
57			print("{0}: {1}".format(daemon_name, daemon_version), file=file)
58			print(file=file)
59			print(COPYRIGHT, file=file)
60
61
62			def init_logging(
63			name: str,
64			log_level: int,
65			*,
66			log_file: Optional[str] = None,
67			foreground: Optional[bool] = False
68			):
69
70			rootlogger = logging.getLogger()
71			rootlogger.setLevel(log_level)
72
73			if foreground:
74			console = logging.StreamHandler()
75			console.setFormatter(
76			logging.Formatter(
77			'%(asctime)s {}: %(levelname)s: (%(name)s) %(message)s'.format(
78			name
79			)
80			)
81			)
82			rootlogger.addHandler(console)
83			elif log_file:
84			logfile = WatchedFileHandler(log_file)
85			logfile.setFormatter(
86			logging.Formatter(
87			'%(asctime)s {}: %(levelname)s: (%(name)s) %(message)s'.format(
88			name
89			)
90			)
91			)
92			rootlogger.addHandler(logfile)
93			else:
94			syslog = SysLogHandler('/dev/log')
95			syslog.setFormatter(
96			logging.Formatter(
97			'{}: %(levelname)s: (%(name)s) %(message)s'.format(name)
98			)
99			)
100			rootlogger.addHandler(syslog)
101			# Duplicate syslog's file descriptor to stout/stderr.
102			syslog_fd = syslog.socket.fileno()
103			os.dup2(syslog_fd, 1)
104			os.dup2(syslog_fd, 2)
105
106
107			def main(
108			name: str,
109			daemon_class: Type[OSPDaemon],
110			parser: Optional[ParserType] = None,
111			):
112			""" OSPD Main function. """
113
114			if not parser:
115			parser = create_parser(name)
116			args = parser.parse_arguments()
117
118			if args.version:
119			args.foreground = True
120
121			init_logging(
122			name, args.log_level, log_file=args.log_file, foreground=args.foreground
123			)
124
125			if args.port == 0:
126			server = UnixSocketServer(
127			args.unix_socket, args.socket_mode, args.stream_timeout,
128			)
129			else:
130			server = TlsServer(
131			args.address,
132			args.port,
133			args.cert_file,
134			args.key_file,
135			args.ca_file,
136			args.stream_timeout,
137			)
138
139			daemon = daemon_class(**vars(args))
140
141			if args.version:
142			print_version(daemon)
143			sys.exit()
144
145			if args.list_commands:
146			print(daemon.get_help_text())
147
148			if not args.foreground:
149			go_to_background()
150
151			if not create_pid(args.pid_file):
152			sys.exit()
153
154			# Set signal handler and cleanup
155			atexit.register(remove_pidfile, pidfile=args.pid_file)
156			signal.signal(signal.SIGTERM, partial(remove_pidfile, args.pid_file))
157
158			if not daemon.check():
159			return 1
160
161			daemon.init()
162
163			daemon.run(server)
164
165			return 0
166

greenbone / ospd

Pull Request — master (#198)

ospd.main.main() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like