Skip to content
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
fix test
  • Loading branch information
IlyaFaer committed Jan 10, 2022
commit c278a43a629bd88c8e568a361add85c0d9b31f07
53 changes: 27 additions & 26 deletions test/test_suite.py
Original file line number Diff line number Diff line change
Expand Up @@ -1606,6 +1606,31 @@ def test_read_only(self):
assert connection.connection.read_only is False


class LimitOffsetTest(fixtures.TestBase):
"""
Check that SQL with an offset and no limit is being generated correctly.
"""

def setUp(self):
self._engine = create_engine(get_db_url(), pool_size=1)
self._metadata = MetaData(bind=self._engine)

self._table = Table(
"users",
self._metadata,
Column("user_id", Integer, primary_key=True),
Column("user_name", String(16), nullable=False),
)

self._metadata.create_all(self._engine)

def test_offset_only(self):
for offset in [1, 7, 10, 100, 1000, 10000]:

with self._engine.connect().execution_options(read_only=True) as connection:
list(connection.execute(self._table.select().offset(offset)).fetchall())


class ExecutionOptionsStalenessTest(fixtures.TestBase):
"""
Check that `execution_options()` method correctly
Expand All @@ -1617,9 +1642,10 @@ def setUp(self):
metadata = MetaData(bind=self._engine)

self._table = Table(
"execution_options_tab",
"options_tab",
metadata,
Column("opt_id", Integer, primary_key=True),
Column("opt_name", String(16), nullable=False),
)

metadata.create_all(self._engine)
Expand All @@ -1641,31 +1667,6 @@ def test_staleness(self):
pass


class LimitOffsetTest(fixtures.TestBase):
"""
Check that SQL with an offset and no limit is being generated correctly.
"""

def setUp(self):
self._engine = create_engine(get_db_url(), pool_size=1)
self._metadata = MetaData(bind=self._engine)

self._table = Table(
"users",
self._metadata,
Column("user_id", Integer, primary_key=True),
Column("user_name", String(16), nullable=False),
)

self._metadata.create_all(self._engine)

def test_offset_only(self):
for offset in [1, 7, 10, 100, 1000, 10000]:

with self._engine.connect().execution_options(read_only=True) as connection:
list(connection.execute(self._table.select().offset(offset)).fetchall())


class TemporaryTableTest(fixtures.TestBase):
"""
Check that temporary tables raise an error on creation.
Expand Down