Add tests

ecosystem/indexer/src/models/transactions.rs

@@ -160,6 +160,7 @@ impl Transaction {
                     .first::<BlockMetadataTransaction>(connection)
                     .optional()?;
             }
+            "genesis_transaction" => {}
             _ => unreachable!("Unknown transaction type: {}", &self.type_),
         };
         Ok((user_transaction, block_metadata_transaction, events))
@@ -306,3 +307,16 @@ fn parse_timestamp(ts: U64, version: U64) -> chrono::NaiveDateTime {
 pub type BlockMetadataTransactionModel = BlockMetadataTransaction;
 pub type TransactionModel = Transaction;
 pub type UserTransactionModel = UserTransaction;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Datelike;
+
+    #[test]
+    fn test_parse_timestamp() {
+        let ts = parse_timestamp(U64::from(1649560602763949), U64::from(1));
+        assert_eq!(ts.timestamp(), 1649560602);
+        assert_eq!(ts.year(), 2022);
+    }
+}

testsuite/forge/src/interface/aptos.rs

@@ -3,7 +3,7 @@
 
 use super::Test;
 use crate::{CoreContext, Result, TestReport};
-use aptos_rest_client::Client as RestClient;
+use aptos_rest_client::{Client as RestClient, PendingTransaction};
 use aptos_sdk::{
     crypto::ed25519::Ed25519PublicKey,
     move_types::identifier::Identifier,
@@ -83,6 +83,24 @@ impl<'t> AptosContext<'t> {
         self.public_info.mint(addr, amount).await
     }
 
+    pub async fn create_and_fund_user_account(&mut self, amount: u64) -> Result<LocalAccount> {
+        let account = self.random_account();
+        self.create_user_account(account.public_key()).await?;
+        self.mint(account.address(), amount).await?;
+        Ok(account)
+    }
+
+    pub async fn transfer(
+        &self,
+        from_account: &mut LocalAccount,
+        to_account: &LocalAccount,
+        amount: u64,
+    ) -> Result<PendingTransaction> {
+        self.public_info
+            .transfer(from_account, to_account, amount)
+            .await
+    }
+
     pub async fn get_balance(&self, address: AccountAddress) -> Option<u64> {
         self.public_info.get_balance(address).await
     }
@@ -137,6 +155,20 @@ impl<'t> AptosPublicInfo<'t> {
         Ok(())
     }
 
+    pub async fn transfer(
+        &self,
+        from_account: &mut LocalAccount,
+        to_account: &LocalAccount,
+        amount: u64,
+    ) -> Result<PendingTransaction> {
+        let tx = from_account.sign_with_transaction_builder(self.transaction_factory().payload(
+            aptos_stdlib::encode_transfer_script_function(to_account.address(), amount),
+        ));
+        let pending_txn = self.rest_client.submit(&tx).await?.into_inner();
+        self.rest_client.wait_for_transaction(&pending_txn).await?;
+        Ok(pending_txn)
+    }
+
     pub fn transaction_factory(&self) -> TransactionFactory {
         TransactionFactory::new(self.chain_id)
             .with_gas_unit_price(1)

testsuite/smoke-test/Cargo.toml

@@ -13,12 +13,15 @@ edition = "2018"
 anyhow = "1.0.52"
 async-trait = "0.1.42"
 bcs = "0.1.2"
+diesel = { version = "1.4.8", features = ["chrono", "postgres", "r2d2", "numeric", "serde_json"] }
+proptest = "1.0.0"
 tokio = { version = "1.8.1", features = ["full"] }
 reqwest = { version = "0.11.2", features = ["json"] }
 
 aptos = { path = "../../crates/aptos" }
 aptos-config = { path = "../../config" }
 aptos-crypto = { path = "../../crates/aptos-crypto" }
+aptos-indexer = { path = "../../ecosystem/indexer" }
 aptos-rest-client = { path = "../../crates/aptos-rest-client" }
 aptos-sdk = { path = "../../sdk" }
 aptos-temppath = { path = "../../crates/aptos-temppath" }
@@ -42,6 +45,7 @@ regex = "1.5.5"
 serde_yaml = "0.8.17"
 futures = "0.3.12"
 
+
 backup-cli = { path = "../../storage/backup/backup-cli" }
 debug-interface = { path = "../../crates/debug-interface" }
 aptos-genesis-tool = {path = "../../config/management/genesis", features = ["testing"] }

testsuite/smoke-test/src/indexer.rs

@@ -0,0 +1,118 @@
+// Copyright (c) Aptos
+// SPDX-License-Identifier: Apache-2.0
+
+use aptos_indexer::{
+    database::{new_db_pool, PgDbPool, PgPoolConnection},
+    default_processor::DefaultTransactionProcessor,
+    indexer::tailer::Tailer,
+    models::transactions::TransactionModel,
+};
+use diesel::connection::Connection;
+use forge::{AptosContext, AptosTest, Result, Test};
+use std::sync::Arc;
+
+pub struct Indexer;
+
+impl Test for Indexer {
+    fn name(&self) -> &'static str {
+        "ecosystem::indexer"
+    }
+}
+
+pub fn wipe_database(conn: &PgPoolConnection) {
+    for table in [
+        "events",
+        "user_transactions",
+        "block_metadata_transactions",
+        "transactions",
+        "processor_statuses",
+        "__diesel_schema_migrations",
+    ] {
+        conn.execute(&format!("DROP TABLE IF EXISTS {}", table))
+            .unwrap();
+    }
+}
+
+pub fn setup_indexer(ctx: &mut AptosContext) -> anyhow::Result<(PgDbPool, Tailer)> {
+    let database_url = std::env::var("INDEXER_DATABASE_URL")
+        .expect("must set 'INDEXER_DATABASE_URL' to run tests!");
+    let conn_pool = new_db_pool(database_url.as_str())?;
+    wipe_database(&conn_pool.get()?);
+
+    let mut tailer = Tailer::new(ctx.url(), conn_pool.clone())?;
+    tailer.run_migrations();
+
+    let pg_transaction_processor = DefaultTransactionProcessor::new(conn_pool.clone());
+    tailer.add_processor(Arc::new(pg_transaction_processor));
+    Ok((conn_pool, tailer))
+}
+
+#[async_trait::async_trait]
+impl AptosTest for Indexer {
+    async fn run<'t>(&self, ctx: &mut AptosContext<'t>) -> Result<()> {
+        let (conn_pool, mut tailer) = setup_indexer(ctx)?;
+
+        let client = ctx.client();
+        client.get_ledger_information().await.unwrap();
+
+        // Set up accounts, generate some traffic
+        let mut account1 = ctx.create_and_fund_user_account(1000).await.unwrap();
+        let account2 = ctx.create_and_fund_user_account(1000).await.unwrap();
+        // This transfer should emit events
+        let t_tx = ctx.transfer(&mut account1, &account2, 717).await.unwrap();
+
+        // Why do this twice? To ensure the idempotency of the tailer :-)
+        let mut version: u64 = 0;
+        for _ in 0..2 {
+            // Process the next versions
+            version = client
+                .get_ledger_information()
+                .await
+                .unwrap()
+                .into_inner()
+                .version;
+            println!("Height: {}", version);
+
+            tailer.process_next_batch((version + 1) as u8).await;
+
+            // This is the genesis transaction
+            let (tx0, ut0, bmt0, events0) =
+                TransactionModel::get_by_version(0, &conn_pool.get()?).unwrap();
+            assert_eq!(tx0.type_, "genesis_transaction");
+            assert!(ut0.is_none());
+            assert!(bmt0.is_none());
+            assert_eq!(events0.len(), 1);
+
+            // This is a block metadata transaction
+            let (tx1, ut1, bmt1, events1) =
+                TransactionModel::get_by_version(3, &conn_pool.get()?).unwrap();
+            assert_eq!(tx1.type_, "block_metadata_transaction");
+            assert!(ut1.is_none());
+            assert!(bmt1.is_some());
+            assert_eq!(events1.len(), 0);
+
+            // This is the transfer
+            let (tx2, ut2, bmt2, events2) =
+                TransactionModel::get_by_hash(t_tx.hash.to_string().as_str(), &conn_pool.get()?)
+                    .unwrap();
+
+            assert_eq!(tx2.hash, t_tx.hash.to_string());
+
+            // This is a user transaction, so the bmt should be None
+            assert!(ut2.is_some());
+            assert!(bmt2.is_none());
+
+            assert_eq!(events2.len(), 2);
+            assert_eq!(events2.get(0).unwrap().type_, "0x1::TestCoin::SentEvent");
+            assert_eq!(
+                events2.get(1).unwrap().type_,
+                "0x1::TestCoin::ReceivedEvent"
+            );
+        }
+
+        let latest_version = tailer.set_fetcher_to_lowest_processor_version().await;
+        assert!(latest_version > version);
+
+        Ok(())
+    }
+}

