bc479802db
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Pipeline #46 surfaced two pre-existing CI bugs once fakeredis was installed and tests could collect: 1. test_models.py:389 asserted "DISCOVERED" in status_col.type.enums, but the model defines KevinVideoStatus with values_callable so .enums returns the lowercase string values, not member names. Asserting "discovered" instead. 2. Four test files use the db_session fixture which requires a real Postgres on localhost:5432. CI has no Postgres, so 10 tests failed with Connect call failed (errno 111). These genuinely need a DB — mirroring tests/integration/* which already use @pytest.mark.integration. Adding module-level pytestmark = pytest.mark.integration to: - tests/shared/models/test_meet_kevin_trading.py - tests/services/kevin_signal_bridge/test_aggregator.py - tests/services/kevin_signal_bridge/test_audit.py - tests/services/kevin_signal_bridge/test_exit_scanner.py CI runs with -m "not integration" so they're now deselected. Local pytest still picks them up by default (no marker filter).
119 lines
3.3 KiB
Python
119 lines
3.3 KiB
Python
"""Tests for the audit writer (upsert into kevin_signal_bridge_state)."""
|
|
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
from services.kevin_signal_bridge.audit import AuditWriter
|
|
from shared.models.meet_kevin import (
|
|
KevinAnalysis,
|
|
KevinChannel,
|
|
KevinMarketOutlook,
|
|
KevinStockMention,
|
|
KevinTickerAction,
|
|
KevinTimeHorizon,
|
|
KevinVideo,
|
|
KevinVideoStatus,
|
|
)
|
|
from shared.models.meet_kevin_trading import KevinSignalBridgeState
|
|
|
|
|
|
def _factory(session: AsyncSession):
|
|
class _StaticSessionFactory:
|
|
async def __aenter__(self):
|
|
return session
|
|
|
|
async def __aexit__(self, *args):
|
|
pass
|
|
|
|
def factory():
|
|
return _StaticSessionFactory()
|
|
|
|
return factory
|
|
|
|
|
|
async def _seed_mention(session: AsyncSession) -> int:
|
|
channel = KevinChannel(youtube_channel_id="UCa", title="t")
|
|
session.add(channel)
|
|
await session.flush()
|
|
video = KevinVideo(
|
|
channel_id=channel.id,
|
|
youtube_video_id="va",
|
|
title="t",
|
|
published_at=datetime.now(timezone.utc),
|
|
status=KevinVideoStatus.ANALYZED,
|
|
)
|
|
session.add(video)
|
|
await session.flush()
|
|
analysis = KevinAnalysis(
|
|
video_id=video.id,
|
|
model="m",
|
|
prompt_version="v1",
|
|
market_outlook_direction=KevinMarketOutlook.NEUTRAL,
|
|
market_outlook_reasoning="x",
|
|
summary="x",
|
|
prompt_tokens=10,
|
|
completion_tokens=10,
|
|
cost_usd=Decimal("0.01"),
|
|
)
|
|
session.add(analysis)
|
|
await session.flush()
|
|
mention = KevinStockMention(
|
|
video_id=video.id,
|
|
analysis_id=analysis.id,
|
|
symbol="NVDA",
|
|
action=KevinTickerAction.BUY,
|
|
conviction=Decimal("0.7"),
|
|
time_horizon=KevinTimeHorizon.WEEKS,
|
|
rationale_quote="x",
|
|
)
|
|
session.add(mention)
|
|
await session.flush()
|
|
return mention.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_writer_inserts_new_row(db_session: AsyncSession):
|
|
mention_id = await _seed_mention(db_session)
|
|
writer = AuditWriter(session_factory=_factory(db_session))
|
|
await writer.write(
|
|
mention_id=mention_id,
|
|
bridge_status="emitted",
|
|
effective_conviction=Decimal("0.75"),
|
|
notes="hello",
|
|
)
|
|
row = (
|
|
await db_session.execute(
|
|
select(KevinSignalBridgeState).where(
|
|
KevinSignalBridgeState.mention_id == mention_id
|
|
)
|
|
)
|
|
).scalar_one()
|
|
assert row.bridge_status.value == "emitted"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_writer_upserts_on_duplicate(db_session: AsyncSession):
|
|
mention_id = await _seed_mention(db_session)
|
|
writer = AuditWriter(session_factory=_factory(db_session))
|
|
await writer.write(mention_id=mention_id, bridge_status="dry_run", notes="first")
|
|
await writer.write(mention_id=mention_id, bridge_status="emitted", notes="second")
|
|
rows = (
|
|
(
|
|
await db_session.execute(
|
|
select(KevinSignalBridgeState).where(
|
|
KevinSignalBridgeState.mention_id == mention_id
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
assert len(rows) == 1
|
|
assert rows[0].bridge_status.value == "emitted"
|
|
assert rows[0].notes == "second"
|