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 13:42 UTC

ospd.main.main() C

↳ Parent: ospd.main

Complexity

Conditions

Size

Total Lines	73
Code Lines	49

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	49
nop	3
dl	0
loc	73
rs	6.3357
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(
        name,
        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			name,
105			args.log_level,
106			log_file=args.log_file,
107			log_config=args.log_config,
108			foreground=args.foreground,
109			)
110
111			if args.port == 0:
112			server = UnixSocketServer(
113			args.unix_socket, args.socket_mode, args.stream_timeout,
114			)
115			else:
116			server = TlsServer(
117			args.address,
118			args.port,
119			args.cert_file,
120			args.key_file,
121			args.ca_file,
122			args.stream_timeout,
123			)
124
125			daemon = daemon_class(**vars(args))
126
127			if args.version:
128			print_version(daemon)
129			sys.exit()
130
131			if args.list_commands:
132			print(daemon.get_help_text())
133			sys.exit()
134
135			if not args.foreground:
136			go_to_background()
137
138			if not create_pid(args.pid_file):
139			sys.exit()
140
141			# Set signal handler and cleanup
142			atexit.register(
143			exit_cleanup, pidfile=args.pid_file, server=server, daemon=daemon
144			)
145			signal.signal(
146			signal.SIGTERM, partial(exit_cleanup, args.pid_file, server, daemon)
147			)
148
149			if not daemon.check():
150			return 1
151
152			LOGGER.info(
153			"Starting %s version %s.",
154			daemon.daemon_info['name'],
155			daemon.daemon_info['version'],
156			)
157
158			daemon.init(server)
159			daemon.run()
160
161			return 0
162

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