# -*- coding: utf-8 -*-
#
# Copyright (C) 2003-2023 Edgewall Software
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution. The terms
# are also available at https://trac.edgewall.org/wiki/TracLicense.
#
# This software consists of voluntary contributions made by many
# individuals. For the exact contribution history, see the revision
# history and logs, available at https://trac.edgewall.org/log/.

from configparser import RawConfigParser
from glob import glob
from subprocess import PIPE, Popen
import inspect
import io
import os
import sys
import textwrap
import unittest

from trac import db_default
from trac.admin.console import TracAdmin
from trac.admin.test import TracAdminTestCaseBase
from trac.api import IEnvironmentSetupParticipant, ISystemInfoProvider
from trac.attachment import Attachment
from trac.config import ConfigurationError, Option
from trac.core import Component, TracError, implements
from trac.db.api import DatabaseManager, get_column_names
from trac.env import Environment, EnvironmentAdmin, open_environment
from trac.test import EnvironmentStub, get_dburi, makeSuite, mkdtemp, rmtree
from trac.util import create_file, extract_zipfile, hex_entropy, read_file
from trac.util.compat import close_fds
from trac.util import create_file


class EnvironmentWithoutDataTestCase(unittest.TestCase):

    def setUp(self):
        self.env_path = mkdtemp()
        self.env = self._create_env(self.env_path)

    def tearDown(self):
        self.env.shutdown() # really closes the db connections
        rmtree(self.env.path)

    def _create_env(self, env_path):
        return Environment(env_path, create=True, default_data=False)

    def test_database_version(self):
        self.assertEqual(db_default.db_version,
                         self.env.database_version)
        self.assertEqual(db_default.db_version,
                         self.env.database_initial_version)

    def test_environment_data(self):
        with self.env.db_query as db:
            for table, columns, default_data in db_default.get_data(db):
                table_data = db("SELECT * FROM %s" % db.quote(table))
                self.assertEqual([], table_data)

    def test_create_nonexistent_parent_directory(self):
        """TracError raised creating Environment at non-existent path."""
        base_dir = os.path.join(self.env_path, 'non-existent')
        env_path = os.path.join(base_dir, 'non-existent')
        with self.assertRaises(TracError) as cm:
            self._create_env(env_path)
        self.assertEqual("Base directory '%s' does not exist. Please create "
                         "it and retry." % base_dir, str(cm.exception))

    def test_create_in_existing_environment(self):
        """TracError raised creating Environment in existing Environment."""
        with self.assertRaises(TracError) as cm:
            self._create_env(self.env_path)
        self.assertEqual("Directory exists and is not empty.",
                         str(cm.exception))


class EnvironmentTestCase(unittest.TestCase):

    stdout = None
    stderr = None
    devnull = None

    @classmethod
    def setUpClass(cls):
        cls.stdout = sys.stdout
        cls.stderr = sys.stderr
        cls.devnull = io.open(os.devnull, 'w', encoding='utf-8')
        sys.stdout = sys.stderr = cls.devnull

    @classmethod
    def tearDownClass(cls):
        cls.devnull.close()
        sys.stdout = cls.stdout
        sys.stderr = cls.stderr

    def setUp(self):
        self.env_path = mkdtemp()
        self.env = Environment(self.env_path, create=True)

    def tearDown(self):
        self.env.shutdown() # really closes the db connections
        rmtree(self.env.path)

    def test_invalid_version_raises_trac_error(self):
        """TracError raised when environment version is invalid."""
        version = 'Trac Environment Version 0'
        create_file(os.path.join(self.env.path, 'VERSION'), version + '\n')

        with self.assertRaises(TracError) as cm:
            Environment(self.env.path)
        self.assertEqual(
            "Unknown Trac environment type 'Trac Environment Version 0'",
            str(cm.exception))

    def test_version_file_empty(self):
        """TracError raised when environment version is empty."""
        create_file(os.path.join(self.env.path, 'VERSION'), '')
        with self.assertRaises(TracError) as cm:
            Environment(self.env.path)
        self.assertEqual("Unknown Trac environment type ''", str(cm.exception))

    def test_version_file_not_found(self):
        """TracError raised when environment version file not found."""
        version_file = os.path.join(self.env.path, 'VERSION')
        os.remove(version_file)
        with self.assertRaises(TracError) as cm:
            Environment(self.env.path)
        self.assertEqual(
            "No Trac environment found at %s\n"
            "FileNotFoundError: [Errno 2] No such file or directory: '%s'"
            % (self.env.path, version_file),
            str(cm.exception))

    def test_missing_config_file_raises_trac_error(self):
        """TracError is raised when config file is missing."""
        os.remove(self.env.config_file_path)
        self.assertRaises(TracError, Environment, self.env.path)

    def test_log_format(self):
        """Configure the log_format and log to a file at WARNING level."""
        self.env.config.set('logging', 'log_type', 'file')
        self.env.config.set('logging', 'log_level', 'WARNING')
        self.env.config.set('logging', 'log_format',
                            'Trac[$(module)s] $(project)s: $(message)s')
        self.env.config.save()
        self.env.shutdown()
        self.env = Environment(self.env_path)  # Reload environment

        self.env.log.warning("The warning message")

        with open(self.env.log_file_path, encoding='utf-8') as f:
            log = f.readlines()
        self.assertEqual("Trac[env] My Project: The warning message\n",
                         log[-1])

    def test_invalid_log_level_raises_exception(self):
        self.env.config.set('logging', 'log_level', 'invalid')
        self.env.config.save()

        self.assertEqual('invalid',
                         self.env.config.get('logging', 'log_level'))
        self.assertRaises(ConfigurationError, open_environment,
                          self.env.path, True)

    def test_invalid_log_type_raises_exception(self):
        self.env.config.set('logging', 'log_type', 'invalid')
        self.env.config.save()

        self.assertEqual('invalid',
                         self.env.config.get('logging', 'log_type'))
        self.assertRaises(ConfigurationError, open_environment,
                          self.env.path, True)


