<commit_before># -----------------------------------------------------------------------------
# Copyright (c) 2014--, The Qiita Development Team.
#
# Distributed under the terms of the BSD 3-clause License.
#
# The full license is in the file LICENSE, distributed with this software.
# -----------------------------------------------------------------------------

from unittest import TestCase, main
from os import environ, close, remove
from tempfile import mkstemp
from functools import partial
import warnings

from qiita_core.exceptions import MissingConfigSection
from qiita_core.configuration_manager import ConfigurationManager

from configparser import ConfigParser


class ConfigurationManagerTests(TestCase):
    def setUp(self):
        self.old_conf_fp = environ.get('QIITA_CONFIG_FP')
        fd, self.conf_fp = mkstemp(suffix='.txt')
        close(fd)
        with open(self.conf_fp, 'w') as f:
            f.write(CONF)
        environ['QIITA_CONFIG_FP'] = self.conf_fp

        self.conf = ConfigParser()
        with open(self.conf_fp, newline=None) as f:
            self.conf.readfp(f)

    def tearDown(self):
        if self.old_conf_fp is not None:
            environ['QIITA_CONFIG_FP'] = self.old_conf_fp
        else:
            del environ['QIITA_CONFIG_FP']
        remove(self.conf_fp)

    def test_init(self):
        obs = ConfigurationManager()
        # Main section
        self.assertEqual(obs.conf_fp, self.conf_fp)
        self.assertTrue(obs.test_environment)
        self.assertEqual(obs.base_data_dir, "/tmp/")
        self.assertEqual(obs.log_dir, "/tmp/")
        self.assertEqual(obs.base_url, "https://localhost")
        self.assertEqual(obs.max_upload_size, 100)
        self.assertTrue(obs.require_approval)
        self.assertEqual(obs.qiita_env, "source activate qiita")
        self.assertEqual(obs.private_launcher, 'qiita-private-launcher')
        self.assertEqual(obs.plugin_launcher, "qiita-plugin-launcher")
        self.assertEqual(obs.plugin_dir, "/tmp/")
        self.assertEqual(
            obs.valid_upload_extension,
            ["fastq", "fastq.gz", "txt", "tsv", "sff", "fna", "qual"])
        self.assertEqual(obs.certificate_file, "/tmp/server.cert")
        self.assertEqual(obs.cookie_secret, "SECRET")
        self.assertEqual(obs.key_file, "/tmp/server.key")

        # job_scheduler section
        self.assertEqual(obs.job_scheduler_owner, "<EMAIL>")
        self.assertEqual(obs.job_scheduler_poll_val, 15)
        self.assertEqual(obs.job_scheduler_dependency_q_cnt, 2)

        # Postgres section
        self.assertEqual(obs.user, "postgres")
        self.assertEqual(obs.admin_user, "postgres")
        self.assertEqual(obs.password, "<PASSWORD>")
        self.assertEqual(obs.admin_password, "<PASSWORD>")
        self.assertEqual(obs.database, "qiita_test")
        self.assertEqual(obs.host, "localhost")
        self.assertEqual(obs.port, 5432)

        # Redis section
        self.assertEqual(obs.redis_host, "localhost")
        self.assertEqual(obs.redis_password, "<PASSWORD>")
        self.assertEqual(obs.redis_db, 13)
        self.assertEqual(obs.redis_port, 6379)

        # SMTP section
        self.assertEqual(obs.smtp_host, "localhost")
        self.assertEqual(obs.smtp_port, 25)
        self.assertEqual(obs.smtp_user, "qiita")
        self.assertEqual(obs.smtp_password, "<PASSWORD>")
        self.assertFalse(obs.smtp_ssl)
        self.assertEqual(obs.smtp_email, "<EMAIL>")

        # EBI section
        self.assertEqual(obs.ebi_seq_xfer_user, "Webin-41528")
        self.assertEqual(obs.ebi_seq_xfer_pass, "<PASSWORD>")
        self.assertEqual(obs.ebi_seq_xfer_url, "webin.ebi.ac.uk")
        self.assertEqual(
            obs.ebi_dropbox_url,
            "https://www-test.ebi.ac.uk/ena/submit/drop-box/submit/")
        self.assertEqual(obs.ebi_center_name, "qiita-test")
        self.assertEqual(obs.ebi_organization_prefix, "example_organization")

        # VAMPS section
        self.assertEqual(obs.vamps_user, "user")
        self.assertEqual(obs.vamps_pass, "password")
        self.assertEqual(obs.vamps_url,
                         "https://vamps.mbl.edu/mobe_workshop/getfile.php")

        # Portal section
        self.assertEqual(obs.portal_fp, "/tmp/portal.cfg")
        self.assertEqual(obs.portal, "QIITA")
        self.assertEqual(obs.portal_dir, "/portal")

        # iframe section
        self.assertEqual(obs.iframe_qiimp, "https://localhost:8898/")

    def test_init_error(self):
        with open(self.conf_fp, 'w') as f:
            f.write("\n")

        with self.assertRaises(MissingConfigSection):
            ConfigurationManager()

    def test_get_main(self):
        obs = ConfigurationManager()

        conf_setter = partial(self.conf.set, 'main')
        conf_setter('COOKIE_SECRET', '')
        conf_setter('JWT_SECRET', '')
        conf_setter('BASE_DATA_DIR', '')
        conf_setter('PLUGIN_DIR', '')
        conf_setter('CERTIFICATE_FILE', '')
        conf_setter('KEY_FILE', '')
        conf_setter('QIITA_ENV', '')

        # Warning raised if No files will be allowed to be uploaded
        # Warning raised if no cookie_secret
        with warnings.catch_warnings(record=True) as warns:
            obs._get_main(self.conf)

            obs_warns = [str(w.message) for w in warns]
            exp_warns = ['Random cookie secret generated.',
                         'Random JWT secret generated.  Non Public Artifact '
                         'Download Links will expire upon system restart.']
            self.assertCountEqual(obs_warns, exp_warns)

        self.assertNotEqual(obs.cookie_secret, "SECRET")
        # Test default base_data_dir
        self.assertTrue(
            obs.base_data_dir.endswith("/qiita_db/support_files/test_data"))
        # Test default plugin dir
        self.assertTrue(obs.plugin_dir.endswith("/.qiita_plugins"))
        # Default certificate_file
        self.assertTrue(
            obs.certificate_file.endswith(
                "/qiita_core/support_files/server.crt"))
        # Default key_file
        self.assertTrue(
            obs.key_file.endswith("/qiita_core/support_files/server.key"))

        # BASE_DATA_DIR does not exist
        conf_setter('BASE_DATA_DIR', '/surprised/if/this/dir/exists')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        # WORKING_DIR does not exist
        conf_setter('BASE_DATA_DIR', '/tmp')
        conf_setter('WORKING_DIR', '/surprised/if/this/dir/exists')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        # PLUGIN_DIR does not exist
        conf_setter('WORKING_DIR', '/tmp')
        conf_setter('PLUGIN_DIR', '/surprised/if/this/dir/exists')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        # No files can be uploaded
        conf_setter('PLUGIN_DIR', '/tmp')
        conf_setter('VALID_UPLOAD_EXTENSION', '')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        self.assertEqual(obs.qiita_env, "")

    def test_get_job_scheduler(self):
        obs = ConfigurationManager()

        conf_setter = partial(self.conf.set, 'job_scheduler')
        conf_setter('JOB_SCHEDULER_JOB_OWNER', '')
        obs.get_job_scheduler(self.conf)
        self.assertIsNone(obs.job_scheduler_owner)

    def test_get_postgres(self):
        obs = ConfigurationManager()

        conf_setter = partial(self.conf.set, 'postgres')<commit_msg>conf.get_job_scheduler -> conf._get_job_scheduler<commit_after># -----------------------------------------------------------------------------
