File size: 4,282 Bytes
4295a74 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Integration test fixtures using real DBs in containers.
Per v4.0 §T21. Mocks prove the code compiles — real DBs prove it works.
"""
import pytest
import pytest_asyncio
@pytest.fixture(scope="session")
def postgres_container():
from testcontainers.postgres import PostgresContainer
pg = PostgresContainer("postgres:16-alpine")
pg.start()
yield pg
pg.stop()
@pytest_asyncio.fixture
async def pg_conn(postgres_container):
import asyncpg
url = postgres_container.get_connection_url()
if url.startswith("postgresql+psycopg2://"):
url = "postgresql://" + url[len("postgresql+psycopg2://"):]
conn = await asyncpg.connect(url)
await _init_pg_schema(conn)
try:
yield conn
finally:
await conn.close()
async def _init_pg_schema(conn) -> None:
await conn.execute("""
CREATE TABLE IF NOT EXISTS tokens (
token_id TEXT PRIMARY KEY,
chain TEXT NOT NULL,
address TEXT NOT NULL,
symbol TEXT,
name TEXT,
decimals INT DEFAULT 18,
deployer_wallet_id TEXT,
deployed_at TIMESTAMPTZ DEFAULT NOW(),
initial_supply BIGINT,
current_supply BIGINT,
is_honeypot BOOLEAN,
is_mintable BOOLEAN,
is_proxy BOOLEAN,
tax_buy_bps INT,
tax_sell_bps INT,
risk_tier TEXT,
risk_score INT,
risk_factors JSONB DEFAULT '[]',
rag_embedding_id TEXT,
UNIQUE(chain, address)
);
CREATE TABLE IF NOT EXISTS wallets (
wallet_id TEXT PRIMARY KEY,
chain TEXT NOT NULL,
address TEXT NOT NULL,
first_seen TIMESTAMPTZ DEFAULT NOW(),
last_seen TIMESTAMPTZ DEFAULT NOW(),
tx_count INT DEFAULT 0,
total_volume_usd DOUBLE PRECISION DEFAULT 0.0,
is_deployer BOOLEAN DEFAULT FALSE,
reputation_score INT
);
CREATE TABLE IF NOT EXISTS news_items (
news_id TEXT PRIMARY KEY,
url TEXT NOT NULL,
title TEXT NOT NULL,
summary TEXT,
published_at TIMESTAMPTZ DEFAULT NOW(),
ingested_at TIMESTAMPTZ DEFAULT NOW(),
source TEXT,
sentiment_score DOUBLE PRECISION,
body_markdown TEXT,
chains_mentioned JSONB DEFAULT '[]',
tokens_mentioned JSONB DEFAULT '[]',
wallets_mentioned JSONB DEFAULT '[]',
ai_analysis TEXT
);
CREATE TABLE IF NOT EXISTS scan_reports (
report_id TEXT PRIMARY KEY,
subject_type TEXT NOT NULL,
subject_id TEXT NOT NULL,
generated_at TIMESTAMPTZ DEFAULT NOW(),
generated_by_model TEXT,
risk_score INT,
risk_tier TEXT,
markdown_url TEXT
);
""")
@pytest.fixture(scope="session")
def redis_container():
from testcontainers.redis import RedisContainer
r = RedisContainer("redis:7-alpine")
r.start()
yield r
r.stop()
@pytest_asyncio.fixture
async def redis_client(redis_container):
import redis.asyncio as aioredis
try:
url = redis_container.get_connection_url()
except AttributeError:
host = redis_container.get_container_host_ip()
port = redis_container.get_exposed_port(6379)
url = f"redis://{host}:{port}"
client = aioredis.from_url(url)
try:
yield client
finally:
try:
await client.flushdb()
except Exception:
pass
await client.aclose()
@pytest.fixture(scope="session")
def neo4j_container():
from testcontainers.neo4j import Neo4jContainer
n = Neo4jContainer("neo4j:5")
n.start()
yield n
n.stop()
@pytest_asyncio.fixture
async def neo4j_driver(neo4j_container):
from neo4j import AsyncGraphDatabase
# testcontainers Neo4jContainer exposes `.password` attribute
password = getattr(neo4j_container, "password", "admin")
uri = neo4j_container.get_connection_url()
driver = AsyncGraphDatabase.driver(uri, auth=("neo4j", password))
try:
yield driver
finally:
await driver.close() |