class EnvironmentDataTestCase(unittest.TestCase):
    """Tests for environment data.

    The test cases don't modify the environment so the Environment can
    be reused to make the tests run faster.
    """

    env = None
    stdout = None
    stderr = None
    devnull = None

    @classmethod
    def setUpClass(cls):
        cls.stdout = sys.stdout
        cls.stderr = sys.stderr
        cls.devnull = io.open(os.devnull, 'w', encoding='utf-8')
        sys.stdout = sys.stderr = cls.devnull
        cls.env = Environment(mkdtemp(), create=True)

    @classmethod
    def tearDownClass(cls):
        cls.env.shutdown()
        rmtree(cls.env.path)
        cls.devnull.close()
        sys.stdout = cls.stdout
        sys.stderr = cls.stderr

    def test_database_version(self):
        self.assertEqual(db_default.db_version,
                         self.env.database_version)
        self.assertEqual(db_default.db_version,
                         self.env.database_initial_version)

    def test_environment_data(self):
        with self.env.db_query as db:
            for table, columns, default_data in db_default.get_data(db):
                table_data = db("""
                    SELECT %s FROM %s
                    """ % (','.join(columns), db.quote(table)))
                self.assertEqual(list(default_data), table_data)

    def test_is_component_enabled(self):
        self.assertTrue(Environment.required)
        self.assertTrue(self.env.is_component_enabled(Environment))

    def test_dumped_values_in_tracini(self):
        parser = RawConfigParser()
        filename = self.env.config.filename
        self.assertEqual([filename], parser.read(filename, 'utf-8'))
        self.assertEqual('#cc0,#0c0,#0cc,#00c,#c0c,#c00',
                         parser.get('revisionlog', 'graph_colors'))
        self.assertEqual('disabled', parser.get('trac', 'secure_cookies'))

    def test_dumped_values_in_tracini_sample(self):
        parser = RawConfigParser()
        filename = self.env.config.filename + '.sample'
        self.assertEqual([filename], parser.read(filename, 'utf-8'))
        self.assertEqual('#cc0,#0c0,#0cc,#00c,#c0c,#c00',
                         parser.get('revisionlog', 'graph_colors'))
        self.assertEqual('disabled', parser.get('trac', 'secure_cookies'))
        self.assertTrue(parser.has_option('logging', 'log_format'))
        self.assertEqual('', parser.get('logging', 'log_format'))

    def test_tracguide_is_readonly(self):
        from trac.wiki.model import WikiPage
        for name in ('TracGuide', 'TracInstall', 'TracUpgrade'):
            self.assertTrue(WikiPage(self.env, name).readonly)
        for name in ('SandBox', 'WikiStart', 'InterMapTxt'):
            self.assertFalse(WikiPage(self.env, name).readonly)


