<commit_before># coding: utf-8

from sqlalchemy.engine import reflection
from sqlalchemy.testing.assertions import eq_, assert_raises, \
    AssertsExecutionResults
from sqlalchemy.testing import fixtures
from sqlalchemy import testing
from sqlalchemy import inspect
from sqlalchemy import Table, Column, MetaData, Integer, String, \
    PrimaryKeyConstraint, ForeignKey, join, Sequence, UniqueConstraint, \
    Index
from sqlalchemy import exc
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import base as postgresql
from sqlalchemy.dialects.postgresql import ARRAY


class ForeignTableReflectionTest(fixtures.TablesTest, AssertsExecutionResults):
    """Test reflection on foreign tables"""

    __requires__ = 'postgresql_test_dblink',
    __only_on__ = 'postgresql >= 9.3'
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        from sqlalchemy.testing import config
        dblink = config.file_config.get(
            'sqla_testing', 'postgres_test_db_link')

        testtable = Table(
            'testtable', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)))

        for ddl in [
            "CREATE SERVER test_server FOREIGN DATA WRAPPER postgres_fdw "
            "OPTIONS (dbname 'test', host '%s')" % dblink,
            "CREATE USER MAPPING FOR public \
            SERVER test_server options (user 'scott', password '<PASSWORD>')",
            "CREATE FOREIGN TABLE test_foreigntable ( "
            "   id          INT, "
            "   data        VARCHAR(30) "
            ") SERVER test_server OPTIONS (table_name 'testtable')",
        ]:
            sa.event.listen(metadata, "after_create", sa.DDL(ddl))

        for ddl in [
            'DROP FOREIGN TABLE test_foreigntable',
            'DROP USER MAPPING FOR public SERVER test_server',
            "DROP SERVER test_server"
        ]:
            sa.event.listen(metadata, "before_drop", sa.DDL(ddl))

    def test_foreign_table_is_reflected(self):
        metadata = MetaData(testing.db)
        table = Table('test_foreigntable', metadata, autoload=True)
        eq_(set(table.columns.keys()), set(['id', 'data']),
            "Columns of reflected foreign table didn't equal expected columns")

    def test_get_foreign_table_names(self):
        inspector = inspect(testing.db)
        with testing.db.connect() as conn:
            ft_names = inspector.get_foreign_table_names()
            eq_(ft_names, ['test_foreigntable'])

    def test_get_table_names_no_foreign(self):
        inspector = inspect(testing.db)
        with testing.db.connect() as conn:
            names = inspector.get_table_names()
            eq_(names, ['testtable'])


class MaterialiedViewReflectionTest(
        fixtures.TablesTest, AssertsExecutionResults):
    """Test reflection on materialized views"""

    __only_on__ = 'postgresql >= 9.3'
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        testtable = Table(
            'testtable', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)))

        # insert data before we create the view
        @sa.event.listens_for(testtable, "after_create")<commit_msg>Fix typo<commit_after># coding: utf-8

from sqlalchemy.engine import reflection
from sqlalchemy.testing.assertions import eq_, assert_raises, \
    AssertsExecutionResults
from sqlalchemy.testing import fixtures
from sqlalchemy import testing
from sqlalchemy import inspect
from sqlalchemy import Table, Column, MetaData, Integer, String, \
    PrimaryKeyConstraint, ForeignKey, join, Sequence, UniqueConstraint, \
    Index
from sqlalchemy import exc
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import base as postgresql
from sqlalchemy.dialects.postgresql import ARRAY


class ForeignTableReflectionTest(fixtures.TablesTest, AssertsExecutionResults):
    """Test reflection on foreign tables"""

    __requires__ = 'postgresql_test_dblink',
    __only_on__ = 'postgresql >= 9.3'
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        from sqlalchemy.testing import config
        dblink = config.file_config.get(
            'sqla_testing', 'postgres_test_db_link')

        testtable = Table(
            'testtable', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)))

        for ddl in [
            "CREATE SERVER test_server FOREIGN DATA WRAPPER postgres_fdw "
            "OPTIONS (dbname 'test', host '%s')" % dblink,
            "CREATE USER MAPPING FOR public \
            SERVER test_server options (user 'scott', password '<PASSWORD>')",
            "CREATE FOREIGN TABLE test_foreigntable ( "
            "   id          INT, "
            "   data        VARCHAR(30) "
            ") SERVER test_server OPTIONS (table_name 'testtable')",
        ]:
            sa.event.listen(metadata, "after_create", sa.DDL(ddl))

        for ddl in [
            'DROP FOREIGN TABLE test_foreigntable',
            'DROP USER MAPPING FOR public SERVER test_server',
            "DROP SERVER test_server"
        ]:
            sa.event.listen(metadata, "before_drop", sa.DDL(ddl))

    def test_foreign_table_is_reflected(self):
        metadata = MetaData(testing.db)
        table = Table('test_foreigntable', metadata, autoload=True)
        eq_(set(table.columns.keys()), set(['id', 'data']),
            "Columns of reflected foreign table didn't equal expected columns")

    def test_get_foreign_table_names(self):
        inspector = inspect(testing.db)
        with testing.db.connect() as conn:
            ft_names = inspector.get_foreign_table_names()
            eq_(ft_names, ['test_foreigntable'])

    def test_get_table_names_no_foreign(self):
        inspector = inspect(testing.db)
        with testing.db.connect() as conn:
            names = inspector.get_table_names()
            eq_(names, ['testtable'])


class MaterializedViewReflectionTest(
        fixtures.TablesTest, AssertsExecutionResults):
    """Test reflection on materialized views"""

    __only_on__ = 'postgresql >= 9.3'
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        testtable = Table(
            'testtable', metadata,
            Column('id', Integer, primary_key=True),
            Column('data', String(30)))

        # insert data before we create the view
        @sa.event.listens_for(testtable, "after_create")
{'fraction_non_alphanumeric': 0.09425852011053117, 'fraction_numerical': 0.0033773411114522566, 'mean_word_length': 2.796620046620047, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, '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': '7648529', 'n_tokens_mistral': 1939, 'n_tokens_neox': 1826, 'n_words': 474}