96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""Tests for starboard database helper functions."""
|
|
|
|
# pylint: disable=wrong-import-position
|
|
|
|
import os
|
|
import sys
|
|
|
|
# Ensure project root is on sys.path
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest # pylint: disable=import-error
|
|
|
|
import settings_manager # pylint: disable=import-error
|
|
|
|
|
|
class DummyBot:
|
|
"""Simple container for a pg_pool mock."""
|
|
|
|
def __init__(self, pg_pool):
|
|
self.pg_pool = pg_pool
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_starboard_entry():
|
|
"""Verify create_starboard_entry executes expected queries."""
|
|
|
|
conn = AsyncMock()
|
|
acquire_cm = AsyncMock()
|
|
acquire_cm.__aenter__.return_value = conn
|
|
acquire_cm.__aexit__.return_value = None
|
|
|
|
pg_pool = AsyncMock()
|
|
pg_pool.acquire.return_value = acquire_cm
|
|
|
|
bot = DummyBot(pg_pool)
|
|
with patch.object(settings_manager, "get_bot_instance", return_value=bot):
|
|
result = await settings_manager.create_starboard_entry(
|
|
guild_id=1,
|
|
original_message_id=2,
|
|
original_channel_id=3,
|
|
starboard_message_id=4,
|
|
author_id=5,
|
|
star_count=6,
|
|
)
|
|
|
|
assert result is True
|
|
pg_pool.acquire.assert_called_once()
|
|
assert conn.execute.await_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_starboard_entry():
|
|
"""Verify update_starboard_entry updates star count."""
|
|
|
|
conn = AsyncMock()
|
|
pg_pool = AsyncMock()
|
|
pg_pool.acquire = AsyncMock(return_value=conn)
|
|
pg_pool.release = AsyncMock()
|
|
|
|
bot = DummyBot(pg_pool)
|
|
with patch.object(settings_manager, "get_bot_instance", return_value=bot):
|
|
result = await settings_manager.update_starboard_entry(
|
|
guild_id=1, original_message_id=2, star_count=3
|
|
)
|
|
|
|
assert result is True
|
|
pg_pool.acquire.assert_called_once()
|
|
conn.execute.assert_awaited_once()
|
|
pg_pool.release.assert_called_once_with(conn)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_starboard_entry():
|
|
"""Verify get_starboard_entry fetches the row and returns a dict."""
|
|
|
|
entry_data = {"guild_id": 1, "original_message_id": 2}
|
|
conn = AsyncMock()
|
|
conn.fetchrow = AsyncMock(return_value=entry_data)
|
|
|
|
acquire_cm = AsyncMock()
|
|
acquire_cm.__aenter__.return_value = conn
|
|
acquire_cm.__aexit__.return_value = None
|
|
|
|
pg_pool = AsyncMock()
|
|
pg_pool.acquire.return_value = acquire_cm
|
|
|
|
bot = DummyBot(pg_pool)
|
|
with patch.object(settings_manager, "get_bot_instance", return_value=bot):
|
|
result = await settings_manager.get_starboard_entry(1, 2)
|
|
|
|
assert result == entry_data
|
|
pg_pool.acquire.assert_called_once()
|
|
conn.fetchrow.assert_awaited_once()
|