class EnvironmentAttributesTestCase(unittest.TestCase):
    """Tests for attributes which don't require a real environment
    on disk, and therefore can be executed against an `EnvironmentStub`
    object (faster execution).
    """

    def setUp(self):
        self.env = EnvironmentStub()
        self.env.config.set('trac', 'base_url',
                            'https://trac.edgewall.org/some/path')

    def test_is_component_enabled(self):
        self.assertFalse(EnvironmentStub.required)
        self.assertIsNone(self.env.is_component_enabled(EnvironmentStub))

    def test_db_exc(self):
        db_exc = self.env.db_exc
        self.assertTrue(hasattr(db_exc, 'IntegrityError'))
        self.assertIs(db_exc, self.env.db_exc)

    def test_abs_href(self):
        abs_href = self.env.abs_href
        self.assertEqual('https://trac.edgewall.org/some/path', abs_href())
        self.assertIs(abs_href, self.env.abs_href)

    def test_href(self):
        href = self.env.href
        self.assertEqual('/some/path', href())
        self.assertIs(href, self.env.href)

    def test_log_file_path_is_relative_path(self):
        log_file_path = self.env.log_file_path
        self.assertEqual(os.path.join(self.env.path, 'log', 'trac.log'),
                         log_file_path)
        self.assertIs(log_file_path, self.env.log_file_path)

    def test_log_file_path_is_absolute_path(self):
        log_file = os.path.join(self.env.path, 'trac.log')
        self.env.config.set('logging', 'log_file', log_file)
        self.assertEqual(log_file, self.env.log_file_path)

    def test_log_level_not_case_sensitive(self):
        """[logging] log_level is not case-sensitive."""
        self.env.config.set('logging', 'log_level', 'warning')
        self.env.config.save()

        self.assertEqual('warning',
                         self.env.config.get('logging', 'log_level'))
        self.assertEqual('WARNING', self.env.log_level)

    def test_log_type_not_case_sensitive(self):
        """[logging] log_type is not case-sensitive."""
        self.env.config.set('logging', 'log_type', 'File')
        self.env.config.save()

        self.assertEqual('File',
                         self.env.config.get('logging', 'log_type'))
        self.assertEqual('file', self.env.log_type)


class EnvironmentUpgradeTestCase(unittest.TestCase):

    def setUp(self):
        self.env = EnvironmentStub()

    def tearDown(self):
        self.env.reset_db()

    def test_multiple_upgrade_participants(self):

        class Participant1(Component):
            implements(IEnvironmentSetupParticipant)
            def environment_created(self):
                pass
            def environment_needs_upgrade(self):
                return True
            def upgrade_environment(self):
                insert_value('value1', 1)

        class Participant2(Component):
            implements(IEnvironmentSetupParticipant)
            def environment_created(self):
                pass
            def environment_needs_upgrade(self):
                return True
            def upgrade_environment(self):
                insert_value('value2', 2)

        def insert_value(name, value):
            with self.env.db_transaction as db:
                db("""
                    INSERT INTO {0} (name, value) VALUES (%s, %s)
                    """.format(db.quote('system')), (name, value))

        def select_value(name):
            with self.env.db_query as db:
                for value, in db("""
                        SELECT value FROM {0} WHERE name=%s
                        """.format(db.quote('system')), (name,)):
                    return value

        self.env.enable_component(Participant1)
        self.env.enable_component(Participant2)

        self.assertTrue(self.env.needs_upgrade())
        self.assertTrue(self.env.upgrade())
        self.assertEqual('1', select_value('value1'))
        self.assertEqual('2', select_value('value2'))

    def test_upgrade_environment(self):
        """EnvironmentSetupParticipants are called only if
        environment_needs_upgrade returns True for the participant.
        """

        class SetupParticipantA(Component):
            implements(IEnvironmentSetupParticipant)

            called = False

            def environment_created(self):
                pass

            def environment_needs_upgrade(self):
                return True

            def upgrade_environment(self):
                self.called = True

        class SetupParticipantB(Component):
            implements(IEnvironmentSetupParticipant)

            called = False

            def environment_created(self):
                pass

            def environment_needs_upgrade(self):
                return False

            def upgrade_environment(self):
                self.called = True

        self.env.enable_component(SetupParticipantA)
        self.env.enable_component(SetupParticipantB)
        participant_a = SetupParticipantA(self.env)
        participant_b = SetupParticipantB(self.env)

        self.assertTrue(self.env.needs_upgrade())
        self.env.upgrade()
        self.assertTrue(participant_a.called)
        self.assertFalse(participant_b.called)


