Repository: megaease/easegress
Pull request #156 (2021-07-30 08:17:21)

add unit tests


Patch:
diff --git a/README.md b/README.md
index ce8889bc..0fd141d0 100644
--- a/README.md
+++ b/README.md
@@ -15,6 +15,7 @@
     - [More Filters](#more-filters)
   - [Documentation](#documentation)
   - [Roadmap](#roadmap)
+  - [Community](#community)
   - [License](#license)
 
 ## What is Easegress
@@ -298,6 +299,12 @@ See [reference](./doc/reference.md) and [developer guide](./doc/developer-guide.
 
 See [Easegress Roadmap](./doc/Roadmap.md) for details.
 
+## Community
+
+- [Slack Channel](https://openmegaease.slack.com/archives/C028Z6SUCU9) for requirement, issue and development.
+- [MegaEase on Twitter](https://twitter.com/megaease)
+
+
 ## License
 
 Easegress is under the Apache 2.0 license. See the [LICENSE](./LICENSE) file for details.
diff --git a/doc/filters.md b/doc/filters.md
index 6d9dac6b..d7c48dfb 100644
--- a/doc/filters.md
+++ b/doc/filters.md
@@ -47,7 +47,7 @@
     - [Configuration](#configuration-14)
     - [Results](#results-14)
   - [Common Types](#common-types)
-    - [apiaggregator.APIProxy](#apiaggregatorapiproxy)
+    - [apiaggregator.Pipeline](#apiaggregatorpipeline)
     - [pathadaptor.Spec](#pathadaptorspec)
     - [pathadaptor.RegexpReplace](#pathadaptorregexpreplace)
     - [httpheader.AdaptSpec](#httpheaderadaptspec)
@@ -79,17 +79,17 @@ A Filter is a request/response processor. Multiple filters can be orchestrated t
 
 ## APIAggregator
 
-The API Aggregator forwards one request to multiple API proxies and aggregates responses.
+The API Aggregator forwards one request to multiple API HTTP Pipelines in the same namespace and aggregates responses.
 
-Below is an example configuration that forwards one request to two proxies, `http-proxy-1` and `http-proxy-2`. When forwarding a request to `http-proxy-2`, the request method is changed to `GET` and a new header `Original-Method` is added with the original request method. The two responses are merged into one before return to the client.
+Below is an example configuration that forwards one request to two pipelines, `http-pipeline-1` and `http-pipeline-2`. When forwarding a request to `http-pipeline-2`, the request method is changed to `GET` and a new header `Original-Method` is added with the original request method. The two responses are merged into one before return to the client.
 
 ```yaml
 kind: APIAggregator
 name: api-aggregator-example
 mergeResponse: true
-apiProxies:
-- httpProxyName: http-proxy-1
-- httpProxyName: http-proxy-2
+pipelines:
+- name: http-pipeline-1
+- name: http-pipeline-2
   method: GET
   header:
     add:
@@ -102,10 +102,10 @@ apiProxies:
 | Name           | Type                                               | Description                                                                                                                                     | Required |
 | -------------- | -------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------- | -------- |
 | maxBodyBytes   | int64                                              | The upper limit of request body size, default is 0 which means no limit                                                                         | No       |
-| partialSucceed | bool                                               | Whether regards the result of the original request as successful or not when a request to some of the API proxies fails, default is false       | No       |
+| partialSucceed | bool                                               | Whether regards the result of the original request as successful or not when a request to some of the API pipelines fails, default is false     | No       |
 | timeout        | string                                             | Timeout duration for requests to API proxies                                                                                                    | No       |
 | mergeResponse  | bool                                               | Whether merging the multiple response objects into one, default is false means the final response is an array of the responses from API proxies | No       |
-| apiProxies     | [][apiaggregator.APIProxy](#apiaggregatorapiproxy) | Configuration of API proxies                                                                                                                    | Yes      |
+| pipelines      | [][apiaggregator.Pipeline](#apiaggregatorpipeline) | Configuration of API proxies                                                                                                                    | Yes      |
 
 ### Results
 
@@ -627,34 +627,35 @@ $ go build -tags=wasmhost
 
 ### Configuration
 
-| Name           | Type              | Description    | Required |
-| -------------- | ----------------- |--------------- | -------- |
-| maxConcurrency | int32             | The maximum requests the filter can process concurrently. Default is 10 and minimum value is 1. | Yes       |
-| code           | string            | The wasm code, can be the base64 encoded code, or path/url of the file which contains the code. | Yes    |
-| timeout        | string            | Timeout for wasm execution, default is 100ms. | Yes     |
-| parameters     | map[string]string | Parameters to initialize the wasm code. | No     |
+| Name           | Type              | Description                                                                                     | Required |
+| -------------- | ----------------- | ----------------------------------------------------------------------------------------------- | -------- |
+| maxConcurrency | int32             | The maximum requests the filter can process concurrently. Default is 10 and minimum value is 1. | Yes      |
+| code           | string            | The wasm code, can be the base64 encoded code, or path/url of the file which contains the code. | Yes      |
+| timeout        | string            | Timeout for wasm execution, default is 100ms.                                                   | Yes      |
+| parameters     | map[string]string | Parameters to initialize the wasm code.                                                         | No       |
+
 
 ### Results
 
-| Value       | Description                         |
-| ----------- | ----------------------------------- |
-| outOfVM     | Can not found an available wasm VM. |
-| wasmError   | An error occurs during the execution of wasm code. |
-| wasmResult1 <td rowspan="3">Results defined and returned by wasm code.</td>
-|     ...      
-| wasmResult9 
+| Value                                                                       | Description                                        |
+| --------------------------------------------------------------------------- | -------------------------------------------------- |
+| outOfVM                                                                     | Can not found an available wasm VM.                |
+| wasmError                                                                   | An error occurs during the execution of wasm code. |
+| wasmResult1 <td rowspan="3">Results defined and returned by wasm code.</td> |
+| ...                                                                         |
+| wasmResult9                                                                 |
 
 ## Common Types
 
-### apiaggregator.APIProxy
+### apiaggregator.Pipeline
 
-| Name          | Type                                         | Description                                                                | Required |
-| ------------- | -------------------------------------------- | -------------------------------------------------------------------------- | -------- |
-| httpProxyName | string                                       | The name of target HTTP pipeline                                           | Yes      |
-| method        | string                                       | Replaces request method with the value of this option when specified       | No       |
-| path          | [pathadaptor.Spec](#pathadaptorSpec)         | Rules to revise request path                                               | No       |
-| header        | [httpheader.AdaptSpec](#httpheaderAdaptSpec) | Rules to revise request header                                             | No       |
-| disableBody   | bool                                         | Whether forwards the body of the original request or not, default is false | No       |
+| Name        | Type                                         | Description                                                                | Required |
+| ----------- | -------------------------------------------- | -------------------------------------------------------------------------- | -------- |
+| name        | string                                       | The name of target HTTP pipeline                                           | Yes      |
+| method      | string                                       | Replaces request method with the value of this option when specified       | No       |
+| path        | [pathadaptor.Spec](#pathadaptorSpec)         | Rules to revise request path                                               | No       |
+| header      | [httpheader.AdaptSpec](#httpheaderAdaptSpec) | Rules to revise request header                                             | No       |
+| disableBody | bool                                         | Whether forwards the body of the original request or not, default is false | No       |
 
 ### pathadaptor.Spec
 
diff --git a/example/sbin/start.sh b/example/sbin/start.sh
index 8fa80e6a..a2706b23 100755
--- a/example/sbin/start.sh
+++ b/example/sbin/start.sh
@@ -18,11 +18,18 @@ fi
 touch  $stdfile
 nohup $server "$@" --config-file $cfgfile   >> $stdfile 2>&1 &
 
-pid=`ps -eo pid,args | grep "$server" | grep -v grep | awk '{print $1}'`
-if [ "$pid" = "" ]; then
-    echo "Error: failed to start $sever"
-    exit 2
-fi
+try_time=0
+pid=""
+while [ "$pid" = "" ]
+do
+    sleep 3
+    try_time=$(($try_time+1))
+    pid=`ps -eo pid,args | grep "$server" | grep -v grep | awk '{print $1}'`
+    if [[ $try_time -ge 2 ]]; then
+        echo "Error: failed to start $sever"
+        exit 2
+    fi
+done
 
 popd > /dev/null
 
diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index fc5c2e79..c1c78385 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -31,7 +31,6 @@ import (
 	"go.etcd.io/etcd/server/v3/embed"
 	yaml "gopkg.in/yaml.v2"
 
-	"github.com/megaease/easegress/pkg/common"
 	"github.com/megaease/easegress/pkg/logger"
 	"github.com/megaease/easegress/pkg/option"
 )
@@ -366,18 +365,6 @@ func (c *cluster) checkClusterName() error {
 	return nil
 }
 
-// removeAndBackupEtcdData is DEPRECATED,
-// will be deleted when it's sure that we won't need it.
-func (c *cluster) removeAndBackupEtcdData() {
-	if !common.IsDirEmpty(c.opt.AbsDataDir) {
-		logger.Infof("backup and clean %s", c.opt.AbsDataDir)
-		err := common.BackupAndCleanDir(c.opt.AbsDataDir)
-		if err != nil {
-			logger.Errorf("backup and clean %s failed: %v", c.opt.AbsDataDir, err)
-		}
-	}
-}
-
 func (c *cluster) getClient() (*clientv3.Client, error) {
 	c.clientMutex.RLock()
 	if c.client != nil {
diff --git a/pkg/common/lazybuf.go b/pkg/common/lazybuf.go
deleted file mode 100644
index f3eb64f0..00000000
--- a/pkg/common/lazybuf.go
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Copyright (c) 2017, MegaEase
- * All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package common
-
-// NOTE: Copy from path/path.go in standard library.
-
-// A Lazybuf is a lazily constructed path buffer.
-// It supports append, reading previously appended bytes,
-// and retrieving the final string. It does not allocate a buffer
-// to hold the output until that output diverges from s.
-type Lazybuf struct {
-	s   string
-	buf []byte
-	w   int
-}
-
-// NewLazybuf returns lazy buffer
-func NewLazybuf(s string) *Lazybuf {
-	return &Lazybuf{
-		s: s,
-	}
-}
-
-// Index returns a byte by its index
-func (b *Lazybuf) Index(i int) byte {
-	if b.buf != nil {
-		return b.buf[i]
-	}
-	return b.s[i]
-}
-
-// Append appends a byte into buffer
-func (b *Lazybuf) Append(c byte) {
-	if b.buf == nil {
-		if b.w < len(b.s) && b.s[b.w] == c {
-			b.w++
-			return
-		}
-		b.buf = make([]byte, len(b.s))
-		copy(b.buf, b.s[:b.w])
-	}
-	b.buf[b.w] = c
-	b.w++
-}
-
-// String returns the string of lazybuf
-func (b *Lazybuf) String() string {
-	if b.buf == nil {
-		return b.s[:b.w]
-	}
-	return string(b.buf[:b.w])
-}
diff --git a/pkg/common/sample.go b/pkg/common/sample.go
deleted file mode 100644
index 51cd9ed9..00000000
--- a/pkg/common/sample.go
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- * Copyright (c) 2017, MegaEase
- * All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package common
-
-import (
-	"math"
-	"sync"
-	"time"
-
-	metrics "github.com/rcrowley/go-metrics"
-)
-
-// ExpDecaySample is the structure use for monitoring metrics(such as: P90, P50 etc)
-type ExpDecaySample struct {
-	bucketIdxLock sync.RWMutex
-	bucketIdx     uint64
-	buckets       []metrics.Sample
-	stop          chan struct{}
-	closeLock     sync.RWMutex
-	closed        bool
-}
-
-// NewExpDecaySample creates ExpDecaySample structure
-func NewExpDecaySample(timeRange time.Duration, secondsForEachBucket int) *ExpDecaySample {
-	if secondsForEachBucket == 0 {
-		secondsForEachBucket = int(timeRange / 2)
-	}
-
-	s := new(ExpDecaySample)
-
-	n := uint64(math.Ceil(timeRange.Seconds()/float64(secondsForEachBucket))) + 1
-
-	var idx uint64
-	for idx < n {
-		s.buckets = append(s.buckets, metrics.NewExpDecaySample(514, 0.015))
-		idx++
-	}
-
-	s.stop = make(chan struct{})
-	switcher := func() {
-		ticker := time.NewTicker(time.Second * time.Duration(secondsForEachBucket))
-		defer ticker.Stop()
-		for {
-			select {
-			case <-ticker.C:
-				s.bucketIdxLock.Lock()
-
-				s.buckets[s.bucketIdx].Clear()
-				s.bucketIdx = (s.bucketIdx + 1) % n
-
-				s.bucketIdxLock.Unlock()
-			case <-s.stop:
-				return
-			}
-		}
-	}
-
-	go switcher()
-
-	return s
-}
-
-// Close cleans the ExpDecaySample.
-func (s *ExpDecaySample) Close() {
-	s.closeLock.Lock()
-	defer s.closeLock.Unlock()
-
-	if s.closed {
-		return
-	}
-
-	close(s.stop)
-	s.closed = true
-
-	// defense, accelerate gc
-	for i := 0; i < len(s.buckets); i++ {
-		s.buckets[i].Clear()
-	}
-}
-
-// Update updates the ExpDecaySample
-func (s *ExpDecaySample) Update(v int64) {
-	for i := 0; i < len(s.buckets); i++ {
-		s.buckets[i].Update(v)
-	}
-}
-
-// Percentile calculates the percentile
-func (s *ExpDecaySample) Percentile(q float64) float64 {
-	s.bucketIdxLock.RLock()
-	defer s.bucketIdxLock.RUnlock()
-
-	idx := (s.bucketIdx + 1) % uint64(len(s.buckets))
-
-	return s.buckets[idx].Percentile(q)
-}
-
-// StdDev return the standard deviation value
-func (s *ExpDecaySample) StdDev() float64 {
-	s.bucketIdxLock.RLock()
-	defer s.bucketIdxLock.RUnlock()
-
-	idx := (s.bucketIdx + 1) % uint64(len(s.buckets))
-
-	return s.buckets[idx].StdDev()
-}
-
-// Max returns the max value
-func (s *ExpDecaySample) Max() int64 {
-	s.bucketIdxLock.RLock()
-	defer s.bucketIdxLock.RUnlock()
-
-	idx := (s.bucketIdx + 1) % uint64(len(s.buckets))
-
-	return s.buckets[idx].Max()
-}
-
-// Min returns the min value
-func (s *ExpDecaySample) Min() int64 {
-	s.bucketIdxLock.RLock()
-	defer s.bucketIdxLock.RUnlock()
-
-	idx := (s.bucketIdx + 1) % uint64(len(s.buckets))
-
-	return s.buckets[idx].Min()
-}
-
-// Count returns the count value
-func (s *ExpDecaySample) Count() int64 {
-	s.bucketIdxLock.RLock()
-	defer s.bucketIdxLock.RUnlock()
-
-	idx := (s.bucketIdx + 1) % uint64(len(s.buckets))
-
-	return s.buckets[idx].Count()
-}
-
-// Variance returns the variance value
-func (s *ExpDecaySample) Variance() float64 {
-	s.bucketIdxLock.RLock()
-	defer s.bucketIdxLock.RUnlock()
-
-	idx := (s.bucketIdx + 1) % uint64(len(s.buckets))
-
-	return s.buckets[idx].Variance()
-}
-
-// Sum returns the sum value
-func (s *ExpDecaySample) Sum() int64 {
-	s.bucketIdxLock.RLock()
-	defer s.bucketIdxLock.RUnlock()
-
-	idx := (s.bucketIdx + 1) % uint64(len(s.buckets))
-
-	return s.buckets[idx].Sum()
-}
diff --git a/pkg/common/sys_proc_others.go b/pkg/common/sys_proc_others.go
deleted file mode 100644
index 084c47b8..00000000
--- a/pkg/common/sys_proc_others.go
+++ /dev/null
@@ -1,29 +0,0 @@
-// +build darwin dragonfly freebsd netbsd openbsd solaris windows plan9
-
-/*
- * Copyright (c) 2017, MegaEase
- * All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package common
-
-import (
-	"syscall"
-)
-
-// SysProcAttr returns system process attribute
-func SysProcAttr() *syscall.SysProcAttr {
-	return nil
-}
diff --git a/pkg/common/utils.go b/pkg/common/utils.go
index ceffcaed..fb91fb88 100644
--- a/pkg/common/utils.go
+++ b/pkg/common/utils.go
@@ -188,24 +188,6 @@ var (
 	FalseStrings = []string{"0", "f", "false", "off", "n", "no"}
 )
 
-// RemoveRepeatedByte removes the repeated bytes.
-func RemoveRepeatedByte(s string, needRemoveByte byte) string {
-	if len(s) < 2 {
-		return s
-	}
-
-	out := NewLazybuf(s)
-	repeatingByte := s[0]
-	out.Append(repeatingByte)
-	for _, c := range []byte(s[1:]) {
-		if c != repeatingByte || c != needRemoveByte {
-			out.Append(c)
-		}
-		repeatingByte = c
-	}
-	return out.String()
-}
-
 // NextNumberPowerOf2 return the number of power of 2
 // Via: https://stackoverflow.com/a/466242/1705845
 //      https://graphics.stanford.edu/~seander/bithacks.html#RoundUpPowerOf2
diff --git a/pkg/context/httptemplate.go b/pkg/context/httptemplate.go
index 2e1f97e0..9242801e 100644
--- a/pkg/context/httptemplate.go
+++ b/pkg/context/httptemplate.go
@@ -185,6 +185,9 @@ func NewHTTPTemplateDummy() *HTTPTemplate {
 
 func readBody(body io.Reader, maxBodySize int64) (*bytes.Buffer, error) {
 	buff := bytes.NewBuffer(nil)
+	if body == nil {
+		return buff, nil
+	}
 	written, err := io.CopyN(buff, body, defaultMaxBodySize+1)
 
 	if err != nil && err != io.EOF {
diff --git a/pkg/filter/apiaggregator/apiaggregator.go b/pkg/filter/apiaggregator/apiaggregator.go
index c3f5d3e6..8fc0828c 100644
--- a/pkg/filter/apiaggregator/apiaggregator.go
+++ b/pkg/filter/apiaggregator/apiaggregator.go
@@ -32,7 +32,7 @@ import (
 	"github.com/megaease/easegress/pkg/context"
 	"github.com/megaease/easegress/pkg/logger"
 	"github.com/megaease/easegress/pkg/object/httppipeline"
-	"github.com/megaease/easegress/pkg/protocol"
+	"github.com/megaease/easegress/pkg/object/rawconfigtrafficcontroller"
 	"github.com/megaease/easegress/pkg/tracing"
 	"github.com/megaease/easegress/pkg/util/httpheader"
 	"github.com/megaease/easegress/pkg/util/pathadaptor"
@@ -48,45 +48,56 @@ const (
 var results = []string{resultFailed}
 
 func init() {
-	// FIXME: Rewrite APIAggregator because the HTTPProxy is eliminated
-	// I(@xxx7xxxx) think we should not empower filter to cross pipelines.
-
-	// httppipeline.Register(&APIAggregator{})
+	httppipeline.Register(&APIAggregator{})
 }
 
 type (
-	// APIAggregator is the entity to complete rate limiting.
+	// APIAggregator is a filter to aggregate several HTTP API responses.
 	APIAggregator struct {
 		filterSpec *httppipeline.FilterSpec
 		spec       *Spec
 
-		muxMapper protocol.MuxMapper
+		// rctc is for getting Pipeline in default namespace.
+		rctc *rawconfigtrafficcontroller.RawConfigTrafficController
 	}
 
-	// Spec describes APIAggregator.
+	// Spec is APIAggregator's spec.
 	Spec struct {
 		// MaxBodyBytes in [0, 10MB]
-		MaxBodyBytes   int64  `yaml:"maxBodyBytes" jsonschema:"omitempty,minimum=0,maximum=102400"`
-		PartialSucceed bool   `yaml:"partialSucceed"`
-		Timeout        string `yaml:"timeout" jsonschema:"omitempty,format=duration"`
-		MergeResponse  bool   `yaml:"mergeResponse"`
+		MaxBodyBytes int64 `yaml:"maxBodyBytes" jsonschema:"omitempty,minimum=0,maximum=102400"`
+
+		// PartialSucceed indicates wether Whether regards the result of the original request as successful
+		// or not when a request to some of the API pipelines fails.
+		PartialSucceed bool `yaml:"partialSucceed"`
+
+		// Timeout is the request duration for each APIs.
+		Timeout string `yaml:"timeout" jsonschema:"omitempty,format=duration"`
 
-		// User describes HTTP service target via an existing HTTPProxy
-		APIProxies []*APIProxy `yaml:"apiProxies" jsonschema:"required"`
+		// MergeResponse indicates whether to merge JSON response bodies or not.
+		MergeResponse bool `yaml:"mergeResponse"`
+
+		// User describes HTTP service target via an existing Pipeline.
+		Pipelines []*Pipeline `yaml:"pipelines" jsonschema:"required"`
 
 		timeout *time.Duration
 	}
 
-	// APIProxy describes the single API in EG's HTTPProxy object.
-	APIProxy struct {
-		// HTTPProxy's name in EG
-		HTTPProxyName string `yaml:"httpProxyName" jsonschema:"required"`
+	// Pipeline is the single API HTTP Pipeline in default namespace.
+	Pipeline struct {
+		// Name is the name of pipeline in EG
+		Name string `yaml:"name" jsonschema:"required"`
+
+		// Method is the HTTP method for requesting this pipeline.
+		Method string `yaml:"method" jsonschema:"omitempty,format=httpmethod"`
+
+		// Path is the HTTP request path adaptor for requesting this pipeline.
+		Path *pathadaptor.Spec `yaml:"path,omitempty" jsonschema:"omitempty"`
 
-		// Describes details about the request-target
-		Method      string                `yaml:"method" jsonschema:"omitempty,format=httpmethod"`
-		Path        *pathadaptor.Spec     `yaml:"path,omitempty" jsonschema:"omitempty"`
-		Header      *httpheader.AdaptSpec `yaml:"header,omitempty" jsonschema:"omitempty"`
-		DisableBody bool                  `yaml:"disableBody" jsonschema:"omitempty"`
+		// Header is the HTTP header adaptor for requestring this pipeline.
+		Header *httpheader.AdaptSpec `yaml:"header,omitempty" jsonschema:"omitempty"`
+
+		// DisableBody discart this pipeline's response body if it set to true.
+		DisableBody bool `yaml:"disableBody" jsonschema:"omitempty"`
 
 		pa *pathadaptor.PathAdaptor
 	}
@@ -118,6 +129,16 @@ func (aa *APIAggregator) Results() []string {
 // Init initializes APIAggregator.
 func (aa *APIAggregator) Init(filterSpec *httppipeline.FilterSpec) {
 	aa.filterSpec, aa.spec = filterSpec, filterSpec.FilterSpec().(*Spec)
+	entity, exists := filterSpec.Super().GetSystemController(rawconfigtrafficcontroller.Kind)
+	if !exists {
+		panic(fmt.Errorf("BUG: raw config traffic controller not found"))
+	}
+
+	rctc, ok := entity.Instance().(*rawconfigtrafficcontroller.RawConfigTrafficController)
+	if !ok {
+		panic(fmt.Errorf("BUG: want *RawConfigTrafficController, got %T", entity.Instance()))
+	}
+	aa.rctc = rctc
 	aa.reload()
 }
 
@@ -138,9 +159,9 @@ func (aa *APIAggregator) reload() {
 		}
 	}
 
-	for _, proxy := range aa.spec.APIProxies {
-		if proxy.Path != nil {
-			proxy.pa = pathadaptor.New(proxy.Path)
+	for _, p := range aa.spec.Pipelines {
+		if p.Path != nil {
+			p.pa = pathadaptor.New(p.Path)
 		}
 	}
 }
@@ -151,11 +172,6 @@ func (aa *APIAggregator) Handle(ctx context.HTTPContext) (result string) {
 	return ctx.CallNextHandler(result)
 }
 
-// InjectMuxMapper injects mux mapper into APIAggregator.
-func (aa *APIAggregator) InjectMuxMapper(mapper protocol.MuxMapper) {
-	aa.muxMapper = mapper
-}
-
 func (aa *APIAggregator) handle(ctx context.HTTPContext) (result string) {
 	buff := bytes.NewBuffer(nil)
 	if aa.spec.MaxBodyBytes > 0 {
@@ -173,34 +189,33 @@ func (aa *APIAggregator) handle(ctx context.HTTPContext) (result string) {
 	}
 
 	wg := &sync.WaitGroup{}
-	wg.Add(len(aa.spec.APIProxies))
+	wg.Add(len(aa.spec.Pipelines))
 
-	httpResps := make([]context.HTTPResponse, len(aa.spec.APIProxies))
-	for i, proxy := range aa.spec.APIProxies {
-		req, err := aa.newHTTPReq(ctx, proxy, buff)
+	httpResps := make([]context.HTTPResponse, len(aa.spec.Pipelines))
+	for i, p := range aa.spec.Pipelines {
+		req, err := aa.newHTTPReq(ctx, p, buff)
 		if err != nil {
-			logger.Errorf("BUG: new HTTPProxy request failed %v proxyname[%d]", err, aa.spec.APIProxies[i].HTTPProxyName)
+			logger.Errorf("BUG: new HTTP request failed: %v, pipelinename: %s", err, aa.spec.Pipelines[i].Name)
 			ctx.Response().SetStatusCode(http.StatusBadRequest)
 			return resultFailed
 		}
 
 		go func(i int, name string, req *http.Request) {
 			defer wg.Done()
-			copyCtx, err := aa.newCtx(ctx, req, buff)
-			if err != nil {
-				httpResps[i] = nil
+			handler, exists := aa.rctc.GetHTTPPipeline(name)
+			if !exists {
+				logger.Errorf("pipeline: %s not found in current namespace", name)
 				return
 			}
+			w := httptest.NewRecorder()
+			copyCtx := context.New(w, req, tracing.NoopTracing, "no trace")
+			handler.Handle(copyCtx)
+			rsp := copyCtx.Response()
 
-			handler, exists := aa.muxMapper.GetHandler(name)
-
-			if !exists {
-				httpResps[i] = nil
-			} else {
-				handler.Handle(copyCtx)
-				httpResps[i] = copyCtx.Response()
+			if rsp != nil && rsp.StatusCode() == http.StatusOK {
+				httpResps[i] = rsp
 			}
-		}(i, proxy.HTTPProxyName, req)
+		}(i, p.Name, req)
 	}
 
 	wg.Wait()
@@ -217,19 +232,17 @@ func (aa *APIAggregator) handle(ctx context.HTTPContext) (result string) {
 
 	data := make(map[string][]byte)
 
-	// Get all HTTPProxy response' body
+	// Get all HTTPPipeline response' body
 	for i, resp := range httpResps {
 		if resp == nil && !aa.spec.PartialSucceed {
-			ctx.AddTag(fmt.Sprintf("apiAggregator: failed in HTTPProxy %s",
-				aa.spec.APIProxies[i].HTTPProxyName))
+			ctx.Response().Std().Header().Set("X-EG-Aggregator", fmt.Sprintf("failed-in-%s",
+				aa.spec.Pipelines[i].Name))
 			ctx.Response().SetStatusCode(http.StatusServiceUnavailable)
 			return resultFailed
 		}
 
-		// call HTTPProxy could be successful even with a no exist backend
-		// so resp is not nil, but resp.Body() is nil.
 		if resp != nil && resp.Body() != nil {
-			if res := aa.copyHTTPBody2Map(resp.Body(), ctx, data, aa.spec.APIProxies[i].HTTPProxyName); len(res) != 0 {
+			if res := aa.copyHTTPBody2Map(resp.Body(), ctx, data, aa.spec.Pipelines[i].Name); len(res) != 0 {
 				return res
 			}
 		}
@@ -238,22 +251,7 @@ func (aa *APIAggregator) handle(ctx context.HTTPContext) (result string) {
 	return aa.formatResponse(ctx, data)
 }
 
-func (aa *APIAggregator) newCtx(ctx context.HTTPContext, req *http.Request, buff *bytes.Buffer) (context.HTTPContext, error) {
-	// Construct a new context for the HTTPProxy
-	// responseWriter is an HTTP responseRecorder, not the original context's real
-	// responseWriter, or these Proxies will overwrite each others
-	w := httptest.NewRecorder()
-	var stdctx stdcontext.Context = ctx
-	if aa.spec.timeout != nil {
-		stdctx, _ = stdcontext.WithTimeout(stdctx, *aa.spec.timeout)
-	}
-
-	copyCtx := context.New(w, req, tracing.NoopTracing, "no trace")
-
-	return copyCtx, nil
-}
-
-func (aa *APIAggregator) newHTTPReq(ctx context.HTTPContext, proxy *APIProxy, buff *bytes.Buffer) (*http.Request, error) {
+func (aa *APIAggregator) newHTTPReq(ctx context.HTTPContext, p *Pipeline, buff *bytes.Buffer) (*http.Request, error) {
 	var stdctx stdcontext.Context = ctx
 	if aa.spec.timeout != nil {
 		// NOTE: Cancel function could be omitted here.
@@ -261,17 +259,17 @@ func (aa *APIAggregator) newHTTPReq(ctx context.HTTPContext, proxy *APIProxy, bu
 	}
 
 	method := ctx.Request().Method()
-	if proxy.Method != "" {
-		method = proxy.Method
+	if p.Method != "" {
+		method = p.Method
 	}
 
 	url := ctx.Request().Std().URL
-	if proxy.pa != nil {
-		url.Path = proxy.pa.Adapt(url.Path)
+	if p.pa != nil {
+		url.Path = p.pa.Adapt(url.Path)
 	}
 
 	var body io.Reader
-	if !proxy.DisableBody {
+	if !p.DisableBody {
 		body = bytes.NewReader(buff.Bytes())
 	}
 
diff --git a/pkg/filter/circuitbreaker/circuitbreaker.go b/pkg/filter/circuitbreaker/circuitbreaker.go
index 5543bc86..40699717 100644
--- a/pkg/filter/circuitbreaker/circuitbreaker.go
+++ b/pkg/filter/circuitbreaker/circuitbreaker.go
@@ -107,7 +107,7 @@ URLLoop:
 	return nil
 }
 
-func (url *URLRule) createCircuitBreaker() {
+func (url *URLRule) buildPolicy() *libcb.Policy {
 	policy := libcb.Policy{
 		FailureRateThreshold:             url.policy.FailureRateThreshold,
 		SlowCallRateThreshold:            url.policy.SlowCallRateThreshold,
@@ -157,7 +157,12 @@ func (url *URLRule) createCircuitBreaker() {
 		policy.WaitDurationInOpen = time.Minute
 	}
 
-	url.cb = libcb.New(&policy)
+	return &policy
+}
+
+func (url *URLRule) createCircuitBreaker() {
+	policy := url.buildPolicy()
+	url.cb = libcb.New(policy)
 }
 
 // Kind returns the kind of CircuitBreaker.
diff --git a/pkg/filter/mock/mock.go b/pkg/filter/mock/mock.go
index 6d2130c6..4c8ad60e 100644
--- a/pkg/filter/mock/mock.go
+++ b/pkg/filter/mock/mock.go
@@ -103,11 +103,7 @@ func (m *Mock) reload() {
 		if r.Delay == "" {
 			continue
 		}
-		var err error
-		r.delay, err = time.ParseDuration(r.Delay)
-		if err != nil {
-			logger.Errorf("BUG: parse duration %s failed: %v", r.Delay, err)
-		}
+		r.delay, _ = time.ParseDuration(r.Delay)
 	}
 }
 
@@ -129,7 +125,7 @@ func (m *Mock) handle(ctx context.HTTPContext) (result string) {
 		w.SetBody(strings.NewReader(rule.Body))
 		result = resultMocked
 
-		if rule.delay < 0 {
+		if rule.delay <= 0 {
 			return
 		}
 
diff --git a/pkg/filter/proxy/compression.go b/pkg/filter/proxy/compression.go
index dbd9c8d6..0a33f6b4 100644
--- a/pkg/filter/proxy/compression.go
+++ b/pkg/filter/proxy/compression.go
@@ -54,7 +54,7 @@ type (
 	}
 )
 
-func newcompression(spec *CompressionSpec) *compression {
+func newCompression(spec *CompressionSpec) *compression {
 	return &compression{
 		spec: spec,
 	}
diff --git a/pkg/filter/proxy/proxy.go b/pkg/filter/proxy/proxy.go
index f083bdf9..3019de11 100644
--- a/pkg/filter/proxy/proxy.go
+++ b/pkg/filter/proxy/proxy.go
@@ -212,7 +212,7 @@ func (b *Proxy) reload() {
 	}
 
 	if b.spec.Compression != nil {
-		b.compression = newcompression(b.spec.Compression)
+		b.compression = newCompression(b.spec.Compression)
 	}
 }
 
diff --git a/pkg/filter/proxy/server.go b/pkg/filter/proxy/server.go
index fb76ef15..343f2a71 100644
--- a/pkg/filter/proxy/server.go
+++ b/pkg/filter/proxy/server.go
@@ -31,8 +31,14 @@ import (
 	"github.com/megaease/easegress/pkg/util/stringtool"
 )
 
+// make it mockable in test
+var fnGetService atomic.Value
+
 func init() {
 	rand.Seed(time.Now().UnixNano())
+	fnGetService.Store(func(serviceRegistry, serviceName string) (*serviceregistry.Service, error) {
+		return serviceregistry.Global.GetService(serviceRegistry, serviceName)
+	})
 }
 
 const (
@@ -180,11 +186,16 @@ func (s *servers) useStaticServers() {
 	s.service = nil
 }
 
+func getService(serviceRegistry, serviceName string) (*serviceregistry.Service, error) {
+	fn := fnGetService.Load().(func(serviceRegistry, serviceName string) (*serviceregistry.Service, error))
+	return fn(serviceRegistry, serviceName)
+}
+
 func (s *servers) useService() (*serviceregistry.Service, error) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
-	service, err := serviceregistry.Global.GetService(s.poolSpec.ServiceRegistry, s.poolSpec.ServiceName)
+	service, err := getService(s.poolSpec.ServiceRegistry, s.poolSpec.ServiceName)
 	if err != nil {
 		return nil, fmt.Errorf("get service %s failed: %v", s.poolSpec.ServiceName, err)
 	}
diff --git a/pkg/filter/responseadaptor/responseadaptor.go b/pkg/filter/responseadaptor/responseadaptor.go
index 15ea5e96..0a1ab348 100644
--- a/pkg/filter/responseadaptor/responseadaptor.go
+++ b/pkg/filter/responseadaptor/responseadaptor.go
@@ -98,18 +98,18 @@ func (ra *ResponseAdaptor) handle(ctx context.HTTPContext) string {
 	hte := ctx.Template()
 	ctx.Response().Header().Adapt(ra.spec.Header, hte)
 
-	if len(ra.spec.Body) != 0 {
-		if hte.HasTemplates(ra.spec.Body) {
-			if body, err := hte.Render(ra.spec.Body); err != nil {
-				logger.Errorf("BUG responseadaptor render body faile , template %s , err %v",
-					ra.spec.Body, err)
-			} else {
-				ctx.Response().SetBody(bytes.NewReader([]byte(body)))
-			}
-		} else {
-			ctx.Response().SetBody(bytes.NewReader([]byte(ra.spec.Body)))
-		}
+	if len(ra.spec.Body) == 0 {
+		return ""
 	}
+
+	if !hte.HasTemplates(ra.spec.Body) {
+		ctx.Response().SetBody(bytes.NewReader([]byte(ra.spec.Body)))
+	} else if body, err := hte.Render(ra.spec.Body); err != nil {
+		logger.Errorf("BUG responseadaptor render body failed, template %s , err %v", ra.spec.Body, err)
+	} else {
+		ctx.Response().SetBody(bytes.NewReader([]byte(body)))
+	}
+
 	return ""
 }
 
diff --git a/pkg/filter/validator/oauth2.go b/pkg/filter/validator/oauth2.go
index 727d0bef..62744265 100644
--- a/pkg/filter/validator/oauth2.go
+++ b/pkg/filter/validator/oauth2.go
@@ -93,6 +93,11 @@ func NewOAuth2Validator(spec *OAuth2ValidatorSpec) *OAuth2Validator {
 	return v
 }
 
+// make it mockable
+var fnSendRequest = func(client *http.Client, r *http.Request) (*http.Response, error) {
+	return client.Do(r)
+}
+
 func (v *OAuth2Validator) introspectToken(tokenStr string) (*tokenInfo, error) {
 	var body bytes.Buffer
 	body.WriteString("token=")
@@ -111,7 +116,7 @@ func (v *OAuth2Validator) introspectToken(tokenStr string) (*tokenInfo, error) {
 		r.Header.Set("Authorization", "Basic "+v.spec.TokenIntrospect.BasicAuth)
 	}
 
-	resp, e := v.client.Do(r)
+	resp, e := fnSendRequest(v.client, r)
 	if e != nil {
 		return nil, e
 	}
diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go
index 3c44aaf0..9c478ce9 100644
--- a/pkg/logger/logger.go
+++ b/pkg/logger/logger.go
@@ -37,6 +37,18 @@ func Init(opt *option.Options) {
 	initRestAPI(opt)
 }
 
+// InitNop initializes all logger as nop, mainly for unit testing
+func InitNop() {
+	nop := zap.NewNop()
+	httpFilterAccessLogger = nop
+	httpFilterDumpLogger = nop
+	restAPILogger = nop
+
+	defaultLogger = nop.Sugar()
+	gressLogger = defaultLogger
+	stderrLogger = defaultLogger
+}
+
 const (
 	stdoutFilename           = "stdout.log"
 	filterHTTPAccessFilename = "filter_http_access.log"
diff --git a/pkg/object/httppipeline/httppipeline.go b/pkg/object/httppipeline/httppipeline.go
index 7fd9f6b1..115113e3 100644
--- a/pkg/object/httppipeline/httppipeline.go
+++ b/pkg/object/httppipeline/httppipeline.go
@@ -444,6 +444,18 @@ func (hp *HTTPPipeline) Handle(ctx context.HTTPContext) {
 	filterStat := &FilterStat{}
 
 	handle := func(lastResult string) string {
+		// For saving the `filterIndex`'s filter generated HTTP Response.
+		// Note: the sequence of pipeline is stack-liked, we save the filter's response into template
+		// at the beginning of the next filter.
+		if filterIndex != -1 {
+			name := hp.runningFilters[filterIndex].spec.Name()
+			if err := ctx.SaveRspToTemplate(name); err != nil {
+				format := "save http rsp failed, dict is %#v err is %v"
+				logger.Errorf(format, ctx.Template().GetDict(), err)
+			}
+			logger.Debugf("filter %s, saved response dict %v", name, ctx.Template().GetDict())
+		}
+
 		// Filters are called recursively as a stack, so we need to save current
 		// state and restore it before return
 		lastIndex := filterIndex
@@ -468,6 +480,7 @@ func (hp *HTTPPipeline) Handle(ctx context.HTTPContext) {
 			logger.Errorf(format, ctx.Template().GetDict(), err)
 		}
 
+		logger.Debugf("filter %s saved request dict %v", name, ctx.Template().GetDict())
 		filterStat = &FilterStat{Name: name, Kind: filter.spec.Kind()}
 
 		startTime := time.Now()
@@ -476,11 +489,6 @@ func (hp *HTTPPipeline) Handle(ctx context.HTTPContext) {
 		filterStat.Duration = time.Since(startTime)
 		filterStat.Result = result
 
-		if err := ctx.SaveRspToTemplate(name); err != nil {
-			format := "save http rsp failed, dict is %#v err is %v"
-			logger.Errorf(format, ctx.Template().GetDict(), err)
-		}
-
 		lastStat.Next = append(lastStat.Next, filterStat)
 		return result
 	}
diff --git a/pkg/object/rawconfigtrafficcontroller/rawconfigtrafficcontroller.go b/pkg/object/rawconfigtrafficcontroller/rawconfigtrafficcontroller.go
index 02fc21f8..425e63e1 100644
--- a/pkg/object/rawconfigtrafficcontroller/rawconfigtrafficcontroller.go
+++ b/pkg/object/rawconfigtrafficcontroller/rawconfigtrafficcontroller.go
@@ -24,6 +24,7 @@ import (
 	"github.com/megaease/easegress/pkg/object/httppipeline"
 	"github.com/megaease/easegress/pkg/object/httpserver"
 	"github.com/megaease/easegress/pkg/object/trafficcontroller"
+	"github.com/megaease/easegress/pkg/protocol"
 	"github.com/megaease/easegress/pkg/supervisor"
 )
 
@@ -52,8 +53,7 @@ type (
 	}
 
 	// Spec describes RawConfigTrafficController.
-	Spec struct {
-	}
+	Spec struct{}
 
 	// Status is the status of RawConfigTrafficController.
 	Status = trafficcontroller.StatusInSameNamespace
@@ -90,6 +90,16 @@ func (rctc *RawConfigTrafficController) Inherit(spec *supervisor.Spec, previousG
 	rctc.Init(spec)
 }
 
+// GetHTTPPipeline gets Pipeline within the default namespace
+func (rctc *RawConfigTrafficController) GetHTTPPipeline(name string) (protocol.HTTPHandler, bool) {
+	p, exist := rctc.tc.GetHTTPPipeline(DefaultNamespace, name)
+	if !exist {
+		return nil, false
+	}
+	handler := p.Instance().(protocol.HTTPHandler)
+	return handler, true
+}
+
 func (rctc *RawConfigTrafficController) reload() {
 	entity, exists := rctc.superSpec.Super().GetSystemController(trafficcontroller.Kind)
 	if !exists {
diff --git a/pkg/util/jmxtool/agent_controller.go b/pkg/util/jmxtool/agent_controller.go
index 620cb8ef..69d2dd97 100644
--- a/pkg/util/jmxtool/agent_controller.go
+++ b/pkg/util/jmxtool/agent_controller.go
@@ -20,7 +20,6 @@ package jmxtool
 import (
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
 	"net/http"
 	"strconv"
 
@@ -75,17 +74,11 @@ func (agent *AgentClient) UpdateService(newService *spec.Service, version int64)
 	}
 
 	url := agent.URL + serviceConfigURL
-	resp, err := handleRequest(http.MethodPut, url, bytes)
+	bodyString, err := handleRequest(http.MethodPut, url, bytes)
 	if err != nil {
 		return fmt.Errorf("handleRequest error: %v", err)
 	}
-
-	bodyBytes, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return fmt.Errorf("read the http body failed: %v", err)
-	}
-	bodyString := string(bodyBytes)
-	logger.Infof("Update Service, URL: %s,request: %s, result: %v", url, string(bytes), bodyString)
+	logger.Infof("Update Service, URL: %s,request: %s, result: %v", url, string(bytes), string(bodyString))
 	return err
 }
 
@@ -108,15 +101,10 @@ func (agent *AgentClient) UpdateCanary(globalHeaders *spec.GlobalCanaryHeaders,
 	}
 
 	url := agent.URL + canaryConfigURL
-	resp, err := handleRequest(http.MethodPut, url, bytes)
+	bodyString, err := handleRequest(http.MethodPut, url, bytes)
 	if err != nil {
 		return fmt.Errorf("handleRequest error: %v", err)
 	}
-	bodyBytes, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return fmt.Errorf("read the http body failed: %v", err)
-	}
-	bodyString := string(bodyBytes)
-	logger.Infof("Update Canary, URL: %s, request %s, result: %v", url, string(bytes), bodyString)
+	logger.Infof("Update Canary, URL: %s,request: %s, result: %v", url, string(bytes), string(bodyString))
 	return err
 }
diff --git a/pkg/util/jmxtool/common.go b/pkg/util/jmxtool/common.go
index a74410bd..dd7b8664 100644
--- a/pkg/util/jmxtool/common.go
+++ b/pkg/util/jmxtool/common.go
@@ -95,7 +95,7 @@ type APIErr struct {
 	Message string `yaml:"message"`
 }
 
-func handleRequest(httpMethod string, url string, reqBody []byte) (*http.Response, error) {
+func handleRequest(httpMethod string, url string, reqBody []byte) ([]byte, error) {
 	req, err := http.NewRequest(httpMethod, url, bytes.NewReader(reqBody))
 	if err != nil {
 		return nil, err
@@ -103,27 +103,27 @@ func handleRequest(httpMethod string, url string, reqBody []byte) (*http.Respons
 
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		return resp, err
+		return nil, err
 	}
 	defer resp.Body.Close()
 
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
-		return resp, err
+		return nil, err
 	}
 
-	if !successfulStatusCode(resp.StatusCode) {
-		msg := string(body)
-		apiErr := &APIErr{}
-		err = yaml.Unmarshal(body, apiErr)
-		if err == nil {
-			msg = apiErr.Message
-		}
+	if successfulStatusCode(resp.StatusCode) {
+		return body, nil
+	}
 
-		return resp, fmt.Errorf("Request failed: Code: %d, Msg: %s ", resp.StatusCode, msg)
+	msg := string(body)
+	apiErr := &APIErr{}
+	err = yaml.Unmarshal(body, apiErr)
+	if err == nil {
+		msg = apiErr.Message
 	}
 
-	return resp, nil
+	return nil, fmt.Errorf("Request failed: Code: %d, Msg: %s ", resp.StatusCode, msg)
 }
 
 func successfulStatusCode(code int) bool {
diff --git a/pkg/common/sys_proc_linux.go b/pkg/util/jmxtool/common_test.go
similarity index 54%
rename from pkg/common/sys_proc_linux.go
rename to pkg/util/jmxtool/common_test.go
index e5691e3f..d693e4ef 100644
--- a/pkg/common/sys_proc_linux.go
+++ b/pkg/util/jmxtool/common_test.go
@@ -1,5 +1,3 @@
-// +build linux
-
 /*
  * Copyright (c) 2017, MegaEase
  * All rights reserved.
@@ -17,16 +15,30 @@
  * limitations under the License.
  */
 
-package common
+package jmxtool
 
 import (
-	"syscall"
+	"reflect"
+	"testing"
 )
 
-// SysProcAttr returns system process attribute
-func SysProcAttr() *syscall.SysProcAttr {
-	return &syscall.SysProcAttr{
-		Cloneflags: syscall.CLONE_NEWNS | syscall.CLONE_NEWIPC |
-			syscall.CLONE_NEWPID | syscall.CLONE_NEWUTS | syscall.CLONE_NEWUSER,
+func Test_extractKVs(t *testing.T) {
+	// extractKVs for int
+	prefix := "test"
+	want := []map[string]string{{prefix: "1"}}
+	got := extractKVs(prefix, 1)
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("error for extract int")
+	}
+
+	// extractKVs for interface array
+	want = []map[string]string{
+		{prefix + ".0": "1"},
+		{prefix + ".1": "2"},
+		{prefix + ".2": "3"},
+	}
+	got = extractKVs(prefix, []interface{}{1, 2, 3})
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("error for extract []interface")
 	}
 }


Test patch:
diff --git a/pkg/cluster/cluster_test.go b/pkg/cluster/cluster_test.go
index 182b2e4f..1e330b41 100644
--- a/pkg/cluster/cluster_test.go
+++ b/pkg/cluster/cluster_test.go
@@ -22,6 +22,8 @@ import (
 	"sync"
 	"testing"
 	"time"
+
+	"go.etcd.io/etcd/api/v3/mvccpb"
 )
 
 func mockClusters(count int) []*cluster {
@@ -65,7 +67,7 @@ func mockClusters(count int) []*cluster {
 		c := cls.(*cluster)
 
 		for {
-			err = c.getReady()
+			_, err := c.getClient()
 			time.Sleep(HeartbeatInterval)
 			if err != nil {
 				continue
@@ -90,6 +92,201 @@ func closeClusters(clusters []*cluster) {
 }
 
 func TestCluster(t *testing.T) {
-	clusters := mockClusters(5)
+	clusters := mockClusters(3)
 	defer closeClusters(clusters)
+	// for testing longRequestContext()
+	clusters[0].longRequestContext()
+
+}
+
+func TestLease(t *testing.T) {
+	_, err := strToLease("266394")
+	if err != nil {
+		t.Errorf("str to lease failed: %v", err)
+	}
+}
+
+func TestLeaseInvalid(t *testing.T) {
+	_, err := strToLease("test")
+	if err == nil {
+		t.Errorf("str to lease should not succ with \"test\" value")
+	}
+}
+
+func TestClusterSyncer(t *testing.T) {
+	opts, _, _ := mockMembers(1)
+	cls, err := New(opts[0])
+
+	if err != nil {
+		t.Errorf("init failed: %v", err)
+	}
+
+	c := cls.(*cluster)
+
+	_, err = c.getClient()
+	if err != nil {
+		t.Errorf("get ready failed: %v", err)
+	}
+
+	syncer, err := c.Syncer(3 * time.Second)
+
+	if err != nil {
+		t.Errorf("new syncer failed: %v", err)
+	}
+
+	if _, err = syncer.Sync("akey"); err != nil {
+		t.Errorf("syncer sync failed: %v", err)
+	}
+
+	if _, err = syncer.SyncRaw("akey"); err != nil {
+		t.Errorf("syncer sync failed: %v", err)
+	}
+
+	if _, err = syncer.SyncPrefix("akey"); err != nil {
+		t.Errorf("syncer sync failed: %v", err)
+	}
+
+	if _, err = syncer.SyncRawPrefix("akey"); err != nil {
+		t.Errorf("syncer sync failed: %v", err)
+	}
+
+	syncer.Close()
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	cls.CloseServer(wg)
+	wg.Wait()
+}
+
+func TestClusterWatcher(t *testing.T) {
+	opts, _, _ := mockMembers(1)
+	cls, err := New(opts[0])
+
+	if err != nil {
+		t.Errorf("init failed: %v", err)
+	}
+
+	c := cls.(*cluster)
+
+	_, err = c.getClient()
+	if err != nil {
+		t.Errorf("get ready failed: %v", err)
+	}
+
+	watcher, err := c.Watcher()
+
+	if err != nil {
+		t.Errorf("new syncer failed: %v", err)
+	}
+
+	if _, err = watcher.Watch("akey"); err != nil {
+		t.Errorf("watcher watch failed: %v", err)
+	}
+
+	if _, err = watcher.WatchRaw("akey"); err != nil {
+		t.Errorf("watcher watch failed: %v", err)
+	}
+
+	if _, err = watcher.WatchPrefix("akey"); err != nil {
+		t.Errorf("watcher watch failed: %v", err)
+	}
+
+	if _, err = watcher.WatchRawPrefix("akey"); err != nil {
+		t.Errorf("watcher watch failed: %v", err)
+	}
+
+	watcher.Close()
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	cls.CloseServer(wg)
+	wg.Wait()
+}
+
+func TestUtil(t *testing.T) {
+	equal := isDataEuqal(map[string]*mvccpb.KeyValue{
+		"aaa": {
+			Key:     []byte("akey"),
+			Version: 11233,
+		},
+	}, map[string]*mvccpb.KeyValue{
+		"aaa": {
+			Key:     []byte("akey"),
+			Version: 11233,
+		},
+	})
+
+	if !equal {
+		t.Error("isDataEqual failed")
+	}
+
+	equal = isDataEuqal(map[string]*mvccpb.KeyValue{
+		"aaa": {
+			Key:     []byte("akey"),
+			Version: 11233,
+		},
+		"bbb": {
+			Key:     []byte("akey"),
+			Version: 11233,
+		},
+	}, map[string]*mvccpb.KeyValue{
+		"aaa": {
+			Key:     []byte("akey"),
+			Version: 11233,
+		},
+	})
+
+	if equal {
+		t.Error("isDataEqual should not equal failed")
+	}
+}
+
+func TestMutexAndOP(t *testing.T) {
+	opts, _, _ := mockMembers(1)
+	cls, err := New(opts[0])
+
+	if err != nil {
+		t.Errorf("init failed: %v", err)
+	}
+
+	c := cls.(*cluster)
+
+	_, err = c.getClient()
+	if err != nil {
+		t.Errorf("get ready failed: %v", err)
+	}
+
+	m, err := c.Mutex("akey")
+	if err != nil {
+		t.Errorf("cluster mutex failed: %v", err)
+	}
+
+	m.Lock()
+	defer m.Unlock()
+
+	value := "a value"
+	err = c.PutAndDeleteUnderLease(map[string]*string{
+		"akey": &value,
+	})
+
+	if err != nil {
+		t.Errorf("PutAndDeleteUnderLease failed: %v", err)
+	}
+
+	err = c.PutAndDelete(map[string]*string{
+		"akey": &value,
+	})
+
+	if err != nil {
+		t.Errorf("PutAndDelete failed :%v", err)
+	}
+
+	if err = c.Delete("akey"); err != nil {
+		t.Errorf("Delete failed: %v", err)
+	}
+
+	err = c.DeletePrefix("akey")
+	if err != nil {
+		t.Errorf("DeletePrefix failed: %v", err)
+	}
 }
diff --git a/pkg/cluster/layout_test.go b/pkg/cluster/layout_test.go
new file mode 100644
index 00000000..2f5e30fc
--- /dev/null
+++ b/pkg/cluster/layout_test.go
@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cluster
+
+import (
+	"testing"
+)
+
+func TestLayout(t *testing.T) {
+	l := Layout{}
+
+	if len(l.OtherLease("member-1")) == 0 {
+		t.Error("other lease  empty")
+	}
+
+	if len(l.StatusObjectPrefix("member-1")) == 0 {
+		t.Error("other lease  empty")
+	}
+
+	if len(l.StatusMemberPrefix()) == 0 {
+		t.Error("StatusMemberPrefix  empty")
+	}
+
+	if len(l.OtherStatusMemberKey("member-1")) == 0 {
+		t.Error("other status member key empty")
+	}
+
+	if len(l.StatusObjectsPrefix()) == 0 {
+		t.Error("StatusObjectsPrefix empty")
+	}
+
+	if len(l.StatusObjectKey("member-1")) == 0 {
+		t.Error("StatusObjectKey key empty")
+	}
+
+	if len(l.ConfigObjectPrefix()) == 0 {
+		t.Error("ConfigObjectPrefix empty")
+	}
+
+	if len(l.ConfigObjectKey("member-1")) == 0 {
+		t.Error("ConfigObjectKey empty")
+	}
+
+	if len(l.ConfigVersion()) == 0 {
+		t.Error("ConfigVersion empty")
+	}
+
+	if len(l.WasmCodeEvent()) == 0 {
+		t.Error("WasmCodeEvent empty")
+	}
+}
diff --git a/pkg/cluster/member_test.go b/pkg/cluster/member_test.go
index 6384ee54..520edeac 100644
--- a/pkg/cluster/member_test.go
+++ b/pkg/cluster/member_test.go
@@ -19,12 +19,14 @@ package cluster
 
 import (
 	"fmt"
-	"io/ioutil"
+	"math/rand"
 	"os"
+	"path"
 	"path/filepath"
 	"reflect"
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/phayes/freeport"
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
@@ -39,21 +41,20 @@ var (
 	tempDir       = os.TempDir()
 )
 
-func TestMain(m *testing.M) {
-	absLogDir, err := ioutil.TempDir(tempDir, "global-log")
-	if err != nil {
-		panic(fmt.Errorf("create tmp dir failed: %v", err))
-	}
+func init() {
+	rand.Seed(time.Now().UnixNano())
+}
 
-	logger.Init(&option.Options{
-		Name:      "member-for-log",
-		AbsLogDir: absLogDir,
-	})
+func getRandomString(n int) string {
+	randBytes := make([]byte, n/2)
+	rand.Read(randBytes)
+	return fmt.Sprintf("%x", randBytes)
+}
 
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	tempDir = path.Join(tempDir, getRandomString(6))
 	code := m.Run()
-
-	logger.Sync()
-	os.RemoveAll(tempDir)
 	os.Exit(code)
 }
 
diff --git a/pkg/common/os_test.go b/pkg/common/os_test.go
new file mode 100644
index 00000000..6de5630d
--- /dev/null
+++ b/pkg/common/os_test.go
@@ -0,0 +1,68 @@
+// +build !windows
+
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package common
+
+import (
+	"bytes"
+	"os"
+	"os/exec"
+	"testing"
+)
+
+func TestNonZeroExit(t *testing.T) {
+	if os.Getenv("BE_TestNonZeroExit") == "1" {
+		Exit(1, "error")
+		return
+	}
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	cmd := exec.Command(os.Args[0], "-test.run=TestNonZeroExit")
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	cmd.Env = append(os.Environ(), "BE_TestNonZeroExit=1")
+	err := cmd.Run()
+	e, ok := err.(*exec.ExitError)
+	if ok && !e.Success() && stderr.String() != "" && stdout.String() == "" {
+		return
+	}
+	t.Fatalf("process ran with err %v, want exit status 1", err)
+}
+
+func TestZeroExit(t *testing.T) {
+	if os.Getenv("BE_TestZeroExit") == "1" {
+		Exit(0, "everythingok")
+		return
+	}
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	cmd := exec.Command(os.Args[0], "-test.run=TestZeroExit")
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	cmd.Env = append(os.Environ(), "BE_TestZeroExit=1")
+	err := cmd.Run()
+	if err == nil && stdout.String() != "" && stderr.String() == "" {
+		return
+	}
+	t.Fatalf("process ran with err %v, want exit status 0", err)
+}
diff --git a/pkg/common/signal_unix_test.go b/pkg/common/signal_unix_test.go
new file mode 100644
index 00000000..295c9b01
--- /dev/null
+++ b/pkg/common/signal_unix_test.go
@@ -0,0 +1,70 @@
+// +build !windows
+
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package common
+
+import (
+	"syscall"
+	"testing"
+)
+
+func TestNotifySignalAndRaiseSignal(t *testing.T) {
+	c := make(chan Signal, 5)
+	var currSig Signal
+	sigs := []Signal{SignalInt, SignalTerm, SingalUsr2}
+
+	// NotifySignal
+	if err := NotifySignal(nil); err == nil {
+		t.Errorf("error for nil chan1")
+	}
+	if err := NotifySignal(nil, SignalInt, SignalTerm); err == nil {
+		t.Errorf("error for nil chan2")
+	}
+	if err := NotifySignal(c); err == nil {
+		t.Errorf("error for empty signal")
+	}
+	if err := NotifySignal(c, SignalInt, "BadSignal", SignalTerm); err == nil {
+		t.Errorf("error for unsupported signal")
+	}
+
+	if err := NotifySignal(c, sigs...); err != nil {
+		t.Fatalf("error with input %+v", sigs)
+	}
+	for _, s := range sigs {
+		syscall.Kill(syscall.Getpid(), signalToOsMap[s])
+		currSig = <-c
+		if currSig != s {
+			t.Fatalf("error signal from chan")
+		}
+	}
+
+	// RaiseSignal
+	if err := RaiseSignal(syscall.Getpid(), "BadSignal"); err == nil {
+		t.Errorf("error with wrong input")
+	}
+	for _, s := range sigs {
+		if err := RaiseSignal(syscall.Getpid(), s); err != nil {
+			t.Fatalf("error raise signal")
+		}
+		currSig = <-c
+		if currSig != s {
+			t.Fatalf("error signal from chan")
+		}
+	}
+}
diff --git a/pkg/common/stat_aggregator_test.go b/pkg/common/stat_aggregator_test.go
index 71f134b4..bac48dee 100644
--- a/pkg/common/stat_aggregator_test.go
+++ b/pkg/common/stat_aggregator_test.go
@@ -70,7 +70,7 @@ func mockWantXGotYKindError(want numericKind, got interface{}) error {
 		reflect.ValueOf(got).Kind())
 }
 
-func TestNumericMaxAggregator(t *testing.T) {
+func TestNumericMaxAggregatorInt(t *testing.T) {
 	em := newErrMapper([]error{
 		mockUnsupportedKindError(""),
 		nil,
@@ -88,7 +88,6 @@ func TestNumericMaxAggregator(t *testing.T) {
 	em.mapNext(t, a.Aggregate(uint(10)))
 	em.mapNext(t, a.Aggregate(int(10)))
 	em.mapNext(t, a.Aggregate(int(-2)))
-
 	em.mapNext(t, a.Aggregate(float64(2.0)))
 
 	wantResult := int64(10)
@@ -96,14 +95,61 @@ func TestNumericMaxAggregator(t *testing.T) {
 	if gotResult != wantResult {
 		t.Fatalf("want %d, got %d", wantResult, gotResult)
 	}
+	name := "numeric_max"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
+
+func TestNumericMaxAggregatorUint(t *testing.T) {
+	em := newErrMapper([]error{
+		mockUnsupportedKindError(""),
+		nil,
+		nil,
+	})
+
+	a := &NumericMaxAggregator{}
+	em.mapNext(t, a.Aggregate(""))
+	em.mapNext(t, a.Aggregate(uint(2)))
+	em.mapNext(t, a.Aggregate(uint(3)))
+
+	wantResult := uint64(3)
+	gotResult := a.Result().(uint64)
+	if gotResult != wantResult {
+		t.Fatalf("want %d, got %d", wantResult, gotResult)
+	}
+	name := "numeric_max"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
+
+func TestNumericMaxAggregatorFloat(t *testing.T) {
+	em := newErrMapper([]error{
+		mockUnsupportedKindError(""),
+		nil,
+		nil,
+		mockWantXGotYKindError(floatNum, uint(0)),
+	})
 
+	a := &NumericMaxAggregator{}
+	em.mapNext(t, a.Aggregate(""))
+	em.mapNext(t, a.Aggregate(float64(100)))
+	em.mapNext(t, a.Aggregate(float64(-2)))
+	em.mapNext(t, a.Aggregate(uint(10)))
+
+	wantResult := float64(100)
+	gotResult := a.Result().(float64)
+	if gotResult != wantResult {
+		t.Fatalf("want %+v, got %+v", wantResult, gotResult)
+	}
 	name := "numeric_max"
 	if a.String() != name {
 		t.Errorf("want %s, got %s", "numeric_max", name)
 	}
 }
 
-func TestNumericMinAggregator(t *testing.T) {
+func TestNumericMinAggregatorFloat(t *testing.T) {
 	em := newErrMapper([]error{
 		nil,
 		mockWantXGotYKindError(floatNum, int(-2)),
@@ -133,7 +179,59 @@ func TestNumericMinAggregator(t *testing.T) {
 	}
 }
 
-func TestNumericSumAggregator(t *testing.T) {
+func TestNumericMinAggregatorInt(t *testing.T) {
+	em := newErrMapper([]error{
+		nil,
+		mockWantXGotYKindError(intNum, uint(3)),
+		mockWantXGotYKindError(intNum, uint(10)),
+		nil,
+	})
+
+	a := &NumericMinAggregator{}
+	em.mapNext(t, a.Aggregate(int(23)))
+	em.mapNext(t, a.Aggregate(uint(3)))
+	em.mapNext(t, a.Aggregate(uint(10)))
+	em.mapNext(t, a.Aggregate(int(-100)))
+
+	wantResult := int64(-100)
+	gotResult := a.Result().(int64)
+	if gotResult != wantResult {
+		t.Fatalf("want %v, got %v", wantResult, gotResult)
+	}
+
+	name := "numeric_min"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
+
+func TestNumericMinAggregatorUint(t *testing.T) {
+	em := newErrMapper([]error{
+		nil,
+		mockWantXGotYKindError(uintNum, int(3)),
+		mockWantXGotYKindError(uintNum, int(10)),
+		nil,
+	})
+
+	a := &NumericMinAggregator{}
+	em.mapNext(t, a.Aggregate(uint(23)))
+	em.mapNext(t, a.Aggregate(int(3)))
+	em.mapNext(t, a.Aggregate(int(10)))
+	em.mapNext(t, a.Aggregate(uint(25)))
+
+	wantResult := uint64(23)
+	gotResult := a.Result().(uint64)
+	if gotResult != wantResult {
+		t.Fatalf("want %v, got %v", wantResult, gotResult)
+	}
+
+	name := "numeric_min"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
+
+func TestNumericSumAggregatorUint(t *testing.T) {
 	em := newErrMapper([]error{
 		mockUnsupportedKindError(map[int]int{0: 1}),
 		nil,
@@ -163,7 +261,61 @@ func TestNumericSumAggregator(t *testing.T) {
 	}
 }
 
-func TestNumericAvgAggregator(t *testing.T) {
+func TestNumericSumAggregatorInt(t *testing.T) {
+	em := newErrMapper([]error{
+		mockUnsupportedKindError(map[int]int{0: 1}),
+		nil,
+		mockWantXGotYKindError(intNum, uint(2)),
+		nil,
+		mockWantXGotYKindError(intNum, float32(10.1)),
+	})
+
+	a := &NumericSumAggregator{}
+	em.mapNext(t, a.Aggregate(map[int]int{0: 1}))
+	em.mapNext(t, a.Aggregate(int(2)))
+	em.mapNext(t, a.Aggregate(uint(2)))
+	em.mapNext(t, a.Aggregate(int(10)))
+	em.mapNext(t, a.Aggregate(float32(10.1)))
+
+	wantResult := int64(12)
+	gotResult := a.Result().(int64)
+	if gotResult != wantResult {
+		t.Fatalf("want %d, got %d", wantResult, gotResult)
+	}
+
+	name := "numeric_sum"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
+
+func TestNumericSumAggregatorFloat(t *testing.T) {
+	em := newErrMapper([]error{
+		mockUnsupportedKindError(map[int]int{0: 1}),
+		nil,
+		mockWantXGotYKindError(floatNum, int(-2)),
+		nil,
+	})
+
+	a := &NumericSumAggregator{}
+	em.mapNext(t, a.Aggregate(map[int]int{0: 1}))
+	em.mapNext(t, a.Aggregate(float64(2)))
+	em.mapNext(t, a.Aggregate(int(-2)))
+	em.mapNext(t, a.Aggregate(float64(10)))
+
+	wantResult := float64(12)
+	gotResult := a.Result().(float64)
+	if gotResult != wantResult {
+		t.Fatalf("want %+v, got %+v", wantResult, gotResult)
+	}
+
+	name := "numeric_sum"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
+
+func TestNumericAvgAggregatorInt(t *testing.T) {
 	em := newErrMapper([]error{
 		mockUnsupportedKindError(make(chan int)),
 		nil,
@@ -196,3 +348,60 @@ func TestNumericAvgAggregator(t *testing.T) {
 		t.Errorf("want %s, got %s", "numeric_max", name)
 	}
 }
+
+func TestNumericAvgAggregatorFloat(t *testing.T) {
+	em := newErrMapper([]error{
+		mockUnsupportedKindError(make(chan int)),
+		nil,
+		nil,
+	})
+
+	a := &NumericAvgAggregator{}
+	em.mapNext(t, a.Aggregate(make(chan int)))
+	em.mapNext(t, a.Aggregate(float64(2)))
+	em.mapNext(t, a.Aggregate(float64(-2)))
+
+	wantResult := float64(0)
+	gotResult := a.Result().(float64)
+	if gotResult != wantResult {
+		t.Fatalf("want %+v, got %+v", wantResult, gotResult)
+	}
+
+	name := "numeric_average"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
+
+func TestNumericAvgAggregatorUint(t *testing.T) {
+	em := newErrMapper([]error{
+		mockUnsupportedKindError(make(chan int)),
+		nil,
+		nil,
+		mockWantXGotYKindError(uintNum, float32(-10.0)),
+		mockWantXGotYKindError(uintNum, int(100)),
+		nil,
+		nil,
+		nil,
+	})
+
+	a := &NumericAvgAggregator{}
+	em.mapNext(t, a.Aggregate(make(chan int)))
+	em.mapNext(t, a.Aggregate(uint(2)))
+	em.mapNext(t, a.Aggregate(uint(4)))
+	em.mapNext(t, a.Aggregate(float32(-10.0)))
+	em.mapNext(t, a.Aggregate(int(100)))
+	em.mapNext(t, a.Aggregate(uint(4)))
+	em.mapNext(t, a.Aggregate(uint(2)))
+
+	wantResult := uint64(3)
+	gotResult := a.Result().(uint64)
+	if gotResult != wantResult {
+		t.Fatalf("want %d, got %d", wantResult, gotResult)
+	}
+
+	name := "numeric_average"
+	if a.String() != name {
+		t.Errorf("want %s, got %s", "numeric_max", name)
+	}
+}
diff --git a/pkg/common/utils_test.go b/pkg/common/utils_test.go
index f4dbf830..bb65a1ae 100644
--- a/pkg/common/utils_test.go
+++ b/pkg/common/utils_test.go
@@ -362,24 +362,6 @@ func TestPanicToErr(t *testing.T) {
 	PanicToErr(f, nil)
 }
 
-func TestRemoveRepeatedByteTest(t *testing.T) {
-	s := "abbcccddd"
-	var b byte = 'b'
-	result := RemoveRepeatedByte(s, b)
-	expected := "abcccddd"
-	if result != expected {
-		t.Errorf("expected %s, result %s", expected, result)
-	}
-
-	s = "a"
-	b = 'a'
-	result = RemoveRepeatedByte(s, b)
-	expected = "a"
-	if result != expected {
-		t.Errorf("expected %s, result %s", expected, result)
-	}
-}
-
 func TestValidateName(t *testing.T) {
 	if err := ValidateName("localhost"); err != nil {
 		t.Errorf("error %v", err)
diff --git a/pkg/context/contexttest/context.go b/pkg/context/contexttest/context.go
new file mode 100644
index 00000000..c11b08b9
--- /dev/null
+++ b/pkg/context/contexttest/context.go
@@ -0,0 +1,233 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package contexttest
+
+import (
+	"time"
+
+	"github.com/megaease/easegress/pkg/context"
+	"github.com/megaease/easegress/pkg/tracing"
+	"github.com/megaease/easegress/pkg/util/httpstat"
+	"github.com/megaease/easegress/pkg/util/texttemplate"
+)
+
+// MockedHTTPContext is the mocked HTTP context
+type MockedHTTPContext struct {
+	MockedLock               func()
+	MockedUnlock             func()
+	MockedSpan               func() tracing.Span
+	MockedRequest            MockedHTTPRequest
+	MockedResponse           MockedHTTPResponse
+	MockedDeadline           func() (time.Time, bool)
+	MockedDone               func() <-chan struct{}
+	MockedErr                func() error
+	MockedValue              func(key interface{}) interface{}
+	MockedCancel             func(err error)
+	MockedCancelled          func() bool
+	MockedClientDisconnected func() bool
+	MockedDuration           func() time.Duration
+	MockedOnFinish           func(func())
+	MockedAddTag             func(tag string)
+	MockedStatMetric         func() *httpstat.Metric
+	MockedLog                func() string
+	MockedFinish             func()
+	MockedTemplate           func() texttemplate.TemplateEngine
+	MockedSetTemplate        func(ht *context.HTTPTemplate)
+	MockedSaveReqToTemplate  func(filterName string) error
+	MockedSaveRspToTemplate  func(filterName string) error
+	MockedCallNextHandler    func(lastResult string) string
+	MockedSetHandlerCaller   func(caller context.HandlerCaller)
+}
+
+// Lock mocks the Lock function of HTTPContext
+func (c *MockedHTTPContext) Lock() {
+	if c.MockedLock != nil {
+		c.MockedLock()
+	}
+}
+
+// Unlock mocks the Unlock function of HTTPContext
+func (c *MockedHTTPContext) Unlock() {
+	if c.MockedUnlock != nil {
+		c.MockedUnlock()
+	}
+}
+
+// Span mocks the Span function of HTTPContext
+func (c *MockedHTTPContext) Span() tracing.Span {
+	if c.MockedSpan != nil {
+		return c.MockedSpan()
+	}
+	return nil
+}
+
+// Request mocks the Request function of HTTPContext
+func (c *MockedHTTPContext) Request() context.HTTPRequest {
+	return &c.MockedRequest
+}
+
+// Response mocks the Response function of HTTPContext
+func (c *MockedHTTPContext) Response() context.HTTPResponse {
+	return &c.MockedResponse
+}
+
+// Deadline mocks the Deadline function of HTTPContext
+func (c *MockedHTTPContext) Deadline() (deadline time.Time, ok bool) {
+	if c.MockedDeadline != nil {
+		return c.MockedDeadline()
+	}
+	return time.Now(), false
+}
+
+// Done mocks the Done function of HTTPContext
+func (c *MockedHTTPContext) Done() <-chan struct{} {
+	if c.MockedDone != nil {
+		return c.MockedDone()
+	}
+	return nil
+}
+
+// Err mocks the Err function of HTTPContext
+func (c *MockedHTTPContext) Err() error {
+	if c.MockedErr != nil {
+		return c.MockedErr()
+	}
+	return nil
+}
+
+// Value mocks the Value function of HTTPContext
+func (c *MockedHTTPContext) Value(key interface{}) interface{} {
+	if c.MockedValue != nil {
+		return c.MockedValue(key)
+	}
+	return nil
+}
+
+// Cancel mocks the Cancel function of HTTPContext
+func (c *MockedHTTPContext) Cancel(err error) {
+	if c.MockedCancel != nil {
+		c.MockedCancel(err)
+	}
+}
+
+// Cancelled mocks the Cancelled function of HTTPContext
+func (c *MockedHTTPContext) Cancelled() bool {
+	if c.MockedCancelled != nil {
+		return c.MockedCancelled()
+	}
+	return false
+}
+
+// ClientDisconnected mocks the ClientDisconnected function of HTTPContext
+func (c *MockedHTTPContext) ClientDisconnected() bool {
+	if c.MockedClientDisconnected != nil {
+		return c.MockedClientDisconnected()
+	}
+	return false
+}
+
+// Duration mocks the Duration function of HTTPContext
+func (c *MockedHTTPContext) Duration() time.Duration {
+	if c.MockedDuration != nil {
+		return c.MockedDuration()
+	}
+	return 0
+}
+
+// OnFinish mocks the OnFinish function of HTTPContext
+func (c *MockedHTTPContext) OnFinish(fn func()) {
+	if c.MockedFinish != nil {
+		c.MockedOnFinish(fn)
+	}
+}
+
+// AddTag mocks the AddTag function of HTTPContext
+func (c *MockedHTTPContext) AddTag(tag string) {
+	if c.MockedAddTag != nil {
+		c.MockedAddTag(tag)
+	}
+}
+
+// StatMetric mocks the StatMetric function of HTTPContext
+func (c *MockedHTTPContext) StatMetric() *httpstat.Metric {
+	if c.MockedStatMetric != nil {
+		return c.MockedStatMetric()
+	}
+	return nil
+}
+
+// Log mocks the Log function of HTTPContext
+func (c *MockedHTTPContext) Log() string {
+	if c.MockedLog != nil {
+		return c.MockedLog()
+	}
+	return ""
+}
+
+// Finish mocks the Finish function of HTTPContext
+func (c *MockedHTTPContext) Finish() {
+	if c.MockedFinish != nil {
+		c.MockedFinish()
+	}
+}
+
+// Template mocks the Template function of HTTPContext
+func (c *MockedHTTPContext) Template() texttemplate.TemplateEngine {
+	if c.MockedTemplate != nil {
+		return c.MockedTemplate()
+	}
+	return nil
+}
+
+// SetTemplate mocks the SetTemplate function of HTTPContext
+func (c *MockedHTTPContext) SetTemplate(ht *context.HTTPTemplate) {
+	if c.MockedSetTemplate != nil {
+		c.MockedSetTemplate(ht)
+	}
+}
+
+// SaveReqToTemplate mocks the SaveReqToTemplate function of HTTPContext
+func (c *MockedHTTPContext) SaveReqToTemplate(filterName string) error {
+	if c.MockedSaveReqToTemplate != nil {
+		return c.MockedSaveReqToTemplate(filterName)
+	}
+	return nil
+}
+
+// SaveRspToTemplate mocks the SaveRspToTemplate function of HTTPContext
+func (c *MockedHTTPContext) SaveRspToTemplate(filterName string) error {
+	if c.MockedSaveRspToTemplate != nil {
+		return c.MockedSaveRspToTemplate(filterName)
+	}
+	return nil
+}
+
+// CallNextHandler mocks the CallNextHandler function of HTTPContext
+func (c *MockedHTTPContext) CallNextHandler(lastResult string) string {
+	if c.MockedCallNextHandler != nil {
+		return c.MockedCallNextHandler(lastResult)
+	}
+	return lastResult
+}
+
+// SetHandlerCaller mocks the SetHandlerCaller function of HTTPContext
+func (c *MockedHTTPContext) SetHandlerCaller(caller context.HandlerCaller) {
+	if c.MockedSetHandlerCaller != nil {
+		c.SetHandlerCaller(caller)
+	}
+}
diff --git a/pkg/context/contexttest/request.go b/pkg/context/contexttest/request.go
new file mode 100644
index 00000000..aac1a914
--- /dev/null
+++ b/pkg/context/contexttest/request.go
@@ -0,0 +1,212 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package contexttest
+
+import (
+	"io"
+	"net/http"
+
+	"github.com/megaease/easegress/pkg/util/httpheader"
+)
+
+// MockedHTTPRequest is the mocked HTTP request
+type MockedHTTPRequest struct {
+	MockedRealIP      func() string
+	MockedMethod      func() string
+	MockedSetMethod   func(method string)
+	MockedScheme      func() string
+	MockedHost        func() string
+	MockedSetHost     func(host string)
+	MockedPath        func() string
+	MockedSetPath     func(path string)
+	MockedEscapedPath func() string
+	MockedQuery       func() string
+	MockedSetQuery    func(query string)
+	MockedFragment    func() string
+	MockedProto       func() string
+	MockedHeader      func() *httpheader.HTTPHeader
+	MockedCookie      func(name string) (*http.Cookie, error)
+	MockedCookies     func() []*http.Cookie
+	MockedAddCookie   func(cookie *http.Cookie)
+	MockedBody        func() io.Reader
+	MockedSetBody     func(io.Reader)
+	MockedStd         func() *http.Request
+	MockedSize        func() uint64
+}
+
+// RealIP mocks the RealIP function of HTTPRequest
+func (r *MockedHTTPRequest) RealIP() string {
+	if r.MockedRealIP != nil {
+		return r.MockedRealIP()
+	}
+	return ""
+}
+
+// Method mocks the Method function of HTTPRequest
+func (r *MockedHTTPRequest) Method() string {
+	if r.MockedMethod != nil {
+		return r.MockedMethod()
+	}
+	return ""
+}
+
+// SetMethod mocks the SetMethod function of HTTPRequest
+func (r *MockedHTTPRequest) SetMethod(method string) {
+	if r.MockedSetMethod != nil {
+		r.MockedSetMethod(method)
+	}
+}
+
+// Scheme mocks the Scheme function of HTTPRequest
+func (r *MockedHTTPRequest) Scheme() string {
+	if r.MockedScheme != nil {
+		return r.MockedScheme()
+	}
+	return ""
+}
+
+// Host mocks the Host function of HTTPRequest
+func (r *MockedHTTPRequest) Host() string {
+	if r.MockedHost != nil {
+		return r.MockedHost()
+	}
+	return ""
+}
+
+// SetHost mocks the SetHost function of HTTPRequest
+func (r *MockedHTTPRequest) SetHost(host string) {
+	if r.MockedSetHost != nil {
+		r.MockedSetHost(host)
+	}
+}
+
+// Path mocks the Path function of HTTPRequest
+func (r *MockedHTTPRequest) Path() string {
+	if r.MockedPath != nil {
+		return r.MockedPath()
+	}
+	return ""
+}
+
+// SetPath mocks the SetPath function of HTTPRequest
+func (r *MockedHTTPRequest) SetPath(path string) {
+	if r.MockedSetPath != nil {
+		r.MockedSetPath(path)
+	}
+}
+
+// EscapedPath mocks the EscapedPath function of HTTPRequest
+func (r *MockedHTTPRequest) EscapedPath() string {
+	if r.MockedEscapedPath != nil {
+		return r.MockedEscapedPath()
+	}
+	return ""
+}
+
+// Query mocks the Query function of HTTPRequest
+func (r *MockedHTTPRequest) Query() string {
+	if r.MockedQuery != nil {
+		return r.MockedQuery()
+	}
+	return ""
+}
+
+// SetQuery mocks the SetQuery function of HTTPRequest
+func (r *MockedHTTPRequest) SetQuery(query string) {
+	if r.MockedSetQuery != nil {
+		r.MockedSetQuery(query)
+	}
+}
+
+// Fragment mocks the Fragment function of HTTPRequest
+func (r *MockedHTTPRequest) Fragment() string {
+	if r.MockedFragment != nil {
+		return r.MockedFragment()
+	}
+	return ""
+}
+
+// Proto mocks the Proto function of HTTPRequest
+func (r *MockedHTTPRequest) Proto() string {
+	if r.MockedProto != nil {
+		return r.MockedProto()
+	}
+	return ""
+}
+
+// Header mocks the Header function of HTTPRequest
+func (r *MockedHTTPRequest) Header() *httpheader.HTTPHeader {
+	if r.MockedHeader != nil {
+		return r.MockedHeader()
+	}
+	return nil
+}
+
+// Cookie mocks the Cookie function of HTTPRequest
+func (r *MockedHTTPRequest) Cookie(name string) (*http.Cookie, error) {
+	if r.MockedCookie != nil {
+		return r.MockedCookie(name)
+	}
+	return nil, nil
+}
+
+// Cookies mocks the Cookies function of HTTPRequest
+func (r *MockedHTTPRequest) Cookies() []*http.Cookie {
+	if r.MockedCookies != nil {
+		return r.MockedCookies()
+	}
+	return nil
+}
+
+// AddCookie mocks the AddCookie function of HTTPRequest
+func (r *MockedHTTPRequest) AddCookie(cookie *http.Cookie) {
+	if r.MockedAddCookie != nil {
+		r.MockedAddCookie(cookie)
+	}
+}
+
+// Body mocks the Body function of HTTPRequest
+func (r *MockedHTTPRequest) Body() io.Reader {
+	if r.MockedBody != nil {
+		return r.MockedBody()
+	}
+	return nil
+}
+
+// SetBody mocks the SetBody function of HTTPRequest
+func (r *MockedHTTPRequest) SetBody(body io.Reader) {
+	if r.MockedSetBody != nil {
+		r.MockedSetBody(body)
+	}
+}
+
+// Std mocks the Std function of HTTPRequest
+func (r *MockedHTTPRequest) Std() *http.Request {
+	if r.MockedStd != nil {
+		return r.MockedStd()
+	}
+	return &http.Request{}
+}
+
+// Size mocks the Size function of HTTPRequest
+func (r *MockedHTTPRequest) Size() uint64 {
+	if r.MockedSize != nil {
+		return r.MockedSize()
+	}
+	return 0
+}
diff --git a/pkg/context/contexttest/response.go b/pkg/context/contexttest/response.go
new file mode 100644
index 00000000..cd2d6a50
--- /dev/null
+++ b/pkg/context/contexttest/response.go
@@ -0,0 +1,107 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package contexttest
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+
+	"github.com/megaease/easegress/pkg/util/httpheader"
+)
+
+// MockedHTTPResponse is the mocked HTTP response
+type MockedHTTPResponse struct {
+	MockedStatusCode    func() int
+	MockedSetStatusCode func(code int)
+	MockedHeader        func() *httpheader.HTTPHeader
+	MockedSetCookie     func(cookie *http.Cookie)
+	MockedSetBody       func(body io.Reader)
+	MockedBody          func() io.Reader
+	MockedOnFlushBody   func(func(body []byte, complete bool) (newBody []byte))
+	MockedStd           func() http.ResponseWriter
+	MockedSize          func() uint64
+}
+
+// StatusCode returns the status code
+func (r *MockedHTTPResponse) StatusCode() int {
+	if r.MockedStatusCode != nil {
+		return r.MockedStatusCode()
+	}
+	return 0
+}
+
+// SetStatusCode set the status code
+func (r *MockedHTTPResponse) SetStatusCode(code int) {
+	if r.MockedSetStatusCode != nil {
+		r.MockedSetStatusCode(code)
+	}
+}
+
+// Header returns the header
+func (r *MockedHTTPResponse) Header() *httpheader.HTTPHeader {
+	if r.MockedHeader != nil {
+		return r.MockedHeader()
+	}
+	return nil
+}
+
+// SetCookie sets a cookie
+func (r *MockedHTTPResponse) SetCookie(cookie *http.Cookie) {
+	if r.MockedSetCookie != nil {
+		r.MockedSetCookie(cookie)
+	}
+}
+
+// SetBody sets the response body
+func (r *MockedHTTPResponse) SetBody(body io.Reader) {
+	if r.MockedSetBody != nil {
+		r.MockedSetBody(body)
+	}
+}
+
+// Body returns the response body
+func (r *MockedHTTPResponse) Body() io.Reader {
+	if r.MockedBody != nil {
+		return r.MockedBody()
+	}
+	return nil
+}
+
+// OnFlushBody registers a callback function on flush body
+func (r *MockedHTTPResponse) OnFlushBody(fn func(body []byte, complete bool) (newBody []byte)) {
+	if r.MockedOnFlushBody != nil {
+		r.MockedOnFlushBody(fn)
+	}
+}
+
+// Std returns the stardard response
+func (r *MockedHTTPResponse) Std() http.ResponseWriter {
+	if r.MockedStd != nil {
+		return r.MockedStd()
+	}
+	return &httptest.ResponseRecorder{}
+}
+
+// Size returns the response size
+func (r *MockedHTTPResponse) Size() uint64 {
+	if r.MockedSize != nil {
+		return r.MockedSize()
+	}
+	return 0
+}
diff --git a/pkg/filter/canary/matcher_test.go b/pkg/filter/canary/matcher_test.go
index 29a59c0e..e04fe080 100644
--- a/pkg/filter/canary/matcher_test.go
+++ b/pkg/filter/canary/matcher_test.go
@@ -20,28 +20,14 @@ package canary
 import (
 	"net/http"
 	"os"
-	"path/filepath"
 	"testing"
 
 	"github.com/megaease/easegress/pkg/logger"
-	"github.com/megaease/easegress/pkg/option"
 )
 
-const tempDir = "/tmp/eg-test"
-
 func TestMain(m *testing.M) {
-	absLogDir := filepath.Join(tempDir, "global-log")
-	os.MkdirAll(absLogDir, 0o755)
-	logger.Init(&option.Options{
-		Name:      "canary-for-log",
-		AbsLogDir: absLogDir,
-	})
-
+	logger.InitNop()
 	code := m.Run()
-
-	logger.Sync()
-	os.RemoveAll(tempDir)
-
 	os.Exit(code)
 }
 
diff --git a/pkg/filter/circuitbreaker/circuitbreaker_test.go b/pkg/filter/circuitbreaker/circuitbreaker_test.go
new file mode 100644
index 00000000..94792ffc
--- /dev/null
+++ b/pkg/filter/circuitbreaker/circuitbreaker_test.go
@@ -0,0 +1,161 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package circuitbreaker
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	libcb "github.com/megaease/easegress/pkg/util/circuitbreaker"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+)
+
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	code := m.Run()
+	os.Exit(code)
+}
+
+func TestCircuitBreaker(t *testing.T) {
+	const yamlSpec = `
+kind: CircuitBreaker
+name: circuitbreaker
+policies:
+- name: default
+  slowCallRateThreshold: 1
+  failureRateThreshold: 50
+  slidingWindowType: COUNT_BASED
+  slidingWindowSize: 10
+  minimumNumberOfCalls: 5
+  failureStatusCodes: [500, 503]
+defaultPolicyRef: default
+urls:
+- methods: []
+  url:
+    exact: /circuitbreak
+    prefix:
+    regex:
+`
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+
+	spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+	if e != nil {
+		t.Errorf("unexpected error: %v", e)
+	}
+
+	cb := &CircuitBreaker{}
+	cb.Init(spec)
+
+	resp := httptest.NewRecorder()
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedMethod = func() string {
+		return http.MethodGet
+	}
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/circuitbreak"
+	}
+	ctx.MockedResponse.MockedStd = func() http.ResponseWriter {
+		return resp
+	}
+	ctx.MockedResponse.MockedStatusCode = func() int {
+		return http.StatusInternalServerError
+	}
+
+	for i := 0; i < 5; i++ {
+		result := cb.Handle(ctx)
+		if result == resultShortCircuited {
+			t.Error("should not be short circuited")
+		}
+	}
+
+	result := cb.Handle(ctx)
+	if result != resultShortCircuited {
+		t.Error("should be short circuited")
+	}
+
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/notcircuitbreak"
+	}
+	result = cb.Handle(ctx)
+	if result == resultShortCircuited {
+		t.Error("should not be short circuited")
+	}
+
+	if cb.Status() != nil {
+		t.Error("behavior changed, please update this case")
+	}
+	cb.Description()
+
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/circuitbreak"
+	}
+	newCb := &CircuitBreaker{}
+	spec, _ = httppipeline.NewFilterSpec(rawSpec, nil)
+	newCb.Inherit(spec, cb)
+	cb.Close()
+	result = newCb.Handle(ctx)
+	if result != resultShortCircuited {
+		t.Error("new circuit breaker should be short circuited")
+	}
+}
+
+func TestBuildPolicy(t *testing.T) {
+	url := &URLRule{
+		policy: &Policy{
+			SlidingWindowType:         "time_based",
+			SlowCallDurationThreshold: "10ms",
+			MaxWaitDurationInHalfOpen: "30s",
+			WaitDurationInOpen:        "60s",
+		},
+	}
+	p := url.buildPolicy()
+	if p.FailureRateThreshold != 50 {
+		t.Error("failure rate threshold is not default")
+	}
+	if p.SlowCallRateThreshold != 100 {
+		t.Error("slow call rate threshold is not default")
+	}
+	if p.SlidingWindowType != libcb.TimeBased {
+		t.Error("sliding window type is incorrect")
+	}
+	if p.SlidingWindowSize != 100 {
+		t.Error("sliding window size is not default")
+	}
+	if p.PermittedNumberOfCallsInHalfOpen != 10 {
+		t.Error("permitted number of calls in half open is not default")
+	}
+	if p.MinimumNumberOfCalls != 100 {
+		t.Error("minimum number of calls is not default")
+	}
+	if p.SlowCallDurationThreshold != 10*time.Millisecond {
+		t.Error("slow call duration threshold is not 10ms")
+	}
+	if p.MaxWaitDurationInHalfOpen != 30*time.Second {
+		t.Error("max wait duration in half open is not 30s")
+	}
+	if p.WaitDurationInOpen != time.Minute {
+		t.Error("wait duration in open is not 1m")
+	}
+}
diff --git a/pkg/filter/corsadaptor/corsadaptor_test.go b/pkg/filter/corsadaptor/corsadaptor_test.go
new file mode 100644
index 00000000..3790363d
--- /dev/null
+++ b/pkg/filter/corsadaptor/corsadaptor_test.go
@@ -0,0 +1,77 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package corsadaptor
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+)
+
+func TestCORSAdaptor(t *testing.T) {
+	const yamlSpec = `
+kind: CORSAdaptor
+name: cors
+`
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+
+	spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+	if e != nil {
+		t.Errorf("unexpected error: %v", e)
+	}
+
+	cors := &CORSAdaptor{}
+	cors.Init(spec)
+
+	header := http.Header{}
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedMethod = func() string {
+		return http.MethodOptions
+	}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	result := cors.Handle(ctx)
+	if result == resultPreflighted {
+		t.Error("request should not be preflighted")
+	}
+
+	header.Add("Access-Control-Request-Method", "abc")
+	result = cors.Handle(ctx)
+	if result != resultPreflighted {
+		t.Error("request should be preflighted")
+	}
+
+	newCors := &CORSAdaptor{}
+	spec, _ = httppipeline.NewFilterSpec(rawSpec, nil)
+	newCors.Inherit(spec, cors)
+	cors.Close()
+	ctx.MockedRequest.MockedMethod = func() string {
+		return http.MethodGet
+	}
+	result = newCors.Handle(ctx)
+	if result == resultPreflighted {
+		t.Error("request should not be preflighted")
+	}
+}
diff --git a/pkg/filter/fallback/fallback_test.go b/pkg/filter/fallback/fallback_test.go
new file mode 100644
index 00000000..1253f187
--- /dev/null
+++ b/pkg/filter/fallback/fallback_test.go
@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package fallback
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+)
+
+func TestFallback(t *testing.T) {
+	const yamlSpec = `
+kind: Fallback
+name: fallback
+mockCode: 203
+mockHeaders:
+  X-Mocked: true
+mockBody: "mocked body"
+`
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+
+	spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+	if e != nil {
+		t.Errorf("unexpected error: %v", e)
+	}
+
+	fb := &Fallback{}
+	fb.Init(spec)
+
+	resp := httptest.NewRecorder()
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedResponse.MockedSetStatusCode = func(code int) {
+		resp.WriteHeader(code)
+	}
+	ctx.MockedResponse.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(resp.Header())
+	}
+	ctx.MockedResponse.MockedSetBody = func(body io.Reader) {
+		data, _ := io.ReadAll(body)
+		resp.Write(data)
+	}
+
+	fb.Handle(ctx)
+	if resp.Code != 203 {
+		t.Error("status code is not correct")
+	}
+	if resp.Body.String() != "mocked body" {
+		t.Error("body is not correct")
+	}
+	if resp.Header().Get("X-Mocked") != "true" {
+		t.Error("header is not correct")
+	}
+
+	newFb := &Fallback{}
+	spec, _ = httppipeline.NewFilterSpec(rawSpec, nil)
+	newFb.Inherit(spec, fb)
+	fb.Close()
+	ctx.MockedRequest.MockedMethod = func() string {
+		return http.MethodGet
+	}
+
+	resp = httptest.NewRecorder()
+	newFb.Handle(ctx)
+	if resp.Code != 203 {
+		t.Error("status code is not correct")
+	}
+	if resp.Body.String() != "mocked body" {
+		t.Error("body is not correct")
+	}
+	if resp.Header().Get("X-Mocked") != "true" {
+		t.Error("header is not correct")
+	}
+}
diff --git a/pkg/filter/mock/mock_test.go b/pkg/filter/mock/mock_test.go
new file mode 100644
index 00000000..e4ed7ed8
--- /dev/null
+++ b/pkg/filter/mock/mock_test.go
@@ -0,0 +1,118 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mock
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+)
+
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	code := m.Run()
+	os.Exit(code)
+}
+
+func TestMock(t *testing.T) {
+	const yamlSpec = `
+kind: Mock
+name: mock
+rules:
+- path: /sales
+  code: 203
+  body: 'mocked body'
+  headers:
+    X-Test: test1
+  delay: 1ms
+- code: 203
+  body: 'mocked body 2'
+  headers:
+    X-Test: test2
+`
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+
+	spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+	if e != nil {
+		t.Errorf("unexpected error: %v", e)
+	}
+
+	m := &Mock{}
+	m.Init(spec)
+
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedMethod = func() string {
+		return http.MethodGet
+	}
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/sales"
+	}
+	resp := httptest.NewRecorder()
+	ctx.MockedResponse.MockedSetStatusCode = func(code int) {
+		resp.WriteHeader(code)
+	}
+	ctx.MockedResponse.MockedSetBody = func(body io.Reader) {
+		data, _ := io.ReadAll(body)
+		resp.Write(data)
+	}
+	ctx.MockedResponse.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(resp.Header())
+	}
+	ctx.MockedCallNextHandler = func(lastResult string) string {
+		return ""
+	}
+
+	m.Handle(ctx)
+	if resp.Code != 203 {
+		t.Error("status code is not 203")
+	}
+
+	if resp.Body.String() != "mocked body" {
+		t.Error("body should be 'mocked body'")
+	}
+
+	if resp.Header().Get("X-Test") != "test1" {
+		t.Error("header 'X-Test' should be 'test1'")
+	}
+
+	if m.Status() != nil {
+		t.Error("behavior changed, please update this case")
+	}
+	m.Description()
+
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/customer"
+	}
+	newM := &Mock{}
+	spec, _ = httppipeline.NewFilterSpec(rawSpec, nil)
+	newM.Inherit(spec, m)
+	m.Close()
+	newM.Handle(ctx)
+	if resp.Code != 203 {
+		t.Error("status code is not 203")
+	}
+}
diff --git a/pkg/filter/proxy/compression_test.go b/pkg/filter/proxy/compression_test.go
new file mode 100644
index 00000000..f4efbbc4
--- /dev/null
+++ b/pkg/filter/proxy/compression_test.go
@@ -0,0 +1,150 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package proxy
+
+import (
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+)
+
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	code := m.Run()
+	os.Exit(code)
+}
+
+func TestAcceptGzip(t *testing.T) {
+	c := newCompression(&CompressionSpec{MinLength: 100})
+
+	header := http.Header{}
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	if !c.acceptGzip(ctx) {
+		t.Error("accept gzip should be true")
+	}
+
+	header.Add(httpheader.KeyAcceptEncoding, "text/text")
+	if c.acceptGzip(ctx) {
+		t.Error("accept gzip should be false")
+	}
+
+	header.Add(httpheader.KeyAcceptEncoding, "*/*")
+	if !c.acceptGzip(ctx) {
+		t.Error("accept gzip should be true")
+	}
+
+	header.Del(httpheader.KeyAcceptEncoding)
+	header.Add(httpheader.KeyAcceptEncoding, "gzip")
+	if !c.acceptGzip(ctx) {
+		t.Error("accept gzip should be true")
+	}
+}
+
+func TestAlreadyGziped(t *testing.T) {
+	c := newCompression(&CompressionSpec{MinLength: 100})
+
+	header := http.Header{}
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedResponse.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	if c.alreadyGziped(ctx) {
+		t.Error("already gziped should be false")
+	}
+
+	header.Add(httpheader.KeyContentEncoding, "text")
+	if c.alreadyGziped(ctx) {
+		t.Error("already gziped should be false")
+	}
+
+	header.Add(httpheader.KeyContentEncoding, "gzip")
+	if !c.alreadyGziped(ctx) {
+		t.Error("already gziped should be true")
+	}
+}
+
+func TestParseContentLength(t *testing.T) {
+	c := newCompression(&CompressionSpec{MinLength: 100})
+
+	header := http.Header{}
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedResponse.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	if c.parseContentLength(ctx) != -1 {
+		t.Error("content length should be -1")
+	}
+
+	header.Set(httpheader.KeyContentLength, "abc")
+	if c.parseContentLength(ctx) != -1 {
+		t.Error("content length should be -1")
+	}
+
+	header.Set(httpheader.KeyContentLength, "100")
+	if c.parseContentLength(ctx) != 100 {
+		t.Error("content length should be 100")
+	}
+}
+
+func TestCompress(t *testing.T) {
+	c := newCompression(&CompressionSpec{MinLength: 100})
+
+	header := http.Header{}
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+	ctx.MockedResponse.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+	header.Set(httpheader.KeyContentLength, "20")
+
+	rawBody := strings.Repeat("this is the raw body. ", 100)
+	sr := strings.NewReader(rawBody)
+	ctx.MockedResponse.MockedBody = func() io.Reader {
+		return sr
+	}
+
+	c.compress(ctx)
+	if header.Get(httpheader.KeyContentEncoding) == "gzip" {
+		t.Error("body should not be gziped")
+	}
+
+	ctx.MockedResponse.MockedSetBody = func(body io.Reader) {
+		io.ReadAll(body)
+	}
+
+	header.Set(httpheader.KeyContentLength, "120")
+	c.compress(ctx)
+	if header.Get(httpheader.KeyContentEncoding) != "gzip" {
+		t.Error("body should be gziped")
+	}
+
+}
diff --git a/pkg/filter/proxy/masterslavereader_test.go b/pkg/filter/proxy/masterslavereader_test.go
new file mode 100644
index 00000000..d8af3061
--- /dev/null
+++ b/pkg/filter/proxy/masterslavereader_test.go
@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package proxy
+
+import (
+	"bytes"
+	"io"
+	"testing"
+	"testing/iotest"
+)
+
+func TestReader(t *testing.T) {
+	var b bytes.Buffer
+	// Write strings to the Buffer.
+	b.WriteString("ABC")
+
+	r := bytes.NewReader(b.Bytes())
+
+	reader1, reader2 := newMasterSlaveReader(r)
+
+	buff := make([]byte, 10)
+	n, _ := reader1.Read(buff)
+
+	if n == 0 {
+		t.Errorf("master read failed")
+	}
+
+	buff1 := make([]byte, 10)
+	n, _ = reader2.Read(buff1)
+	if n == 0 {
+		t.Errorf("slave read failed")
+	}
+
+	if string(buff) != string(buff1) {
+		t.Errorf("buff: %s, buff1: %s not the same ", string(buff), string(buff1))
+	}
+
+	reader1.Close()
+}
+
+func TestEOFReader(t *testing.T) {
+	eofReader := iotest.ErrReader(io.EOF)
+
+	reader1, reader2 := newMasterSlaveReader(eofReader)
+
+	buff := make([]byte, 10)
+	_, err := reader1.Read(buff)
+
+	if err == nil {
+		t.Errorf("master read should failed, but succ")
+	}
+
+	buff1 := make([]byte, 10)
+	_, err = reader2.Read(buff1)
+	if err == nil {
+		t.Errorf("slave read should failed but succ ")
+	}
+
+	if string(buff) != string(buff1) {
+		t.Errorf("buff: %s, buff1: %s not the same ", string(buff), string(buff1))
+	}
+
+	reader1.Close()
+}
diff --git a/pkg/filter/proxy/request_test.go b/pkg/filter/proxy/request_test.go
new file mode 100644
index 00000000..69f4e2a4
--- /dev/null
+++ b/pkg/filter/proxy/request_test.go
@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package proxy
+
+import (
+	"bytes"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+)
+
+func TestRequest(t *testing.T) {
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/abc"
+	}
+	ctx.MockedRequest.MockedQuery = func() string {
+		return "state=1"
+	}
+	ctx.MockedRequest.MockedMethod = func() string {
+		return http.MethodGet
+	}
+	ctx.MockedRequest.MockedHost = func() string {
+		return "megaease.com"
+	}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(http.Header{})
+	}
+
+	server := Server{
+		URL: "http://192.168.1.2",
+	}
+
+	p := pool{}
+	sr := strings.NewReader("this is the raw body")
+	req, _ := p.newRequest(ctx, &server, sr)
+
+	req.start()
+	tm := req.startTime()
+
+	time.Sleep(time.Millisecond)
+	req.start()
+	if tm != req.startTime() {
+		t.Error("start time should not change")
+	}
+
+	req.total()
+
+	time.Sleep(time.Millisecond)
+	req.finish()
+	tm = req.endTime()
+	time.Sleep(time.Millisecond)
+	req.finish()
+
+	if tm != req.endTime() {
+		t.Error("end time should not change")
+	}
+
+	req.total()
+	if req.detail() == "" {
+		t.Error("detail should not be empty")
+	}
+}
+
+func TestResultState(t *testing.T) {
+	rs := &resultState{buff: &bytes.Buffer{}}
+	if n, b := rs.Width(); n != 0 || b {
+		t.Error("implementation changed, this case should be updated")
+	}
+	if n, b := rs.Precision(); n != 0 || b {
+		t.Error("implementation changed, this case should be updated")
+	}
+	if rs.Flag(1) {
+		t.Error("implementation changed, this case should be updated")
+	}
+}
diff --git a/pkg/filter/proxy/server_test.go b/pkg/filter/proxy/server_test.go
index 0a6e5ffe..7699caff 100644
--- a/pkg/filter/proxy/server_test.go
+++ b/pkg/filter/proxy/server_test.go
@@ -18,8 +18,16 @@
 package proxy
 
 import (
+	"fmt"
+	"net/http"
 	"reflect"
 	"testing"
+	"time"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/object/serviceregistry"
+	"github.com/megaease/easegress/pkg/util/hashtool"
+	"github.com/megaease/easegress/pkg/util/httpheader"
 )
 
 func TestPickservers(t *testing.T) {
@@ -201,3 +209,228 @@ func TestPickservers(t *testing.T) {
 		})
 	}
 }
+
+func TestStaticServers(t *testing.T) {
+	servers := []*Server{
+		{
+			URL:    "http://127.0.0.1:9090",
+			Tags:   []string{"d1", "v1", "green"},
+			Weight: 1,
+		},
+		{
+			URL:    "http://127.0.0.1:9091",
+			Tags:   []string{"v1", "d1", "green"},
+			Weight: 2,
+		},
+		{
+			URL:    "http://127.0.0.1:9092",
+			Tags:   []string{"green", "d1", "v1"},
+			Weight: 3,
+		},
+		{
+			URL:    "http://127.0.0.1:9093",
+			Tags:   []string{"v1"},
+			Weight: 4,
+		},
+		{
+			URL:    "http://127.0.0.1:9094",
+			Tags:   []string{"v1", "v3"},
+			Weight: 5,
+		},
+	}
+
+	for _, s := range servers {
+		str := s.String()
+		if str != fmt.Sprintf("%s,%v,%d", s.URL, s.Tags, s.Weight) {
+			t.Error("s.String returns unexpected result")
+		}
+	}
+
+	ss := newStaticServers(servers, []string{}, nil)
+	if ss.len() != len(servers) {
+		t.Errorf("ss.len() is not %d", len(servers))
+	}
+
+	ctx := &contexttest.MockedHTTPContext{}
+	for i := 0; i < len(servers); i++ {
+		if ss.next(ctx) != servers[i] {
+			t.Errorf("ss.next() returns unexpected server")
+		}
+	}
+
+	ss.lb.Policy = "badPolicy"
+	for i := 0; i < len(servers); i++ {
+		if ss.next(ctx) != servers[i] {
+			t.Errorf("ss.next() returns unexpected server")
+		}
+	}
+
+	ss.lb.Policy = PolicyRandom
+	for i := 0; i < len(servers); i++ {
+		ss.next(ctx)
+	}
+
+	ss.lb.Policy = PolicyWeightedRandom
+	for i := 0; i < len(servers); i++ {
+		ss.next(ctx)
+	}
+
+	ip := ""
+	ctx.MockedRequest.MockedRealIP = func() string {
+		return ip
+	}
+	ss.lb.Policy = PolicyIPHash
+	for i := 0; i < len(servers)*5; i++ {
+		ip = fmt.Sprintf("111.222.111.%d", i)
+		sum32 := int(hashtool.Hash32(ip))
+		s := servers[sum32%len(servers)]
+		if ss.next(ctx) != s {
+			t.Errorf("ss.next() returns unexpected server")
+		}
+	}
+
+	header := http.Header{}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+	ss.lb.Policy = PolicyHeaderHash
+	if ss.lb.Validate() == nil {
+		t.Error("LoadBalance.Validate should fail")
+	}
+
+	ss.lb.HeaderHashKey = "X-Megaease"
+	if ss.lb.Validate() != nil {
+		t.Error("LoadBalance.Validate should succeed")
+	}
+	for i := 0; i < len(servers)*5; i++ {
+		v := fmt.Sprintf("value-%d", i)
+		header.Set(ss.lb.HeaderHashKey, v)
+		sum32 := int(hashtool.Hash32(v))
+		s := servers[sum32%len(servers)]
+		if ss.next(ctx) != s {
+			t.Errorf("ss.next() returns unexpected server")
+		}
+	}
+}
+
+func TestServers(t *testing.T) {
+	servers := []*Server{
+		{
+			URL:    "http://127.0.0.1:9090",
+			Tags:   []string{"d1", "v1", "green"},
+			Weight: 1,
+		},
+		{
+			URL:    "http://127.0.0.1:9091",
+			Tags:   []string{"v1", "d1", "green"},
+			Weight: 2,
+		},
+		{
+			URL:    "http://127.0.0.1:9092",
+			Tags:   []string{"green", "d1", "v1"},
+			Weight: 3,
+		},
+		{
+			URL:    "http://127.0.0.1:9093",
+			Tags:   []string{"v1"},
+			Weight: 4,
+		},
+		{
+			URL:    "http://127.0.0.1:9094",
+			Tags:   []string{"v1", "v3"},
+			Weight: 5,
+		},
+	}
+	ps := &PoolSpec{
+		ServersTags:     []string{},
+		ServiceRegistry: "service registry",
+		ServiceName:     "service name",
+		Servers:         []*Server{},
+	}
+
+	ctx := &contexttest.MockedHTTPContext{}
+
+	s := newServers(ps)
+	if s.len() != 0 {
+		t.Errorf("servers.len() should be 0")
+	}
+	s.close()
+
+	ps = &PoolSpec{
+		ServersTags:     []string{},
+		ServiceRegistry: "service registry",
+		ServiceName:     "service name",
+		Servers:         servers,
+	}
+	s = newServers(ps)
+	if s.len() != len(servers) {
+		t.Errorf("servers.len() is not %d", len(servers))
+	}
+	for i := 0; i < len(servers); i++ {
+		if svr, e := s.next(ctx); e != nil || svr != servers[i] {
+			t.Errorf("ss.next() returns unexpected server")
+		}
+	}
+	s.close()
+
+	ps = &PoolSpec{
+		ServersTags:     []string{},
+		ServiceRegistry: "service registry",
+		Servers:         []*Server{},
+		ServiceName:     "testservice",
+	}
+	fnGetService.Store(func(serviceRegistry, serviceName string) (*serviceregistry.Service, error) {
+		return nil, fmt.Errorf("dummy error")
+	})
+	s = newServers(ps)
+	if s.len() != 0 {
+		t.Errorf("servers.len() should be 0")
+	}
+	s.close()
+
+	svcservers := []*serviceregistry.Server{
+		{
+			ServiceName: "testservice",
+			Hostname:    "server1",
+			HostIP:      "192.168.1.1",
+			Port:        80,
+		},
+		{
+			ServiceName: "testservice",
+			Hostname:    "server2",
+			HostIP:      "192.168.1.2",
+			Port:        80,
+		},
+		{
+			ServiceName: "testservice",
+			Hostname:    "server3",
+			HostIP:      "192.168.1.3",
+			Port:        80,
+		},
+	}
+	service, _ := serviceregistry.NewService("testservice", svcservers)
+
+	fnGetService.Store(func(serviceRegistry, serviceName string) (*serviceregistry.Service, error) {
+		return service, nil
+	})
+	s = newServers(ps)
+	if s.len() != len(service.Servers()) {
+		t.Errorf("servers.len() is not %d", len(service.Servers()))
+	}
+
+	svcservers = append(svcservers, &serviceregistry.Server{
+		ServiceName: "testservice",
+		Hostname:    "server4",
+		HostIP:      "192.168.1.4",
+		Port:        80,
+	})
+	service.Update(svcservers)
+
+	time.Sleep(100 * time.Millisecond)
+	if s.len() != len(service.Servers()) {
+		t.Errorf("servers.len() is not %d", len(service.Servers()))
+	}
+
+	service.Close("close")
+	s.close()
+}
diff --git a/pkg/filter/responseadaptor/responseadaptor_test.go b/pkg/filter/responseadaptor/responseadaptor_test.go
new file mode 100644
index 00000000..5a45885f
--- /dev/null
+++ b/pkg/filter/responseadaptor/responseadaptor_test.go
@@ -0,0 +1,129 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package responseadaptor
+
+import (
+	"io"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+	"github.com/megaease/easegress/pkg/util/texttemplate"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+)
+
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	code := m.Run()
+	os.Exit(code)
+}
+
+func TestResponseAdaptor(t *testing.T) {
+	yamlSpec := `
+kind: ResponseAdaptor
+name: ra
+header:
+  del: ["X-Del"]
+  add:
+    "[[headerName]]": "mockedHeaderValue"
+body: "copyright [[name]]"
+`
+
+	ra := doTest(t, yamlSpec, nil)
+
+	yamlSpec = `
+kind: ResponseAdaptor
+name: ra
+header:
+  del: ["X-Del"]
+  add:
+    "[[headerName]]": "mockedHeaderValue"
+body: "copyright megaease"
+`
+	doTest(t, yamlSpec, ra)
+
+	yamlSpec = `
+kind: ResponseAdaptor
+name: ra
+header:
+  del: ["X-Del"]
+  add:
+    "[[headerName]]": "mockedHeaderValue"
+`
+	doTest(t, yamlSpec, nil)
+}
+
+func doTest(t *testing.T, yamlSpec string, prev *ResponseAdaptor) *ResponseAdaptor {
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+
+	spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+	if e != nil {
+		t.Errorf("unexpected error: %v", e)
+	}
+
+	ra := &ResponseAdaptor{}
+	if prev == nil {
+		ra.Init(spec)
+	} else {
+		ra.Inherit(spec, prev)
+	}
+
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedTemplate = func() texttemplate.TemplateEngine {
+		tt, _ := texttemplate.NewDefault([]string{"name", "headerName"})
+		tt.SetDict("name", "megaease")
+		tt.SetDict("headerName", "mockedHeader")
+		return tt
+	}
+	resp := httptest.NewRecorder()
+	resp.Header().Add("X-Del", "deleted")
+
+	ctx.MockedResponse.MockedSetBody = func(body io.Reader) {
+		data, _ := io.ReadAll(body)
+		resp.Write(data)
+	}
+	ctx.MockedResponse.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(resp.Header())
+	}
+	ctx.MockedCallNextHandler = func(lastResult string) string {
+		return ""
+	}
+
+	ra.Handle(ctx)
+
+	if v := resp.Header().Get("mockedHeader"); v != "mockedHeaderValue" {
+		t.Error("unexpected header name or value: ", v)
+	}
+
+	if v := resp.Header().Get("X-Del"); v != "" {
+		t.Error("header 'X-Del' should not exist")
+	}
+
+	if v := resp.Body.String(); ra.spec.Body != "" && v != "copyright megaease" {
+		t.Error("unexpected body:", v)
+	}
+
+	ra.Status()
+	ra.Description()
+	return ra
+}
diff --git a/pkg/filter/timelimiter/timelimiter_test.go b/pkg/filter/timelimiter/timelimiter_test.go
new file mode 100644
index 00000000..91cd4200
--- /dev/null
+++ b/pkg/filter/timelimiter/timelimiter_test.go
@@ -0,0 +1,125 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package timelimiter
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+)
+
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	code := m.Run()
+	os.Exit(code)
+}
+
+func TestTimeLimiter(t *testing.T) {
+	const yamlSpec = `
+kind: TimeLimiter
+name: timelimiter
+defaultTimeoutDuration: 456ms
+urls:
+  - timeoutDuration: 10ms
+    methods: [GET]
+    url:
+      exact: /timelimit
+  - methods: [POST]
+    url:
+      prefix: /customer
+`
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+
+	spec, e := httppipeline.NewFilterSpec(rawSpec, nil)
+	if e != nil {
+		t.Errorf("unexpected error: %v", e)
+	}
+
+	tl := &TimeLimiter{}
+	tl.Init(spec)
+
+	if tl.spec.defaultTimeout != 456*time.Millisecond {
+		t.Error("default timeout duration is not the value in spec")
+	}
+
+	if len(tl.spec.URLs) != 2 {
+		t.Error("the length of 'urls' should be 2")
+	} else if tl.spec.URLs[0].timeout != 10*time.Millisecond {
+		t.Error("timeout duration is not the value in spec")
+	}
+
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedMethod = func() string {
+		return http.MethodGet
+	}
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/timelimit"
+	}
+	ctx.MockedResponse.MockedStd = func() http.ResponseWriter {
+		return &httptest.ResponseRecorder{}
+	}
+	ctx.MockedCallNextHandler = func(lastResult string) string {
+		return ""
+	}
+
+	result := tl.Handle(ctx)
+	if result == resultTimeout {
+		t.Error("timeout should not happen")
+	}
+
+	ctx.MockedCallNextHandler = func(lastResult string) string {
+		time.Sleep(20 * time.Millisecond)
+		return ""
+	}
+
+	result = tl.Handle(ctx)
+	if result != resultTimeout {
+		t.Error("expected timeout didn't happen")
+	}
+
+	ctx.MockedRequest.MockedPath = func() string {
+		return "/notimelimit"
+	}
+
+	result = tl.Handle(ctx)
+	if result == resultTimeout {
+		t.Error("request path doesn't match, timeout should not happen")
+	}
+
+	if tl.Status() != nil {
+		t.Error("behavior changed, please update this case")
+	}
+	tl.Description()
+
+	newTl := &TimeLimiter{}
+	spec, _ = httppipeline.NewFilterSpec(rawSpec, nil)
+	newTl.Inherit(spec, tl)
+	tl.Close()
+	result = newTl.Handle(ctx)
+	if result == resultTimeout {
+		t.Error("request path doesn't match, timeout should not happen")
+	}
+}
diff --git a/pkg/filter/validator/validator_test.go b/pkg/filter/validator/validator_test.go
new file mode 100644
index 00000000..054abb98
--- /dev/null
+++ b/pkg/filter/validator/validator_test.go
@@ -0,0 +1,291 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package validator
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/megaease/easegress/pkg/context/contexttest"
+	"github.com/megaease/easegress/pkg/logger"
+	"github.com/megaease/easegress/pkg/object/httppipeline"
+	"github.com/megaease/easegress/pkg/util/httpheader"
+	"github.com/megaease/easegress/pkg/util/yamltool"
+)
+
+func TestMain(m *testing.M) {
+	logger.InitNop()
+	code := m.Run()
+	os.Exit(code)
+}
+
+func createValidator(yamlSpec string, prev *Validator) *Validator {
+	rawSpec := make(map[string]interface{})
+	yamltool.Unmarshal([]byte(yamlSpec), &rawSpec)
+	spec, _ := httppipeline.NewFilterSpec(rawSpec, nil)
+	v := &Validator{}
+	if prev == nil {
+		v.Init(spec)
+	} else {
+		v.Inherit(spec, prev)
+	}
+	return v
+}
+
+func TestHeaders(t *testing.T) {
+	const yamlSpec = `
+kind: Validator
+name: validator
+headers:
+  Is-Valid:
+    values: ["abc", "goodplan"]
+    regexp: "^ok-.+$"
+`
+
+	v := createValidator(yamlSpec, nil)
+
+	header := http.Header{}
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	result := v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("request has no header 'Is-Valid', should be invalid")
+	}
+
+	header.Add("Is-Valid", "Invalid")
+	result = v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("request has header 'Is-Valid', but value is incorrect, should be invalid")
+	}
+
+	header.Set("Is-Valid", "goodplan")
+	result = v.Handle(ctx)
+	if result == resultInvalid {
+		t.Errorf("request has header 'Is-Valid' and value is correct, should be valid")
+	}
+
+	header.Set("Is-Valid", "ok-1")
+	result = v.Handle(ctx)
+	if result == resultInvalid {
+		t.Errorf("request has header 'Is-Valid' and matches the regular expression, should be valid")
+	}
+}
+
+func TestJWT(t *testing.T) {
+	const yamlSpec = `
+kind: Validator
+name: validator
+jwt:
+  cookieName: auth
+  algorithm: HS256
+  secret: 313233343536
+`
+	v := createValidator(yamlSpec, nil)
+
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedCookie = func(name string) (*http.Cookie, error) {
+		return nil, fmt.Errorf("not exist")
+	}
+	header := http.Header{}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	token := "eyJhbGciOiJIUzM4NCIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.3Ywq9NlR3cBST4nfcdbR-fcZ8374RHzU50X6flKvG-tnWFMalMaHRm3cMpXs1NrZ"
+	header.Set("Authorization", "Bearer "+token)
+	result := v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("the jwt token in header should be invalid")
+	}
+
+	token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.keH6T3x1z7mmhKL1T3r9sQdAxxdzB6siemGMr_6ZOwU"
+	header.Set("Authorization", "Bearer "+token)
+	result = v.Handle(ctx)
+	if result == resultInvalid {
+		t.Errorf("the jwt token in header should be valid")
+	}
+
+	header.Set("Authorization", "not Bearer "+token)
+	result = v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("the jwt token in header should be invalid")
+	}
+
+	header.Set("Authorization", "Bearer "+token+"abc")
+	result = v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("the jwt token in header should be invalid")
+	}
+
+	header.Del("Authorization")
+	ctx.MockedRequest.MockedCookie = func(name string) (*http.Cookie, error) {
+		return &http.Cookie{Value: token}, nil
+	}
+	result = v.Handle(ctx)
+	if result == resultInvalid {
+		t.Errorf("the jwt token in cookie should be valid")
+	}
+
+	v = createValidator(yamlSpec, v)
+	result = v.Handle(ctx)
+	if result == resultInvalid {
+		t.Errorf("the jwt token in cookie should be valid")
+	}
+
+	if v.Status() != nil {
+		t.Error("behavior changed, please update this case")
+	}
+	v.Description()
+}
+
+func TestOAuth2JWT(t *testing.T) {
+	const yamlSpec = `
+kind: Validator
+name: validator
+oauth2:
+  jwt:
+    algorithm: HS256
+    secret: 313233343536
+`
+	v := createValidator(yamlSpec, nil)
+
+	ctx := &contexttest.MockedHTTPContext{}
+
+	header := http.Header{}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	token := "eyJhbGciOiJIUzM4NCIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.3Ywq9NlR3cBST4nfcdbR-fcZ8374RHzU50X6flKvG-tnWFMalMaHRm3cMpXs1NrZ"
+	header.Set("Authorization", "Bearer "+token)
+	result := v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("OAuth/2 Authorization should fail")
+	}
+
+	token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyLCJzY29wZSI6Im1lZ2FlYXNlIn0.HRcRwN6zLJnubaUnZhZ5jC-j-rRiT-5mY8emJW6h6so"
+	header.Set("Authorization", "Bearer "+token)
+	result = v.Handle(ctx)
+	if result == resultInvalid {
+		t.Errorf("OAuth/2 Authorization should succeed")
+	}
+
+	header.Set("Authorization", "not Bearer "+token)
+	result = v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("OAuth/2 Authorization should fail")
+	}
+
+	header.Set("Authorization", "Bearer "+token+"abc")
+	result = v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("OAuth/2 Authorization should fail")
+	}
+}
+
+func TestOAuth2TokenIntrospect(t *testing.T) {
+	yamlSpec := `
+kind: Validator
+name: validator
+oauth2:
+  tokenIntrospect:
+    endPoint: http://oauth2.megaease.com/
+    insecureTls: true
+    clientId: megaease
+    clientSecret: secret
+`
+	v := createValidator(yamlSpec, nil)
+	ctx := &contexttest.MockedHTTPContext{}
+
+	header := http.Header{}
+	ctx.MockedRequest.MockedHeader = func() *httpheader.HTTPHeader {
+		return httpheader.New(header)
+	}
+
+	token := "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyLCJzY29wZSI6Im1lZ2FlYXNlIn0.HRcRwN6zLJnubaUnZhZ5jC-j-rRiT-5mY8emJW6h6so"
+	header.Set("Authorization", "Bearer "+token)
+
+	body := `{
+			"subject":"megaease.com",
+			"scope":"read,write",
+			"active": false
+		}`
+	fnSendRequest = func(client *http.Client, r *http.Request) (*http.Response, error) {
+		reader := strings.NewReader(body)
+		return &http.Response{
+			Body: io.NopCloser(reader),
+		}, nil
+	}
+	result := v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("OAuth/2 Authorization should fail")
+	}
+
+	yamlSpec = `
+kind: Validator
+name: validator
+oauth2:
+  tokenIntrospect:
+    endPoint: http://oauth2.megaease.com/
+    clientId: megaease
+    clientSecret: secret
+    basicAuth: megaease@megaease
+`
+	v = createValidator(yamlSpec, nil)
+
+	body = `{
+			"subject":"megaease.com",
+			"scope":"read,write",
+			"active": true
+		}`
+	result = v.Handle(ctx)
+	if result == resultInvalid {
+		t.Errorf("OAuth/2 Authorization should succeed")
+	}
+}
+
+func TestSignature(t *testing.T) {
+	// This test is almost covered by signer
+
+	const yamlSpec = `
+kind: Validator
+name: validator
+signature:
+  accessKeys:
+    AKID: SECRET
+`
+	v := createValidator(yamlSpec, nil)
+
+	ctx := &contexttest.MockedHTTPContext{}
+	ctx.MockedRequest.MockedStd = func() *http.Request {
+		r, _ := http.NewRequest(http.MethodGet, "http://megaease.com", nil)
+		return r
+	}
+
+	result := v.Handle(ctx)
+	if result != resultInvalid {
+		t.Errorf("OAuth/2 Authorization should fail")
+	}
+}
diff --git a/pkg/object/function/spec/function_test.go b/pkg/object/function/spec/function_test.go
index 0934ba72..979ad276 100644
--- a/pkg/object/function/spec/function_test.go
+++ b/pkg/object/function/spec/function_test.go
@@ -45,6 +45,24 @@ func TestInValidSpec(t *testing.T) {
 }
 
 func TestInvalidKnative(t *testing.T) {
+	spec := Spec{
+		Name:           "demo",
+		Image:          "helloworld:1.0",
+		Port:           10000,
+		AutoScaleType:  AutoScaleMetricCPU,
+		AutoScaleValue: "0.1",
+		MinReplica:     1,
+		MaxReplica:     2,
+		LimitCPU:       "aaa",
+		LimitMemory:    "80Mi",
+		RequestCPU:     "100m",
+		RequestMemory:  "60Mi",
+	}
+
+	var err error
+	if err = spec.Validate(); err == nil {
+		t.Errorf("test failed spec should not be valid, err: %v", err)
+	}
 }
 
 func TestValidSpec(t *testing.T) {
@@ -67,3 +85,102 @@ func TestValidSpec(t *testing.T) {
 		t.Errorf("test failed spec should not be valid, err: %v", err)
 	}
 }
+
+func TestInValidSpecReplica(t *testing.T) {
+	spec := Spec{
+		Name:           "demo",
+		Image:          "helloworld:1.0",
+		Port:           10000,
+		AutoScaleType:  AutoScaleMetricCPU,
+		AutoScaleValue: "0.1",
+		MinReplica:     2,
+		MaxReplica:     1,
+		LimitCPU:       "180m",
+		LimitMemory:    "80Mi",
+		RequestCPU:     "100m",
+		RequestMemory:  "60Mi",
+	}
+
+	var err error
+	if err = spec.Validate(); err == nil {
+		t.Errorf("test failed spec should not be valid, err: %v", err)
+	}
+}
+
+func TestInValidSpecAutoScaleType(t *testing.T) {
+	spec := Spec{
+		Name:           "demo",
+		Image:          "helloworld:1.0",
+		Port:           10000,
+		AutoScaleType:  "nothing",
+		AutoScaleValue: "0.1",
+		MinReplica:     2,
+		MaxReplica:     1,
+		LimitCPU:       "180m",
+		LimitMemory:    "80Mi",
+		RequestCPU:     "100m",
+		RequestMemory:  "60Mi",
+	}
+
+	var err error
+	if err = spec.Validate(); err == nil {
+		t.Errorf("test failed spec should not be invalid, err: %v", err)
+	}
+	fmt.Println("err is ", err)
+}
+
+func TestNext(t *testing.T) {
+	fsm, _ := InitFSM(InitState())
+
+	f := &Function{
+		Spec: &Spec{
+			Name:           "demo",
+			Image:          "helloworld:1.0",
+			Port:           10000,
+			AutoScaleValue: "0.1",
+			MinReplica:     1,
+			MaxReplica:     2,
+			LimitCPU:       "180m",
+			LimitMemory:    "80Mi",
+			RequestCPU:     "100m",
+			RequestMemory:  "60Mi",
+		},
+		Fsm:    fsm,
+		Status: &Status{},
+	}
+
+	updated, err := f.Next(ReadyEvent)
+
+	if err != nil {
+		t.Errorf("test failed Next should succ, err: %v", err)
+	}
+
+	t.Logf("updated: %v, new status: %s\n", updated, f.Fsm.Current())
+}
+
+func TestInvalidNext(t *testing.T) {
+	fsm, _ := InitFSM(InitState())
+
+	f := &Function{
+		Spec: &Spec{
+			Name:           "demo",
+			Image:          "helloworld:1.0",
+			Port:           10000,
+			AutoScaleValue: "0.1",
+			MinReplica:     1,
+			MaxReplica:     2,
+			LimitCPU:       "180m",
+			LimitMemory:    "80Mi",
+			RequestCPU:     "100m",
+			RequestMemory:  "60Mi",
+		},
+		Fsm:    fsm,
+		Status: &Status{},
+	}
+
+	_, err := f.Next(StartEvent)
+
+	if err == nil {
+		t.Errorf("test failed Next should failed, start event will be rejected in initial state")
+	}
+}
diff --git a/pkg/object/meshcontroller/spec/spec_test.go b/pkg/object/meshcontroller/spec/spec_test.go
index fbfd8086..028577db 100644
--- a/pkg/object/meshcontroller/spec/spec_test.go
+++ b/pkg/object/meshcontroller/spec/spec_test.go
@@ -20,7 +20,6 @@ package spec
 import (
 	"fmt"
 	"os"
-	"path/filepath"
 	"testing"
 
 	"github.com/megaease/easegress/pkg/filter/circuitbreaker"
@@ -29,25 +28,13 @@ import (
 	"github.com/megaease/easegress/pkg/filter/retryer"
 	"github.com/megaease/easegress/pkg/filter/timelimiter"
 	"github.com/megaease/easegress/pkg/logger"
-	"github.com/megaease/easegress/pkg/option"
+	_ "github.com/megaease/easegress/pkg/object/httpserver"
 	"github.com/megaease/easegress/pkg/util/urlrule"
 )
 
-const tempDir = "/tmp/eg-test"
-
 func TestMain(m *testing.M) {
-	absLogDir := filepath.Join(tempDir, "global-log")
-	os.MkdirAll(absLogDir, 0o755)
-	logger.Init(&option.Options{
-		Name:      "meshspec-for-log",
-		AbsLogDir: absLogDir,
-	})
-
+	logger.InitNop()
 	code := m.Run()
-
-	logger.Sync()
-	os.RemoveAll(tempDir)
-
 	os.Exit(code)
 }
 
@@ -73,6 +60,32 @@ func TestSideCarIngressPipelineSpec(t *testing.T) {
 	fmt.Println(superSpec.YAMLConfig())
 }
 
+func TestAdminInValidat(t *testing.T) {
+	a := Admin{
+		RegistryType:      "unknow",
+		HeartbeatInterval: "10s",
+	}
+
+	err := a.Validate()
+
+	if err == nil {
+		t.Errorf("registry type is invalid, should failed")
+	}
+}
+
+func TestAdminValidat(t *testing.T) {
+	a := Admin{
+		RegistryType:      "eureka",
+		HeartbeatInterval: "10s",
+	}
+
+	err := a.Validate()
+
+	if err != nil {
+		t.Errorf("registry type is valid, err: %v", err)
+	}
+}
+
 func TestSideCarEgressPipelineSpec(t *testing.T) {
 	s := &Service{
 		Name: "order-001",
@@ -184,6 +197,99 @@ func TestSideCarEgressPipelineWithCanarySpec(t *testing.T) {
 	fmt.Println(superSpec.YAMLConfig())
 }
 
+func TestSideCarEgressPipelneNotLoadBalancer(t *testing.T) {
+	s := &Service{
+		Name: "order-003-canary-array",
+		Sidecar: &Sidecar{
+			Address:         "127.0.0.1",
+			IngressPort:     8080,
+			IngressProtocol: "http",
+			EgressPort:      9090,
+			EgressProtocol:  "http",
+		},
+
+		Canary: &Canary{
+			CanaryRules: []*CanaryRule{
+				{
+					Headers: map[string]*urlrule.StringMatch{
+						"X-canary": {
+							Exact: "lv1",
+						},
+					},
+					URLs: []*urlrule.URLRule{
+						{
+							Methods: []string{
+								"GET",
+								"POST",
+							},
+							URL: urlrule.StringMatch{
+								Prefix: "/",
+							},
+						},
+					},
+					ServiceInstanceLabels: map[string]string{
+						"version": "v1",
+					},
+				},
+				{
+					Headers: map[string]*urlrule.StringMatch{
+						"X-canary": {
+							Exact: "ams",
+						},
+					},
+					URLs: []*urlrule.URLRule{
+						{
+							Methods: []string{
+								"GET",
+								"POST",
+							},
+							URL: urlrule.StringMatch{
+								Prefix: "/",
+							},
+						},
+					},
+					ServiceInstanceLabels: map[string]string{
+						"version": "v2",
+					},
+				},
+			},
+		},
+	}
+
+	instanceSpecs := []*ServiceInstanceSpec{
+		{
+			ServiceName: "fake-001",
+			InstanceID:  "xxx-89757",
+			IP:          "192.168.0.110",
+			Port:        80,
+			Status:      "UP",
+		},
+		{
+			ServiceName: "fake-002-canary",
+			InstanceID:  "zzz-73597",
+			IP:          "192.168.0.120",
+			Port:        80,
+			Status:      "UP",
+			Labels: map[string]string{
+				"version": "v1",
+			},
+		},
+		{
+			ServiceName: "fake-003-canary-no-match",
+			InstanceID:  "yyy-73587",
+			IP:          "192.168.0.121",
+			Port:        80,
+			Status:      "UP",
+			Labels: map[string]string{
+				"version": "v2",
+			},
+		},
+	}
+
+	superSpec, _ := s.SideCarEgressPipelineSpec(instanceSpecs)
+	fmt.Println(superSpec.YAMLConfig())
+}
+
 func TestSideCarEgressPipelineWithMultipleCanarySpec(t *testing.T) {
 	s := &Service{
 		Name: "order-003-canary-array",
@@ -429,6 +535,26 @@ func TestSideCarEgressPipelineWithCanaryInstanceMultipleLabelSpec(t *testing.T)
 	fmt.Println(superSpec.YAMLConfig())
 }
 
+func TestIngressHTTPServerSpec(t *testing.T) {
+	rule := []*IngressRule{
+		{
+			Host: "megaease.com",
+			Paths: []*IngressPath{
+				{
+					Path:    "/",
+					Backend: "portal",
+				},
+			},
+		},
+	}
+
+	_, err := IngressHTTPServerSpec(1233, rule)
+
+	if err != nil {
+		t.Errorf("ingress http server spec failed: %v", err)
+	}
+
+}
 func TestSideCarIngressWithResiliencePipelineSpec(t *testing.T) {
 	s := &Service{
 		Name: "order-001",
@@ -575,3 +701,234 @@ func TestSideCarEgressResiliencePipelineSpec(t *testing.T) {
 	superSpec, _ := s.SideCarEgressPipelineSpec(instanceSpecs)
 	fmt.Println(superSpec.YAMLConfig())
 }
+
+func TestPipelineBuilderFailed(t *testing.T) {
+	builder := newPipelineSpecBuilder("abc")
+
+	builder.appendRateLimiter(nil)
+
+	builder.appendCircuitBreaker(nil)
+
+	builder.appendRetryer(nil)
+
+	builder.appendTimeLimiter(nil)
+
+	yamlStr := builder.yamlConfig()
+	if len(yamlStr) == 0 {
+		t.Errorf("builder append nil resilience filter failed")
+	}
+}
+
+func TestPipelineBuilder(t *testing.T) {
+	builder := newPipelineSpecBuilder("abc")
+
+	rateLimiter := &ratelimiter.Spec{
+		Policies: []*ratelimiter.Policy{{
+			Name:               "default",
+			TimeoutDuration:    "100ms",
+			LimitForPeriod:     50,
+			LimitRefreshPeriod: "10ms",
+		}},
+		DefaultPolicyRef: "default",
+		URLs: []*ratelimiter.URLRule{{
+			URLRule: urlrule.URLRule{
+				Methods: []string{"GET"},
+				URL: urlrule.StringMatch{
+					Exact:  "/path1",
+					Prefix: "/path2/",
+					RegEx:  "^/path3/[0-9]+$",
+				},
+				PolicyRef: "default",
+			},
+		}},
+	}
+
+	builder.appendRateLimiter(rateLimiter)
+	yaml := builder.yamlConfig()
+
+	if len(yaml) == 0 {
+		t.Errorf("pipeline builder yamlconfig failed")
+	}
+
+}
+
+func TestIngressPipelineSpec(t *testing.T) {
+	s := &Service{
+		Name: "order-001",
+		LoadBalance: &LoadBalance{
+			Policy: proxy.PolicyRandom,
+		},
+		Sidecar: &Sidecar{
+			Address:         "127.0.0.1",
+			IngressPort:     8080,
+			IngressProtocol: "http",
+			EgressPort:      9090,
+			EgressProtocol:  "http",
+		},
+	}
+	instanceSpecs := []*ServiceInstanceSpec{
+		{
+			ServiceName: "fake-001",
+			InstanceID:  "xxx-89757",
+			IP:          "192.168.0.110",
+			Port:        80,
+			Status:      "UP",
+		},
+		{
+			ServiceName: "fake-002",
+			InstanceID:  "zzz-73597",
+			IP:          "192.168.0.120",
+			Port:        80,
+			Status:      "UP",
+		},
+	}
+	superSpec, err := s.IngressPipelineSpec(instanceSpecs)
+
+	if err != nil {
+		t.Fatalf("%v", err)
+	}
+	fmt.Println(superSpec.YAMLConfig())
+}
+
+func TestSidecarIngressPipelineSpec(t *testing.T) {
+	s := &Service{
+		Name: "order-001",
+		LoadBalance: &LoadBalance{
+			Policy: proxy.PolicyRandom,
+		},
+		Sidecar: &Sidecar{
+			Address:         "127.0.0.1",
+			IngressPort:     8080,
+			IngressProtocol: "http",
+			EgressPort:      9090,
+			EgressProtocol:  "http",
+		},
+	}
+
+	superSpec, err := s.SideCarIngressHTTPServerSpec()
+
+	if err != nil {
+		t.Fatalf("ingress http server spec failed: %v", err)
+	}
+	fmt.Println(superSpec.YAMLConfig())
+
+	superSpec, err = s.SideCarEgressHTTPServerSpec()
+
+	if err != nil {
+		t.Fatalf("egress http server spec failed: %v", err)
+	}
+
+	fmt.Println(superSpec.YAMLConfig())
+}
+
+func TestUniqueCanaryHeadersEmpty(t *testing.T) {
+	s := &Service{
+		Name: "order-001",
+		LoadBalance: &LoadBalance{
+			Policy: proxy.PolicyRandom,
+		},
+		Sidecar: &Sidecar{
+			Address:         "127.0.0.1",
+			IngressPort:     8080,
+			IngressProtocol: "http",
+			EgressPort:      9090,
+			EgressProtocol:  "http",
+		},
+	}
+
+	val := s.UniqueCanaryHeaders()
+	if len(val) > 0 {
+		t.Errorf("canary header should be none")
+	}
+
+}
+
+func TestUniqueCanaryHeaders(t *testing.T) {
+	s := &Service{
+		Name: "order-002-canary",
+		LoadBalance: &LoadBalance{
+			Policy: proxy.PolicyIPHash,
+		},
+		Sidecar: &Sidecar{
+			Address:         "127.0.0.1",
+			IngressPort:     8080,
+			IngressProtocol: "http",
+			EgressPort:      9090,
+			EgressProtocol:  "http",
+		},
+
+		Canary: &Canary{
+			CanaryRules: []*CanaryRule{
+				{
+					Headers: map[string]*urlrule.StringMatch{
+						"X-canary": {
+							Exact: "lv1",
+						},
+					},
+					URLs: []*urlrule.URLRule{
+						{
+							Methods: []string{
+								"GET",
+								"POST",
+							},
+							URL: urlrule.StringMatch{
+								Prefix: "/",
+							},
+						},
+					},
+					ServiceInstanceLabels: map[string]string{
+						"version": "v1",
+					},
+				},
+			},
+		},
+	}
+
+	val := s.UniqueCanaryHeaders()
+	if len(val) == 0 {
+		t.Errorf("canary header should not be none")
+	}
+}
+
+func TestEgressName(t *testing.T) {
+	s := &Service{
+		Name: "order-001",
+		LoadBalance: &LoadBalance{
+			Policy: proxy.PolicyRandom,
+		},
+		Sidecar: &Sidecar{
+			Address:         "127.0.0.1",
+			IngressPort:     8080,
+			IngressProtocol: "http",
+			EgressPort:      9090,
+			EgressProtocol:  "http",
+		},
+	}
+	if len(s.EgressHTTPServerName()) == 0 {
+		t.Error("egress httpserver name should not be none")
+	}
+
+	if len(s.EgressHandlerName()) == 0 {
+		t.Error("egress httpserver handler should not be none")
+	}
+
+	if len(s.IngressHTTPServerName()) == 0 {
+		t.Error("ingress httpserver handler should not be none")
+	}
+
+	if len(s.IngressHandlerName()) == 0 {
+		t.Error("ingress httpserver handler should not be none")
+	}
+
+	if len(s.BackendName()) == 0 {
+		t.Error("backend name should not be none")
+	}
+
+	if len(s.IngressEndpoint()) == 0 {
+		t.Error("ingress endpoint name should not be none")
+	}
+
+	if len(s.EgressEndpoint()) == 0 {
+		t.Error("egress endpoint name should not be none")
+	}
+}
diff --git a/pkg/object/statussynccontroller/timer_test.go b/pkg/object/statussynccontroller/timer_test.go
deleted file mode 100644
index 45e4e850..00000000
--- a/pkg/object/statussynccontroller/timer_test.go
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Copyright (c) 2017, MegaEase
- * All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package statussynccontroller
-
-import (
-	"reflect"
-	"testing"
-	"time"
-)
-
-func TestNextDuration(t *testing.T) {
-	type parameters struct {
-		t             time.Time
-		paceInSeconds int
-	}
-
-	tests := []struct {
-		name  string
-		want  time.Duration
-		input parameters
-	}{
-		{
-			name: "test-1",
-			want: 5*time.Second - 555*time.Nanosecond,
-			input: parameters{
-				t:             time.Unix(0, 555),
-				paceInSeconds: 5,
-			},
-		},
-		{
-			name: "test-2",
-			want: 778 * time.Millisecond,
-			input: parameters{
-				t:             time.Unix(199994, 222000000),
-				paceInSeconds: 5,
-			},
-		},
-		{
-			name: "test-3",
-			want: 5*time.Second - 333*time.Nanosecond,
-			input: parameters{
-				t:             time.Unix(38383835, 333),
-				paceInSeconds: 5,
-			},
-		},
-		{
-			name: "test-4",
-			want: 9*time.Second - 888*time.Nanosecond,
-			input: parameters{
-				t:             time.Unix(99999991, 888),
-				paceInSeconds: 10,
-			},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := nextDuration(tt.input.t, tt.input.paceInSeconds)
-			if !reflect.DeepEqual(tt.want, got) {
-				t.Fatalf("want %v, got %v", tt.want, got)
-			}
-		})
-	}
-}
diff --git a/pkg/util/jmxtool/agent_controller_test.go b/pkg/util/jmxtool/agent_controller_test.go
index 9dee30d2..1e6f8615 100644
--- a/pkg/util/jmxtool/agent_controller_test.go
+++ b/pkg/util/jmxtool/agent_controller_test.go
@@ -26,19 +26,25 @@ import (
 	"time"
 
 	"github.com/megaease/easegress/pkg/filter/proxy"
+	"github.com/megaease/easegress/pkg/logger"
 	"github.com/megaease/easegress/pkg/object/meshcontroller/spec"
 )
 
-func httpServer(finished chan bool) {
+func httpServer(finished chan bool, notFoundFlag bool) {
 	m := http.NewServeMux()
 	s := http.Server{Addr: ":8181", Handler: m}
 	m.HandleFunc("/shutdown", func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintf(w, "Goobey, %q", html.EscapeString(r.URL.Path))
 		s.Shutdown(context.Background())
 	})
-	m.HandleFunc(serviceConfigURL, func(w http.ResponseWriter, r *http.Request) {
-		fmt.Fprintf(w, "Hello, %q", html.EscapeString(r.URL.Path))
-	})
+	if !notFoundFlag {
+		m.HandleFunc(serviceConfigURL, func(w http.ResponseWriter, r *http.Request) {
+			fmt.Fprintf(w, "Hello, %q", html.EscapeString(r.URL.Path))
+		})
+		m.HandleFunc(canaryConfigURL, func(w http.ResponseWriter, r *http.Request) {
+			fmt.Fprintf(w, "Hello, %q", html.EscapeString(r.URL.Path))
+		})
+	}
 	if err := s.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 		fmt.Println(err)
 	}
@@ -46,14 +52,8 @@ func httpServer(finished chan bool) {
 	finished <- true
 }
 
-func TestAgentClient(t *testing.T) {
-	finished := make(chan bool)
-	go httpServer(finished)
-
-	agent := NewAgentClient("127.0.0.1", "8181")
-	fmt.Printf("%v\n", agent)
-
-	service := &spec.Service{
+func getTestService() spec.Service {
+	service := spec.Service{
 		Name: "agent",
 		LoadBalance: &spec.LoadBalance{
 			Policy: proxy.PolicyRandom,
@@ -66,17 +66,75 @@ func TestAgentClient(t *testing.T) {
 			EgressProtocol:  "http",
 		},
 	}
+	return service
+}
+
+func TestAgentClientSuccess(t *testing.T) {
+	logger.InitNop()
+	finished := make(chan bool)
+	go httpServer(finished, false)
+
+	agent := NewAgentClient("127.0.0.1", "8181")
+	fmt.Printf("%+v\n", agent)
+
+	service := getTestService()
+	// UpdateService check
+	err := agent.UpdateService(&service, 1)
+	if err != nil {
+		t.Errorf("agent update service failed\n")
+	}
 
-	agent.UpdateService(service, 1)
+	// UpdateCanary
 	header := &spec.GlobalCanaryHeaders{
 		ServiceHeaders: map[string][]string{},
 	}
-	agent.UpdateCanary(header, 1)
+	err = agent.UpdateCanary(header, 1)
+	if err != nil {
+		t.Errorf("agent update canary failed\n")
+	}
 
+	// shutdown
 	var client = &http.Client{
 		Timeout: time.Second,
 	}
 	client.Get("http://127.0.0.1:8181/shutdown")
+	<-finished
+}
 
+func TestAgentClientFail(t *testing.T) {
+	logger.InitNop()
+	agent := NewAgentClient("127.0.0.1", "8181")
+	service := getTestService()
+
+	// test without available service
+	err := agent.UpdateService(&service, 1)
+	if err == nil {
+		t.Errorf("agent should fail\n")
+	}
+	header := &spec.GlobalCanaryHeaders{
+		ServiceHeaders: map[string][]string{},
+	}
+	err = agent.UpdateCanary(header, 1)
+	if err == nil {
+		t.Errorf("agent should fail\n")
+	}
+
+	// test with 404
+	finished := make(chan bool)
+	go httpServer(finished, true)
+
+	err = agent.UpdateService(&service, 1)
+	if err == nil {
+		t.Errorf("agent shoudl fail\n")
+	}
+	err = agent.UpdateCanary(header, 1)
+	if err == nil {
+		t.Errorf("agent shoudl fail\n")
+	}
+	// shutdown
+	var client = &http.Client{
+		Timeout: time.Second,
+	}
+	client.Get("http://127.0.0.1:8181/shutdown")
 	<-finished
 }