Adriano Dal Pastro
197c9db74d
Usa il group 'audit' esistente (sibling di 'audit verify' per la hash chain). Opzioni: --db, --since DAYS, --asset ETH|BTC, --json. Output stdout Rich di default, dump JSON con --json. Esce con codice 2 su sqlite3.OperationalError (DB malformato/schema mancante). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""Smoke tests for the ``cerbero-bite audit data`` CLI subcommand."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from click.testing import CliRunner
|
|
|
|
from cerbero_bite.cli import main as cli_main
|
|
from cerbero_bite.state import connect, run_migrations, transaction
|
|
|
|
|
|
def _seed_minimal_db(db_path: Path) -> None:
|
|
conn = connect(db_path)
|
|
try:
|
|
run_migrations(conn)
|
|
now = datetime.now(UTC).replace(microsecond=0)
|
|
with transaction(conn):
|
|
for asset, spot in (("ETH", "3000"), ("BTC", "80000")):
|
|
conn.execute(
|
|
"INSERT INTO market_snapshots(timestamp, asset, spot, "
|
|
"dvol, fetch_ok) VALUES (?,?,?,?,1)",
|
|
(now.isoformat(), asset, spot, "55"),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO option_chain_snapshots(timestamp, asset, "
|
|
"instrument_name, strike, expiry, option_type, bid, ask, iv) "
|
|
"VALUES (?, 'ETH', 'ETH-X', '3000', ?, 'C', '0.01', '0.02', '0.7')",
|
|
(now.isoformat(), now.isoformat()),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_audit_data_human_output(tmp_path: Path) -> None:
|
|
db = tmp_path / "state.sqlite"
|
|
_seed_minimal_db(db)
|
|
result = CliRunner().invoke(
|
|
cli_main, ["audit", "data", "--db", str(db), "--since", "1"]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "ETH — market_snapshots" in result.output
|
|
assert "BTC — market_snapshots" in result.output
|
|
assert "ETH — option_chain_snapshots" in result.output
|
|
|
|
|
|
def test_audit_data_json_output(tmp_path: Path) -> None:
|
|
db = tmp_path / "state.sqlite"
|
|
_seed_minimal_db(db)
|
|
result = CliRunner().invoke(
|
|
cli_main,
|
|
["audit", "data", "--db", str(db), "--since", "1", "--json"],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert "since" in payload
|
|
assert "until" in payload
|
|
assert set(payload["assets"]) == {"ETH", "BTC"}
|
|
assert "market" in payload["assets"]["ETH"]
|
|
assert "chain" in payload["assets"]["ETH"]
|
|
assert payload["assets"]["ETH"]["market"]["asset"] == "ETH"
|
|
|
|
|
|
def test_audit_data_single_asset_filter(tmp_path: Path) -> None:
|
|
db = tmp_path / "state.sqlite"
|
|
_seed_minimal_db(db)
|
|
result = CliRunner().invoke(
|
|
cli_main,
|
|
["audit", "data", "--db", str(db), "--asset", "BTC", "--json"],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert set(payload["assets"]) == {"BTC"}
|
|
|
|
|
|
def test_audit_data_missing_db_exits_2(tmp_path: Path) -> None:
|
|
# click.Path(exists=True) returns exit_code=2 (UsageError) for missing files.
|
|
result = CliRunner().invoke(
|
|
cli_main,
|
|
["audit", "data", "--db", str(tmp_path / "absent.sqlite")],
|
|
)
|
|
assert result.exit_code == 2
|