fix tests bigquery provider + ruff

Author: MoonBoi9001

pyproject.toml

@@ -58,10 +58,7 @@ explicit_package_bases = true
 
 [tool.pytest.ini_options]
 minversion = "6.0"
-addopts = "-ra -q"
-testpaths = [
-    "tests",
-]
-pythonpath = [
-  "."
-]
+addopts = "--cov=src --cov-report=term-missing -v"
+testpaths = ["tests"]
+python_files = "test_*.py"
+pythonpath = ["."]

requirements.txt

@@ -34,6 +34,7 @@ aiohttp>=3.9.0  # For async HTTP requests (used by web3)
 pytest>=8.0.0
 pytest-cov>=6.0.0
 pytest-mock>=3.0.0
+pytest-snapshot>=0.9.0
 mypy>=1.0.0
 types-pytz  # Type stubs for pytz
 types-requests  # Type stubs for requests

src/models/bigquery_provider.py

@@ -185,7 +185,7 @@ def fetch_indexer_issuance_eligibility_data(self, start_date: date, end_date: da
                     - eligible_for_indexing_rewards: Whether the indexer is eligible for indexing rewards.
         """
         # Construct the query
-        query = self._get_indexer_eligibility_query(start_date, end_date)
+        query = self._get_indexer_eligibility_query(start_date=start_date, end_date=end_date)
 
         # Return the results df
         return self._read_gbq_dataframe(query)

src/models/blockchain_client.py

@@ -324,7 +324,7 @@ def _get_gas_prices(self, replace: bool) -> Tuple[int, int]:
             latest_block = cast(BlockData, latest_block_data)
             base_fee_hex = latest_block["baseFeePerGas"]
             base_fee = int(base_fee_hex) if isinstance(base_fee_hex, int) else int(str(base_fee_hex), 16)
-            logger.info(f"Latest block base fee: {base_fee/1e9:.2f} gwei")
+            logger.info(f"Latest block base fee: {base_fee / 1e9:.2f} gwei")
 
         # If the base fee cannot be retrieved, use a fallback value
         except Exception as e:
@@ -334,7 +334,7 @@ def _get_gas_prices(self, replace: bool) -> Tuple[int, int]:
         # Try to get the max priority fee
         try:
             max_priority_fee = self._execute_rpc_call(lambda: self.w3.eth.max_priority_fee)
-            logger.info(f"Max priority fee: {max_priority_fee/1e9:.2f} gwei")
+            logger.info(f"Max priority fee: {max_priority_fee / 1e9:.2f} gwei")
 
         # If the max priority fee cannot be retrieved, use a fallback value
         except Exception as e:
@@ -364,15 +364,15 @@ def _build_transaction_params(
             max_priority_fee_per_gas = max_priority_fee * 2
             tx_params["maxFeePerGas"] = max_fee_per_gas
             tx_params["maxPriorityFeePerGas"] = max_priority_fee_per_gas
-            logger.info(f"High gas for replacement: {max_fee_per_gas/1e9:.2f} gwei")
+            logger.info(f"High gas for replacement: {max_fee_per_gas / 1e9:.2f} gwei")
 
         # If we are not replacing a pending transaction, use a lower gas price
         else:
             max_fee_per_gas = base_fee * 2 + max_priority_fee
             max_priority_fee_per_gas = max_priority_fee
             tx_params["maxFeePerGas"] = max_fee_per_gas
             tx_params["maxPriorityFeePerGas"] = max_priority_fee_per_gas
-            logger.info(f"Standard gas: {max_fee_per_gas/1e9:.2f} gwei")
+            logger.info(f"Standard gas: {max_fee_per_gas / 1e9:.2f} gwei")
 
         logger.info(f"Transaction parameters: nonce={nonce}, gas={gas_limit}, chain_id={chain_id}")
         return tx_params

src/models/eligibility_pipeline.py

@@ -175,7 +175,7 @@ def validate_dataframe_structure(self, df: pd.DataFrame, required_columns: List[
         # If any required columns are missing, raise an error
         if missing_columns:
             raise ValueError(
-                f"DataFrame missing required columns: {missing_columns}. " f"Found columns: {list(df.columns)}"
+                f"DataFrame missing required columns: {missing_columns}. Found columns: {list(df.columns)}"
             )
 
         # If all required columns are present, return True

src/utils/slack_notifier.py

@@ -129,7 +129,7 @@ def send_success_notification(
 
         # Add transaction links if provided
         if transaction_links:
-            tx_links = "\n".join([f"Batch {i+1}: {link}" for i, link in enumerate(transaction_links)])
+            tx_links = "\n".join([f"Batch {i + 1}: {link}" for i, link in enumerate(transaction_links)])
             fields.append({"title": "Transactions", "value": tx_links, "short": False})
 
         # Create message payload
@@ -179,7 +179,7 @@ def send_failure_notification(
 
         # Add partial transaction links if any succeeded before failure
         if partial_transaction_links:
-            tx_links = "\n".join([f"Batch {i+1}: {link}" for i, link in enumerate(partial_transaction_links)])
+            tx_links = "\n".join([f"Batch {i + 1}: {link}" for i, link in enumerate(partial_transaction_links)])
             fields.append({"title": "Partial Transactions", "value": tx_links, "short": False})
 
         # Truncate error message if too long

tests/snapshots/test_bigquery_provider/test_get_indexer_eligibility_query_matches_snapshot/indexer_eligibility_query.sql

@@ -0,0 +1,83 @@
+
+        WITH
+        -- Get daily query metrics per indexer
+        DailyMetrics AS (
+            SELECT
+                day_partition AS day,
+                indexer,
+                COUNT(*) AS query_attempts,
+                SUM(CASE
+                    WHEN status = '200 OK'
+                    AND response_time_ms < 5000
+                    AND blocks_behind < 50000
+                    THEN 1
+                    ELSE 0
+                END) AS good_responses,
+                COUNT(DISTINCT deployment) AS unique_subgraphs_served
+            FROM
+                test.dataset.table
+            WHERE
+                day_partition BETWEEN '2025-01-01' AND '2025-01-28'
+            GROUP BY
+                day_partition, indexer
+        ),
+        -- Determine which days count as 'online' (>= 1 good query on >= 10 subgraphs)
+        DaysOnline AS (
+            SELECT
+                indexer,
+                day,
+                unique_subgraphs_served,
+                CASE WHEN good_responses >= 1 AND unique_subgraphs_served >= 10
+                    THEN 1 ELSE 0
+                END AS is_online_day
+            FROM
+                DailyMetrics
+        ),
+        -- Calculate unique subgraphs served with at least one good query
+        UniqueSubgraphs AS (
+            SELECT
+                indexer,
+                COUNT(DISTINCT deployment) AS unique_good_response_subgraphs
+            FROM
+                test.dataset.table
+            WHERE
+                day_partition BETWEEN '2025-01-01' AND '2025-01-28'
+                AND status = '200 OK'
+                AND response_time_ms < 5000
+                AND blocks_behind < 50000
+            GROUP BY
+                indexer
+        ),
+        -- Calculate overall metrics per indexer
+        IndexerMetrics AS (
+            SELECT
+                d.indexer,
+                SUM(m.query_attempts) AS total_query_attempts,
+                SUM(m.good_responses) AS total_good_responses,
+                SUM(d.is_online_day) AS total_good_days_online,
+                ds.unique_good_response_subgraphs
+            FROM
+                DailyMetrics m
+            JOIN
+                DaysOnline d USING (indexer, day)
+            LEFT JOIN
+                UniqueSubgraphs ds ON m.indexer = ds.indexer
+            GROUP BY
+                d.indexer, ds.unique_good_response_subgraphs
+        )
+        -- Final result with eligibility determination
+        SELECT
+            indexer,
+            total_query_attempts AS query_attempts,
+            total_good_responses AS good_responses,
+            total_good_days_online,
+            unique_good_response_subgraphs,
+            CASE
+                WHEN total_good_days_online >= 5 THEN 1
+                ELSE 0
+            END AS eligible_for_indexing_rewards
+        FROM
+            IndexerMetrics
+        ORDER BY
+            total_good_days_online DESC, good_responses DESC
+