87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
"""
|
||
pytest configuration and shared fixtures for zhiyi tests.
|
||
"""
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# Add src to path for imports
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from datetime import datetime
|
||
from typing import List
|
||
|
||
from models.episode import Episode
|
||
from models.distilled import Distilled, TYPE_FACT
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fake embedder (list-based, no real bge-m3 model)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class FakeEmbedder:
|
||
"""Simple list-based fake embedder for fast tests.
|
||
|
||
Returns a fixed-size vector of ones. The embedding dimension is
|
||
configurable (default 4) so tests run fast without loading the real
|
||
bge-m3 model.
|
||
"""
|
||
|
||
def __init__(self, dim: int = 4):
|
||
self.dim = dim
|
||
|
||
def encode(self, texts: List[str], **kwargs) -> List[List[float]]:
|
||
"""Return fake vectors – one per input text."""
|
||
return [[1.0] * self.dim for _ in texts]
|
||
|
||
async def encode_async(self, texts: List[str], **kwargs) -> List[List[float]]:
|
||
"""Async wrapper – delegates to encode()."""
|
||
return self.encode(texts, **kwargs)
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_embedder():
|
||
"""Provides a FakeEmbedder instance."""
|
||
return FakeEmbedder(dim=4)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Shared data fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.fixture
|
||
def sample_episode():
|
||
"""A basic Episode instance for use across tests."""
|
||
return Episode.create(
|
||
content='Alice met Bob at the cafe yesterday.',
|
||
source='test',
|
||
entities=['Alice', 'Bob'],
|
||
facts=['Alice met Bob at a cafe'],
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_distilled():
|
||
"""A basic Distilled instance for use across tests."""
|
||
ep = Episode.create(content='Test content')
|
||
return Distilled(
|
||
id='distilled-1',
|
||
episode_id=ep.id,
|
||
type=TYPE_FACT,
|
||
summary='Test summary of the distilled content.',
|
||
importance=0.5,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Async event loop fixture
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest_asyncio.fixture
|
||
async def async_event_loop():
|
||
"""Provides a fresh event loop for each async test."""
|
||
import asyncio
|
||
loop = asyncio.new_event_loop()
|
||
yield loop
|
||
loop.close() |