Merge pull request #2 from gocardless/lawrence-integration-tests

pkg/pgreplay/integration

.circleci/config.yml

@@ -5,6 +5,14 @@ references:
   docker_golang: &docker_golang
     docker:
       - image: golang:1.11
+        environment:
+          PGHOST: "127.0.0.1"
+          PGUSER: "postgres"
+      - image: postgres:11.2
+        environment:
+          POSTGRES_USER: postgres
+          POSTGRES_DB: pgreplay_test
+          POSTGRES_PASSWORD: ""
     working_directory: /go/src/github.com/gocardless/pgreplay-go
 
 jobs:
@@ -15,9 +23,15 @@ jobs:
       - run:
           name: Install ginkgo test runner
           command: go get github.com/onsi/ginkgo/ginkgo
+      - run:
+          name: Install Postgres
+          command: apt-get update && apt-get install -y postgresql-client libpq-dev
+      - run:
+          name: Create test database
+          command: make recreatedb
       - run:
           name: Run tests
-          command: ginkgo -race -r -v
+          command: ginkgo -race -r
 
 workflows:
   version: 2

Makefile

@@ -15,6 +15,21 @@ bin/%.linux_amd64:
 bin/%:
 	$(BUILD_COMMAND) -o $@ cmd/$*/main.go
 
+createdb:
+	psql postgres -c "DROP GROUP IF EXISTS pgreplay_test_users; CREATE GROUP pgreplay_test_users WITH LOGIN CREATEDB;"
+	psql postgres -U pgreplay_test_users -c "CREATE DATABASE pgreplay_test;"
+	psql pgreplay_test -c "DROP ROLE IF EXISTS alice; CREATE ROLE alice LOGIN;"
+	psql pgreplay_test -c "DROP ROLE IF EXISTS bob;   CREATE ROLE bob   LOGIN;"
+	psql pgreplay_test -c "ALTER GROUP pgreplay_test_users ADD USER alice, bob;"
+
+dropdb:
+	psql postgres -c "DROP DATABASE IF EXISTS pgreplay_test;"
+
+structure:
+	psql pgreplay_test -U pgreplay_test_users -f pkg/pgreplay/integration/testdata/structure.sql
+
+recreatedb: dropdb createdb structure
+
 # go get -u github.com/onsi/ginkgo/ginkgo
 test:
 	ginkgo -v -r

README.md