testsuite/smoke-test/src/lib.rs

@@ -4,6 +4,7 @@
 // Defines Forge Tests
 pub mod aptos;
 pub mod fullnode;
+pub mod indexer;
 pub mod nft_transaction;
 pub mod rest_api;
 pub mod transaction;

testsuite/smoke-test/src/rest_api.rs

@@ -37,12 +37,8 @@ impl AptosTest for BasicClient {
         let client = ctx.client();
         client.get_ledger_information().await?;
 
-        let mut account1 = ctx.random_account();
-        ctx.create_user_account(account1.public_key()).await?;
-        ctx.mint(account1.address(), 10_000).await?;
-        let account2 = ctx.random_account();
-        ctx.create_user_account(account2.public_key()).await?;
-        ctx.mint(account2.address(), 10_000).await?;
+        let mut account1 = ctx.create_and_fund_user_account(10_000).await?;
+        let account2 = ctx.create_and_fund_user_account(10_000).await?;
 
         let tx = account1.sign_with_transaction_builder(ctx.transaction_factory().payload(
             aptos_stdlib::encode_transfer_script_function(account2.address(), 1),

testsuite/smoke-test/tests/forge.rs

@@ -2,14 +2,15 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use forge::{forge_main, ForgeConfig, LocalFactory, Options, Result};
-use smoke_test::{
-    fullnode::LaunchFullnode,
-    rest_api::{self, GetIndex},
-};
+use smoke_test::{fullnode::LaunchFullnode, indexer, rest_api};
 
 fn main() -> Result<()> {
     let tests = ForgeConfig::default()
-        .with_aptos_tests(&[&GetIndex, &rest_api::BasicClient])
+        .with_aptos_tests(&[
+            &rest_api::GetIndex,
+            &rest_api::BasicClient,
+            &indexer::Indexer,
+        ])
         //TODO Re-enable these tests once we fix how the move compiler is invoked
         // .with_admin_tests(&[&MalformedScript, &ExecuteCustomModuleAndScript])
         .with_network_tests(&[&LaunchFullnode]);