LogDog: Support per-namespace expired archival.

appengine/logdog/coordinator/backend/archiveCron_test.go

Index: appengine/logdog/coordinator/backend/archiveCron_test.go
diff --git a/appengine/logdog/coordinator/backend/archiveCron_test.go b/appengine/logdog/coordinator/backend/archiveCron_test.go
index 99aab32142ff24811f2991b753ae5f26c6772a7a..c280eaaf3a6b549e9367d6c914c27f7fe5e6aabd 100644
--- a/appengine/logdog/coordinator/backend/archiveCron_test.go
+++ b/appengine/logdog/coordinator/backend/archiveCron_test.go
@@ -8,16 +8,19 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"sort"
 	"testing"
 	"time"
 
 	"github.com/julienschmidt/httprouter"
 	ds "github.com/luci/gae/service/datastore"
+	tq "github.com/luci/gae/service/taskqueue"
 	"github.com/luci/luci-go/appengine/gaetesting"
 	"github.com/luci/luci-go/appengine/logdog/coordinator"
 	ct "github.com/luci/luci-go/appengine/logdog/coordinator/coordinatorTest"
 	"github.com/luci/luci-go/common/clock"
 	"github.com/luci/luci-go/common/clock/testclock"
+	memoryConfig "github.com/luci/luci-go/common/config/impl/memory"
 	"github.com/luci/luci-go/common/errors"
 	"github.com/luci/luci-go/common/proto/google"
 
