kytos.utils.config.KytosConfig.check_sections() - Code Metrics - Inspection of "Import setuptools before distutils" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#323)

unknown

created 2022-11-12 13:05 UTC

kytos.utils.config.KytosConfig.check_sections() A

↳ Parent: kytos.utils.config

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
eloc	6
nop	1
dl	0
loc	7
ccs	0
cts	5
cp	0
crap	12
rs	10
c	0
b	0
f	0

"""Kytos utils configuration."""
# This file is part of kytos-utils.
#
# Copyright (c) 2016 Kytos Team
#
# Authors:
#    Beraldo Leal <beraldo AT ncc DOT unesp DOT br>

import json
import logging
import os
import re
import shutil
from collections import namedtuple
from configparser import ConfigParser
from pathlib import Path
from urllib.error import URLError
from urllib.request import urlopen

LOG = logging.getLogger(__name__)


def create_skel_dir():
    """Install data_files in the /etc directory."""
    base_env = Path(os.environ.get('VIRTUAL_ENV', '/'))
    etc_kytos = base_env / 'etc' / 'kytos'

    # kytos-utils/kytos/utils/config.py -> kytos-utils/kytos
    parent_dir = Path(__file__).resolve().parent.parent

    if not etc_kytos.exists():
        os.makedirs(etc_kytos)

    src = parent_dir / 'templates' / 'skel'
    dst = etc_kytos / 'skel'

    if dst.exists():
        if not next(dst.iterdir(), None):
            # Path already exists but it's empty, so we'll populate it
            # We remove it first to avoid an exception from copytree
            dst.rmdir()
            shutil.copytree(str(src), str(dst))
    else:
        shutil.copytree(str(src), str(dst))


class KytosConfig():
    """Kytos Configs.

    Read the config file for kytos utils and/or request data for the user in
    order to get the correct paths and links.
    """

    def __init__(self, config_file='~/.kytosrc'):
        """Init method.

        Receive the config_file as argument.
        """
        create_skel_dir()
        self.config_file = os.path.expanduser(config_file)
        self.debug = False
        if self.debug:
            LOG.setLevel(logging.DEBUG)

        # allow_no_value=True is used to keep the comments on the config file.
        self.config = ConfigParser(allow_no_value=True)

        # Parse the config file. If no config file was found, then create some
        # default sections on the config variable.
        self.config.read(self.config_file)
        self.check_sections(self.config)

        self.set_env_or_defaults()

        if not os.path.exists(self.config_file):
            LOG.warning("Config file %s not found.", self.config_file)
            LOG.warning("Creating a new empty config file.")
            with open(self.config_file, 'w') as output_file:
                os.chmod(self.config_file, 0o0600)
                self.config.write(output_file)

    def log_configs(self):
        """Log the read configs if debug is enabled."""
        for sec in self.config.sections():
            LOG.debug('   %s: %s', sec, self.config.options(sec))

    def set_env_or_defaults(self):
        """Read some environment variables and set them on the config.

        If no environment variable is found and the config section/key is
        empty, then set some default values.
        """
        option = namedtuple('Option', ['section', 'name', 'env_var',
                                       'default_value'])

        options = [option('auth', 'user', 'NAPPS_USER', None),
                   option('auth', 'token', 'NAPPS_TOKEN', None),
                   option('napps', 'api', 'NAPPS_API_URI',
                          'https://napps.kytos.io/api/'),
                   option('napps', 'repo', 'NAPPS_REPO_URI',
                          'https://napps.kytos.io/repo'),
                   option('kytos', 'api', 'KYTOS_API',
                          'http://localhost:8181/')]

        for option in options:
            if not self.config.has_option(option.section, option.name):
                env_value = os.environ.get(option.env_var,
                                           option.default_value)
                if env_value:
                    self.config.set(option.section, option.name, env_value)

        self.config.set('global', 'debug', str(self.debug))

    @staticmethod
    def check_sections(config):
        """Create a empty config file."""
        default_sections = ['global', 'auth', 'napps', 'kytos']
        for section in default_sections:
            if not config.has_section(section):
                config.add_section(section)

    def save_token(self, user, token):
        """Save the token on the config file."""
        self.config.set('auth', 'user', user)
        self.config.set('auth', 'token', token)
        # allow_no_value=True is used to keep the comments on the config file.
        new_config = ConfigParser(allow_no_value=True)

        # Parse the config file. If no config file was found, then create some
        # default sections on the config variable.
        new_config.read(self.config_file)
        self.check_sections(new_config)

        new_config.set('auth', 'user', user)
        new_config.set('auth', 'token', token)
        filename = os.path.expanduser(self.config_file)
        with open(filename, 'w') as out_file:
            os.chmod(filename, 0o0600)
            new_config.write(out_file)

    def clear_token(self):
        """Clear Token information on config file."""
        # allow_no_value=True is used to keep the comments on the config file.
        new_config = ConfigParser(allow_no_value=True)

        # Parse the config file. If no config file was found, then create some
        # default sections on the config variable.
        new_config.read(self.config_file)
        self.check_sections(new_config)

        new_config.remove_option('auth', 'user')
        new_config.remove_option('auth', 'token')
        filename = os.path.expanduser(self.config_file)
        with open(filename, 'w') as out_file:
            os.chmod(filename, 0o0600)
            new_config.write(out_file)

    @classmethod
    def get_metadata(cls):
        """Return kytos-utils metadata."""
        meta_path = ("%s/metadata.py" % os.path.dirname(__file__))
        meta_file = open(meta_path).read()
        metadata = dict(re.findall(r"(__[a-z]+__)\s*=\s*'([^']+)'", meta_file))
        return metadata

    @classmethod
    def get_remote_metadata(cls):
        """Return kytos metadata."""
        kytos_api = KytosConfig().config.get('kytos', 'api')
        meta_uri = kytos_api + 'api/kytos/core/metadata/'
        meta_file = urlopen(meta_uri).read()
        metadata = json.loads(meta_file)
        return metadata

    @classmethod
    def check_versions(cls):
        """Check if kytos and kytos-utils metadata are compatible."""
        try:
            kytos_metadata = cls.get_remote_metadata()
            kytos_version = kytos_metadata.get('__version__')
        except URLError as exc:
            LOG.debug('Couldn\'t connect to kytos server: %s', exc)
        else:
            kutils_metadata = cls.get_metadata()
            kutils_version = kutils_metadata.get('__version__')

            if kytos_version != kutils_version:
                logger = logging.getLogger()
                logger.warning('kytos (%s) and kytos utils (%s) versions '
                               'are not equal.', kytos_version, kutils_version)


