210 lines
7.5 KiB
Python
210 lines
7.5 KiB
Python
import pytest
|
|
from nanoid import generate as generate_nanoid
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src import crud, models, schemas
|
|
from src.deriver.enqueue import create_dream_record
|
|
from src.exceptions import ResourceNotFoundException
|
|
|
|
|
|
class TestSessionCRUD:
|
|
"""Test suite for session CRUD operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_peer_configuration(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_data: tuple[models.Workspace, models.Peer],
|
|
):
|
|
"""Test retrieving peer configuration data from session"""
|
|
test_workspace, test_peer = sample_data
|
|
|
|
# Create another peer
|
|
peer2 = models.Peer(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
db_session.add(peer2)
|
|
await db_session.flush()
|
|
|
|
# Create session
|
|
test_session = models.Session(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
db_session.add(test_session)
|
|
await db_session.flush()
|
|
|
|
# Add peers to session with different configurations
|
|
peer_configs = {
|
|
test_peer.name: schemas.SessionPeerConfig(
|
|
observe_others=True, observe_me=False
|
|
),
|
|
peer2.name: schemas.SessionPeerConfig(
|
|
observe_others=False, observe_me=True
|
|
),
|
|
}
|
|
|
|
# Set up peers in session
|
|
await crud.set_peers_for_session(
|
|
db_session,
|
|
workspace_name=test_workspace.name,
|
|
session_name=test_session.name,
|
|
peer_names=peer_configs,
|
|
)
|
|
|
|
# Test the get_session_peer_configuration function
|
|
stmt = await crud.get_session_peer_configuration(
|
|
workspace_name=test_workspace.name,
|
|
session_name=test_session.name,
|
|
)
|
|
result = await db_session.execute(stmt)
|
|
configurations = result.all()
|
|
|
|
# Should return configurations for all active peers
|
|
assert len(configurations) == 2
|
|
|
|
# Verify the structure of returned data
|
|
for peer_name, peer_config, session_peer_config, is_active in configurations:
|
|
assert isinstance(peer_name, str)
|
|
assert isinstance(peer_config, dict) or peer_config is None
|
|
assert isinstance(session_peer_config, dict)
|
|
assert isinstance(is_active, bool)
|
|
|
|
# Check that session_peer_config matches what we set
|
|
expected_config = peer_configs[peer_name]
|
|
assert (
|
|
session_peer_config["observe_others"] == expected_config.observe_others
|
|
)
|
|
assert session_peer_config["observe_me"] == expected_config.observe_me
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_not_found(self, db_session: AsyncSession):
|
|
"""Test get_session with non-existent session raises ResourceNotFoundException"""
|
|
with pytest.raises(ResourceNotFoundException):
|
|
await crud.get_session(db_session, "nonexistent", "nonexistent_workspace")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_peer_config_not_found(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_data: tuple[models.Workspace, models.Peer],
|
|
):
|
|
"""Test get_peer_config with non-existent peer raises ResourceNotFoundException"""
|
|
test_workspace, _test_peer = sample_data
|
|
|
|
# Create session
|
|
test_session = models.Session(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
db_session.add(test_session)
|
|
await db_session.flush()
|
|
|
|
with pytest.raises(ResourceNotFoundException):
|
|
await crud.get_peer_config(
|
|
db_session, test_workspace.name, test_session.name, "nonexistent_peer"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_session_not_found(self, db_session: AsyncSession):
|
|
"""Test clone_session with non-existent session raises ResourceNotFoundException"""
|
|
with pytest.raises(ResourceNotFoundException):
|
|
await crud.clone_session(db_session, "workspace", "nonexistent_session")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_session_invalid_cutoff_message(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_data: tuple[models.Workspace, models.Peer],
|
|
):
|
|
"""Test clone_session with invalid cutoff message raises ValueError"""
|
|
test_workspace, _test_peer = sample_data
|
|
|
|
# Create session
|
|
test_session = models.Session(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
db_session.add(test_session)
|
|
await db_session.flush()
|
|
|
|
# Try to clone with invalid cutoff message ID
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="Message not found or doesn't belong to the specified session",
|
|
):
|
|
await crud.clone_session(
|
|
db_session, test_workspace.name, test_session.name, "invalid_message_id"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_session_clears_dangling_dream_session_hint(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_data: tuple[models.Workspace, models.Peer],
|
|
):
|
|
"""delete_session drops payload.session_name from dreams referencing it.
|
|
|
|
Dream queue items carry session_id=None and a work_unit_key without the
|
|
session name, so they survive the deletes above and would fail resolving
|
|
the gone session. Other dreams must be left untouched.
|
|
"""
|
|
test_workspace, test_peer = sample_data
|
|
|
|
test_session = models.Session(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
other_session = models.Session(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
db_session.add_all([test_session, other_session])
|
|
await db_session.flush()
|
|
|
|
def dream(observed: str, session_name: str | None) -> models.QueueItem:
|
|
# Distinct observed peers: the dream work_unit_key omits session_name,
|
|
# so same-peer dreams would collide on the pending-dream unique index.
|
|
return models.QueueItem(
|
|
**create_dream_record(
|
|
test_workspace.name,
|
|
observer=test_peer.name,
|
|
observed=observed,
|
|
dream_type=schemas.DreamType.OMNI,
|
|
session_name=session_name,
|
|
)
|
|
)
|
|
|
|
db_session.add_all(
|
|
[
|
|
dream("observed-deleted", test_session.name),
|
|
dream("observed-other", other_session.name),
|
|
dream("observed-global", None),
|
|
]
|
|
)
|
|
await db_session.flush()
|
|
|
|
await crud.delete_session(
|
|
db_session,
|
|
workspace_name=test_workspace.name,
|
|
session_name=test_session.name,
|
|
)
|
|
|
|
dreams = (
|
|
(
|
|
await db_session.execute(
|
|
select(models.QueueItem).where(
|
|
models.QueueItem.workspace_name == test_workspace.name
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
# All three dreams survive; only the one pointing at the deleted session
|
|
# loses its hint, leaving it workspace-scoped.
|
|
assert {
|
|
item.payload["observed"]: item.payload.get("session_name")
|
|
for item in dreams
|
|
} == {
|
|
"observed-deleted": None,
|
|
"observed-other": other_session.name,
|
|
"observed-global": None,
|
|
}
|