class KnownUsersTestCase(unittest.TestCase):

    def setUp(self):
        self.env = EnvironmentStub()
        users = [
            ('123', None, 'a@example.com', 0),
            ('jane', 'Jane', None, 1),
            ('joe', None, 'joe@example.com', 1),
            ('tom', 'Tom', 'tom@example.com', 1)
        ]
        self.env.insert_users(users)
        self.expected = [user[:3] for user in users if user[3] == 1]

    def tearDown(self):
        self.env.reset_db()

    def test_get_known_users_as_list_of_tuples(self):
        users = list(self.env.get_known_users())

        i = 0
        for i, user in enumerate(users):
            self.assertEqual(self.expected[i], user)
        else:
            self.assertEqual(2, i)

    def test_get_known_users_as_dict(self):
        users = self.env.get_known_users(as_dict=True)

        self.assertEqual(3, len(users))
        for exp in self.expected:
            self.assertEqual(exp[1:], users[exp[0]])

    def test_get_known_users_is_cached(self):
        self.env.get_known_users()
        self.env.get_known_users(as_dict=True)
        self.env.insert_users([('user4', None, None)])

        users_list = list(self.env.get_known_users())
        users_dict = self.env.get_known_users(as_dict=True)

        i = 0
        for i, user in enumerate(users_list):
            self.assertEqual(self.expected[i], user)
            self.assertIn(self.expected[i][0], users_dict)
        else:
            self.assertEqual(2, i)
            self.assertEqual(3, len(users_dict))

    def test_invalidate_known_users_cache(self):
        self.env.get_known_users()
        self.env.get_known_users(as_dict=True)
        user = ('user4', 'User Four', 'user4@example.net')
        self.env.insert_users([user])
        self.expected.append(user[:3])

        self.env.invalidate_known_users_cache()
        users_list = self.env.get_known_users()
        users_dict = self.env.get_known_users(as_dict=True)

        i = 0
        for i, user in enumerate(users_list):
            self.assertEqual(self.expected[i], user)
            self.assertIn(self.expected[i][0], users_dict)
        else:
            self.assertEqual(3, i)
            self.assertEqual(4, len(users_dict))


class SystemInfoTestCase(unittest.TestCase):

    def setUp(self):
        self.env = EnvironmentStub()

    def tearDown(self):
        self.env.reset_db()

    def test_database_backend_version(self):
        """Database backend is returned in system_info."""
        # closes the pooled connections to use `DatabaseManager.shutdown()`
        # instead of `env.shutdown()` to avoid `log.shutdown()`
        DatabaseManager(self.env).shutdown()
        info_before = self.env.system_info
        self.env.db_query("SELECT 42")  # just connects database
        info_after = self.env.system_info

        def get_info(system_info, name):
            for info in system_info:
                if info[0] == name:
                    return info[1]
            self.fail('Missing %r' % name)

        if self.env.dburi.startswith('mysql'):
            self.assertRegex(get_info(info_before, 'MySQL'),
                             r'^server: \(not-connected\), '
                             r'client: "\d+(\.\d+)+([-.].+)?", '
                             r'thread-safe: True$')
            self.assertRegex(get_info(info_after, 'MySQL'),
                             r'^server: "\d+(\.\d+)+([-.].+)?", '
                             r'client: "\d+(\.\d+)+([-.].+)?", '
                             r'thread-safe: True$')
            self.assertRegex(get_info(info_before, 'pymysql'),
                             r'^\d+(\.\d+)+$')
            self.assertRegex(get_info(info_after, 'pymysql'),
                             r'^\d+(\.\d+)+$')
        elif self.env.dburi.startswith('postgres'):
            self.assertRegex(get_info(info_before, 'PostgreSQL'),
                             r'^server: \(not-connected\), '
                             r'client: (\d+(\.\d+)+|\(unknown\))$')
            self.assertRegex(get_info(info_after, 'PostgreSQL'),
                             r'^server: \d+(\.\d+)+, '
                             r'client: (\d+(\.\d+)+|\(unknown\))$')
            self.assertRegex(get_info(info_before, 'psycopg2'),
                             r'^\d+(\.\d+)+$')
            self.assertRegex(get_info(info_after, 'psycopg2'),
                             r'^\d+(\.\d+)+$')
        elif self.env.dburi.startswith('sqlite'):
            self.assertEqual(info_before, info_after)
            self.assertRegex(get_info(info_before, 'SQLite'),
                             r'^\d+(\.\d+)+$')
            self.assertRegex(get_info(info_before, 'pysqlite'),
                             r'^\d+(\.\d+)+$')
        else:
            self.fail("Unknown value for dburi %s" % self.env.dburi)