@@ -27,6 +27,8 @@ You can turn on logging in a compatible format like so:
 
 ```sql
 ALTER SYSTEM SET log_directory='/postgres-logs';
+ALTER SYSTEM SET log_connections='on';
+ALTER SYSTEM SET log_disconnections='on';
 ALTER SYSTEM SET log_line_prefix='%m|%u|%d|%c|';
 ALTER SYSTEM SET log_min_error_statement='log';
 ALTER SYSTEM SET log_min_messages='error';

pkg/pgreplay/integration/integration_test.go

@@ -0,0 +1,130 @@
+package integration
+
+import (
+	"os"
+	"strconv"
+
+	kitlog "github.com/go-kit/kit/log"
+	"github.com/gocardless/pgreplay-go/pkg/pgreplay"
+	"github.com/jackc/pgx"
+	"github.com/onsi/gomega/types"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/ginkgo/extensions/table"
+	. "github.com/onsi/gomega"
+	. "github.com/onsi/gomega/gstruct"
+)
+
+var _ = Describe("pgreplay", func() {
+	var (
+		conn   *pgx.Conn
+		logger = kitlog.NewLogfmtLogger(GinkgoWriter)
+		err    error
+
+		// We expect a Postgres database to be running for integration tests, and that
+		// environment variables are appropriately configured to permit access.
+		cfg = pgx.ConnConfig{
+			Database: tryEnviron("PGDATABASE", "pgreplay_test"),
+			Host:     tryEnviron("PGHOST", "127.0.0.1"),
+			User:     tryEnviron("PGUSER", "pgreplay_test_users"),
+			Password: tryEnviron("PGPASSWORD", ""),
+			Port:     uint16(mustAtoi(tryEnviron("PGPORT", "5432"))),
+		}
+	)
+
+	DescribeTable("Replaying logfiles",
+		func(errlogfixture string, matchLogs []types.GomegaMatcher) {
+			conn, err = pgx.Connect(cfg)
+			Expect(err).NotTo(HaveOccurred(), "failed to connect to postgres")
+
+			_, err = conn.Exec(`TRUNCATE logs;`)
+			Expect(err).NotTo(HaveOccurred(), "failed to truncate logs table")
+
+			database, err := pgreplay.NewDatabase(cfg)
+			Expect(err).NotTo(HaveOccurred())
+
+			errlog, err := os.Open(errlogfixture)
+			Expect(err).NotTo(HaveOccurred())
+
+			items, logerrs, parsingDone := pgreplay.Parse(errlog)
+			go func() {
+				defer GinkgoRecover()
+				for err := range logerrs {
+					logger.Log("event", "parse.error", "error", err)
+				}
+			}()
+
+			stream, err := pgreplay.NewStreamer(nil, nil).Stream(items, 1.0)
+			Expect(err).NotTo(HaveOccurred())
+
+			errs, consumeDone := database.Consume(stream)
+
+			// Expect that we finish with no errors
+			Eventually(consumeDone).Should(BeClosed())
+			Eventually(errs).Should(BeClosed())
+
+			// Parsing should complete
+			Eventually(parsingDone).Should(BeClosed())
+
+			// Extract the logs that our test will have placed in the database
+			logs, err := getLogs(conn)
+
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(logs)).To(Equal(len(matchLogs)))
+
+			for idx, matchLog := range matchLogs {
+				Expect(logs[idx]).To(matchLog)
+			}
+		},
+		Entry("Single user", "testdata/single_user.log", []types.GomegaMatcher{
+			matchLog("alice", "says hello"),
+			matchLog("alice", "sees 1 logs"),
+			matchLog("alice", "sees 2 of alice's logs"),
+			matchLog("alice", "sees 0 of bob's logs"),
+		}),
+	)
+})
+
+func tryEnviron(key, otherwise string) string {
+	if value, found := os.LookupEnv(key); found {
+		return value
+	}
+
+	return otherwise
+}
+
+func mustAtoi(numstr string) int {
+	num, err := strconv.Atoi(numstr)
+	if err != nil {
+		panic(err)
+	}
+
+	return num
+}
+
+func getLogs(conn *pgx.Conn) ([]interface{}, error) {
+	rows, err := conn.Query(`SELECT id::text, author, message FROM logs ORDER BY id;`)
+	if err != nil {
+		return nil, err
+	}
+
+	defer rows.Close()
+	var id, author, message string
+	var logs = []interface{}{}
+
+	for rows.Next() {
+		if err := rows.Scan(&id, &author, &message); err != nil {
+			return nil, err
+		}
+
+		logs = append(logs, &struct{ ID, Author, Message string }{id, author, message})
+	}
+
+	return logs, nil
+}
+
+func matchLog(author, message string) types.GomegaMatcher {
+	return PointTo(
+		MatchFields(IgnoreExtras, Fields{"Author": Equal(author), "Message": Equal(message)}),
+	)
+}

pkg/pgreplay/integration/suite_test.go

@@ -0,0 +1,13 @@
+package integration
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestSuite(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "pkg/pgreplay/integration")
+}

pkg/pgreplay/integration/testdata/single_user.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"github.com/jackc/pgx"
+)
+
+func connect(user string) *pgx.Conn {
+	conn, _ := pgx.Connect(
+		pgx.ConnConfig{
+			Host: "127.0.0.1", Port: 5432, Database: "pgreplay_test", User: user,
+		},
+	)
+
+	return conn
+}
+
+var someoneSeesUser = `insert into logs (author, message) (
+  select $1, format('sees %s of %s''s logs', count(*), $2::text) from logs where author = $2
+);`
+
+func main() {
+	alice := connect("alice")
+	alice.Exec(`insert into logs (author, message) values ('alice', 'says hello');`)
+	alice.Exec(`insert into logs (author, message) (
+  select 'alice', format('sees %s logs', count(*)) from logs
+);`)
+
+	// Named prepared statement
+	alice.Prepare("someone_sees_user", someoneSeesUser)
+	alice.Exec("someone_sees_user", "alice", "alice")
+
+	// Unnamed prepared statement
+	alice.Exec(someoneSeesUser, "alice", "bob")
+}