1			"""Kytos utils configuration."""
2			# This file is part of kytos-utils.
3			#
4			# Copyright (c) 2016 Kytos Team
5			#
6			# Authors:
7			# Beraldo Leal <beraldo AT ncc DOT unesp DOT br>
8
9			import json
10			import logging
11			import os
12			import re
13			import shutil
14			from collections import namedtuple
15			from configparser import ConfigParser
16			from pathlib import Path
17			from urllib.error import URLError
18			from urllib.request import urlopen
19
20			LOG = logging.getLogger(__name__)
21
22
23			def create_skel_dir():
24			"""Install data_files in the /etc directory."""
25			base_env = Path(os.environ.get('VIRTUAL_ENV', '/'))
26			etc_kytos = base_env / 'etc' / 'kytos'
27
28			# kytos-utils/kytos/utils/config.py -> kytos-utils/kytos
29			parent_dir = Path(__file__).resolve().parent.parent
30
31			if not etc_kytos.exists():
32			os.makedirs(etc_kytos)
33
34			src = parent_dir / 'templates' / 'skel'
35			dst = etc_kytos / 'skel'
36
37			if dst.exists():
38			if not next(dst.iterdir(), None):
39			# Path already exists but it's empty, so we'll populate it
40			# We remove it first to avoid an exception from copytree
41			dst.rmdir()
42			shutil.copytree(str(src), str(dst))
43			else:
44			shutil.copytree(str(src), str(dst))
45
46
47			class KytosConfig():
48			"""Kytos Configs.
49
50			Read the config file for kytos utils and/or request data for the user in
51			order to get the correct paths and links.
52			"""
53
54			def __init__(self, config_file='~/.kytosrc'):
55			"""Init method.
56
57			Receive the config_file as argument.
58			"""
59			create_skel_dir()
60			self.config_file = os.path.expanduser(config_file)
61			self.debug = False
62			if self.debug:
63			LOG.setLevel(logging.DEBUG)
64
65			# allow_no_value=True is used to keep the comments on the config file.
66			self.config = ConfigParser(allow_no_value=True)
67
68			# Parse the config file. If no config file was found, then create some
69			# default sections on the config variable.
70			self.config.read(self.config_file)
71			self.check_sections(self.config)
72
73			self.set_env_or_defaults()
74
75			if not os.path.exists(self.config_file):
76			LOG.warning("Config file %s not found.", self.config_file)
77			LOG.warning("Creating a new empty config file.")
78			with open(self.config_file, 'w') as output_file:
79			os.chmod(self.config_file, 0o0600)
80			self.config.write(output_file)
81
82			def log_configs(self):
83			"""Log the read configs if debug is enabled."""
84			for sec in self.config.sections():
85			LOG.debug(' %s: %s', sec, self.config.options(sec))
86
87			def set_env_or_defaults(self):
88			"""Read some environment variables and set them on the config.
89
90			If no environment variable is found and the config section/key is
91			empty, then set some default values.
92			"""
93			option = namedtuple('Option', ['section', 'name', 'env_var',
94			'default_value'])
95
96			options = [option('auth', 'user', 'NAPPS_USER', None),
97			option('auth', 'token', 'NAPPS_TOKEN', None),
98			option('napps', 'api', 'NAPPS_API_URI',
99			'https://napps.kytos.io/api/'),
100			option('napps', 'repo', 'NAPPS_REPO_URI',
101			'https://napps.kytos.io/repo'),
102			option('kytos', 'api', 'KYTOS_API',
103			'http://localhost:8181/')]
104
105			for option in options:
106			if not self.config.has_option(option.section, option.name):
107			env_value = os.environ.get(option.env_var,
108			option.default_value)
109			if env_value:
110			self.config.set(option.section, option.name, env_value)
111
112			self.config.set('global', 'debug', str(self.debug))
113
114			@staticmethod
115			def check_sections(config):
116			"""Create a empty config file."""
117			default_sections = ['global', 'auth', 'napps', 'kytos']
118			for section in default_sections:
119			if not config.has_section(section):
120			config.add_section(section)
121
122			def save_token(self, user, token):
123			"""Save the token on the config file."""
124			self.config.set('auth', 'user', user)
125			self.config.set('auth', 'token', token)
126			# allow_no_value=True is used to keep the comments on the config file.
127			new_config = ConfigParser(allow_no_value=True)
128
129			# Parse the config file. If no config file was found, then create some
130			# default sections on the config variable.
131			new_config.read(self.config_file)
132			self.check_sections(new_config)
133
134			new_config.set('auth', 'user', user)
135			new_config.set('auth', 'token', token)
136			filename = os.path.expanduser(self.config_file)
137			with open(filename, 'w') as out_file:
138			os.chmod(filename, 0o0600)
139			new_config.write(out_file)
140
141			def clear_token(self):
142			"""Clear Token information on config file."""
143			# allow_no_value=True is used to keep the comments on the config file.
144			new_config = ConfigParser(allow_no_value=True)
145
146			# Parse the config file. If no config file was found, then create some
147			# default sections on the config variable.
148			new_config.read(self.config_file)
149			self.check_sections(new_config)
150
151			new_config.remove_option('auth', 'user')
152			new_config.remove_option('auth', 'token')
153			filename = os.path.expanduser(self.config_file)
154			with open(filename, 'w') as out_file:
155			os.chmod(filename, 0o0600)
156			new_config.write(out_file)
157
158			@classmethod
159			def get_metadata(cls):
160			"""Return kytos-utils metadata."""
161			meta_path = ("%s/metadata.py" % os.path.dirname(__file__))
162			meta_file = open(meta_path).read()
163			metadata = dict(re.findall(r"(__[a-z]+__)\s=\s'([^']+)'", meta_file))
164			return metadata
165
166			@classmethod
167			def get_remote_metadata(cls):
168			"""Return kytos metadata."""
169			kytos_api = KytosConfig().config.get('kytos', 'api')
170			meta_uri = kytos_api + 'api/kytos/core/metadata/'
171			meta_file = urlopen(meta_uri).read()
172			metadata = json.loads(meta_file)
173			return metadata
174
175			@classmethod
176			def check_versions(cls):
177			"""Check if kytos and kytos-utils metadata are compatible."""
178			try:
179			kytos_metadata = cls.get_remote_metadata()
180			kytos_version = kytos_metadata.get('__version__')
181			except URLError as exc:
182			LOG.debug('Couldn\'t connect to kytos server: %s', exc)
183			else:
184			kutils_metadata = cls.get_metadata()
185			kutils_version = kutils_metadata.get('__version__')
186
187			if kytos_version != kutils_version:
188			logger = logging.getLogger()
189			logger.warning('kytos (%s) and kytos utils (%s) versions '
190			'are not equal.', kytos_version, kutils_version)
191

kytos / kytos-utils

Pull Request — master (#323)

kytos.utils.config.KytosConfig.check_sections() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like