53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
import os
|
|
import sqlite3
|
|
import sys
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
PROJECT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
if PROJECT_DIR not in sys.path:
|
|
sys.path.insert(0, PROJECT_DIR)
|
|
|
|
from app.db import altcoin_db
|
|
from app.services.event_driven_screener import init_event_tables
|
|
from app.web import web_server
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db(monkeypatch, tmp_path):
|
|
db_path = tmp_path / "altcoin_monitor.db"
|
|
monkeypatch.setattr(altcoin_db, "DB_PATH", str(db_path))
|
|
monkeypatch.setattr(web_server, "init_db", altcoin_db.init_db)
|
|
monkeypatch.setenv("ALPHAX_DB_PATH", str(db_path))
|
|
altcoin_db.init_db()
|
|
init_event_tables()
|
|
return db_path
|
|
|
|
|
|
def _insert_event(db_path, title, event_type, symbol="SOL/USDT"):
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO event_news (
|
|
event_hash, source, symbol, title, url, published_at, detected_at,
|
|
importance, event_type, raw_json, processed, decision, tech_score, rec_id, pushed
|
|
) VALUES (?, ?, ?, ?, ?, datetime('now'), datetime('now'), ?, ?, '{}', 0, '', 0, 0, 0)
|
|
""",
|
|
(f"hash-{event_type}-{title}", "coingecko", symbol, title, "https://example.com", "A", event_type),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def test_sentiment_api_hides_internal_theme_expansion_events(temp_db):
|
|
_insert_event(temp_db, "[主题扩散:solana_meme] SOL(Solana) enters CoinGecko Trending #4", "theme_expansion")
|
|
_insert_event(temp_db, "Binance Will List ABCUSDT Perpetual Contracts", "major_listing_or_contract", "ABC/USDT")
|
|
|
|
client = TestClient(web_server.app)
|
|
resp = client.get("/api/sentiment?hours=24")
|
|
assert resp.status_code == 200
|
|
titles = [item["title"] for item in resp.json()["events"]]
|
|
assert "Binance Will List ABCUSDT Perpetual Contracts" in titles
|
|
assert not any("主题扩散" in title for title in titles)
|