class ConvertDatabaseTestCase(unittest.TestCase):

    stdout = None
    stderr = None

    @classmethod
    def setUpClass(cls):
        cls.stdout = sys.stdout
        cls.stderr = sys.stderr
        cls.devnull = io.open(os.devnull, 'w', encoding='utf-8')
        sys.stdout = sys.stderr = cls.devnull

    @classmethod
    def tearDownClass(cls):
        cls.devnull.close()
        sys.stdout = cls.stdout
        sys.stderr = cls.stderr

    def setUp(self):
        self.path = mkdtemp()
        self.src_path = os.path.join(self.path, 'src')
        self.dst_path = os.path.join(self.path, 'dst')
        self.src_env = None
        self.dst_env = None
        self._destroy_db()

    def tearDown(self):
        if self.src_env:
            self.src_env.shutdown()
        if self.dst_env:
            self.dst_env.shutdown()
        rmtree(self.path)

    def _create_env(self, path, dburi):
        env = Environment(path, True,
                          [('trac', 'database', dburi),
                           ('trac', 'base_url', 'http://localhost/'),
                           ('project', 'name', 'Pŕójéćŧ Ńáḿé')])
        dbm = DatabaseManager(env)
        dbm.set_database_version(21, 'initial_database_version')
        att = Attachment(env, 'wiki', 'WikiStart')
        att.insert('filename.txt', io.BytesIO(b'test'), 4)
        env.shutdown()

    if 'destroying' in inspect.signature(EnvironmentStub.__init__).parameters:
        def _destroy_db(self):
            EnvironmentStub(path=self.path, destroying=True).destroy_db()
    else:
        def _destroy_db(self):
            EnvironmentStub(path=self.path).destroy_db()

    def _get_all_records(self, env):
        def primary(row, columns):
            if len(columns) == 1:
                return row[columns[0]]
            else:
                return tuple(row[col] for col in columns)

        records = {}
        with env.db_query as db:
            cursor = db.cursor()
            for table in db_default.schema:
                primary_cols = ','.join(db.quote(col) for col in table.key)
                query = "SELECT * FROM %s ORDER BY %s" \
                        % (db.quote(table.name), primary_cols)
                cursor.execute(query)
                columns = get_column_names(cursor)
                rows = {}
                for row in cursor:
                    row = dict(zip(columns, row))
                    rows[primary(row, table.key)] = row
                records[table.name] = rows
        return records

    def _generate_module_name(self):
        return 'trac_convert_db_' + hex_entropy(16)

    def _build_egg_file(self):
        module_name = self._generate_module_name()
        plugin_src = os.path.join(self.path, 'plugin_src')
        os.mkdir(plugin_src)
        os.mkdir(os.path.join(plugin_src, module_name))
        create_file(os.path.join(plugin_src, 'setup.py'),
                    _setup_py % {'name': module_name})
        create_file(os.path.join(plugin_src, module_name, '__init__.py'),
                    _plugin_py)
        proc = Popen((sys.executable, 'setup.py', 'bdist_egg'), cwd=plugin_src,
                     stdin=PIPE, stdout=PIPE, stderr=PIPE, close_fds=close_fds)
        proc.communicate(input='')
        for f in (proc.stdin, proc.stdout, proc.stderr):
            f.close()
        for filename in glob(os.path.join(plugin_src, 'dist', '*-*.egg')):
            return filename

    def _convert_db(self, env, dburi, path):
        EnvironmentAdmin(env)._do_convert_db(dburi, path)

    def _convert_db_inplace(self, env, dburi):
        self._convert_db(env, dburi, None)

    def _compare_records(self, expected, actual):
        self.assertEqual(list(expected), list(actual))
        for table in db_default.schema:
            name = table.name
            if name == 'report':
                self.assertEqual(list(expected[name]), list(actual[name]))
            else:
                self.assertEqual(expected[name], actual[name])

    def _get_options(self, env):
        config = env.config
        return [(section, name, self._option_dumps(section, name, value))
                for section in sorted(config.sections())
                for name, value in sorted(config.options(section))
                if (section, name) != ('trac', 'database')]

    def _option_dumps(self, section, name, value):
        try:
            option = Option.registry[(section, name)]
        except KeyError:
            pass
        else:
            value = option.dumps(value)
        return value

    def test_convert_from_sqlite_to_env(self):
        self._create_env(self.src_path, 'sqlite:db/trac.db')
        dburi = get_dburi()
        if dburi == 'sqlite::memory:':
            dburi = 'sqlite:db/trac.db'

        self.src_env = Environment(self.src_path)
        src_options = self._get_options(self.src_env)
        src_records = self._get_all_records(self.src_env)
        self._convert_db(self.src_env, dburi, self.dst_path)
        self.dst_env = Environment(self.dst_path)
        dst_options = self._get_options(self.dst_env)
        dst_records = self._get_all_records(self.dst_env)
        self.assertEqual({'name': 'initial_database_version', 'value': '21'},
                         dst_records['system']['initial_database_version'])
        self._compare_records(src_records, dst_records)
        self.assertEqual(src_options, dst_options)
        att = Attachment(self.dst_env, 'wiki', 'WikiStart', 'filename.txt')
        self.assertEqual('test', read_file(att.path))

    def test_convert_from_sqlite_inplace(self):
        self._create_env(self.src_path, 'sqlite:db/trac.db')
        dburi = get_dburi()
        if dburi in ('sqlite::memory:', 'sqlite:db/trac.db'):
            dburi = 'sqlite:db/trac-convert.db'

        self.src_env = Environment(self.src_path)
        src_options = self._get_options(self.src_env)
        src_records = self._get_all_records(self.src_env)
        self._convert_db_inplace(self.src_env, dburi)
        self.src_env.shutdown()
        self.src_env = Environment(self.src_path)
        dst_options = self._get_options(self.src_env)
        dst_records = self._get_all_records(self.src_env)
        self.assertEqual({'name': 'initial_database_version', 'value': '21'},
                         dst_records['system']['initial_database_version'])
        self._compare_records(src_records, dst_records)
        self.assertEqual(src_options, dst_options)

    def test_convert_to_sqlite_env(self):
        dburi = get_dburi()
        if dburi == 'sqlite::memory:':
            dburi = 'sqlite:db/trac.db'
        self._create_env(self.src_path, dburi)

        self.src_env = Environment(self.src_path)
        src_options = self._get_options(self.src_env)
        src_records = self._get_all_records(self.src_env)
        self._convert_db(self.src_env, 'sqlite:db/trac.db', self.dst_path)
        self.dst_env = Environment(self.dst_path)
        dst_options = self._get_options(self.dst_env)
        dst_records = self._get_all_records(self.dst_env)
        self.assertEqual({'name': 'initial_database_version', 'value': '21'},
                         dst_records['system']['initial_database_version'])
        self._compare_records(src_records, dst_records)
        self.assertEqual(src_options, dst_options)
        att = Attachment(self.dst_env, 'wiki', 'WikiStart', 'filename.txt')
        self.assertEqual('test', read_file(att.path))

    def test_convert_to_sqlite_inplace(self):
        dburi = get_dburi()
        if dburi in ('sqlite::memory:', 'sqlite:db/trac.db'):
            dburi = 'sqlite:db/trac-convert.db'
        self._create_env(self.src_path, dburi)

        self.src_env = Environment(self.src_path)
        src_options = self._get_options(self.src_env)
        src_records = self._get_all_records(self.src_env)
        self._convert_db_inplace(self.src_env, 'sqlite:db/trac.db')
        self.src_env.shutdown()
        self.src_env = Environment(self.src_path)
        dst_options = self._get_options(self.src_env)
        dst_records = self._get_all_records(self.src_env)
        self.assertEqual({'name': 'initial_database_version', 'value': '21'},
                         dst_records['system']['initial_database_version'])
        self._compare_records(src_records, dst_records)
        self.assertEqual(src_options, dst_options)

    def _test_convert_with_plugin_to_sqlite_env(self):
        self.src_env = Environment(self.src_path)
        self.assertTrue(self.src_env.needs_upgrade())
        self.src_env.upgrade()
        self.assertFalse(self.src_env.needs_upgrade())
        src_options = self._get_options(self.src_env)
        src_records = self._get_all_records(self.src_env)

        self._convert_db(self.src_env, 'sqlite:db/trac.db', self.dst_path)
        self.dst_env = Environment(self.dst_path)
        self.assertFalse(self.dst_env.needs_upgrade())
        self.assertFalse(os.path.exists(os.path.join(self.dst_env.log_dir,
                                                     'created')))
        self.assertTrue(os.path.exists(os.path.join(self.dst_env.log_dir,
                                                    'upgraded')))
        dst_options = self._get_options(self.dst_env)
        dst_records = self._get_all_records(self.dst_env)
        self.assertEqual({'name': 'initial_database_version', 'value': '21'},
                         dst_records['system']['initial_database_version'])
        self._compare_records(src_records, dst_records)
        self.assertEqual(src_options, dst_options)
        att = Attachment(self.dst_env, 'wiki', 'WikiStart', 'filename.txt')
        self.assertEqual('test', read_file(att.path))

    def test_convert_with_plugin_py_to_sqlite_env(self):
        dburi = get_dburi()
        if dburi == 'sqlite::memory:':
            dburi = 'sqlite:db/trac.db'
        self._create_env(self.src_path, dburi)
        plugin_name = self._generate_module_name() + '.py'
        create_file(os.path.join(self.src_path, 'plugins', plugin_name),
                    _plugin_py)
        self._test_convert_with_plugin_to_sqlite_env()

    def test_convert_with_plugin_egg_to_sqlite_env(self):
        dburi = get_dburi()
        if dburi == 'sqlite::memory:':
            dburi = 'sqlite:db/trac.db'
        self._create_env(self.src_path, dburi)
        extract_zipfile(self._build_egg_file(),
                        os.path.join(self.src_path, 'plugins',
                                     'trac_convert_db_test.egg'))
        self._test_convert_with_plugin_to_sqlite_env()


