trumpsignal-backend/tests/test_kol_x.py

"""Tests for the X (Twitter) KOL ingester (kol_x).

The twitterapi.io fetch and the AI scorer are mocked so we exercise the
storage / dedup / mapping logic deterministically — no network, no AI spend.

The contract these lock down:
  - tweets land as KolPost(source="twitter")
  - kol_handle is the CANONICAL handle (joins to KolWallet), not the X username
  - bare retweets are skipped before scoring
  - tickers_json keeps the {ticker, action, conviction} shape kol_divergence reads
  - tier / post_type / talks_vs_trades_flag / sentiment are persisted (migration 027)
  - re-running is idempotent (dedup by tweet id)
  - page-level early-stop fires when a full page is all-dedup
  - no twitterapi key → full no-op
"""

from __future__ import annotations

import json
from typing import AsyncGenerator

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

from app.config import settings
from app.models import Base, KolPost
from app.services import kol_x


_FAKE_TWEETS = [
    {  # actionable position statement → should be stored + scored
        "id": "111",
        "text": "I just dumped my entire $HYPE position",
        "url": "https://x.com/CryptoHayes/status/111",
        "createdAt": "Thu Jun 04 05:49:13 +0000 2026",
        "author": {"followers": 797330},
    },
    {  # bare retweet → skipped before the AI call
        "id": "222",
        "text": "RT @someone: not my words",
        "createdAt": "Thu Jun 04 06:00:00 +0000 2026",
        "author": {"followers": 797330},
    },
    {  # low-signal but original → stored (AI decides noise/not)
        "id": "333",
        "text": "gm crypto fam",
        "createdAt": "Thu Jun 04 07:00:00 +0000 2026",
        "author": {"followers": 797330},
    },
]

_FAKE_SCORE = {
    "post_type": "original",
    "tier": "trade_signal",
    "summary": "Dumped HYPE",
    "tickers": [{"ticker": "HYPE", "action": "sell", "conviction": 0.95}],
    "talks_vs_trades_flag": True,
    "sentiment": "bearish",
    "model": "test-model",
    "version": "x-test",
    "error": None,
}

_KOL = {"handle": "cryptohayes", "x_username": "CryptoHayes", "display_name": "Hayes"}


