<commit_before>#!/usr/bin/env python3
"""Test the configuration module."""

import multiprocessing
import os
import sys
import os.path
import unittest
import shutil
import random
import string
import tempfile
import yaml

# Try to create a working PYTHONPATH
TEST_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
SWITCHMAP_DIRECTORY = os.path.abspath(os.path.join(TEST_DIRECTORY, os.pardir))
ROOT_DIRECTORY = os.path.abspath(os.path.join(SWITCHMAP_DIRECTORY, os.pardir))
if TEST_DIRECTORY.endswith('/switchmap-ng/switchmap/test') is True:
    sys.path.append(ROOT_DIRECTORY)
else:
    print(
        'This script is not installed in the "switchmap-ng/bin" directory. '
        'Please fix.')
    sys.exit(2)

from switchmap.utils import configuration


class TestConfig(unittest.TestCase):
    """Checks all functions and methods."""

    #########################################################################
    # General object setup
    #########################################################################

    random_string = ''.join([random.choice(
        string.ascii_letters + string.digits) for n in range(9)])

    log_directory = tempfile.mkdtemp()
    cache_directory = tempfile.mkdtemp()
    good_config = ("""\
main:
    log_directory: %s
    cache_directory: %s
    agent_threads: 20
    bind_port: 3000
    hostnames:
    - 192.168.1.1
    - 192.168.1.2
    - 192.168.1.3
    - 192.168.1.4
    listen_address: 0.0.0.0
    log_level: debug
    polling_interval: 20
""") % (log_directory, cache_directory)

    # Convert good_config to dictionary
    good_dict = yaml.load(bytes(good_config, 'utf-8'))

    # Set the environmental variable for the configuration directory
    directory = tempfile.mkdtemp()
    os.environ['SWITCHMAP_CONFIGDIR'] = directory
    config_file = ('%s/test_config.yaml') % (directory)

    # Write good_config to file
    with open(config_file, 'w') as f_handle:
        yaml.dump(good_dict, f_handle, default_flow_style=True)

    # Create configuration object
    config = configuration.Config()

    @classmethod
    def tearDownClass(cls):
        """Post test cleanup."""
        os.rmdir(cls.log_directory)
        os.rmdir(cls.config.topology_directory())
        os.rmdir(cls.cache_directory)
        os.remove(cls.config_file)
        os.rmdir(cls.directory)

    def test_init(self):
        """Testing method init."""
        # Testing with non-existant directory
        directory = 'bogus'
        os.environ['SWITCHMAP_CONFIGDIR'] = directory
        with self.assertRaises(SystemExit):
            configuration.Config()

        # Testing with an empty directory
        empty_directory = tempfile.mkdtemp()
        os.environ['SWITCHMAP_CONFIGDIR'] = empty_directory
        with self.assertRaises(SystemExit):
            configuration.Config()

        # Write bad_config to file
        empty_config_file = ('%s/test_config.yaml') % (empty_directory)
        with open(empty_config_file, 'w') as f_handle:
            f_handle.write('')

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.log_file()

        # Cleanup files in temp directories
        _delete_files(directory)

    def test_log_file(self):
        """Testing method log_file."""
        # Test the log_file with a good_dict
        # good key and key_value
        result = self.config.log_file()
        self.assertEqual(
            result, ('%s/switchmap-ng.log') % (self.log_directory))

    def test_web_log_file(self):
        """Testing method web_log_file ."""
        # Testing web_log_file with a good dictionary.
        result = self.config.web_log_file()
        self.assertEqual(
            result, ('%s/switchmap-ng-api.log') % (self.log_directory))

    def test_log_level(self):
        """Testing method log_level."""
        # Tesing with a good_dictionary
        # good key and good key_value
        result = self.config.log_level()
        self.assertEqual(result, 'debug')
        self.assertEqual(result, self.good_dict['main']['log_level'])

        # Set the environmental variable for the configuration directory
        directory = tempfile.mkdtemp()
        os.environ['SWITCHMAP_CONFIGDIR'] = directory
        config_file = ('%s/test_config.yaml') % (directory)

        # Testing log_level with blank key and blank key_value
        key = ''
        key_value = ''
        bad_config = ("""\
main:
    %s %s
""") % (key, key_value)
        bad_dict = yaml.load(bytes(bad_config, 'utf-8'))

        # Write bad_config to file
        with open(config_file, 'w') as f_handle:
            yaml.dump(bad_dict, f_handle, default_flow_style=True)

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.log_level()

        # Testing log_level with good key and blank key_value
        key = 'log_level:'
        key_value = ''
        bad_config = ("""\
main:
    %s %s
""") % (key, key_value)
        bad_dict = yaml.load(bytes(bad_config, 'utf-8'))

        # Write bad_config to file
        with open(config_file, 'w') as f_handle:
            yaml.dump(bad_dict, f_handle, default_flow_style=True)

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.log_level()

        # Cleanup files in temp directories
        _delete_files(directory)

    def test_cache_directory(self):
        """Testing method cache_directory."""
        # Testing cache_directory with temp directory
        # Set the environmental variable for the configuration directory
        directory = tempfile.mkdtemp()
        os.environ['SWITCHMAP_CONFIGDIR'] = directory
        config_file = ('%s/test_config.yaml') % (directory)

        # Testing cache_directory with blank key_value(filepath)
        key = ''
        key_value = ''
        bad_config = ("""\
main:
    %s %s
""") % (key, key_value)
        bad_dict = yaml.load(bytes(bad_config, 'utf-8'))

        with open(config_file, 'w') as f_handle:
            yaml.dump(bad_dict, f_handle, default_flow_style=True)

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.cache_directory()

        # Cleanup files in temp directories
        _delete_files(directory)

    def test_agent_threads(self):
        """Testing method agent_threads."""
        # Testing agent_threads with good_dict
        # good key and key_value
        result = self.config.agent_threads()

        # We don't want a value that's too big that the CPU cannot cope
        expected = min(result, (multiprocessing.cpu_count() * 2) + 1)

        self.assertEqual(result, expected)
        self.assertEqual(result, self.good_dict['main']['agent_threads'])

    def test_polling_interval(self):
        """Testing method polling_interval."""
        # Testing polling_interval with good_dictionary
        # good key and key_value<commit_msg>Fixes #100<commit_after>#!/usr/bin/env python3
