Add tests for CSV file mode

.github/workflows/tests.yml

@@ -53,32 +53,9 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v2
 
-      - uses: actions/checkout@v3
-        with:
-          ref: ${{ env.stack-orchestrator-ref }}
-          path: "./stack-orchestrator/"
-          repository: vulcanize/stack-orchestrator
-          fetch-depth: 0
-
-      - uses: actions/checkout@v3
-        with:
-          ref: ${{ env.ipld-eth-db-ref }}
-          repository: vulcanize/ipld-eth-db
-          path: "./ipld-eth-db/"
-          fetch-depth: 0
-
-      - name: Create config file
-        run: |
-          echo vulcanize_ipld_eth_db=$GITHUB_WORKSPACE/ipld-eth-db/ > $GITHUB_WORKSPACE/config.sh
-          echo db_write=true >> $GITHUB_WORKSPACE/config.sh
-          cat $GITHUB_WORKSPACE/config.sh
-
       - name: Run docker compose
         run: |
-          docker-compose  \
-          -f "$GITHUB_WORKSPACE/stack-orchestrator/docker/local/docker-compose-db-sharding.yml" \
-          --env-file $GITHUB_WORKSPACE/config.sh \
-          up -d --build
+          docker-compose up -d
 
       - name: Give the migration a few seconds
         run: sleep 30;

docker-compose.yml

@@ -1,14 +1,27 @@
-version: "3.2"
+version: '3.2'
 
 services:
-  ipld-eth-db:
+  migrations:
     restart: on-failure
     depends_on:
-      - access-node
-    image: vulcanize/ipld-eth-db:v4.1.1-alpha
+      - ipld-eth-db
+    image: vulcanize/ipld-eth-db:v4.1.4-alpha
     environment:
       DATABASE_USER: "vdbm"
-      DATABASE_NAME: "vulcanize_testing_v4"
+      DATABASE_NAME: "vulcanize_testing"
       DATABASE_PASSWORD: "password"
-      DATABASE_HOSTNAME: "access-node"
+      DATABASE_HOSTNAME: "ipld-eth-db"
       DATABASE_PORT: 5432
+
+  ipld-eth-db:
+    image: timescale/timescaledb:latest-pg14
+    restart: always
+    command: ["postgres", "-c", "log_statement=all"]
+    environment:
+      POSTGRES_USER: "vdbm"
+      POSTGRES_DB: "vulcanize_testing"
+      POSTGRES_PASSWORD: "password"
+    ports:
+      - "127.0.0.1:8077:5432"
+    volumes:
+      - ./statediff/indexer/database/file:/file

statediff/indexer/database/file/indexer_legacy_test.go → statediff/indexer/database/file/csv_indexer_legacy_test.go

@@ -1,5 +1,5 @@
 // VulcanizeDB
-// Copyright © 2019 Vulcanize
+// Copyright © 2022 Vulcanize
 
 // This program is free software: you can redistribute it and/or modify
 // it under the terms of the GNU Affero General Public License as published by
@@ -21,34 +21,33 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
+	"strings"
 	"testing"
 
-	"github.com/ipfs/go-cid"
 	"github.com/jmoiron/sqlx"
 	"github.com/multiformats/go-multihash"
 	"github.com/stretchr/testify/require"
 
-	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/file"
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
-	"github.com/ethereum/go-ethereum/statediff/indexer/mocks"
 )
 
-var (
-	legacyData      = mocks.NewLegacyData()
-	mockLegacyBlock *types.Block
-	legacyHeaderCID cid.Cid
-)
+const dbDirectory = "/file"
 
-func setupLegacy(t *testing.T) {
+func setupCSVLegacy(t *testing.T) {
 	mockLegacyBlock = legacyData.MockBlock
 	legacyHeaderCID, _ = ipld.RawdataToCid(ipld.MEthHeader, legacyData.MockHeaderRlp, multihash.KECCAK_256)
+	file.TestConfig.Mode = file.CSV
+	file.TestConfig.OutputDir = "./statediffing_legacy_test"
+
 	if _, err := os.Stat(file.TestConfig.OutputDir); !errors.Is(err, os.ErrNotExist) {
-		err := os.Remove(file.TestConfig.OutputDir)
+		err := os.RemoveAll(file.TestConfig.OutputDir)
 		require.NoError(t, err)
 	}
+
 	ind, err := file.NewStateDiffIndexer(context.Background(), legacyData.Config, file.TestConfig)
 	require.NoError(t, err)
 	var tx interfaces.Batch
@@ -81,37 +80,29 @@ func setupLegacy(t *testing.T) {
 	}
 }
 
-func dumpFileData(t *testing.T) {
+func dumpCSVFileData(t *testing.T) {
 	pgCopyStatement := `COPY %s FROM '%s' CSV`
+	outputDir := filepath.Join(dbDirectory, file.TestConfig.OutputDir)
 
 	for _, tbl := range file.Tables {
-		stm := fmt.Sprintf(pgCopyStatement, tbl.Name, file.TableFile(file.TestConfig.OutputDir, tbl.Name))
+		stm := fmt.Sprintf(pgCopyStatement, tbl.Name, file.TableFile(outputDir, tbl.Name))
+
+		varcharColumns := tbl.VarcharColumns()
+		if len(varcharColumns) > 0 {
+			stm = fmt.Sprintf(
+				pgCopyStatement+" FORCE NOT NULL %s",
+				tbl.Name,
+				file.TableFile(outputDir, tbl.Name),
+				strings.Join(varcharColumns, ", "),
+			)
+		}
+
 		_, err = sqlxdb.Exec(stm)
 		require.NoError(t, err)
 	}
 }
 
-func resetAndDumpWatchedAddressesFileData(t *testing.T) {
-	resetDB(t)
-
-	sqlFileBytes, err := os.ReadFile(file.TestConfig.WatchedAddressesFilePath)
-	require.NoError(t, err)
-
-	_, err = sqlxdb.Exec(string(sqlFileBytes))
-	require.NoError(t, err)
-}
-
-func resetDB(t *testing.T) {
-	file.TearDownDB(t, sqlxdb)
-
-	connStr := postgres.DefaultConfig.DbConnectionString()
-	sqlxdb, err = sqlx.Connect("postgres", connStr)
-	if err != nil {
-		t.Fatalf("failed to connect to db with connection string: %s err: %v", connStr, err)
-	}
-}
-
-func tearDown(t *testing.T) {
+func tearDownCSV(t *testing.T) {
 	file.TearDownDB(t, sqlxdb)
 
 	err := os.RemoveAll(file.TestConfig.OutputDir)
@@ -125,17 +116,11 @@ func tearDown(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func expectTrue(t *testing.T, value bool) {
-	if !value {
-		t.Fatalf("Assertion failed")
-	}
-}
-
-func TestFileIndexerLegacy(t *testing.T) {
+func TestCSVFileIndexerLegacy(t *testing.T) {
 	t.Run("Publish and index header IPLDs", func(t *testing.T) {
-		setupLegacy(t)
-		dumpFileData(t)
-		defer tearDown(t)
+		setupCSVLegacy(t)
+		dumpCSVFileData(t)
+		defer tearDownCSV(t)
 		pgStr := `SELECT cid, td, reward, block_hash, coinbase
 				FROM eth.header_cids
 				WHERE block_number = $1`