_setup_py = """\
from setuptools import setup, find_packages

setup(
    name = '%(name)s',
    version = '0.1.0',
    description = '',
    license = '',
    install_requires = ['Trac'],
    packages = find_packages(exclude=['*.tests*']),
    entry_points = {'trac.plugins': ['%(name)s = %(name)s']})
"""


_plugin_py = """\
import os.path
from trac.core import Component, implements
from trac.env import IEnvironmentSetupParticipant
from trac.util import create_file

class Setup(Component):

    implements(IEnvironmentSetupParticipant)

    def __init__(self):
        self._created_file = os.path.join(self.env.path, 'log', 'created')
        self._upgraded_file = os.path.join(self.env.path, 'log', 'upgraded')

    def environment_created(self):
        create_file(self._created_file)

    def environment_needs_upgrade(self):
        return not os.path.exists(self._upgraded_file)

    def upgrade_environment(self):
        create_file(self._upgraded_file)
"""


class SystemInfoProviderTestCase(unittest.TestCase):

    system_info_providers = []

    @classmethod
    def setUpClass(cls):
        class SystemInfoProvider1(Component):
            implements(ISystemInfoProvider)

            def get_system_info(self):
                yield 'pkg1', 1.0
                yield 'pkg2', 2.0

        class SystemInfoProvider2(Component):
            implements(ISystemInfoProvider)

            def get_system_info(self):
                yield 'pkg1', 1.0

        cls.system_info_providers = [SystemInfoProvider1, SystemInfoProvider2]

    @classmethod
    def tearDownClass(cls):
        from trac.core import ComponentMeta
        for component in cls.system_info_providers:
            ComponentMeta.deregister(component)

    def setUp(self):
        self.env = EnvironmentStub(enable=self.system_info_providers)

    def test_system_info_property(self):
        """The system_info property returns a list of all tuples
        generated by ISystemInfoProvider implementations.
        """
        system_info = self.env.system_info
        self.assertIn(('pkg1', 1.0), system_info)
        self.assertIn(('pkg2', 2.0), system_info)

    def test_duplicate_entries_are_removed(self):
        """Duplicate entries are removed."""
        system_info = self.env.system_info
        self.assertIn(('pkg1', 1.0), system_info)
        self.assertEqual(len(system_info), len(set(system_info)))


