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

Passed

Pull Request — master (#293)

by Juan José

created 2020-06-30 14:44 UTC

ospd.main.main() C

↳ Parent: ospd.main

Complexity

Conditions

Size

Total Lines	72
Code Lines	48

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	48
nop	3
dl	0
loc	72
rs	6.3684
c	0
b	0
f	0

How to fix Long Method

# Copyright (C) 2014-2020 Greenbone Networks GmbH
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import logging

import os
import sys
import atexit
import signal

from functools import partial

from typing import Type, Optional
from pathlib import Path

from ospd.misc import go_to_background, create_pid
from ospd.ospd import OSPDaemon
from ospd.parser import create_parser, ParserType
from ospd.server import TlsServer, UnixSocketServer, BaseServer
from ospd.logger import init_logging


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."""

LOGGER = logging.getLogger(__name__)


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 exit_cleanup(
    pidfile: str,
    server: BaseServer,
    daemon: OSPDaemon,
    _signum=None,
    _frame=None,
) -> None:
    """ Removes the pidfile before ending the daemon. """
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    pidpath = Path(pidfile)

    daemon.daemon_exit_cleanup()

    if not pidpath.is_file():
        return

    with pidpath.open() as f:
        if int(f.read()) == os.getpid():
            LOGGER.info("Shutting-down server ...")
            server.close()
            LOGGER.debug("Finishing daemon process")
            pidpath.unlink()
            sys.exit()


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(
        args.log_level,
        log_file=args.log_file,
        log_config=args.log_config,
        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())
        sys.exit()

    if not args.foreground:
        go_to_background()

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

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

    if not daemon.check():
        return 1

    LOGGER.info(
        "Starting %s version %s.",
        daemon.daemon_info['name'],
        daemon.daemon_info['version'],
    )

    daemon.init(server)
    daemon.run()

    return 0


1			# Copyright (C) 2014-2020 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: AGPL-3.0-or-later
4			#
5			# This program is free software: you can redistribute it and/or modify
6			# it under the terms of the GNU Affero General Public License as
7			# published by the Free Software Foundation, either version 3 of the
8			# 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 Affero General Public License for more details.
14			#
15			# You should have received a copy of the GNU Affero General Public License
16			# along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18			import logging
19
20			import os
21			import sys
22			import atexit
23			import signal
24
25			from functools import partial
26
27			from typing import Type, Optional
28			from pathlib import Path
29
30			from ospd.misc import go_to_background, create_pid
31			from ospd.ospd import OSPDaemon
32			from ospd.parser import create_parser, ParserType
33			from ospd.server import TlsServer, UnixSocketServer, BaseServer
34			from ospd.logger import init_logging
35
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			LOGGER = logging.getLogger(__name__)
43
44
45			def print_version(daemon: OSPDaemon, file=sys.stdout):
46			""" Prints the server version and license information."""
47
48			scanner_name = daemon.get_scanner_name()
49			server_version = daemon.get_server_version()
50			protocol_version = daemon.get_protocol_version()
51			daemon_name = daemon.get_daemon_name()
52			daemon_version = daemon.get_daemon_version()
53
54			print(
55			"OSP Server for {0}: {1}".format(scanner_name, server_version),
56			file=file,
57			)
58			print("OSP: {0}".format(protocol_version), file=file)
59			print("{0}: {1}".format(daemon_name, daemon_version), file=file)
60			print(file=file)
61			print(COPYRIGHT, file=file)
62
63
64			def exit_cleanup(
65			pidfile: str,
66			server: BaseServer,
67			daemon: OSPDaemon,
68			_signum=None,
69			_frame=None,
70			) -> None:
71			""" Removes the pidfile before ending the daemon. """
72			signal.signal(signal.SIGINT, signal.SIG_IGN)
73			pidpath = Path(pidfile)
74
75			daemon.daemon_exit_cleanup()
76
77			if not pidpath.is_file():
78			return
79
80			with pidpath.open() as f:
81			if int(f.read()) == os.getpid():
82			LOGGER.info("Shutting-down server ...")
83			server.close()
84			LOGGER.debug("Finishing daemon process")
85			pidpath.unlink()
86			sys.exit()
87
88
89			def main(
90			name: str,
91			daemon_class: Type[OSPDaemon],
92			parser: Optional[ParserType] = None,
93			):
94			""" OSPD Main function. """
95
96			if not parser:
97			parser = create_parser(name)
98			args = parser.parse_arguments()
99
100			if args.version:
101			args.foreground = True
102
103			init_logging(
104			args.log_level,
105			log_file=args.log_file,
106			log_config=args.log_config,
107			foreground=args.foreground,
108			)
109
110			if args.port == 0:
111			server = UnixSocketServer(
112			args.unix_socket, args.socket_mode, args.stream_timeout,
113			)
114			else:
115			server = TlsServer(
116			args.address,
117			args.port,
118			args.cert_file,
119			args.key_file,
120			args.ca_file,
121			args.stream_timeout,
122			)
123
124			daemon = daemon_class(**vars(args))
125
126			if args.version:
127			print_version(daemon)
128			sys.exit()
129
130			if args.list_commands:
131			print(daemon.get_help_text())
132			sys.exit()
133
134			if not args.foreground:
135			go_to_background()
136
137			if not create_pid(args.pid_file):
138			sys.exit()
139
140			# Set signal handler and cleanup
141			atexit.register(
142			exit_cleanup, pidfile=args.pid_file, server=server, daemon=daemon
143			)
144			signal.signal(
145			signal.SIGTERM, partial(exit_cleanup, args.pid_file, server, daemon)
146			)
147
148			if not daemon.check():
149			return 1
150
151			LOGGER.info(
152			"Starting %s version %s.",
153			daemon.daemon_info['name'],
154			daemon.daemon_info['version'],
155			)
156
157			daemon.init(server)
158			daemon.run()
159
160			return 0
161

greenbone / ospd

Pull Request — master (#293)

ospd.main.main() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like