add rethinkdb -> postgresql integration test, fix connection leak in …

…pg client (#293)

.travis.yml

@@ -18,6 +18,7 @@ env:
   - TESTDIR=integration_tests/mongo_to_mongo
   - TESTDIR=integration_tests/mongo_to_es
   - TESTDIR=integration_tests/mongo_to_rethink
+  - TESTDIR=integration_tests/rethink_to_postgres
 before_install:
 - "./scripts/before_install.sh"
 cache:

CHANGELOG.md

@@ -7,6 +7,13 @@
 
 ### Bugfixes
 
+## v0.2.1 [2017-03-07]
+
+### Features
+- added RethinkDB -> PostgreSQL integration test
+
+### Bugfixes
+- fixed connection leak in PostgreSQL client
 
 ## v0.2.0 [2017-02-28]
 

adaptor/postgres/client.go

@@ -61,8 +61,11 @@ func (c *Client) Close() {
 
 // Connect initializes the Postgres connection
 func (c *Client) Connect() (client.Session, error) {
-	// there's really no way for this to error because we know the driver we're passing is
-	// available.
-	c.pqSession, _ = sql.Open("postgres", c.uri)
-	return &Session{c.pqSession}, nil
+	if c.pqSession == nil {
+		// there's really no way for this to error because we know the driver we're passing is
+		// available.
+		c.pqSession, _ = sql.Open("postgres", c.uri)
+	}
+	err := c.pqSession.Ping()
+	return &Session{c.pqSession}, err
 }

adaptor/postgres/writer.go

@@ -42,7 +42,7 @@ func (w *Writer) Write(msg message.Msg) func(client.Session) error {
 }
 
 func insertMsg(m message.Msg, s *sql.DB) error {
-	log.Infof("Write INSERT to Postgres %v", m.Namespace())
+	log.With("table", m.Namespace()).Debugln("INSERT")
 	var (
 		keys         []string
 		placeholders []string
@@ -71,7 +71,7 @@ func insertMsg(m message.Msg, s *sql.DB) error {
 }
 
 func deleteMsg(m message.Msg, s *sql.DB) error {
-	fmt.Printf("Write DELETE to Postgres %v values %v\n", m.Namespace(), m.Data())
+	log.With("table", m.Namespace()).With("values", m.Data()).Debugln("DELETE")
 	var (
 		ckeys []string
 		vals  []interface{}
@@ -103,7 +103,7 @@ func deleteMsg(m message.Msg, s *sql.DB) error {
 }
 
 func updateMsg(m message.Msg, s *sql.DB) error {
-	fmt.Printf("Write UPDATE to Postgres %v\n", m.Namespace())
+	log.With("table", m.Namespace()).Debugln("UPDATE")
 	var (
 		ckeys []string
 		ukeys []string

integration_tests/config.yml

@@ -13,7 +13,14 @@ nodes:
   enron_sink_es:
     type: elasticsearch
     uri: https://${ES_ENRON_SINK_USER}:${ES_ENRON_SINK_PASSWORD}@${ES_ENRON_SINK_URI}
+  enron_source_rethink:
+    type: rethinkdb
+    uri: rethink://admin:${RETHINKDB_ENRON_SOURCE_PASSWORD}@${RETHINKDB_ENRON_SOURCE_URI}/enron
+    ssl: true
   enron_sink_rethink:
     type: rethinkdb
-    uri: rethink://${RETHINKDB_ENRON_SINK_USER}:${RETHINKDB_ENRON_SINK_PASSWORD}@${RETHINKDB_ENRON_SINK_URI}/enron
+    uri: rethink://admin:${RETHINKDB_ENRON_SINK_PASSWORD}@${RETHINKDB_ENRON_SINK_URI}/enron
     ssl: true
+  enron_sink_postgres:
+    type: postgres
+    uri: postgres://${POSTGRES_ENRON_SINK_USER}:${POSTGRES_ENRON_SINK_PASSWORD}@${POSTGRES_ENRON_SINK_URI}

integration_tests/mongo_to_es/transporter_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestMongoToElasticsearchDocCount(t *testing.T) {
@@ -22,6 +23,9 @@ func TestMongoToElasticsearchDocCount(t *testing.T) {
 		t.Fatalf("unexpected error, %s", err)
 	}
 
+	// give the cluster a bit to breathe
+	time.Sleep(30 * time.Second)
+
 	req, _ = http.NewRequest(
 		http.MethodGet,
 		fmt.Sprintf("https://%s/enron/emails/_count", strings.Split(os.Getenv("ES_ENRON_SINK_URI"), ",")[0]),

integration_tests/rethink_to_postgres/app.js

@@ -0,0 +1,2 @@
+Source({name:"enron_source_rethink", namespace: 'enron.emails'})
+  .save({name:"enron_sink_postgres", namespace: 'enron.emails'});

integration_tests/rethink_to_postgres/main_test.go

@@ -0,0 +1,72 @@
+// +build integration
+
+package integration_test
+
+import (
+	"database/sql"
+	"flag"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/compose/transporter/log"
+
+	_ "github.com/lib/pq" // import pq driver
+)
+
+const (
+	emailsSchema = `id varchar(24),
+  body TEXT,
+  filename varchar(255),
+  headers jsonb,
+  mailbox varchar(255),
+  subFolder varchar(255),
+
+  PRIMARY KEY (id)`
+)
+
+var (
+	postgresSourceSession *sql.DB
+	cleanup               = flag.Bool("cleanup", false, "used to determine whether or not to run cleanup function")
+)
+
+func setup() {
+	log.Infoln("setting up tests")
+	u := fmt.Sprintf("postgres://%s:%s@%s",
+		os.Getenv("POSTGRES_ENRON_SINK_USER"),
+		os.Getenv("POSTGRES_ENRON_SINK_PASSWORD"),
+		os.Getenv("POSTGRES_ENRON_SINK_URI"))
+	postgresSourceSession, _ = sql.Open("postgres", u)
+}
+
+func cleanupData() {
+	log.Infoln("cleaning up data")
+
+	if _, err := postgresSourceSession.Exec("DROP TABLE IF EXISTS emails;"); err != nil {
+		log.Errorf("unable to drop table, could affect tests, %s", err)
+	}
+
+	_, err := postgresSourceSession.Exec(fmt.Sprintf("CREATE TABLE emails ( %s );", emailsSchema))
+	if err != nil {
+		log.Errorf("unable to create table, could affect tests, %s", err)
+	}
+}
+
+func TestMain(m *testing.M) {
+	flag.Parse()
+	setup()
+	if *cleanup {
+		cleanupData()
+		shutdown()
+		os.Exit(0)
+		return
+	}
+	code := m.Run()
+	shutdown()
+	os.Exit(code)
+}
+
+func shutdown() {
+	log.Infoln("shutting down tests")
+	log.Infoln("tests shutdown complete")
+}

integration_tests/rethink_to_postgres/transporter_test.go

@@ -0,0 +1,18 @@
+// +build integration
+
+package integration_test
+
+import (
+	"testing"
+)
+
+func TestRethinkToPostgresDocCount(t *testing.T) {
+	var count int
+	err := postgresSourceSession.QueryRow("SELECT COUNT(id) FROM emails;").Scan(&count)
+	if err != nil {
+		t.Errorf("unable to count table, %s", err)
+	}
+	if count != 5477 {
+		t.Errorf("bad emailCount, expected 5477, got %d", count)
+	}
+}