class TracAdminDeployTestCase(TracAdminTestCaseBase):
    """Tests for the trac-admin deploy command."""

    stdout = None
    stderr = None
    devnull = None

    @classmethod
    def setUpClass(cls):
        cls.stdout = sys.stdout
        cls.stderr = sys.stderr
        cls.devnull = io.open(os.devnull, 'w', encoding='utf-8')
        sys.stdout = sys.stderr = cls.devnull

    @classmethod
    def tearDownClass(cls):
        cls.devnull.close()
        sys.stdout = cls.stdout
        sys.stderr = cls.stderr

    def setUp(self):
        self.env = Environment(path=mkdtemp(), create=True)
        self.admin = TracAdmin(self.env.path)
        self.admin.env_set('', self.env)

    def tearDown(self):
        self.env.shutdown()
        rmtree(self.env.path)

    def test_deploy(self):
        target = os.path.join(self.env.path, 'www')
        shebang = ('#!' + sys.executable).encode('utf-8')
        posix = os.name == 'posix'
        if posix:
            saved = os.umask(0o27)
        try:
            rv, output = self.execute('deploy %s' % target)
        finally:
            if posix:
                os.umask(saved)

        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output)
        self.assertTrue(os.path.exists(os.path.join(target, 'cgi-bin')))
        self.assertTrue(os.path.exists(os.path.join(target, 'htdocs')))
        self.assertTrue(os.path.exists(os.path.join(target, 'htdocs',
                                                    'common')))
        self.assertTrue(os.path.exists(os.path.join(target, 'htdocs',
                                                    'site')))
        self.assertTrue(os.path.isfile(os.path.join(
            target, 'htdocs', 'common', 'js', 'trac.js')))
        self.assertTrue(os.path.isfile(os.path.join(
            target, 'htdocs', 'common', 'css', 'trac.css')))
        for ext in ('cgi', 'fcgi', 'wsgi'):
            filename = os.path.join(target, 'cgi-bin', 'trac.%s' % ext)
            content = read_file(filename, 'rb')
            self.assertIn(shebang, content)
            self.assertEqual(0, content.index(shebang))
            self.assertIn(repr(self.env.path).encode('ascii'), content)
            if posix:
                self.assertEqual(0o750, os.stat(filename).st_mode & 0o7777)

    def test_deploy_to_invalid_target_raises_error(self):
        """Running deploy with target directory equal to or below the source
        directory raises AdminCommandError.
        """
        rv, output = self.execute('deploy %s' % self.env.htdocs_dir)

        self.assertEqual(2, rv, output)
        self.assertExpectedResult(output)