# Copyright (c) 2014--, The Qiita Development Team.
#
# Distributed under the terms of the BSD 3-clause License.
#
# The full license is in the file LICENSE, distributed with this software.
# -----------------------------------------------------------------------------

from unittest import TestCase, main
from os import environ, close, remove
from tempfile import mkstemp
from functools import partial
import warnings

from qiita_core.exceptions import MissingConfigSection
from qiita_core.configuration_manager import ConfigurationManager

from configparser import ConfigParser


class ConfigurationManagerTests(TestCase):
    def setUp(self):
        self.old_conf_fp = environ.get('QIITA_CONFIG_FP')
        fd, self.conf_fp = mkstemp(suffix='.txt')
        close(fd)
        with open(self.conf_fp, 'w') as f:
            f.write(CONF)
        environ['QIITA_CONFIG_FP'] = self.conf_fp

        self.conf = ConfigParser()
        with open(self.conf_fp, newline=None) as f:
            self.conf.readfp(f)

    def tearDown(self):
        if self.old_conf_fp is not None:
            environ['QIITA_CONFIG_FP'] = self.old_conf_fp
        else:
            del environ['QIITA_CONFIG_FP']
        remove(self.conf_fp)

    def test_init(self):
        obs = ConfigurationManager()
        # Main section
        self.assertEqual(obs.conf_fp, self.conf_fp)
        self.assertTrue(obs.test_environment)
        self.assertEqual(obs.base_data_dir, "/tmp/")
        self.assertEqual(obs.log_dir, "/tmp/")
        self.assertEqual(obs.base_url, "https://localhost")
        self.assertEqual(obs.max_upload_size, 100)
        self.assertTrue(obs.require_approval)
        self.assertEqual(obs.qiita_env, "source activate qiita")
        self.assertEqual(obs.private_launcher, 'qiita-private-launcher')
        self.assertEqual(obs.plugin_launcher, "qiita-plugin-launcher")
        self.assertEqual(obs.plugin_dir, "/tmp/")
        self.assertEqual(
            obs.valid_upload_extension,
            ["fastq", "fastq.gz", "txt", "tsv", "sff", "fna", "qual"])
        self.assertEqual(obs.certificate_file, "/tmp/server.cert")
        self.assertEqual(obs.cookie_secret, "SECRET")
        self.assertEqual(obs.key_file, "/tmp/server.key")

        # job_scheduler section
        self.assertEqual(obs.job_scheduler_owner, "<EMAIL>")
        self.assertEqual(obs.job_scheduler_poll_val, 15)
        self.assertEqual(obs.job_scheduler_dependency_q_cnt, 2)

        # Postgres section
        self.assertEqual(obs.user, "postgres")
        self.assertEqual(obs.admin_user, "postgres")
        self.assertEqual(obs.password, "<PASSWORD>")
        self.assertEqual(obs.admin_password, "<PASSWORD>")
        self.assertEqual(obs.database, "qiita_test")
        self.assertEqual(obs.host, "localhost")
        self.assertEqual(obs.port, 5432)

        # Redis section
        self.assertEqual(obs.redis_host, "localhost")
        self.assertEqual(obs.redis_password, "<PASSWORD>")
        self.assertEqual(obs.redis_db, 13)
        self.assertEqual(obs.redis_port, 6379)

        # SMTP section
        self.assertEqual(obs.smtp_host, "localhost")
        self.assertEqual(obs.smtp_port, 25)
        self.assertEqual(obs.smtp_user, "qiita")
        self.assertEqual(obs.smtp_password, "<PASSWORD>")
        self.assertFalse(obs.smtp_ssl)
        self.assertEqual(obs.smtp_email, "<EMAIL>")

        # EBI section
        self.assertEqual(obs.ebi_seq_xfer_user, "Webin-41528")
        self.assertEqual(obs.ebi_seq_xfer_pass, "<PASSWORD>")
        self.assertEqual(obs.ebi_seq_xfer_url, "webin.ebi.ac.uk")
        self.assertEqual(
            obs.ebi_dropbox_url,
            "https://www-test.ebi.ac.uk/ena/submit/drop-box/submit/")
        self.assertEqual(obs.ebi_center_name, "qiita-test")
        self.assertEqual(obs.ebi_organization_prefix, "example_organization")

        # VAMPS section
        self.assertEqual(obs.vamps_user, "user")
        self.assertEqual(obs.vamps_pass, "password")
        self.assertEqual(obs.vamps_url,
                         "https://vamps.mbl.edu/mobe_workshop/getfile.php")

        # Portal section
        self.assertEqual(obs.portal_fp, "/tmp/portal.cfg")
        self.assertEqual(obs.portal, "QIITA")
        self.assertEqual(obs.portal_dir, "/portal")

        # iframe section
        self.assertEqual(obs.iframe_qiimp, "https://localhost:8898/")

    def test_init_error(self):
        with open(self.conf_fp, 'w') as f:
            f.write("\n")

        with self.assertRaises(MissingConfigSection):
            ConfigurationManager()

    def test_get_main(self):
        obs = ConfigurationManager()

        conf_setter = partial(self.conf.set, 'main')
        conf_setter('COOKIE_SECRET', '')
        conf_setter('JWT_SECRET', '')
        conf_setter('BASE_DATA_DIR', '')
        conf_setter('PLUGIN_DIR', '')
        conf_setter('CERTIFICATE_FILE', '')
        conf_setter('KEY_FILE', '')
        conf_setter('QIITA_ENV', '')

        # Warning raised if No files will be allowed to be uploaded
        # Warning raised if no cookie_secret
        with warnings.catch_warnings(record=True) as warns:
            obs._get_main(self.conf)

            obs_warns = [str(w.message) for w in warns]
            exp_warns = ['Random cookie secret generated.',
                         'Random JWT secret generated.  Non Public Artifact '
                         'Download Links will expire upon system restart.']
            self.assertCountEqual(obs_warns, exp_warns)

        self.assertNotEqual(obs.cookie_secret, "SECRET")
        # Test default base_data_dir
        self.assertTrue(
            obs.base_data_dir.endswith("/qiita_db/support_files/test_data"))
        # Test default plugin dir
        self.assertTrue(obs.plugin_dir.endswith("/.qiita_plugins"))
        # Default certificate_file
        self.assertTrue(
            obs.certificate_file.endswith(
                "/qiita_core/support_files/server.crt"))
        # Default key_file
        self.assertTrue(
            obs.key_file.endswith("/qiita_core/support_files/server.key"))

        # BASE_DATA_DIR does not exist
        conf_setter('BASE_DATA_DIR', '/surprised/if/this/dir/exists')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        # WORKING_DIR does not exist
        conf_setter('BASE_DATA_DIR', '/tmp')
        conf_setter('WORKING_DIR', '/surprised/if/this/dir/exists')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        # PLUGIN_DIR does not exist
        conf_setter('WORKING_DIR', '/tmp')
        conf_setter('PLUGIN_DIR', '/surprised/if/this/dir/exists')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        # No files can be uploaded
        conf_setter('PLUGIN_DIR', '/tmp')
        conf_setter('VALID_UPLOAD_EXTENSION', '')
        with self.assertRaises(ValueError):
            obs._get_main(self.conf)

        self.assertEqual(obs.qiita_env, "")

    def test_get_job_scheduler(self):
        obs = ConfigurationManager()

        conf_setter = partial(self.conf.set, 'job_scheduler')
        conf_setter('JOB_SCHEDULER_JOB_OWNER', '')
        obs._get_job_scheduler(self.conf)
        self.assertIsNone(obs.job_scheduler_owner)

    def test_get_postgres(self):
        obs = ConfigurationManager()

        conf_setter = partial(self.conf.set, 'postgres')
{'fraction_non_alphanumeric': 0.13295296225038147, 'fraction_numerical': 0.004246002786439328, 'mean_word_length': 2.499071494893222, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 18, 'https://': 8, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2550605', 'n_tokens_mistral': 4743, 'n_tokens_neox': 4558, 'n_words': 760}