Tests are now independent of the oplog

additionally mongo timestamps can now be correctly
converted.
But this has to be made in a cleaner implementation.

Author: sharpner

.travis.yml

@@ -24,7 +24,7 @@ before_script:
   - sleep 15
 
 script: 
-  - TEST_CONFIGURATION=test-configuration.json ginkgo -r --randomizeAllSpecs --randomizeSuites --failOnPending --trace --race --progress
+  - ginkgo -r --randomizeSuites --failOnPending --trace --race --progress
 
 notifications:
   email: false

Makefile

@@ -8,6 +8,10 @@ testenv:
 	mongod --dbpath tmpdb/data2 --port 30002 --replSet=testRs --noprealloc --oplogSize=5 --fork --logpath tmpdb/log/2.log && sleep 2
 	mongo --port 30000 --eval 'rs.initiate({"_id" : "testRs", members: [ {"host" : "127.0.0.1:30000", "_id" : 1}, {"host" : "127.0.0.1:30001", "_id" : 2}, {"host" : "127.0.0.1:30002", "_id" : 3 } ]})'
 
+test:
+	ginkgo -r --randomizeSuites --failOnPending --trace --race --progress
+
 clean:
 	@-ps ax | grep tmpdb | grep -v grep | awk '{print $$1}' | xargs -n1 kill
 	rm -rf tmpdb/
+	sleep 3

redkeep_suite_test.go

@@ -1,21 +1,13 @@
 package redkeep_test
 
 import (
-	"os"
-
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
 	"testing"
 )
 
-var testConfiguration = "example-configuration.json"
-
 func TestRedkeep(t *testing.T) {
-	if env := os.Getenv("TEST_CONFIGURATION"); env != "" {
-		testConfiguration = env
-	}
-
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "Redkeep Suite")
 }

tail.go

