Tolan/backend/tests/test_energy_stones.py

import sys
from pathlib import Path

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy import select

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from models_db import Base, User, ConversationEnergyStone  # noqa: E402
from services.energy_stone_service import EnergyStoneService  # noqa: E402


class DummyDBManager:
    """Minimal DB manager compatible with EnergyStoneService."""

    def __init__(self, session_factory: async_sessionmaker[AsyncSession]):
        self._session_factory = session_factory

    def get_session(self) -> AsyncSession:
        return self._session_factory()


@pytest_asyncio.fixture
async def db_manager():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    manager = DummyDBManager(session_factory)
    try:
        yield manager
    finally:
        await engine.dispose()


@pytest_asyncio.fixture
async def seeded_user(db_manager: DummyDBManager):
    user_id = "test-user"
    async with db_manager.get_session() as session:
        user = User(id=1, user_id=user_id, username="Tester")
        session.add(user)
        await session.commit()
    return user_id


def build_service(interval_seconds: int = 30) -> EnergyStoneService:
    service = EnergyStoneService()
    service.conversation_interval_ms = interval_seconds * 1000
    return service


@pytest.mark.asyncio
async def test_conversation_stone_not_created_before_threshold(db_manager, seeded_user):
    service = build_service()
    await service.set_db_manager(db_manager)

    stones = await service.create_conversation_stones_if_needed(
        user_id=seeded_user,
        request_id="req-before-threshold",
        total_duration_ms=15_000,
        duration_increment_ms=15_000,
    )

    assert stones == []


@pytest.mark.asyncio
async def test_conversation_stone_created_at_30s_threshold(db_manager, seeded_user):
    service = build_service()
    await service.set_db_manager(db_manager)

    stones = await service.create_conversation_stones_if_needed(
        user_id=seeded_user,
        request_id="req-threshold",
        total_duration_ms=30_000,
        duration_increment_ms=30_000,
    )

    assert len(stones) == 1
    payload = stones[0]
    assert payload["stone_type"] == "conversation"
    assert payload["metadata"]["threshold_index"] == 1
    assert payload["metadata"]["total_duration_ms"] == 30_000

    async with db_manager.get_session() as session:
        result = await session.execute(
            select(ConversationEnergyStone).where(
                ConversationEnergyStone.user_id == seeded_user
            )
        )
        stored = result.scalars().all()
        assert len(stored) == 1
        assert stored[0].threshold_index == 1
        assert stored[0].total_duration_ms == 30_000