<commit_before># -*- coding: utf-8 -*-
"""
Test maintain_schema contextmanager
"""
try:
    from unittest import mock
except:
    import mock
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm.exc import FlushError
from sqlalchemy.orm import Session, scoped_session

from sqlalchemy_sqlschema import maintain_schema
from sqlalchemy_sqlschema.maintain_schema import SchemaContextManager
from sqlalchemy_sqlschema.sql import GetSchema, SetSchema

def test_decorator():
    """Test that using as a decorator triggers the context manager"""
    from sqlalchemy_sqlschema.maintain_schema import SchemaContextManager
    with mock.patch.object(
            SchemaContextManager, "__enter__", mock.Mock(return_value='foo')), \
         mock.patch.object(
             SchemaContextManager, "__exit__", mock.Mock(return_value=False)):

        m = maintain_schema("schema", None)

        @m
        def a_function():
            pass

        a_function()

        assert m.__enter__.call_count == 1
        assert m.__exit__.call_count == 1


@pytest.fixture(scope="session")
def engine():
    return create_engine('sqlite://')

def _mock_session(engine):
    """Return an sqlite session where get and set schema execution is mocked
    out."""
    session = Session(engine)
    original_execute = session.execute
    class GetSchemaResult:
        def scalar(self):
            return "default_schema"

    def execute(stmt, *args, **kwargs):
        if isinstance(stmt, GetSchema):
            return GetSchemaResult()
        elif isinstance(stmt, SetSchema):
            return original_execute("select 1")
        else:
            return original_execute(stmt, *args, **kwargs)

    patcher = mock.patch.object(session, "execute", autospec=True, side_effect=execute)
    return session, patcher

@pytest.yield_fixture
def mock_session(engine):
    session, patcher = _mock_session(engine)
    patcher.start()
    yield session
    patcher.stop()
    session.close()

@pytest.yield_fixture
def mock_session2(engine):
    session, patcher = _mock_session(engine)
    patcher.start()
    yield session
    patcher.stop()
    session.close()

@pytest.yield_fixture
def Model(engine):
    from sqlalchemy import Column, Integer
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class Model(Base):
        __tablename__ = "model"<commit_msg>change fixture scope<commit_after># -*- coding: utf-8 -*-
"""
Test maintain_schema contextmanager
"""
try:
    from unittest import mock
except:
    import mock
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm.exc import FlushError
from sqlalchemy.orm import Session, scoped_session

from sqlalchemy_sqlschema import maintain_schema
from sqlalchemy_sqlschema.maintain_schema import SchemaContextManager
from sqlalchemy_sqlschema.sql import GetSchema, SetSchema

def test_decorator():
    """Test that using as a decorator triggers the context manager"""
    from sqlalchemy_sqlschema.maintain_schema import SchemaContextManager
    with mock.patch.object(
            SchemaContextManager, "__enter__", mock.Mock(return_value='foo')), \
         mock.patch.object(
             SchemaContextManager, "__exit__", mock.Mock(return_value=False)):

        m = maintain_schema("schema", None)

        @m
        def a_function():
            pass

        a_function()

        assert m.__enter__.call_count == 1
        assert m.__exit__.call_count == 1


@pytest.fixture(scope="session")
def engine():
    return create_engine('sqlite://')

def _mock_session(engine):
    """Return an sqlite session where get and set schema execution is mocked
    out."""
    session = Session(engine)
    original_execute = session.execute
    class GetSchemaResult:
        def scalar(self):
            return "default_schema"

    def execute(stmt, *args, **kwargs):
        if isinstance(stmt, GetSchema):
            return GetSchemaResult()
        elif isinstance(stmt, SetSchema):
            return original_execute("select 1")
        else:
            return original_execute(stmt, *args, **kwargs)

    patcher = mock.patch.object(session, "execute", autospec=True, side_effect=execute)
    return session, patcher

@pytest.yield_fixture
def mock_session(engine):
    session, patcher = _mock_session(engine)
    patcher.start()
    yield session
    patcher.stop()
    session.close()

@pytest.yield_fixture
def mock_session2(engine):
    session, patcher = _mock_session(engine)
    patcher.start()
    yield session
    patcher.stop()
    session.close()

@pytest.yield_fixture(scope="session")
def Model(engine):
    from sqlalchemy import Column, Integer
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class Model(Base):
        __tablename__ = "model"
{'fraction_non_alphanumeric': 0.09220886862826357, 'fraction_numerical': 0.0020721094073767096, 'mean_word_length': 3.4325068870523414, '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': '6514305', 'n_tokens_mistral': 1404, 'n_tokens_neox': 1351, 'n_words': 359}