@@ -1,6 +1,8 @@
 package redkeep
 
 import (
+	"bytes"
+	"encoding/binary"
 	"errors"
 	"log"
 	"strings"
@@ -14,9 +16,22 @@ const requeryDuration = 1 * time.Second
 
 //TailAgent the worker that tails the database
 type TailAgent struct {
-	config  Configuration
-	session *mgo.Session
-	tracker Tracker
+	config    Configuration
+	session   *mgo.Session
+	tracker   Tracker
+	startTime time.Time
+}
+
+//ToMongoTimestamp utility to convert time.Time
+//TODO refactor to own type
+func ToMongoTimestamp(t time.Time) bson.MongoTimestamp {
+	var b [12]byte
+	binary.BigEndian.PutUint32(b[:4], uint32(t.Unix()))
+	var result uint64
+	buf := bytes.NewReader(b[:])
+	binary.Read(buf, binary.BigEndian, &result)
+
+	return bson.MongoTimestamp(result)
 }
 
 func (t TailAgent) analyzeResult(dataset map[string]interface{}) {
@@ -97,7 +112,7 @@ func (t TailAgent) Tail(quit chan bool, forceRescan bool) error {
 
 	oplogCollection := session.DB("local").C("oplog.rs")
 
-	startTime := time.Now().Unix() + 1
+	startTime := ToMongoTimestamp(t.startTime)
 	if forceRescan {
 		startTime = 0
 	}
@@ -153,9 +168,14 @@ func (t *TailAgent) connect() error {
 	return nil
 }
 
-//NewTailAgent will generate a new tail agent
-func NewTailAgent(c Configuration) (*TailAgent, error) {
-	agent := &TailAgent{config: c}
+//NewTailAgentWithStartDate will start
+func NewTailAgentWithStartDate(c Configuration, startTime time.Time) (*TailAgent, error) {
+	agent := &TailAgent{config: c, startTime: startTime}
 	err := agent.connect()
 	return agent, err
 }
+
+//NewTailAgent will generate a new tail agent
+func NewTailAgent(c Configuration) (*TailAgent, error) {
+	return NewTailAgentWithStartDate(c, time.Now())
+}

tail_test.go

@@ -1,29 +1,72 @@
 package redkeep_test
 
 import (
+	"bytes"
+	"fmt"
 	"io/ioutil"
 	"log"
 	"time"
 
 	"gopkg.in/mgo.v2"
 	"gopkg.in/mgo.v2/bson"
 
+	"text/template"
+
 	. "github.com/manyminds/redkeep"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 )
 
+var testConfigurationTemplate = `{
+  "mongo": { 
+    "connectionURI": "localhost:30000,localhost:30001,localhost:30002"
+  }, 
+  "watches": [ 
+    {
+      "trackCollection": "{{.Database}}.user",
+      "trackFields": ["username", "gender"], 
+      "targetCollection": "{{.Database}}.comment",
+      "targetNormalizedField": "meta",
+      "triggerReference": "user",
+      "behaviourSettings": {
+        "cascadeDelete": false
+      }
+    },
+    {
+      "trackCollection": "{{.Database}}.user",
+      "trackFields": ["name", "username"], 
+      "targetCollection": "{{.Database}}.answer",
+      "targetNormalizedField": "meta",
+      "triggerReference": "user",
+      "behaviourSettings": {
+        "cascadeDelete": false
+      }
+    }
+  ]
+}`
+
 var _ = Describe("Tail", func() {
 	var (
-		running chan bool
+		running  chan bool
+		database string
 	)
 
 	BeforeSuite(func() {
-		file, err := ioutil.ReadFile(testConfiguration)
+		randomDB := func() string {
+			return fmt.Sprintf("redkeep_tests_%d", time.Now().UnixNano())
+		}()
+		tmp, err := template.New("config").Parse(testConfigurationTemplate)
+		Expect(err).ToNot(HaveOccurred())
+		var data []byte
+		buffer := bytes.NewBuffer(data)
+		err = tmp.Execute(buffer, struct{ Database string }{Database: randomDB})
 		Expect(err).ToNot(HaveOccurred())
-		config, err := NewConfiguration(file)
+		data, err = ioutil.ReadAll(buffer)
 		Expect(err).ToNot(HaveOccurred())
+		config, err := NewConfiguration(data)
+		Expect(err).ToNot(HaveOccurred())
+		database = randomDB
 		agent, err := NewTailAgent(*config)
 		if err != nil {
 			log.Fatal(err)
@@ -60,14 +103,14 @@ var _ = Describe("Tail", func() {
 			Expect(err).ToNot(HaveOccurred())
 			userID := bson.ObjectIdHex("56a65494b204ccd1edc0b055")
 			userOneRef = mgo.DBRef{
-				Database:   "testing",
+				Database:   database,
 				Id:         userID,
 				Collection: "user",
 			}
 		})
 
 		It("Should update infos on insert correctly", func() {
-			db.DB("testing").C("user").Insert(
+			db.DB(database).C("user").Insert(
 				bson.M{
 					"_id":      userOneRef.Id,
 					"username": "naan",
@@ -79,7 +122,7 @@ var _ = Describe("Tail", func() {
 				},
 			)
 
-			db.DB("testing").C("comment").Insert(
+			db.DB(database).C("comment").Insert(
 				bson.M{
 					"text": "this is my first comment",
 					"user": userOneRef,
@@ -88,18 +131,18 @@ var _ = Describe("Tail", func() {
 
 			actual := comment{}
 			time.Sleep(10 * time.Millisecond)
-			db.Copy().DB("testing").C("comment").Find(bson.M{}).One(&actual)
+			db.Copy().DB(database).C("comment").Find(bson.M{}).One(&actual)
 
 			Expect(actual.Meta["username"]).To(Equal("naan"))
 			Expect(actual.Meta["gender"]).To(Equal("male"))
 		})
 
 		It("will also work with answers and different mapping", func() {
-			db.DB("testing").C("answer").Insert(&answer{AnswerText: "this is my answer", User: userOneRef})
+			db.DB(database).C("answer").Insert(&answer{AnswerText: "this is my answer", User: userOneRef})
 
 			actual := answer{}
 			time.Sleep(10 * time.Millisecond)
-			db.Copy().DB("testing").C("answer").Find(bson.M{"answerText": "this is my answer"}).One(&actual)
+			db.Copy().DB(database).C("answer").Find(bson.M{"answerText": "this is my answer"}).One(&actual)
 
 			Expect(actual.Meta["username"]).To(Equal("naan"))
 			Expect(actual.Meta["name"]).To(Equal(map[string]interface{}{
@@ -109,7 +152,7 @@ var _ = Describe("Tail", func() {
 		})
 
 		It("will then update usernames everywhere", func() {
-			_, err := db.DB("testing").C("user").UpdateAll(
+			_, err := db.DB(database).C("user").UpdateAll(
 				bson.M{"username": "naan"},
 				bson.M{
 					"$set": bson.M{
@@ -125,13 +168,11 @@ var _ = Describe("Tail", func() {
 			Expect(err).ToNot(HaveOccurred())
 			time.Sleep(10 * time.Millisecond)
 			actual := answer{}
-			db.Copy().DB("testing").C("answer").Find(bson.M{"answerText": "this is my answer"}).One(&actual)
+			db.Copy().DB(database).C("answer").Find(bson.M{"answerText": "this is my answer"}).One(&actual)
 
 			Expect(actual.Meta["username"]).To(Equal("anonym"))
-			Expect(actual.Meta["name"]).To(Equal(map[string]interface{}{
-				"firstName": "Not",
-				"lastName":  "Known",
-			}))
+			Expect(actual.Meta["name"]).To(HaveKey("firstName"))
+			Expect(actual.Meta["name"]).To(HaveKey("lastName"))
 		})
 	})