"""Test the configuration module."""

import multiprocessing
import os
import sys
import os.path
import unittest
import shutil
import random
import string
import tempfile
import yaml

# Try to create a working PYTHONPATH
TEST_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
SWITCHMAP_DIRECTORY = os.path.abspath(os.path.join(TEST_DIRECTORY, os.pardir))
ROOT_DIRECTORY = os.path.abspath(os.path.join(SWITCHMAP_DIRECTORY, os.pardir))
if TEST_DIRECTORY.endswith('/switchmap-ng/switchmap/test') is True:
    sys.path.append(ROOT_DIRECTORY)
else:
    print(
        'This script is not installed in the "switchmap-ng/bin" directory. '
        'Please fix.')
    sys.exit(2)

from switchmap.utils import configuration


class TestConfig(unittest.TestCase):
    """Checks all functions and methods."""

    #########################################################################
    # General object setup
    #########################################################################

    random_string = ''.join([random.choice(
        string.ascii_letters + string.digits) for n in range(9)])

    log_directory = tempfile.mkdtemp()
    cache_directory = tempfile.mkdtemp()
    good_config = ("""\
main:
    log_directory: %s
    cache_directory: %s
    agent_threads: 20
    bind_port: 3000
    hostnames:
    - 192.168.1.1
    - 192.168.1.2
    - 192.168.1.3
    - 192.168.1.4
    listen_address: 0.0.0.0
    log_level: debug
    polling_interval: 20
""") % (log_directory, cache_directory)

    # Convert good_config to dictionary
    good_dict = yaml.load(bytes(good_config, 'utf-8'))

    # Set the environmental variable for the configuration directory
    directory = tempfile.mkdtemp()
    os.environ['SWITCHMAP_CONFIGDIR'] = directory
    config_file = ('%s/test_config.yaml') % (directory)

    # Write good_config to file
    with open(config_file, 'w') as f_handle:
        yaml.dump(good_dict, f_handle, default_flow_style=True)

    # Create configuration object
    config = configuration.Config()

    @classmethod
    def tearDownClass(cls):
        """Post test cleanup."""
        os.rmdir(cls.log_directory)
        os.rmdir(cls.config.topology_directory())
        os.rmdir(cls.config.idle_directory())
        os.rmdir(cls.cache_directory)
        os.remove(cls.config_file)
        os.rmdir(cls.directory)

    def test_init(self):
        """Testing method init."""
        # Testing with non-existant directory
        directory = 'bogus'
        os.environ['SWITCHMAP_CONFIGDIR'] = directory
        with self.assertRaises(SystemExit):
            configuration.Config()

        # Testing with an empty directory
        empty_directory = tempfile.mkdtemp()
        os.environ['SWITCHMAP_CONFIGDIR'] = empty_directory
        with self.assertRaises(SystemExit):
            configuration.Config()

        # Write bad_config to file
        empty_config_file = ('%s/test_config.yaml') % (empty_directory)
        with open(empty_config_file, 'w') as f_handle:
            f_handle.write('')

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.log_file()

        # Cleanup files in temp directories
        _delete_files(directory)

    def test_log_file(self):
        """Testing method log_file."""
        # Test the log_file with a good_dict
        # good key and key_value
        result = self.config.log_file()
        self.assertEqual(
            result, ('%s/switchmap-ng.log') % (self.log_directory))

    def test_web_log_file(self):
        """Testing method web_log_file ."""
        # Testing web_log_file with a good dictionary.
        result = self.config.web_log_file()
        self.assertEqual(
            result, ('%s/switchmap-ng-api.log') % (self.log_directory))

    def test_log_level(self):
        """Testing method log_level."""
        # Tesing with a good_dictionary
        # good key and good key_value
        result = self.config.log_level()
        self.assertEqual(result, 'debug')
        self.assertEqual(result, self.good_dict['main']['log_level'])

        # Set the environmental variable for the configuration directory
        directory = tempfile.mkdtemp()
        os.environ['SWITCHMAP_CONFIGDIR'] = directory
        config_file = ('%s/test_config.yaml') % (directory)

        # Testing log_level with blank key and blank key_value
        key = ''
        key_value = ''
        bad_config = ("""\
main:
    %s %s
""") % (key, key_value)
        bad_dict = yaml.load(bytes(bad_config, 'utf-8'))

        # Write bad_config to file
        with open(config_file, 'w') as f_handle:
            yaml.dump(bad_dict, f_handle, default_flow_style=True)

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.log_level()

        # Testing log_level with good key and blank key_value
        key = 'log_level:'
        key_value = ''
        bad_config = ("""\
main:
    %s %s
""") % (key, key_value)
        bad_dict = yaml.load(bytes(bad_config, 'utf-8'))

        # Write bad_config to file
        with open(config_file, 'w') as f_handle:
            yaml.dump(bad_dict, f_handle, default_flow_style=True)

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.log_level()

        # Cleanup files in temp directories
        _delete_files(directory)

    def test_cache_directory(self):
        """Testing method cache_directory."""
        # Testing cache_directory with temp directory
        # Set the environmental variable for the configuration directory
        directory = tempfile.mkdtemp()
        os.environ['SWITCHMAP_CONFIGDIR'] = directory
        config_file = ('%s/test_config.yaml') % (directory)

        # Testing cache_directory with blank key_value(filepath)
        key = ''
        key_value = ''
        bad_config = ("""\
main:
    %s %s
""") % (key, key_value)
        bad_dict = yaml.load(bytes(bad_config, 'utf-8'))

        with open(config_file, 'w') as f_handle:
            yaml.dump(bad_dict, f_handle, default_flow_style=True)

        # Create configuration object
        config = configuration.Config()
        with self.assertRaises(SystemExit):
            config.cache_directory()

        # Cleanup files in temp directories
        _delete_files(directory)

    def test_agent_threads(self):
        """Testing method agent_threads."""
        # Testing agent_threads with good_dict
        # good key and key_value
        result = self.config.agent_threads()

        # We don't want a value that's too big that the CPU cannot cope
        expected = min(result, (multiprocessing.cpu_count() * 2) + 1)

        self.assertEqual(result, expected)

    def test_polling_interval(self):
        """Testing method polling_interval."""
        # Testing polling_interval with good_dictionary
        # good key and key_value
{'fraction_non_alphanumeric': 0.12371790337412464, 'fraction_numerical': 0.007710263846643559, 'mean_word_length': 2.754115772703133, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 10, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6026803', 'n_tokens_mistral': 4259, 'n_tokens_neox': 4045, 'n_words': 969}