55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
|
|
import unittest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.engine.cursor import CursorResult
|
|
from sqlalchemy.engine.result import ChunkedIteratorResult
|
|
|
|
from db import DatabaseManager, FeedChannelModel, AuditModel
|
|
|
|
|
|
class TestDatabaseConnections(unittest.IsolatedAsyncioTestCase):
|
|
"""The purpose of this test, is to ensure that the database connections function properly."""
|
|
|
|
async def test_select__feed_channel_model(self):
|
|
"""This test runs a select query on the `FeedChannelModel`"""
|
|
|
|
async with DatabaseManager() as database:
|
|
query = select(FeedChannelModel).limit(1000)
|
|
result = await database.session.execute(query)
|
|
|
|
self.assertIsInstance(
|
|
result,
|
|
ChunkedIteratorResult,
|
|
f"Result should be `ChunkedIteratorResult`, not {type(result)!r}"
|
|
)
|
|
|
|
async def test_select__rss_source_model(self):
|
|
"""This test runs a select query on the `RssSourceModel`"""
|
|
|
|
async with DatabaseManager() as database:
|
|
query = select(RssSourceModel).limit(1000)
|
|
result = await database.session.execute(query)
|
|
|
|
self.assertIsInstance(
|
|
result,
|
|
ChunkedIteratorResult,
|
|
f"Result should be `ChunkedIteratorResult`, not {type(result)!r}"
|
|
)
|
|
|
|
async def test_select__audit_model(self):
|
|
"""This test runs a select query on the `AuditModel`"""
|
|
|
|
async with DatabaseManager() as database:
|
|
query = select(AuditModel).limit(1000)
|
|
result = await database.session.execute(query)
|
|
|
|
self.assertIsInstance(
|
|
result,
|
|
ChunkedIteratorResult,
|
|
f"Result should be `ChunkedIteratorResult`, not {type(result)!r}"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|