class TracAdminInitenvTestCase(TracAdminTestCaseBase):

    def setUp(self):
        self.parent_dir = mkdtemp()
        self.env_path = os.path.join(self.parent_dir, 'trac')
        self.admin = TracAdmin(self.env_path)

    def tearDown(self):
        if os.path.isfile(os.path.join(self.env_path, 'VERSION')):
            self.admin.env.shutdown()
        rmtree(self.parent_dir)

    def test_default_values(self):
        tracini_path = os.path.join(self.env_path, 'conf', 'trac.ini')
        rv, output = self.execute('initenv project1 sqlite:db/sqlite.db')

        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, {
            'env_path': self.env_path,
            'tracini_path': tracini_path
        })

        rv, output = self.execute('component list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_component_list')
        rv, output = self.execute('milestone list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_milestone_list')
        rv, output = self.execute('version list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_version_list')
        rv, output = self.execute('resolution list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_resolution_list')
        rv, output = self.execute('priority list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_priority_list')
        rv, output = self.execute('ticket_type list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_ticket_type_list')
        rv, output = self.execute('permission list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_permission_list')

    def test_no_default_data_argument(self):
        tracini_path = os.path.join(self.env_path, 'conf', 'trac.ini')
        rv, output = self.execute('initenv project1 sqlite:db/sqlite.db '
                                  '--no-default-data')

        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, {
            'env_path': self.env_path,
            'tracini_path': tracini_path
        })

        rv, output = self.execute('component list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_component_list')
        rv, output = self.execute('milestone list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_milestone_list')
        rv, output = self.execute('version list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_version_list')
        rv, output = self.execute('resolution list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_resolution_list')
        rv, output = self.execute('priority list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_priority_list')
        rv, output = self.execute('ticket_type list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_ticket_type_list')
        rv, output = self.execute('permission list')
        self.assertEqual(0, rv, output)
        self.assertExpectedResult(output, suffix='_permission_list')

    def test_config_argument(self):
        """Options contained in file specified by the --config argument
        are written to trac.ini.
        """
        config_file = os.path.join(self.parent_dir, 'config.ini')
        create_file(config_file, textwrap.dedent("""\
            [the-plugin]
            option_a = 1
            option_b = 2
            [components]
            the_plugin.* = enabled
            [project]
            name = project2
            """))
        rv, output = self.execute('initenv project1 sqlite:db/sqlite.db '
                                   '--config=%s' % config_file)
        env = Environment(self.env_path)
        cfile = env.config.parser

        self.assertEqual(0, rv, output)
        self.assertEqual('1', cfile.get('the-plugin', 'option_a'))
        self.assertEqual('2', cfile.get('the-plugin', 'option_b'))
        self.assertEqual('enabled', cfile.get('components', 'the_plugin.*'))
        self.assertEqual('project1', cfile.get('project', 'name'))
        self.assertEqual('sqlite:db/sqlite.db', cfile.get('trac', 'database'))
        for (section, name), option in \
                Option.get_registry(env.compmgr).items():
            if (section, name) not in \
                    (('trac', 'database'), ('project', 'name')):
                self.assertEqual(option.default, cfile.get(section, name))

    def test_config_argument_has_invalid_path(self):
        """Exception is raised when --config argument is an invalid path."""
        config_file = os.path.join(self.parent_dir, 'config.ini')
        rv, output = self.execute('initenv project1 sqlite:db/sqlite.db '
                                   '--config=%s' % config_file)

        self.assertEqual(2, rv, output)
        self.assertExpectedResult(output, {
            'env_path': self.env_path,
            'config_file': config_file,
        })

    def test_config_argument_has_invalid_value(self):
        """Exception is raised when --config argument specifies a malformed
        configuration file.
        """
        config_file = os.path.join(self.parent_dir, 'config.ini')
        create_file(config_file, textwrap.dedent("""\
            [the-plugin]
            option_a = 1
            [components
            the_plugin.* = enabled
            """))
        rv, output = self.execute('initenv project1 sqlite:db/sqlite.db '
                                   '--config=%s' % config_file)

        self.assertEqual(2, rv, output)
        self.assertExpectedResult(output, {
            'env_path': self.env_path,
            'config_file': config_file,
        })


def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(makeSuite(EnvironmentWithoutDataTestCase))
    suite.addTest(makeSuite(EnvironmentDataTestCase))
    suite.addTest(makeSuite(EnvironmentTestCase))
    suite.addTest(makeSuite(EnvironmentAttributesTestCase))
    suite.addTest(makeSuite(EnvironmentUpgradeTestCase))
    suite.addTest(makeSuite(KnownUsersTestCase))
    suite.addTest(makeSuite(SystemInfoTestCase))
    suite.addTest(makeSuite(ConvertDatabaseTestCase))
    suite.addTest(makeSuite(SystemInfoProviderTestCase))
    suite.addTest(makeSuite(TracAdminDeployTestCase))
    suite.addTest(makeSuite(TracAdminInitenvTestCase))
    return suite

if __name__ == '__main__':
    unittest.main(defaultTest='test_suite')