async def _fresh_session_factory():
    """Each test gets its own isolated in-memory DB."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


def _make_page_iter(pages: list[list[dict]]):
    """Return an async generator that yields one page at a time from `pages`."""
    async def fake_iter(_username, max_pages=3) -> AsyncGenerator[list[dict], None]:
        for page in pages:
            yield page
    return fake_iter


def _patch_io(monkeypatch, pages: list[list[dict]] | None = None):
    if pages is None:
        pages = [list(_FAKE_TWEETS)]  # one page containing all fake tweets

    async def fake_score(**_kw):
        return dict(_FAKE_SCORE)

    monkeypatch.setattr(settings, "twitterapi_io_key", "test-key")
    monkeypatch.setattr(kol_x, "_iter_tweet_pages", _make_page_iter(pages))
    monkeypatch.setattr(kol_x.x_analysis, "analyze_x_post", fake_score)


# ── Core contract ─────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_run_x_poll_noop_without_key(monkeypatch):
    monkeypatch.setattr(settings, "twitterapi_io_key", "")
    assert await kol_x.run_x_poll() == []


@pytest.mark.asyncio
async def test_ingest_writes_divergence_ready_kolposts(monkeypatch):
    _patch_io(monkeypatch)
    sf = await _fresh_session_factory()

    async with sf() as s:
        stats = await kol_x._ingest_kol_x(s, _KOL)

        # 111 + 333 stored; 222 (bare RT) skipped before scoring
        assert stats["new"] == 2
        assert stats["skipped"] == 1
        assert stats["analyzed"] == 2
        assert stats["errors"] == 0

        rows = (await s.execute(
            select(KolPost).where(KolPost.source == "twitter")
        )).scalars().all()
        assert {r.external_id for r in rows} == {"111", "333"}
        # canonical handle (joins to KolWallet), NOT the X screen name
        assert all(r.kol_handle == "cryptohayes" for r in rows)

        signal = next(r for r in rows if r.external_id == "111")
        tk = json.loads(signal.tickers_json)
        assert tk[0]["ticker"] == "HYPE"
        assert tk[0]["action"] == "sell"        # in kol_divergence._POST_SHORT
        assert tk[0]["conviction"] == 0.95
        assert signal.analysis_model == "test-model"
        assert signal.analysis_version == "x-test"
        # published_at parsed from the X date string into a naive datetime
        assert signal.published_at.year == 2026 and signal.published_at.month == 6


@pytest.mark.asyncio
async def test_ingest_stores_extended_analysis_fields(monkeypatch):
    """migration 027 fields — tier / post_type / talks_vs_trades_flag / sentiment."""
    _patch_io(monkeypatch)
    sf = await _fresh_session_factory()

    async with sf() as s:
        await kol_x._ingest_kol_x(s, _KOL)
        signal = (await s.execute(
            select(KolPost).where(KolPost.external_id == "111")
        )).scalar_one()

        assert signal.tier == "trade_signal"
        assert signal.post_type == "original"
        assert signal.talks_vs_trades_flag is True
        assert signal.sentiment == "bearish"


@pytest.mark.asyncio
async def test_ingest_is_idempotent_on_rerun(monkeypatch):
    _patch_io(monkeypatch)
    sf = await _fresh_session_factory()

    async with sf() as s:
        first = await kol_x._ingest_kol_x(s, _KOL)
        second = await kol_x._ingest_kol_x(s, _KOL)

    assert first["new"] == 2
    assert second["new"] == 0            # everything already stored
    assert second["skipped"] == 3        # 2 existing + 1 bare RT


@pytest.mark.asyncio
async def test_fetch_failure_does_not_raise(monkeypatch):
    """A dead/blocked KOL must not abort the run — empty pages → empty stats."""
    monkeypatch.setattr(settings, "twitterapi_io_key", "test-key")
    _patch_io(monkeypatch, pages=[])     # generator yields nothing
    sf = await _fresh_session_factory()
    async with sf() as s:
        stats = await kol_x._ingest_kol_x(s, _KOL)
    assert stats["new"] == 0 and stats["errors"] == 0


# ── Multi-page & early-stop ───────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_multipage_fetches_all_new_tweets(monkeypatch):
    """Two pages of distinct tweets → both pages are stored."""
    page1 = [{"id": "p1_1", "text": "SOL is the play", "createdAt": "Thu Jun 04 05:00:00 +0000 2026", "author": {"followers": 100000}}]
    page2 = [{"id": "p2_1", "text": "ETH too slow ngmi", "createdAt": "Wed Jun 03 10:00:00 +0000 2026", "author": {"followers": 100000}}]
    _patch_io(monkeypatch, pages=[page1, page2])
    sf = await _fresh_session_factory()

    async with sf() as s:
        stats = await kol_x._ingest_kol_x(s, _KOL)

    assert stats["new"] == 2


@pytest.mark.asyncio
async def test_early_stop_when_page_fully_deduped(monkeypatch):
    """After page1 is stored, a second run should detect all-dedup on page1
    and break before page2 is consumed — verified by keeping a 'new' tweet
    on page2 that must NOT appear in the DB if early-stop fired correctly."""
    page1 = [{"id": "old1", "text": "BTC to the moon",
               "createdAt": "Thu Jun 04 05:00:00 +0000 2026", "author": {}}]
    page2 = [{"id": "brand_new", "text": "Sold everything",
               "createdAt": "Wed Jun 03 09:00:00 +0000 2026", "author": {}}]

    run = 0
    pages_yielded: list[list] = []

    async def counting_iter(_username, max_pages=3):
        nonlocal run
        run += 1
        if run == 1:
            # First run: only page1 (simulates single-page normal daily run)
            pages_yielded.append(page1)
            yield page1
        else:
            # Second run: both pages available, but early-stop should fire at page1
            for page in [page1, page2]:
                pages_yielded.append(page)
                yield page

    async def fake_score(**_kw):
        return dict(_FAKE_SCORE)

    monkeypatch.setattr(settings, "twitterapi_io_key", "test-key")
    monkeypatch.setattr(kol_x, "_iter_tweet_pages", counting_iter)
    monkeypatch.setattr(kol_x.x_analysis, "analyze_x_post", fake_score)

    sf = await _fresh_session_factory()
    async with sf() as s:
        first = await kol_x._ingest_kol_x(s, _KOL)
        assert first["new"] == 1          # page1 stored (1 tweet)

        pages_yielded.clear()
        second = await kol_x._ingest_kol_x(s, _KOL)
        assert second["new"] == 0         # page1 all deduped → early stop
        # Only page1 should have been yielded — early-stop broke before page2
        assert pages_yielded == [page1]

        # page2's tweet must NOT be in the DB (generator closed before it was yielded)
        rows = (await s.execute(
            select(KolPost).where(KolPost.source == "twitter")
        )).scalars().all()
        assert {r.external_id for r in rows} == {"old1"}


@pytest.mark.asyncio
async def test_all_retweet_page_does_not_early_stop(monkeypatch):
    """A page that is ALL bare retweets (page_new=0, page_deduped=0) must NOT
    trigger early-stop — the next page can still hold original posts. Regression
    guard: previously 'page_new == 0' alone stopped paging on RT-heavy pages,
    silently dropping originals on older pages."""
    page1 = [
        {"id": "rt1", "text": "RT @a: gm",    "createdAt": "Thu Jun 04 05:00:00 +0000 2026", "author": {}},
        {"id": "rt2", "text": "RT @b: wagmi", "createdAt": "Thu Jun 04 04:00:00 +0000 2026", "author": {}},
    ]
    page2 = [
        {"id": "orig1", "text": "Just sold all my $SOL", "createdAt": "Wed Jun 03 09:00:00 +0000 2026", "author": {}},
    ]
    _patch_io(monkeypatch, pages=[page1, page2])
    sf = await _fresh_session_factory()

    async with sf() as s:
        stats = await kol_x._ingest_kol_x(s, _KOL)
        # page1 all-RT → skipped 2, NO early-stop → page2 consumed → orig1 stored
        assert stats["new"] == 1
        assert stats["skipped"] == 2
        rows = (await s.execute(
            select(KolPost).where(KolPost.source == "twitter")
        )).scalars().all()
        assert {r.external_id for r in rows} == {"orig1"}