@@ -27,10 +30,71 @@ import (
 func TestHandleArchiveCron(t *testing.T) {
 	t.Parallel()
 
+	const queueName = "test-archive-scan-queue"
+
+	Convey(`A testing environment`, t, func() {
+		c := gaetesting.TestingContext()
+
+		mcfg := map[string]memoryConfig.ConfigSet{}
+		c = memoryConfig.Use(c, mcfg)
+
+		svcStub := ct.Services{}
+		svcStub.InitConfig()
+		svcStub.ServiceConfig.Coordinator.ArchiveScanProjectQueueName = queueName
+		c = coordinator.WithServices(c, &svcStub)
+
+		b := Backend{}
+
+		tb := testBase{Context: c}
+		r := httprouter.New()
+		b.InstallHandlers(r, tb.base)
+
+		s := httptest.NewServer(r)
+		defer s.Close()
+
+		endpoint := fmt.Sprintf("%s/archive/cron", s.URL)
+
+		Convey(`When projects are registered`, func() {
+			mcfg["projects/foo"] = memoryConfig.ConfigSet{}
+			mcfg["projects/bar"] = memoryConfig.ConfigSet{}
+			mcfg["projects/baz"] = memoryConfig.ConfigSet{}
+
+			Convey(`When the testing queue exists`, func() {
+				tq.Get(c).Testable().CreateQueue(queueName)
+
+				Convey(`Will dispatch archive tasks for each registered project.`, func() {
+					resp, err := http.Get(endpoint)
+					So(err, ShouldBeNil)
+					So(resp.StatusCode, ShouldEqual, http.StatusOK)
+
+					var namespaces []string
+					for _, task := range tq.Get(c).Testable().GetScheduledTasks()[queueName] {
+						namespaces = append(namespaces, task.Header["X-Appengine-Current-Namespace"]...)
+					}
+					sort.Strings(namespaces)
+
+					pns := coordinator.ProjectNamespace
+					So(namespaces, ShouldResemble, []string{pns("bar"), pns("baz"), pns("foo")})
+				})
+			})
+		})
+	})
+}
+
+func TestHandleArchiveScanTask(t *testing.T) {
+	t.Parallel()
+
 	Convey(`A testing environment`, t, func() {
 		c := gaetesting.TestingContext()
 		c, _ = testclock.UseTime(c, testclock.TestTimeUTC)
 
+		// In production, this task will be called with the namespace of is task
+		// queue entry installed. However, since we're calling it manually, we'll
+		// just fix the namespace.
+		if err := coordinator.WithProjectNamespace(&c, "test"); err != nil {
+			panic(err)
+		}
+
 		// Add the archival index from "index.yaml".
 		ds.Get(c).Testable().AddIndexes(
 			&ds.IndexDefinition{
@@ -89,10 +153,10 @@ func TestHandleArchiveCron(t *testing.T) {
 		s := httptest.NewServer(r)
 		defer s.Close()
 
-		endpoint := fmt.Sprintf("%s/archive/cron", s.URL)
+		endpoint := fmt.Sprintf("%s/archive/cron/scan", s.URL)
 
 		Convey(`With no configuration loaded, the endpoint will fail.`, func() {
-			resp, err := http.Get(endpoint)
+			resp, err := http.Post(endpoint, "", nil)
 			So(err, ShouldBeNil)
 			So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
 		})
@@ -100,7 +164,7 @@ func TestHandleArchiveCron(t *testing.T) {
 		Convey(`With no task topic configured, the endpoint will fail.`, func() {
 			svcStub.InitConfig()
 
-			resp, err := http.Get(endpoint)
+			resp, err := http.Post(endpoint, "", nil)
 			So(err, ShouldBeNil)
 			So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
 		})
@@ -112,7 +176,7 @@ func TestHandleArchiveCron(t *testing.T) {
 			svcStub.ServiceConfig.Coordinator.ArchiveDelayMax = google.NewDuration(10 * time.Minute)
 
 			Convey(`A request to the endpoint will be successful.`, func() {
-				resp, err := http.Get(endpoint)
+				resp, err := http.Post(endpoint, "", nil)
 				So(err, ShouldBeNil)
 				So(resp.StatusCode, ShouldEqual, http.StatusOK)
 
@@ -121,7 +185,7 @@ func TestHandleArchiveCron(t *testing.T) {
 
 				// Hit the endpoint again, no additional tasks should be scheduled.
 				Convey(`A subsequent endpoint hit will not schedule any additional tasks.`, func() {
-					resp, err = http.Get(endpoint)
+					resp, err = http.Post(endpoint, "", nil)
 					So(err, ShouldBeNil)
 					So(resp.StatusCode, ShouldEqual, http.StatusOK)
 					So(tap.StreamNames(), ShouldResemble, []string{"bar", "quux"})
@@ -153,7 +217,7 @@ func TestHandleArchiveCron(t *testing.T) {
 
 				Convey(`Will schedule all pages properly.`, func() {
 					// Ensure that all of these tasks get added to the task queue.
-					resp, err := http.Get(endpoint)
+					resp, err := http.Post(endpoint, "", nil)
 					So(err, ShouldBeNil)
 					So(resp.StatusCode, ShouldEqual, http.StatusOK)
 					So(tap.StreamNames(), ShouldResemble, names)
@@ -166,7 +230,7 @@ func TestHandleArchiveCron(t *testing.T) {
 					}
 
 					Convey(`Will not schedule additional tasks on the next run.`, func() {
-						resp, err := http.Get(endpoint)
+						resp, err := http.Post(endpoint, "", nil)
 						So(err, ShouldBeNil)
 						So(resp.StatusCode, ShouldEqual, http.StatusOK)
 						So(tap.StreamNames(), ShouldResemble, names)
@@ -177,7 +241,7 @@ func TestHandleArchiveCron(t *testing.T) {
 					tap.Err = errors.New("test error")
 
 					// Ensure that all of these tasks get added to the task queue.
-					resp, err := http.Get(endpoint)
+					resp, err := http.Post(endpoint, "", nil)
 					So(err, ShouldBeNil)
 					So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
 					So(tap.StreamNames(), ShouldResemble, []string{})
@@ -185,7 +249,7 @@ func TestHandleArchiveCron(t *testing.T) {
 					Convey(`And will schedule streams next run.`, func() {
 						tap.Err = nil
 
-						resp, err := http.Get(endpoint)
+						resp, err := http.Post(endpoint, "", nil)
 						So(err, ShouldBeNil)
 						So(resp.StatusCode, ShouldEqual, http.StatusOK)
 						So(tap.StreamNames(), ShouldResemble, names)