test: add tap agent test

Author: carlosvdr

crates/tap-agent/src/lib.rs

@@ -21,7 +21,8 @@ pub mod cli;
 pub mod database;
 pub mod metrics;
 pub mod tap;
-pub mod tracker;
-
+#[cfg(test)]
+pub mod tap_agent_test;
 #[cfg(test)]
 pub mod test;
+pub mod tracker;

crates/tap-agent/src/tap_agent_test.rs

@@ -0,0 +1,213 @@
+// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
+// SPDX-License-Identifier: Apache-2.0
+
+use bip39::Mnemonic;
+use indexer_config::{
+    Config, DatabaseConfig, EscrowSubgraphConfig, GraphNodeConfig, IndexerConfig,
+    NetworkSubgraphConfig, NonZeroGRT, RavRequestConfig, SubgraphConfig, SubgraphsConfig,
+    TapConfig,
+};
+use indexer_monitor::{indexer_allocations, DeploymentDetails, SubgraphClient};
+use ractor::{concurrency::JoinHandle, Actor, ActorRef};
+use serde_json::json;
+use wiremock::matchers::body_string_contains;
+use wiremock::MockGuard;
+
+use crate::agent::{
+    sender_account::SenderAccountConfig, sender_accounts_manager::SenderAccountsManager,
+};
+use crate::{
+    agent::sender_accounts_manager::{SenderAccountsManagerArgs, SenderAccountsManagerMessage},
+    database, CONFIG, EIP_712_DOMAIN,
+};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+
+use indexer_monitor::EscrowAccounts;
+
+use reqwest::Url;
+use sqlx::PgPool;
+use tap_aggregator::server::run_server;
+use test_assets::INDEXER_ALLOCATIONS;
+use thegraph_core::alloy::primitives::Address;
+use tokio::sync::watch;
+
+use wiremock::{
+    matchers::{method, path},
+    Mock, MockServer, ResponseTemplate,
+};
+
+async fn mock_escrow_subgraph_empty_response() -> (MockServer, MockGuard) {
+    let mock_ecrow_subgraph_server: MockServer = MockServer::start().await;
+    let _mock_ecrow_subgraph = mock_ecrow_subgraph_server
+        .register_as_scoped(
+            Mock::given(method("POST"))
+                .and(body_string_contains("TapTransactions"))
+                .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "data": {
+                        "transactions": [],
+                    }
+                }))),
+        )
+        .await;
+    (mock_ecrow_subgraph_server, _mock_ecrow_subgraph)
+}
+pub async fn start_agent() -> (ActorRef<SenderAccountsManagerMessage>, JoinHandle<()>) {
+    let allocation = INDEXER_ALLOCATIONS.values().next().unwrap().clone();
+    let deployment = allocation.subgraph_deployment.id;
+
+    let (escrow_subgraph_mock_server, _escrow_subgraph_guard) =
+        mock_escrow_subgraph_empty_response().await;
+    let graph_node_mock_server = MockServer::start().await;
+    let network_subgraph_mock_server = MockServer::start().await;
+    let graph_node_mock_server = MockServer::start().await;
+    let mock = Mock::given(method("POST"))
+        .and(path(format!("/subgraphs/id/{deployment}")))
+        .respond_with(ResponseTemplate::new(200).set_body_raw(
+            r#"
+                {
+                    "data": {
+                        "graphNetwork": {
+                            "currentEpoch": 960
+                        }
+                    }
+                }
+                "#,
+            "application/json",
+        ));
+    graph_node_mock_server.register(mock).await;
+
+    // Start a TAP aggregator server.
+    let (handle, aggregator_endpoint) = run_server(
+        0,
+        test_assets::TAP_SIGNER.0.clone(),
+        vec![test_assets::TAP_SIGNER.1].into_iter().collect(),
+        test_assets::TAP_EIP712_DOMAIN.clone(),
+        100 * 1024,
+        100 * 1024,
+        1,
+    )
+    .await
+    .unwrap();
+
+    let (_escrow_tx, escrow_accounts) = watch::channel(EscrowAccounts::new(
+        test_assets::ESCROW_ACCOUNTS_BALANCES.clone(),
+        test_assets::ESCROW_ACCOUNTS_SENDERS_TO_SIGNERS.clone(),
+    ));
+    let (_dispute_tx, dispute_manager) = watch::channel(Address::ZERO);
+
+    let (_allocations_tx, allocations) = watch::channel(test_assets::INDEXER_ALLOCATIONS.clone());
+
+    let graph_node_url = Url::parse(&graph_node_mock_server.uri()).unwrap();
+    let pool_connection: str = "";
+    let mut sender_aggregator_endpoints: HashMap<Address, Url> = HashMap::new();
+    sender_aggregator_endpoints.insert(
+        test_assets::TAP_SENDER.1,
+        Url::from_str(&format!("http://{}", aggregator_endpoint)).expect("This should not fail"),
+    );
+    let config = Config {
+        indexer: IndexerConfig {
+            indexer_address: test_assets::INDEXER_ADDRESS,
+            operator_mnemonic: test_assets::INDEXER_MNEMONIC.clone(),
+        },
+        graph_node: GraphNodeConfig {
+            status_url: graph_node_url.clone(),
+            query_url: graph_node_url.clone(),
+        },
+        database: DatabaseConfig::PostgresUrl {
+            postgres_url: pool_connection,
+        },
+        subgraphs: SubgraphsConfig {
+            network: NetworkSubgraphConfig {
+                config: SubgraphConfig {
+                    query_url: Url::from_str(&network_subgraph_mock_server.uri()).unwrap(),
+                    query_auth_token: None,
+                    deployment_id: None,
+                    syncing_interval_secs: Duration::from_secs(30),
+                },
+                recently_closed_allocation_buffer_secs: Duration::from_secs(30),
+            },
+            escrow: EscrowSubgraphConfig {
+                config: SubgraphConfig {
+                    query_url: Url::from_str(&escrow_subgraph_mock_server.uri()).unwrap(),
+                    query_auth_token: None,
+                    deployment_id: None,
+                    syncing_interval_secs: Duration::from_secs(30),
+                },
+            },
+        },
+        tap: TapConfig {
+            // TODO: replace with a proper implementation once the gateway registry contract is ready
+            max_amount_willing_to_lose_grt: NonZeroGRT::new(1).expect("Should be able to convert"),
+            rav_request: RavRequestConfig {
+                trigger_value_divisor: todo!(),
+                timestamp_buffer_secs: todo!(),
+                request_timeout_secs: todo!(),
+                max_receipts_per_request: todo!(),
+            },
+            sender_timeout_secs: Duration::from_secs(60),
+            sender_aggregator_endpoints,
+        },
+        metrics: todo!(),
+        blockchain: todo!(),
+        service: todo!(),
+        dips: todo!(),
+    };
+    let pgpool = database::connect(database.clone()).await;
+
+    let http_client = reqwest::Client::new();
+
+    let network_subgraph = Box::leak(Box::new(
+        SubgraphClient::new(
+            http_client.clone(),
+            None,
+            DeploymentDetails::for_query_url(&network_subgraph_mock_server.uri()).unwrap(),
+        )
+        .await,
+    ));
+
+    let indexer_allocations = indexer_allocations(
+        network_subgraph,
+        test_assets::INDEXER_ADDRESS,
+        Duration::from_secs(30),
+        Duration::from_secs(30),
+    )
+    .await
+    .expect("Failed to initialize indexer_allocations watcher");
+
+    let escrow_subgraph = Box::leak(Box::new(
+        SubgraphClient::new(
+            http_client.clone(),
+            None,
+            DeploymentDetails::for_query_url(&escrow_subgraph_mock_server.uri()).unwrap(),
+        )
+        .await,
+    ));
+
+    // let escrow_accounts = escrow_accounts(
+    //     escrow_subgraph,
+    //     test_assets::INDEXER_ADDRESS,
+    //     Duration::from_secs(30),
+    //     false,
+    // )
+    // .await
+    // .expect("Error creating escrow_accounts channel");
+
+    let config = Box::leak(Box::new(SenderAccountConfig::from_config(&CONFIG)));
+
+    let args = SenderAccountsManagerArgs {
+        config,
+        domain_separator: EIP_712_DOMAIN.clone(),
+        pgpool,
+        indexer_allocations,
+        escrow_accounts,
+        escrow_subgraph,
+        network_subgraph,
+        sender_aggregator_endpoints: sender_aggregator_endpoints.clone(),
+        prefix: None,
+    };
+
+    SenderAccountsManager::spawn(None, SenderAccountsManager, args)
+        .await
+        .expect("Failed to start sender accounts manager actor.")
+}