<commit_before># -*- coding: utf-8 -*-
"""Test database functionality."""

import sqlite3
import tempfile
import unittest

from contextlib import closing
from datetime import datetime

from sqlalchemy.exc import NoSuchTableError
from sqlalchemy.types import (
    INTEGER,
    TEXT,
)


from pic2map.db import (
    Database,
    transform_metadata_to_row,
)


class DatabaseTest(unittest.TestCase):

    """Database wrapper test cases."""

    def test_get_table_metadata(self):
        """Table metadata can be retrieved using index notation."""
        with tempfile.NamedTemporaryFile() as db_file:
            with closing(sqlite3.connect(db_file.name)) as connection:
                with closing(connection.cursor()) as cursor:
                    cursor.execute(
                        'CREATE TABLE messages (id INTEGER, message TEXT)')

            database = Database(db_file.name)
            table = database['messages']
            schema = {column.name: type(column.type)
                      for column in table.columns}
            self.assertDictEqual(
                schema,
                {'id': INTEGER, 'message': TEXT})

    def test_get_unknown_table_metadata(self):
        """NoSuchTableError raised when table name is not found."""
        with tempfile.NamedTemporaryFile() as db_file:
            with closing(sqlite3.connect(db_file.name)) as connection:
                with closing(connection.cursor()) as cursor:
                    cursor.execute(
                        'CREATE TABLE messages (id INTEGER, message TEXT)')

            database = Database(db_file.name)

            with self.assertRaises(NoSuchTableError):
                database['unknown']

    def test_type_error_on_wrong_table_name(self):
        """TypeError raised when table name is not a string."""
        with tempfile.NamedTemporaryFile() as db_file:
            with closing(sqlite3.connect(db_file.name)) as connection:
                with closing(connection.cursor()) as cursor:
                    cursor.execute(
                        'CREATE TABLE messages (id INTEGER, message TEXT)')

            database = Database(db_file.name)

            with self.assertRaises(TypeError):
                database[0]

    def test_context_manager(self):
        """Connection is opened/closed when used as a context manager."""
        database = Database(':memory:')

        # Connection is None when database object is created
        self.assertIsNone(database.connection)

        with database:
            # Connection is not closed inside the context
            self.assertFalse(database.connection.closed)

        # Connection is closed outside the context
        self.assertTrue(database.connection.closed)


class TransformMetadataToRowTest(unittest.TestCase):

    """EXIF metadata to database row transformation tests."""

    def test_transform_metadata(self):
        """Transform metadata to row."""
        metadata = {
            'SourceFile': 'a.jpg',<commit_msg>Add location database tests<commit_after># -*- coding: utf-8 -*-
"""Test database functionality."""

import os
import shutil
import sqlite3
import tempfile
import unittest

from contextlib import closing
from datetime import datetime

from mock import patch

from sqlalchemy.exc import NoSuchTableError
from sqlalchemy.types import (
    INTEGER,
    TEXT,
)


from pic2map.db import (
    Database,
    LocationDB,
    transform_metadata_to_row,
)


class DatabaseTest(unittest.TestCase):

    """Database wrapper test cases."""

    def test_get_table_metadata(self):
        """Table metadata can be retrieved using index notation."""
        with tempfile.NamedTemporaryFile() as db_file:
            with closing(sqlite3.connect(db_file.name)) as connection:
                with closing(connection.cursor()) as cursor:
                    cursor.execute(
                        'CREATE TABLE messages (id INTEGER, message TEXT)')

            database = Database(db_file.name)
            table = database['messages']
            schema = {column.name: type(column.type)
                      for column in table.columns}
            self.assertDictEqual(
                schema,
                {'id': INTEGER, 'message': TEXT})

    def test_get_unknown_table_metadata(self):
        """NoSuchTableError raised when table name is not found."""
        with tempfile.NamedTemporaryFile() as db_file:
            with closing(sqlite3.connect(db_file.name)) as connection:
                with closing(connection.cursor()) as cursor:
                    cursor.execute(
                        'CREATE TABLE messages (id INTEGER, message TEXT)')

            database = Database(db_file.name)

            with self.assertRaises(NoSuchTableError):
                database['unknown']

    def test_type_error_on_wrong_table_name(self):
        """TypeError raised when table name is not a string."""
        with tempfile.NamedTemporaryFile() as db_file:
            with closing(sqlite3.connect(db_file.name)) as connection:
                with closing(connection.cursor()) as cursor:
                    cursor.execute(
                        'CREATE TABLE messages (id INTEGER, message TEXT)')

            database = Database(db_file.name)

            with self.assertRaises(TypeError):
                database[0]

    def test_context_manager(self):
        """Connection is opened/closed when used as a context manager."""
        database = Database(':memory:')

        # Connection is None when database object is created
        self.assertIsNone(database.connection)

        with database:
            # Connection is not closed inside the context
            self.assertFalse(database.connection.closed)

        # Connection is closed outside the context
        self.assertTrue(database.connection.closed)


class LocationDBTest(unittest.TestCase):

    """Location database tests."""

    def setUp(self):
        """Create temporary directory."""
        self.directory = tempfile.mkdtemp()

    def tearDown(self):
        """Remove temporary directory."""
        shutil.rmtree(self.directory)

    def test_database_exists(self):
        """Database not create if exists."""
        filename = os.path.join(self.directory, 'location.db')
        with closing(sqlite3.connect(filename)) as connection:
            with closing(connection.cursor()) as cursor:
                cursor.execute(
                    'CREATE TABLE location (column_1 TEXT, column_2 TEXT)')

        with patch('pic2map.db.BaseDirectory') as BaseDirectory:
            BaseDirectory.save_data_path.return_value = (
                os.path.dirname(filename))
            location_db = LocationDB()
            self.assertListEqual(
                location_db.location_table.columns.keys(),
                ['column_1', 'column_2'],
            )

    def test_create_database(self):
        """Create database file."""
        with patch('pic2map.db.BaseDirectory') as BaseDirectory:
            BaseDirectory.save_data_path.return_value = self.directory
            LocationDB()
            filename = os.path.join(self.directory, 'location.db')
            self.assertTrue(os.path.isfile(filename))

    def test_insert(self):
        """Insert records in database."""
        rows = [
            {
                'filename': 'a.jpg',
                'latitude': 1.2,
                'longitude': 2.1,
                'datetime': datetime(2015, 1, 1, 12, 34, 56)
            },
            {
                'filename': 'b.jpg',
                'latitude': 3.4,
                'longitude': 4.3,
                'datetime': datetime(2015, 1, 1, 12, 34, 56)
            },
        ]
        with patch('pic2map.db.BaseDirectory') as BaseDirectory:
            BaseDirectory.save_data_path.return_value = self.directory
            with LocationDB() as location_db:
                location_db.insert(rows)

            filename = os.path.join(self.directory, 'location.db')
            with closing(sqlite3.connect(filename)) as connection:
                with closing(connection.cursor()) as cursor:
                    result = cursor.execute('SELECT COUNT(*) FROM location')
                    self.assertListEqual(result.fetchall(), [(2,)])


class TransformMetadataToRowTest(unittest.TestCase):

    """EXIF metadata to database row transformation tests."""

    def test_transform_metadata(self):
        """Transform metadata to row."""
        metadata = {
            'SourceFile': 'a.jpg',
{'fraction_non_alphanumeric': 0.09628933771723815, 'fraction_numerical': 0.006575857209957727, 'mean_word_length': 2.18392523364486, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '301522', 'n_tokens_mistral': 2222, 'n_tokens_neox': 2077, 'n_words': 575}