GAE/GoでTaskQueueのテスト（単純なPUSHキューAddだけ）

aets.go

// Test Support Context
package aets

import (
	"appengine"
	"appengine/taskqueue"
	"appengine_internal"
	tqpb "appengine_internal/taskqueue"
	"sync"
	"time"
)

type Context struct {
	appengine.Context
	taskQueues map[string][]*taskqueue.Task
	sync.Mutex
}

func (cw Context) Call(service, method string, in, out appengine_internal.ProtoMessage, opts *appengine_internal.CallOptions) error {
	if service == "taskqueue" {
		cw.hookTaskQueue(in)
	}
	return cw.Context.Call(service, method, in, out, opts)
}

func (cw Context) AddTasks(queueName string, tasks ...*taskqueue.Task) {
	cw.Lock()
	defer cw.Unlock()
	cw.taskQueues[queueName] = append(cw.taskQueues[queueName], tasks...)
}

func (cw Context) TaskQueue(queueName string) []*taskqueue.Task {
	cw.Lock()
	defer cw.Unlock()
	q := make([]*taskqueue.Task, len(cw.taskQueues[queueName]))
	copy(q, cw.taskQueues[queueName])
	return q
}

func (cw Context) hookTaskQueue(pm appengine_internal.ProtoMessage) {
	// Now only supports some part of PUSH queue
	if req, ok := pm.(*tqpb.TaskQueueAddRequest); ok && req.GetMode() == tqpb.TaskQueueMode_PUSH {

		task := taskqueue.Task{
			Path:    string(req.Url),
			Payload: req.Body,
			Header:  nil, // TODO: http.Header
			Method:  req.GetMethod().String(),
			Name:    string(req.TaskName),
			Delay:   0, // TODO time.Duration
			ETA: time.Unix(
				int64(*req.EtaUsec/1e6),
				int64((*req.EtaUsec%1e6)*1e3)), // TODO: concider delay
			RetryCount:   0,
			Tag:          "",
			RetryOptions: nil, // TODO
		}

		cw.AddTasks(string(req.QueueName), &task)

	} else {
		// TODO support other taskqueue features
		return
	}
}

func NewContext(c appengine.Context) Context {
	queues := make(map[string][]*taskqueue.Task)
	return Context{
		Context:    c,
		taskQueues: queues,
	}
}

tqadd.go

// Testee
package tqtest

import (
	"appengine"
	"appengine/datastore"
	"appengine/taskqueue"
	"net/http"
	"net/url"
)

func init() {
	http.HandleFunc("/", handler)
	http.HandleFunc("/worker", work)

}

func handler(w http.ResponseWriter, r *http.Request) {
	c := appengine.NewContext(r)

	if err := process(c); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
	}
}

func process(c appengine.Context) error {
	if err := datastore.RunInTransaction(c, func(c appengine.Context) error {
		t := taskqueue.NewPOSTTask("/worker", url.Values{
			"test": []string{"hoge", "moke"},
		})
		if _, err := taskqueue.Add(c, t, "foo"); err != nil {
			return err
		}
		return nil
	}, nil); err != nil {
		return err
	}
	return nil
}

func work(w http.ResponseWriter, r *http.Request) {
	c := appengine.NewContext(r)

	if err := r.ParseForm(); err != nil {
		c.Errorf("parse form error: %s", err.Error())
		return
	}
	c.Infof("form: %v", r.Form)
}

tqadd_test.go

// Tester
package tqtest

import (
	"aets"
	"appengine/aetest"
	"testing"
)

func TestAddTask(t *testing.T) {
	c, err := aetest.NewContext(nil)
	if err != nil {
		t.Fatal(err)
	}
	defer c.Close()

	c2 := aets.NewContext(c)

	process(c2)

	queue := c2.TaskQueue("foo")

	if len(queue) != 1 {
		t.Fatalf("unexpected task added. %d", len(queue))
	}

	task := queue[0]

	if task.Path != "/worker" {
		t.Fatalf("unexpected task.Path %s")
	}

	if string(task.Payload) != "test=hoge&test=moke" {
		t.Fatalf("unexpected task.Payload %s", task.Path)
	}

	c.Infof("done!!!!!")
}