latte
abef368a68
Some checks failed
CI/CD Pipeline / Code Quality Checks (push) Failing after 4m49s
CI/CD Pipeline / Security Scanning (push) Successful in 15s
CI/CD Pipeline / Tests (3.11) (push) Successful in 9m41s
CI/CD Pipeline / Tests (3.12) (push) Successful in 9m36s
CI/CD Pipeline / Build Docker Image (push) Has been skipped
Dependency Updates / Update Dependencies (push) Successful in 29s
386 lines
11 KiB
Python
386 lines
11 KiB
Python
"""Pytest fixtures for GuardDen tests."""
|
|
|
|
import asyncio
|
|
import inspect
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import AsyncGenerator
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, event, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parents[1]
|
|
SRC_DIR = ROOT_DIR / "src"
|
|
if str(SRC_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SRC_DIR))
|
|
|
|
# Import after path setup
|
|
from guardden.config import Settings
|
|
from guardden.models.base import Base
|
|
from guardden.models.guild import BannedWord, Guild, GuildSettings
|
|
from guardden.models.moderation import ModerationLog, Strike, UserNote
|
|
from guardden.services.database import Database
|
|
|
|
|
|
def pytest_addoption(parser: pytest.Parser) -> None:
|
|
parser.addini("asyncio_mode", "Asyncio mode for tests", default="auto")
|
|
|
|
|
|
def pytest_configure(config: pytest.Config) -> None:
|
|
config.addinivalue_line("markers", "asyncio: mark async tests")
|
|
|
|
|
|
def pytest_pyfunc_call(pyfuncitem: pytest.Function) -> bool | None:
|
|
test_function = pyfuncitem.obj
|
|
if inspect.iscoroutinefunction(test_function):
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_until_complete(test_function(**pyfuncitem.funcargs))
|
|
loop.close()
|
|
asyncio.set_event_loop(None)
|
|
return True
|
|
return None
|
|
|
|
|
|
# ==============================================================================
|
|
# Basic Test Fixtures
|
|
# ==============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_guild_id() -> int:
|
|
"""Return a sample Discord guild ID."""
|
|
return 123456789012345678
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_user_id() -> int:
|
|
"""Return a sample Discord user ID."""
|
|
return 987654321098765432
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_moderator_id() -> int:
|
|
"""Return a sample Discord moderator ID."""
|
|
return 111111111111111111
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_owner_id() -> int:
|
|
"""Return a sample Discord owner ID."""
|
|
return 222222222222222222
|
|
|
|
|
|
# ==============================================================================
|
|
# Configuration Fixtures
|
|
# ==============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def test_settings() -> Settings:
|
|
"""Return test configuration settings."""
|
|
return Settings(
|
|
discord_token="a" * 60,
|
|
discord_prefix="!test",
|
|
database_url="sqlite+aiosqlite:///test.db",
|
|
database_pool_min=1,
|
|
database_pool_max=1,
|
|
ai_provider="none",
|
|
log_level="DEBUG",
|
|
allowed_guilds=[],
|
|
owner_ids=[],
|
|
data_dir=Path("/tmp/guardden_test"),
|
|
)
|
|
|
|
|
|
# ==============================================================================
|
|
# Database Fixtures
|
|
# ==============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_database(test_settings: Settings) -> AsyncGenerator[Database, None]:
|
|
"""Create a test database with in-memory SQLite."""
|
|
# Use in-memory SQLite for tests
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
echo=False,
|
|
)
|
|
|
|
@event.listens_for(engine.sync_engine, "connect")
|
|
def _enable_sqlite_foreign_keys(dbapi_connection, connection_record):
|
|
cursor = dbapi_connection.cursor()
|
|
cursor.execute("PRAGMA foreign_keys=ON")
|
|
cursor.close()
|
|
|
|
# Create all tables
|
|
async with engine.begin() as conn:
|
|
await conn.execute(text("PRAGMA foreign_keys=ON"))
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
database = Database(test_settings)
|
|
database._engine = engine
|
|
database._session_factory = async_sessionmaker(
|
|
engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
|
|
yield database
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session(test_database: Database) -> AsyncGenerator[AsyncSession, None]:
|
|
"""Create a database session for testing."""
|
|
async with test_database.session() as session:
|
|
yield session
|
|
|
|
|
|
# ==============================================================================
|
|
# Model Fixtures
|
|
# ==============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_guild(db_session: AsyncSession, sample_guild_id: int, sample_owner_id: int) -> Guild:
|
|
"""Create a test guild with settings."""
|
|
guild = Guild(
|
|
id=sample_guild_id,
|
|
name="Test Guild",
|
|
owner_id=sample_owner_id,
|
|
premium=False,
|
|
)
|
|
db_session.add(guild)
|
|
|
|
# Create associated settings
|
|
settings = GuildSettings(
|
|
guild_id=sample_guild_id,
|
|
prefix="!",
|
|
automod_enabled=True,
|
|
ai_moderation_enabled=False,
|
|
verification_enabled=False,
|
|
)
|
|
db_session.add(settings)
|
|
|
|
await db_session.commit()
|
|
await db_session.refresh(guild)
|
|
return guild
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_banned_word(
|
|
db_session: AsyncSession, test_guild: Guild, sample_moderator_id: int
|
|
) -> BannedWord:
|
|
"""Create a test banned word."""
|
|
banned_word = BannedWord(
|
|
guild_id=test_guild.id,
|
|
pattern="badword",
|
|
is_regex=False,
|
|
action="delete",
|
|
reason="Inappropriate content",
|
|
added_by=sample_moderator_id,
|
|
)
|
|
db_session.add(banned_word)
|
|
await db_session.commit()
|
|
await db_session.refresh(banned_word)
|
|
return banned_word
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_moderation_log(
|
|
db_session: AsyncSession, test_guild: Guild, sample_user_id: int, sample_moderator_id: int
|
|
) -> ModerationLog:
|
|
"""Create a test moderation log entry."""
|
|
mod_log = ModerationLog(
|
|
guild_id=test_guild.id,
|
|
target_id=sample_user_id,
|
|
target_name="TestUser",
|
|
moderator_id=sample_moderator_id,
|
|
moderator_name="TestModerator",
|
|
action="warn",
|
|
reason="Test warning",
|
|
is_automatic=False,
|
|
)
|
|
db_session.add(mod_log)
|
|
await db_session.commit()
|
|
await db_session.refresh(mod_log)
|
|
return mod_log
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_strike(
|
|
db_session: AsyncSession, test_guild: Guild, sample_user_id: int, sample_moderator_id: int
|
|
) -> Strike:
|
|
"""Create a test strike."""
|
|
strike = Strike(
|
|
guild_id=test_guild.id,
|
|
user_id=sample_user_id,
|
|
user_name="TestUser",
|
|
moderator_id=sample_moderator_id,
|
|
reason="Test strike",
|
|
points=1,
|
|
is_active=True,
|
|
)
|
|
db_session.add(strike)
|
|
await db_session.commit()
|
|
await db_session.refresh(strike)
|
|
return strike
|
|
|
|
|
|
# ==============================================================================
|
|
# Discord Mock Fixtures
|
|
# ==============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_discord_user(sample_user_id: int) -> MagicMock:
|
|
"""Create a mock Discord user."""
|
|
user = MagicMock()
|
|
user.id = sample_user_id
|
|
user.name = "TestUser"
|
|
user.display_name = "Test User"
|
|
user.mention = f"<@{sample_user_id}>"
|
|
user.avatar = None
|
|
user.bot = False
|
|
user.send = AsyncMock()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_discord_member(mock_discord_user: MagicMock) -> MagicMock:
|
|
"""Create a mock Discord member."""
|
|
member = MagicMock()
|
|
member.id = mock_discord_user.id
|
|
member.name = mock_discord_user.name
|
|
member.display_name = mock_discord_user.display_name
|
|
member.mention = mock_discord_user.mention
|
|
member.avatar = mock_discord_user.avatar
|
|
member.bot = mock_discord_user.bot
|
|
member.send = mock_discord_user.send
|
|
|
|
# Member-specific attributes
|
|
member.guild = MagicMock()
|
|
member.top_role = MagicMock()
|
|
member.top_role.position = 1
|
|
member.roles = [MagicMock()]
|
|
member.joined_at = datetime.now(timezone.utc)
|
|
member.kick = AsyncMock()
|
|
member.ban = AsyncMock()
|
|
member.timeout = AsyncMock()
|
|
|
|
return member
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_discord_guild(sample_guild_id: int, sample_owner_id: int) -> MagicMock:
|
|
"""Create a mock Discord guild."""
|
|
guild = MagicMock()
|
|
guild.id = sample_guild_id
|
|
guild.name = "Test Guild"
|
|
guild.owner_id = sample_owner_id
|
|
guild.member_count = 100
|
|
guild.premium_tier = 0
|
|
|
|
# Methods
|
|
guild.get_member = MagicMock(return_value=None)
|
|
guild.get_channel = MagicMock(return_value=None)
|
|
guild.leave = AsyncMock()
|
|
guild.ban = AsyncMock()
|
|
guild.unban = AsyncMock()
|
|
|
|
return guild
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_discord_channel() -> MagicMock:
|
|
"""Create a mock Discord channel."""
|
|
channel = MagicMock()
|
|
channel.id = 333333333333333333
|
|
channel.name = "test-channel"
|
|
channel.mention = "<#333333333333333333>"
|
|
channel.send = AsyncMock()
|
|
channel.delete_messages = AsyncMock()
|
|
return channel
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_discord_message(
|
|
mock_discord_member: MagicMock, mock_discord_channel: MagicMock
|
|
) -> MagicMock:
|
|
"""Create a mock Discord message."""
|
|
message = MagicMock()
|
|
message.id = 444444444444444444
|
|
message.content = "Test message content"
|
|
message.author = mock_discord_member
|
|
message.channel = mock_discord_channel
|
|
message.guild = mock_discord_member.guild
|
|
message.created_at = datetime.now(timezone.utc)
|
|
message.delete = AsyncMock()
|
|
message.reply = AsyncMock()
|
|
message.add_reaction = AsyncMock()
|
|
return message
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_discord_context(
|
|
mock_discord_member: MagicMock, mock_discord_guild: MagicMock, mock_discord_channel: MagicMock
|
|
) -> MagicMock:
|
|
"""Create a mock Discord command context."""
|
|
ctx = MagicMock()
|
|
ctx.author = mock_discord_member
|
|
ctx.guild = mock_discord_guild
|
|
ctx.channel = mock_discord_channel
|
|
ctx.send = AsyncMock()
|
|
ctx.reply = AsyncMock()
|
|
return ctx
|
|
|
|
|
|
# ==============================================================================
|
|
# Bot and Service Fixtures
|
|
# ==============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_bot(test_database: Database) -> MagicMock:
|
|
"""Create a mock GuardDen bot."""
|
|
bot = MagicMock()
|
|
bot.database = test_database
|
|
bot.guild_config = MagicMock()
|
|
bot.ai_provider = MagicMock()
|
|
bot.rate_limiter = MagicMock()
|
|
bot.user = MagicMock()
|
|
bot.user.id = 555555555555555555
|
|
bot.user.name = "GuardDen"
|
|
return bot
|
|
|
|
|
|
# ==============================================================================
|
|
# Test Environment Setup
|
|
# ==============================================================================
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_test_environment() -> None:
|
|
"""Set up test environment variables."""
|
|
# Set test environment variables
|
|
os.environ["GUARDDEN_DISCORD_TOKEN"] = "test_token_12345678901234567890"
|
|
os.environ["GUARDDEN_DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
|
|
os.environ["GUARDDEN_AI_PROVIDER"] = "none"
|
|
os.environ["GUARDDEN_LOG_LEVEL"] = "DEBUG"
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create an instance of the default event loop for the test session."""
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|