Initial distributor implementation

appengine/cmd/dm/mutate/add_deps_test.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package mutate
 
 import (
         "testing"
 
         "github.com/luci/gae/impl/memory"
         "github.com/luci/gae/service/datastore"
+        "github.com/luci/luci-go/appengine/cmd/dm/distributor"
         "github.com/luci/luci-go/appengine/cmd/dm/model"
         "github.com/luci/luci-go/common/api/dm/service/v1"
         . "github.com/luci/luci-go/common/testing/assertions"
         . "github.com/smartystreets/goconvey/convey"
         "golang.org/x/net/context"
 )
 
 func TestAddDeps(t *testing.T) {
         t.Parallel()
 
@@ skipping 29 matching lines @@
                 })
 
                 Convey("RollForward", func() {
 
                         Convey("Bad", func() {
                                 Convey("Bad ExecutionKey", func() {
                                         So(ds.Put(a, ex), ShouldBeNil)
 
                                         ad.Auth.Token = []byte("nerp")
                                         muts, err := ad.RollForward(c)
-                                        So(err, ShouldBeRPCUnauthenticated, "execution Auth")
+                                        So(err, ShouldBeRPCPermissionDenied, "execution Auth")
                                         So(muts, ShouldBeEmpty)
                                 })
                         })
 
                         Convey("Good", func() {
                                 So(ds.Put(a, ex), ShouldBeNil)
 
                                 Convey("All added already", func() {
                                         So(ds.Put(fds), ShouldBeNil)
 
                                         muts, err := ad.RollForward(c)
                                         So(err, ShouldBeNil)
                                         So(muts, ShouldBeEmpty)
                                 })
 
                                 Convey("None added already", func() {
                                         muts, err := ad.RollForward(c)
                                         So(err, ShouldBeNil)
                                         So(len(muts), ShouldEqual, len(fds))
 
                                         So(muts[0], ShouldResemble, &AddBackDep{
                                                 Dep: fds[0].Edge(), NeedsAck: true})
 
                                         So(ds.Get(a, fds), ShouldBeNil)
-                                        So(a.AddingDepsBitmap.Size(), ShouldEqual, len(fds))
-                                        So(a.WaitingDepBitmap.Size(), ShouldEqual, len(fds))
-                                        So(a.State, ShouldEqual, dm.Attempt_ADDING_DEPS)
+                                        So(a.DepMap.Size(), ShouldEqual, len(fds))
+                                        So(a.State, ShouldEqual, dm.Attempt_EXECUTING)
                                         So(fds[0].ForExecution, ShouldEqual, 1)
+
+                                        muts, err = (&FinishExecution{
+                                                ad.Auth.Id, &distributor.TaskResult{PersistentState: []byte("hi")},
+                                        }).RollForward(c)
+                                        So(err, ShouldBeNil)
+                                        So(muts, ShouldBeNil)
+
+                                        So(ds.Get(a), ShouldBeNil)
+                                        So(a.State, ShouldEqual, dm.Attempt_WAITING)
                                 })
 
                                 Convey("adding new Attempts at the same time", func() {
-                                        ad.Atmpts = dm.NewAttemptList(map[string][]uint32{
+                                        ad.Attempts = dm.NewAttemptList(map[string][]uint32{
                                                 "to": {2, 3},
                                                 "tp": {1},
                                         })
 
                                         muts, err := ad.RollForward(c)
                                         So(err, ShouldBeNil)
                                         So(len(muts), ShouldEqual, len(fds)+3)
 
                                         So(muts[0], ShouldResemble, &EnsureAttempt{dm.NewAttemptID("to", 3)})
                                         So(muts[1], ShouldResemble, &AddBackDep{
                                                 Dep: fds[0].Edge(), NeedsAck: true})
 
                                         So(ds.Get(a, fds), ShouldBeNil)
-                                        So(a.AddingDepsBitmap.Size(), ShouldEqual, len(fds))
-                                        So(a.WaitingDepBitmap.Size(), ShouldEqual, len(fds))
-                                        So(a.State, ShouldEqual, dm.Attempt_ADDING_DEPS)
+                                        So(a.DepMap.Size(), ShouldEqual, len(fds))
+                                        So(a.State, ShouldEqual, dm.Attempt_EXECUTING)
                                         So(fds[0].ForExecution, ShouldEqual, 1)
+
+                                        muts, err = (&FinishExecution{
+                                                ad.Auth.Id, &distributor.TaskResult{PersistentState: []byte("hi")},
+                                        }).RollForward(c)
+                                        So(err, ShouldBeNil)
+                                        So(muts, ShouldBeNil)
+
+                                        So(ds.Get(a), ShouldBeNil)
+                                        So(a.State, ShouldEqual, dm.Attempt_WAITING)
                                 })
                         })
                 })
         })
 }