diff --git a/automation/jobs.yaml b/automation/jobs.yaml index b60f0f5..9db0ea9 100644 --- a/automation/jobs.yaml +++ b/automation/jobs.yaml @@ -61,6 +61,7 @@ profiles: - { name: "stables-oracles", script: protocols/stables/oracles.py } - { name: "cap-status", script: protocols/cap/status.py } - { name: "yearn-alert-large-flows", script: protocols/yearn/alert_large_flows.py } + - { name: "yearn-alert-small-parent-flows", script: protocols/yearn/alert_small_parent_flows.py } # Cache: tks-trigger-cache.json under $CACHE_DIR (check_stuck_triggers.DEFAULT_CACHE_FILE). - { name: "yearn-check-stuck-triggers", script: protocols/yearn/check_stuck_triggers.py, enabled: false } - { name: "maple", script: protocols/maple/main.py, enabled: false } diff --git a/monitoring.yaml b/monitoring.yaml index 09a6755..3bbeb10 100644 --- a/monitoring.yaml +++ b/monitoring.yaml @@ -460,6 +460,7 @@ protocols: tasks: - protocols/yearn/lender_borrower.py - protocols/yearn/alert_large_flows.py + - protocols/yearn/alert_small_parent_flows.py - protocols/yearn/check_timelock_delay.py - protocols/yearn/check_indexer_freshness.py monitors: @@ -470,6 +471,8 @@ protocols: severity: "MEDIUM" - name: "Large Flows" description: "Deposit/withdrawal flows >=$500k USD (Katana withdrawals >=$50k; or 10% of vault totalSupply fallback for unpriced tokens)" + - name: "Small Parent Vault Flows" + description: "Individual deposits and withdrawals with a raw ERC-4626 assets value below 10,000 for active Yearn v3 parent vaults" - name: "Timelock Delay" description: "Yearn TimelockController getMinDelay() < 7 days across Mainnet, Base, Arbitrum, Polygon, Optimism, Katana" - name: "Timelock Events" diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 27a5d09..f99a444 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -55,6 +55,31 @@ Optional flags: - `--chain-ids` (default: all vault chain IDs — `1,8453,42161,747474`) - `--no-cache` (disable caching) +## Small Parent Vault Flows + +The script `yearn/alert_small_parent_flows.py` alerts on every deposit or withdrawal whose raw ERC-4626 `assets` value is strictly between 0 and 10,000 for an active Yearn v3 parent vault. The comparison happens before decimal normalization: 10,000 raw units equals 0.01 USDC, 0.0001 WBTC, or 0.00000000000001 WETH. + +### Data Sources + +- **Parent vault discovery**: Kong GraphQL, filtered to Yearn v3 `vaultType: 1` vaults and excluding retired or hidden entries. +- **Flow events**: Envio `Deposit` and `Withdraw` entities. Alerts include the ERC-4626 owner and sender, the transaction initiator, and the asset receiver for withdrawals. +- **Token decimals**: the parent vault's underlying asset metadata from Kong, used only to show a human-readable amount alongside the raw value. + +Deposits and withdrawals are processed with independent per-chain `(blockNumber, logIndex)` cursors stored in the monitoring database. A cursor advances only after an event is successfully evaluated and, when applicable, delivered to Telegram. A new deployment starts each stream with a two-hour lookback. + +### Usage + +```bash +uv run protocols/yearn/alert_small_parent_flows.py +``` + +Optional flags: + +- `--threshold-raw` (default: `10000`) +- `--lookback-seconds` (default: `7200`, used only before a chain cursor exists) +- `--page-size` (default: `1000`) +- `--chain-ids` (default: `1,10,8453,42161,137,747474`) + ======= ## Shadow Debt Check diff --git a/protocols/yearn/alert_small_parent_flows.py b/protocols/yearn/alert_small_parent_flows.py new file mode 100644 index 0000000..3eb4e79 --- /dev/null +++ b/protocols/yearn/alert_small_parent_flows.py @@ -0,0 +1,417 @@ +#!/usr/bin/env python3 +"""Alert on small deposits and withdrawals from Yearn v3 parent vaults.""" + +from __future__ import annotations + +import argparse +import json +import logging +import os +import sys +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from decimal import Decimal, getcontext +from typing import Callable + +from dotenv import load_dotenv + +from protocols.yearn.kong import fetch_kong_parent_vaults +from utils import store +from utils.alert import Alert, AlertSeverity, send_alert +from utils.chains import EXPLORER_URLS, Chain +from utils.logger import get_logger +from utils.telegram import send_envio_error_message + +load_dotenv() + +getcontext().prec = 60 + +ENVIO_GRAPHQL_URL = os.getenv("ENVIO_GRAPHQL_URL") +DEFAULT_LOG_LEVEL = os.getenv("SMALL_PARENT_FLOWS_LOG_LEVEL", "INFO") +DEFAULT_THRESHOLD_RAW = 10_000 +DEFAULT_LOOKBACK_SECONDS = 7200 +DEFAULT_PAGE_SIZE = 1000 +PROTOCOL = "yearn" +STATE_NAMESPACE = "yearn.small_parent_flows" +FLOW_TYPES = ("deposit", "withdrawal") +FLOW_ENTITY = {"deposit": "Deposit", "withdrawal": "Withdraw"} + +logger = get_logger("yearn.alert_small_parent_flows") + + +@dataclass(frozen=True, order=True) +class EventCursor: + """Per-chain Envio event cursor.""" + + block_number: int + log_index: int + + +def http_json(url: str, body: dict) -> dict: + """POST a JSON body and return the decoded response.""" + request = urllib.request.Request( + url, + data=json.dumps(body).encode("utf-8"), + headers={"Accept": "application/json", "Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=30) as response: + payload: object = json.loads(response.read().decode("utf-8")) + if not isinstance(payload, dict): + raise ValueError("Envio returned a non-object JSON response") + return payload + + +def gql_request(query: str, variables: dict) -> dict | None: + """Execute an Envio GraphQL query, routing failures to its ops channel.""" + if not ENVIO_GRAPHQL_URL: + raise RuntimeError("ENVIO_GRAPHQL_URL is not set") + + try: + payload = http_json(ENVIO_GRAPHQL_URL, {"query": query, "variables": variables}) + except (urllib.error.HTTPError, urllib.error.URLError, ConnectionError, OSError, ValueError) as exc: + send_envio_error_message( + f"Small parent flow monitor: Envio GraphQL request failed ({exc}). Skipping this run.", + PROTOCOL, + source="small_parent_flows", + ) + logger.error("Envio request failed: %s", exc) + return None + + if payload.get("errors"): + send_envio_error_message( + f"Small parent flow monitor: Envio GraphQL errors: {payload['errors']}", + PROTOCOL, + source="small_parent_flows", + ) + logger.error("Envio GraphQL errors: %s", payload["errors"]) + return None + return payload + + +def load_events( + flow_type: str, + chain_id: int, + vault_addresses: list[str], + cursor: EventCursor, + since_ts: int, + limit: int, +) -> list[dict] | None: + """Load one ordered page of parent-vault flow events after ``cursor``.""" + try: + entity = FLOW_ENTITY[flow_type] + except KeyError as exc: + raise ValueError(f"Unknown flow type: {flow_type}") from exc + + receiver_field = "receiver" if flow_type == "withdrawal" else "" + query = """ + query SmallParentFlows( + $chainId: Int! + $addresses: [String!]! + $lastBlock: Int! + $lastLogIndex: Int! + $sinceTs: Int! + $limit: Int! + ) { + events: __ENTITY__( + where: { + chainId: { _eq: $chainId } + vaultAddress: { _in: $addresses } + _or: [ + { blockNumber: { _gt: $lastBlock }, blockTimestamp: { _gte: $sinceTs } } + { blockNumber: { _eq: $lastBlock }, logIndex: { _gt: $lastLogIndex } } + ] + } + order_by: { blockNumber: asc, logIndex: asc } + limit: $limit + ) { + id + vaultAddress + chainId + blockNumber + blockTimestamp + transactionHash + transactionFrom + logIndex + sender + owner + __RECEIVER_FIELD__ + assets + shares + } + } + """ + query = query.replace("__ENTITY__", entity).replace("__RECEIVER_FIELD__", receiver_field) + variables = { + "chainId": chain_id, + "addresses": vault_addresses, + "lastBlock": cursor.block_number, + "lastLogIndex": cursor.log_index, + "sinceTs": since_ts, + "limit": limit, + } + response = gql_request(query, variables) + if response is None: + return None + events = response.get("data", {}).get("events") + if not isinstance(events, list): + raise RuntimeError(f"Envio response missing {entity} list") + return [{**event, "flow_type": flow_type} for event in events] + + +def format_units(raw_assets: str | int, decimals: int) -> Decimal: + """Convert an integer asset amount into normalized token units.""" + return Decimal(str(raw_assets)) / (Decimal(10) ** decimals) + + +def is_small_flow(raw_assets: str | int, threshold_raw: int) -> bool: + """Return whether a positive raw asset amount is below the threshold.""" + amount_raw = int(str(raw_assets)) + return 0 < amount_raw < threshold_raw + + +def format_amount(amount: Decimal) -> str: + """Format a token amount without scientific notation or trailing zeroes.""" + rendered = f"{amount:,.18f}".rstrip("0").rstrip(".") + return rendered or "0" + + +def address_link(address: str, explorer: str | None) -> str: + """Return a full address, linked to the chain explorer when available.""" + if explorer: + return f"[{address}]({explorer}/address/{address})" + return address + + +def build_alert_message( + event: dict, + vault: dict, + raw_assets: int, + amount: Decimal, + threshold_raw: int, +) -> str: + """Build the Telegram message for one qualifying flow.""" + chain_id = int(event["chainId"]) + chain = Chain.from_chain_id(chain_id) + explorer = EXPLORER_URLS.get(chain_id) + vault_address = str(event["vaultAddress"]) + tx_hash = str(event["transactionHash"]) + tx = f"[{tx_hash}]({explorer}/tx/{tx_hash})" if explorer else tx_hash + flow_type = str(event["flow_type"]) + + lines = [ + f"Small parent-vault {flow_type}", + f"🏦 Vault: {address_link(vault_address, explorer)} ({vault['symbol']})", + f"🔢 Raw Assets: {raw_assets:,}", + f"🪙 Normalized: {format_amount(amount)} {vault['asset_symbol']}", + f"📏 Raw Threshold: < {threshold_raw:,}", + f"⛓️ Chain: {chain.network_name}", + f"👤 Owner: {address_link(str(event['owner']), explorer)}", + f"💳 Sender: {address_link(str(event['sender']), explorer)}", + ] + receiver = event.get("receiver") + if receiver: + lines.append(f"📥 Receiver: {address_link(str(receiver), explorer)}") + transaction_from = event.get("transactionFrom") + if transaction_from: + lines.append(f"🚀 Tx From: {address_link(str(transaction_from), explorer)}") + lines.append(f"🔗 Tx: {tx}") + return "\n".join(lines) + + +def cursor_from_event(event: dict) -> EventCursor: + """Return the sortable cursor represented by an Envio event.""" + return EventCursor(int(event["blockNumber"]), int(event["logIndex"])) + + +def state_key(chain_id: int, flow_type: str) -> str: + """Return the persistent-state key for one chain and flow type.""" + if flow_type not in FLOW_ENTITY: + raise ValueError(f"Unknown flow type: {flow_type}") + return f"{chain_id}:{flow_type}" + + +def load_cursor(chain_id: int, flow_type: str) -> EventCursor | None: + """Load a chain/flow cursor from persistent monitor state.""" + key = state_key(chain_id, flow_type) + raw = store.state_get(STATE_NAMESPACE, key) + if raw is None: + return None + try: + payload = json.loads(raw) + return EventCursor(int(payload["block_number"]), int(payload["log_index"])) + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise RuntimeError(f"Invalid small-flow cursor for {key}: {raw}") from exc + + +def save_cursor(chain_id: int, flow_type: str, cursor: EventCursor) -> None: + """Persist a successfully processed chain/flow cursor.""" + store.state_set( + STATE_NAMESPACE, + state_key(chain_id, flow_type), + json.dumps({"block_number": cursor.block_number, "log_index": cursor.log_index}), + ) + + +def process_event( + event: dict, + vaults_by_address: dict[str, dict], + threshold_raw: int, + alert_sender: Callable[[Alert], None] = send_alert, +) -> bool: + """Evaluate one flow, sending an alert when it is below the threshold.""" + vault_address = str(event["vaultAddress"]).lower() + vault = vaults_by_address.get(vault_address) + if vault is None: + raise RuntimeError(f"Envio returned unknown parent vault {event['vaultAddress']}") + + raw_assets = int(str(event["assets"])) + if not is_small_flow(raw_assets, threshold_raw): + return False + + amount = format_units(raw_assets, int(vault["asset_decimals"])) + message = build_alert_message(event, vault, raw_assets, amount, threshold_raw) + alert_sender(Alert(AlertSeverity.LOW, message, PROTOCOL)) + return True + + +def monitor_flow_type( + chain_id: int, + flow_type: str, + addresses: list[str], + vaults_by_address: dict[str, dict], + threshold_raw: int, + lookback_seconds: int, + page_size: int, + now: int | None = None, +) -> tuple[int, int]: + """Fetch and process all new events of one type for a chain.""" + persisted_cursor = load_cursor(chain_id, flow_type) + cursor = persisted_cursor or EventCursor(0, -1) + since_ts = 0 if persisted_cursor else (now or int(time.time())) - lookback_seconds + processed = 0 + alerted = 0 + + while True: + events = load_events(flow_type, chain_id, addresses, cursor, since_ts, page_size) + if events is None: + break + if not events: + break + + for event in events: + event_cursor = cursor_from_event(event) + if event_cursor <= cursor: + continue + if process_event(event, vaults_by_address, threshold_raw): + alerted += 1 + save_cursor(chain_id, flow_type, event_cursor) + cursor = event_cursor + processed += 1 + + if len(events) < page_size: + break + + return processed, alerted + + +def monitor_chain( + chain: Chain, + threshold_raw: int, + lookback_seconds: int, + page_size: int, + now: int | None = None, +) -> tuple[int, int]: + """Fetch and process deposits and withdrawals for one chain.""" + vaults = fetch_kong_parent_vaults(chain) + if not vaults: + logger.warning("No active parent vaults returned for %s", chain.network_name) + return 0, 0 + + vaults_by_address = {str(vault["address"]).lower(): vault for vault in vaults} + # Envio stores checksummed addresses, while older rows or deployments may + # use lowercase. Supplying both forms keeps the case-sensitive filter safe. + addresses = sorted( + {address for vault in vaults for address in (str(vault["address"]), str(vault["address"]).lower())} + ) + + processed = 0 + alerted = 0 + for flow_type in FLOW_TYPES: + flow_processed, flow_alerted = monitor_flow_type( + chain.chain_id, + flow_type, + addresses, + vaults_by_address, + threshold_raw, + lookback_seconds, + page_size, + now, + ) + processed += flow_processed + alerted += flow_alerted + logger.info( + "%s %s: processed=%d alerted=%d", + chain.network_name, + flow_type, + flow_processed, + flow_alerted, + ) + return processed, alerted + + +def parse_chain_ids(raw: str) -> list[Chain]: + """Parse a comma-separated chain-ID list.""" + chains: list[Chain] = [] + for value in raw.split(","): + if value.strip(): + chains.append(Chain.from_chain_id(int(value.strip()))) + return chains + + +def main() -> None: + """Run the small parent-vault flow monitor.""" + default_chain_ids = ",".join(str(chain.chain_id) for chain in Chain) + parser = argparse.ArgumentParser( + description="Alert on Yearn v3 parent-vault deposits and withdrawals below a raw-assets threshold." + ) + parser.add_argument("--threshold-raw", type=int, default=DEFAULT_THRESHOLD_RAW) + parser.add_argument("--lookback-seconds", type=int, default=DEFAULT_LOOKBACK_SECONDS) + parser.add_argument("--page-size", type=int, default=DEFAULT_PAGE_SIZE) + parser.add_argument("--chain-ids", default=default_chain_ids) + parser.add_argument("--log-level", default=DEFAULT_LOG_LEVEL) + args = parser.parse_args() + + logging.basicConfig( + level=args.log_level.upper(), + format="[%(name)s] %(levelname)s %(message)s", + stream=sys.stderr, + ) + if args.threshold_raw <= 0: + parser.error("--threshold-raw must be positive") + if args.lookback_seconds < 0: + parser.error("--lookback-seconds must be non-negative") + if args.page_size <= 0: + parser.error("--page-size must be positive") + + total_processed = 0 + total_alerted = 0 + for chain in parse_chain_ids(args.chain_ids): + processed, alerted = monitor_chain( + chain, + args.threshold_raw, + args.lookback_seconds, + args.page_size, + ) + total_processed += processed + total_alerted += alerted + logger.info("%s: processed=%d alerted=%d", chain.network_name, processed, alerted) + logger.info("complete: processed=%d alerted=%d", total_processed, total_alerted) + + +if __name__ == "__main__": + from utils.runner import run_with_alert + + run_with_alert(main, PROTOCOL) diff --git a/protocols/yearn/kong.py b/protocols/yearn/kong.py index 279d016..2f20b94 100644 --- a/protocols/yearn/kong.py +++ b/protocols/yearn/kong.py @@ -21,6 +21,26 @@ } } """ +KONG_PARENT_VAULTS_QUERY = """ +query YearnParentVaults($chainId: Int) { + vaults(chainId: $chainId, v3: true, yearn: true, vaultType: 1) { + address + name + symbol + decimals + vaultType + asset { + address + symbol + decimals + } + meta { + isRetired + isHidden + } + } +} +""" STRATEGY_SOURCE_ALL = "strategies" STRATEGY_SOURCE_DEFAULT_QUEUE = "default_queue" @@ -86,6 +106,14 @@ def _is_retired(vault: Dict[str, Any]) -> bool: return bool(meta.get("isRetired")) +def _is_hidden(vault: Dict[str, Any]) -> bool: + """Return whether Kong metadata marks the vault hidden.""" + meta = vault.get("meta") + if not isinstance(meta, dict): + return False + return bool(meta.get("isHidden")) + + def _strategy_field(strategy_source: str) -> str: """Return the Kong field backing the requested strategy source.""" if strategy_source == STRATEGY_SOURCE_ALL: @@ -138,3 +166,54 @@ def fetch_kong_vaults( ) return result + + +def fetch_kong_parent_vaults(chain: Chain) -> List[Dict[str, object]]: + """Fetch active Yearn v3 parent/allocator vault metadata from Kong. + + Kong's ``vaultType: 1`` identifies parent/allocator vaults, while + ``vaultType: 2`` identifies strategy vaults. Retired and hidden vaults are + excluded locally so callers only monitor active user-facing parents. + + Args: + chain: Chain to fetch. + + Returns: + Parent vault dicts containing vault and underlying-asset metadata. + + Raises: + KongRequestError: If Kong omits required address or decimal metadata. + """ + data = _post_graphql(KONG_PARENT_VAULTS_QUERY, {"chainId": chain.chain_id}) + vaults = data.get("vaults") + if not isinstance(vaults, list): + raise KongRequestError("Kong response missing vaults list") + + result: List[Dict[str, object]] = [] + for vault in vaults: + if not isinstance(vault, dict) or _is_retired(vault) or _is_hidden(vault): + continue + + address = vault.get("address") + asset = vault.get("asset") + if not isinstance(address, str) or not isinstance(asset, dict): + raise KongRequestError("Kong parent vault missing address or asset metadata") + + asset_address = asset.get("address") + asset_decimals = _parse_decimals(asset.get("decimals")) + if not isinstance(asset_address, str) or asset_decimals is None: + raise KongRequestError(f"Kong parent vault {address} missing asset address or decimals") + + result.append( + { + "address": address, + "name": vault.get("name") or vault.get("symbol") or "UNKNOWN", + "symbol": vault.get("symbol") or "UNKNOWN", + "decimals": _parse_decimals(vault.get("decimals")), + "asset_address": asset_address, + "asset_symbol": asset.get("symbol") or "UNKNOWN", + "asset_decimals": asset_decimals, + } + ) + + return result diff --git a/tests/test_small_parent_flows.py b/tests/test_small_parent_flows.py new file mode 100644 index 0000000..aadc094 --- /dev/null +++ b/tests/test_small_parent_flows.py @@ -0,0 +1,193 @@ +from decimal import Decimal + +from protocols.yearn import alert_small_parent_flows as monitor +from utils.alert import AlertSeverity +from utils.chains import Chain + +VAULT = { + "address": "0xParent", + "name": "USDC yVault", + "symbol": "yvUSDC", + "decimals": 6, + "asset_address": "0xAsset", + "asset_symbol": "USDC", + "asset_decimals": 6, +} + + +def make_event( + *, + flow_type: str = "deposit", + assets: str = "5000", + block_number: int = 100, + log_index: int = 2, +) -> dict: + event = { + "id": f"1_{block_number}_{log_index}", + "flow_type": flow_type, + "vaultAddress": "0xParent", + "chainId": 1, + "blockNumber": block_number, + "blockTimestamp": 1_700_000_000, + "transactionHash": "0xTransaction", + "transactionFrom": "0xTransactionFrom", + "logIndex": log_index, + "sender": "0xSender", + "owner": "0xOwner", + "assets": assets, + "shares": assets, + } + if flow_type == "withdrawal": + event["receiver"] = "0xReceiver" + return event + + +def test_small_flow_uses_raw_asset_units() -> None: + assert monitor.format_units("1234567", 6) == Decimal("1.234567") + assert monitor.is_small_flow("9999", 10_000) + assert not monitor.is_small_flow("10000", 10_000) + assert not monitor.is_small_flow("10001", 10_000) + assert not monitor.is_small_flow("0", 10_000) + + +def test_process_deposit_sends_low_alert_with_all_addresses() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(), + {"0xparent": VAULT}, + 10_000, + alert_sender=alerts.append, + ) + + assert did_alert + assert len(alerts) == 1 + alert = alerts[0] + assert alert.severity is AlertSeverity.LOW + assert alert.protocol == "yearn" + assert "Small parent-vault deposit" in alert.message + assert "Raw Assets: 5,000" in alert.message + assert "Normalized: 0.005 USDC" in alert.message + assert "Raw Threshold: < 10,000" in alert.message + assert "0xOwner" in alert.message + assert "0xSender" in alert.message + assert "0xTransactionFrom" in alert.message + assert "0xTransaction" in alert.message + assert "Receiver" not in alert.message + + +def test_process_withdrawal_includes_asset_receiver() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(flow_type="withdrawal"), + {"0xparent": VAULT}, + 10_000, + alert_sender=alerts.append, + ) + + assert did_alert + assert len(alerts) == 1 + assert "Small parent-vault withdrawal" in alerts[0].message + assert "Receiver" in alerts[0].message + assert "0xReceiver" in alerts[0].message + + +def test_process_event_does_not_alert_at_threshold() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(assets="10000"), + {"0xparent": VAULT}, + 10_000, + alert_sender=alerts.append, + ) + + assert not did_alert + assert alerts == [] + + +def test_load_events_selects_envio_entity_and_receiver(monkeypatch) -> None: + queries = [] + + def fake_gql(query, variables): + queries.append((query, variables)) + return {"data": {"events": [make_event(flow_type="withdrawal")]}} + + monkeypatch.setattr(monitor, "gql_request", fake_gql) + + events = monitor.load_events( + "withdrawal", + 1, + ["0xParent"], + monitor.EventCursor(10, 2), + 1_700_000_000, + 100, + ) + + assert "events: Withdraw(" in queries[0][0] + assert "receiver" in queries[0][0] + assert queries[0][1]["lastBlock"] == 10 + assert events and events[0]["flow_type"] == "withdrawal" + + +def test_monitor_flow_type_pages_and_persists_each_processed_event(monkeypatch) -> None: + first = make_event(flow_type="withdrawal", block_number=100, log_index=2) + second = make_event(flow_type="withdrawal", block_number=101, log_index=3) + calls = [] + saved = [] + + monkeypatch.setattr(monitor, "load_cursor", lambda _chain_id, _flow_type: None) + monkeypatch.setattr( + monitor, + "save_cursor", + lambda chain_id, flow_type, cursor: saved.append((chain_id, flow_type, cursor)), + ) + monkeypatch.setattr(monitor, "process_event", lambda event, *_args: event is first) + + def fake_load(flow_type, chain_id, addresses, cursor, since_ts, limit): + calls.append((flow_type, chain_id, addresses, cursor, since_ts, limit)) + if len(calls) == 1: + return [first, second] + return [] + + monkeypatch.setattr(monitor, "load_events", fake_load) + + processed, alerted = monitor.monitor_flow_type( + 1, + "withdrawal", + ["0xParent"], + {"0xparent": VAULT}, + 10_000, + lookback_seconds=7200, + page_size=2, + now=1_700_010_000, + ) + + assert (processed, alerted) == (2, 1) + assert calls[0][0] == "withdrawal" + assert calls[0][1] == 1 + assert calls[0][3] == monitor.EventCursor(0, -1) + assert calls[0][4] == 1_700_002_800 + assert calls[1][3] == monitor.EventCursor(101, 3) + assert saved == [ + (1, "withdrawal", monitor.EventCursor(100, 2)), + (1, "withdrawal", monitor.EventCursor(101, 3)), + ] + + +def test_monitor_chain_runs_deposit_and_withdrawal_streams(monkeypatch) -> None: + flow_types = [] + + monkeypatch.setattr(monitor, "fetch_kong_parent_vaults", lambda _chain: [VAULT]) + + def fake_monitor(_chain_id, flow_type, *_args): + flow_types.append(flow_type) + return 1, 1 + + monkeypatch.setattr(monitor, "monitor_flow_type", fake_monitor) + + result = monitor.monitor_chain(Chain.MAINNET, 10_000, 7200, 1000) + + assert result == (2, 2) + assert flow_types == ["deposit", "withdrawal"] diff --git a/tests/test_yearn_kong.py b/tests/test_yearn_kong.py index 4a64d61..a744213 100644 --- a/tests/test_yearn_kong.py +++ b/tests/test_yearn_kong.py @@ -87,3 +87,62 @@ def test_fetch_kong_vaults_raises_on_graphql_errors(monkeypatch) -> None: with pytest.raises(kong.KongRequestError): kong.fetch_kong_vaults(Chain.MAINNET) + + +def test_fetch_kong_parent_vaults_filters_inactive_vaults(monkeypatch) -> None: + payload = { + "data": { + "vaults": [ + { + "address": "0xParent", + "name": "USDC yVault", + "symbol": "yvUSDC", + "decimals": "6", + "vaultType": 1, + "asset": {"address": "0xAsset", "symbol": "USDC", "decimals": 6}, + "meta": {"isRetired": False, "isHidden": False}, + }, + { + "address": "0xRetired", + "name": "Old vault", + "symbol": "yvOLD", + "decimals": "18", + "vaultType": 1, + "asset": {"address": "0xOldAsset", "symbol": "OLD", "decimals": 18}, + "meta": {"isRetired": True, "isHidden": False}, + }, + { + "address": "0xHidden", + "name": "Hidden vault", + "symbol": "yvHIDDEN", + "decimals": "18", + "vaultType": 1, + "asset": {"address": "0xHiddenAsset", "symbol": "HIDDEN", "decimals": 18}, + "meta": {"isRetired": False, "isHidden": True}, + }, + ] + } + } + calls = [] + + def fake_post(url: str, json: dict, timeout: int) -> FakeResponse: + calls.append((url, json, timeout)) + return FakeResponse(payload) + + monkeypatch.setattr(kong.requests, "post", fake_post) + + vaults = kong.fetch_kong_parent_vaults(Chain.MAINNET) + + assert "vaultType: 1" in calls[0][1]["query"] + assert calls[0][1]["variables"] == {"chainId": 1} + assert vaults == [ + { + "address": "0xParent", + "name": "USDC yVault", + "symbol": "yvUSDC", + "decimals": 6, + "asset_address": "0xAsset", + "asset_symbol": "USDC", + "asset_decimals": 6, + } + ]