<commit_before># -*- coding: utf-8 -*-
# ###
# Copyright (c) 2013, Rice University
# This software is subject to the provisions of the GNU Affero General
# Public License version 3 (AGPLv3).
# See LICENCE.txt for details.
# ###
import datetime
import hashlib
import json
import os
import sys
import time
import unittest

import psycopg2

from . import testing


class InitializeDBTestCase(unittest.TestCase):
    fixture = testing.schema_fixture

    def setUp(self):
        self.settings = testing.integration_test_settings()

    def tearDown(self):
        self.fixture.tearDown()

    @property
    def target(self):
        from ..database import initdb
        return initdb

    @testing.db_connect
    def test_success(self, cursor):
        self.target(self.settings)
        # Simply test for the 'licenses' table.
        # This isn't comprehensive, but it tests that the schema was put in place
        # and static data (constants) values have been added to tables.
        cursor.execute("SELECT code, version FROM licenses ORDER BY code, version ASC;")
        try:
            license = cursor.fetchone()
        except Exception as exc:
            self.fail(exc)
        self.assertEqual(license, ('by', '1.0'))

    def test_initdb_on_already_initialized_db(self):
        """Testing the ``initdb`` raises a discernible error when the
        database is already initialized.
        """
        self.fixture.setUp()
        # The fixture has initialized the database, so we only need to
        #   run the function.
        with self.assertRaises(psycopg2.InternalError) as caught_exception:
            self.target(self.settings)
        self.assertEqual(caught_exception.exception.message,
                         'Database is already initialized.\n')

    @testing.db_connect
    def test_venv(self, cursor):
        """Testing to veryify the venv schema is properly created and the correct
        instance of python is used
        """
        cursor.execute("CREATE FUNCTION pypath() RETURNS text LANGUAGE "
                       "plpythonu AS $$import sys;return sys.prefix$$")
        cursor.execute("SELECT pypath()")
        db_pypath = cursor.fetchone()[0]

        self.assertEqual(True, os.path.samefile(db_pypath, sys.prefix))


class MiscellaneousFunctionsTestCase(unittest.TestCase):
    fixture = testing.schema_fixture

    @testing.db_connect
    def setUp(self, cursor):
        self.fixture.setUp()

    def tearDown(self):
        self.fixture.tearDown()

    @testing.db_connect
    def test_iso8601(self, cursor):
        # Exams the iso8601 SQL function.
        cursor.execute("SELECT date_trunc('second', current_timestamp), iso8601(current_timestamp);")
        current, iso8601 = cursor.fetchone()
        # We need to make the date timezone aware, which is what the 'Z' does,
        #   but the parsing statement can't do anything with that.
        # We'll use psycopg2's tzinfo classes for simplicity.
        from psycopg2.tz import FixedOffsetTimezone
        value = datetime.datetime(*time.strptime(iso8601,
                                  "%Y-%m-%dT%H:%M:%SZ")[:6], tzinfo=FixedOffsetTimezone())
        self.assertEqual(current, value)

    @unittest.skipIf(not testing.is_venv(),
                     "Not within a virtualenv")
    @testing.db_connect
    def test_pypath(self, cursor):
        site_packages = testing.getsitepackages()
        # Examine the results of the pypath SQL function.
        cursor.execute("SELECT unnest(pypath())")
        paths = [row[0] for row in cursor.fetchall()]

        for site_pkg in site_packages:
            self.assertIn(os.path.abspath(site_pkg), paths)

    @testing.db_connect
    def test_pyimport(self, cursor):
        target_name = 'cnxarchive.database'
        import cnxarchive.database as target
        # Check the results of calling pyimport on cnxarchive.
        cursor.execute("SELECT import, directory, file_path "
                       "FROM pyimport(%s)", (target_name,))
        import_, directory, file_path = cursor.fetchone()

        self.assertEqual(import_, target_name)
        expected_directory = os.path.abspath(os.path.dirname(target.__file__))
        self.assertEqual(directory, expected_directory)
        self.assertEqual(file_path, target.__file__)
<commit_msg>Change the db py_import test to work with source and installed paths<commit_after># -*- coding: utf-8 -*-
# ###
# Copyright (c) 2013, Rice University
# This software is subject to the provisions of the GNU Affero General
# Public License version 3 (AGPLv3).
# See LICENCE.txt for details.
# ###
import datetime
import hashlib
import json
import os
import sys
import time
import unittest

import psycopg2

from . import testing


class InitializeDBTestCase(unittest.TestCase):
    fixture = testing.schema_fixture

    def setUp(self):
        self.settings = testing.integration_test_settings()

    def tearDown(self):
        self.fixture.tearDown()

    @property
    def target(self):
        from ..database import initdb
        return initdb

    @testing.db_connect
    def test_success(self, cursor):
        self.target(self.settings)
        # Simply test for the 'licenses' table.
        # This isn't comprehensive, but it tests that the schema was put in place
        # and static data (constants) values have been added to tables.
        cursor.execute("SELECT code, version FROM licenses ORDER BY code, version ASC;")
        try:
            license = cursor.fetchone()
        except Exception as exc:
            self.fail(exc)
        self.assertEqual(license, ('by', '1.0'))

    def test_initdb_on_already_initialized_db(self):
        """Testing the ``initdb`` raises a discernible error when the
        database is already initialized.
        """
        self.fixture.setUp()
        # The fixture has initialized the database, so we only need to
        #   run the function.
        with self.assertRaises(psycopg2.InternalError) as caught_exception:
            self.target(self.settings)
        self.assertEqual(caught_exception.exception.message,
                         'Database is already initialized.\n')

    @testing.db_connect
    def test_venv(self, cursor):
        """Testing to veryify the venv schema is properly created and the correct
        instance of python is used
        """
        cursor.execute("CREATE FUNCTION pypath() RETURNS text LANGUAGE "
                       "plpythonu AS $$import sys;return sys.prefix$$")
        cursor.execute("SELECT pypath()")
        db_pypath = cursor.fetchone()[0]

        self.assertEqual(True, os.path.samefile(db_pypath, sys.prefix))


class MiscellaneousFunctionsTestCase(unittest.TestCase):
    fixture = testing.schema_fixture

    @testing.db_connect
    def setUp(self, cursor):
        self.fixture.setUp()

    def tearDown(self):
        self.fixture.tearDown()

    @testing.db_connect
    def test_iso8601(self, cursor):
        # Exams the iso8601 SQL function.
        cursor.execute("SELECT date_trunc('second', current_timestamp), iso8601(current_timestamp);")
        current, iso8601 = cursor.fetchone()
        # We need to make the date timezone aware, which is what the 'Z' does,
        #   but the parsing statement can't do anything with that.
        # We'll use psycopg2's tzinfo classes for simplicity.
        from psycopg2.tz import FixedOffsetTimezone
        value = datetime.datetime(*time.strptime(iso8601,
                                  "%Y-%m-%dT%H:%M:%SZ")[:6], tzinfo=FixedOffsetTimezone())
        self.assertEqual(current, value)

    @unittest.skipIf(not testing.is_venv(),
                     "Not within a virtualenv")
    @testing.db_connect
    def test_pypath(self, cursor):
        site_packages = testing.getsitepackages()
        # Examine the results of the pypath SQL function.
        cursor.execute("SELECT unnest(pypath())")
        paths = [row[0] for row in cursor.fetchall()]

        for site_pkg in site_packages:
            self.assertIn(os.path.abspath(site_pkg), paths)

    @testing.db_connect
    def test_pyimport(self, cursor):
        target_name = 'cnxarchive.database'

        # Import the module from current directory
        import cnxarchive.database as target_source

        # Remove current directory from sys.path
        cwd = os.getcwd()
        sys_path = sys.path
        modified_path = [i for i in sys_path if i and i != cwd]
        sys.path = modified_path
        self.addCleanup(setattr, sys, 'path', sys_path)

        # Remove all cnxarchive modules from sys.modules
        sys_modules = sys.modules.copy()
        self.addCleanup(sys.modules.update, sys_modules)
        for module in sys.modules.keys():
            if module.startswith('cnxarchive'):
                del sys.modules[module]

        # Import the installed version
        import cnxarchive.database as target_installed

        # Depending on whether "setup.py develop" or "setup.py install" is
        # used, there are different expected directories and file paths.
        targets = [target_source, target_installed]
        expected_directories = [
            os.path.abspath(os.path.dirname(target.__file__))
            for target in targets]
        expected_file_paths = [
            target.__file__ for target in targets]

        # Check the results of calling pyimport on cnxarchive.
        cursor.execute("SELECT import, directory, file_path "
                       "FROM pyimport(%s)", (target_name,))
        import_, directory, file_path = cursor.fetchone()

        self.assertEqual(import_, target_name)
        self.assertIn(directory, expected_directories)
        self.assertIn(file_path, expected_file_paths)

{'fraction_non_alphanumeric': 0.09029066171923315, 'fraction_numerical': 0.0074211502782931356, 'mean_word_length': 2.6178225205070844, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1343823', 'n_tokens_mistral': 2757, 'n_tokens_neox': 2655, 'n_words': 782}