/usr/share/gocode/src/github.com/influxdata/influxdb/services/opentsdb/service_test.go is in golang-github-influxdb-influxdb-dev 1.0.2+dfsg1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | package opentsdb_test
import (
"io/ioutil"
"log"
"net"
"net/http"
"reflect"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/davecgh/go-spew/spew"
"github.com/influxdata/influxdb/models"
"github.com/influxdata/influxdb/services/meta"
"github.com/influxdata/influxdb/services/opentsdb"
)
// Ensure a point can be written via the telnet protocol.
func TestService_Telnet(t *testing.T) {
t.Parallel()
s := NewService("db0")
if err := s.Open(); err != nil {
t.Fatal(err)
}
defer s.Close()
// Mock points writer.
var called int32
s.PointsWriter.WritePointsFn = func(database, retentionPolicy string, consistencyLevel models.ConsistencyLevel, points []models.Point) error {
atomic.StoreInt32(&called, 1)
if database != "db0" {
t.Fatalf("unexpected database: %s", database)
} else if retentionPolicy != "" {
t.Fatalf("unexpected retention policy: %s", retentionPolicy)
} else if !reflect.DeepEqual(points, []models.Point{
models.MustNewPoint(
"sys.cpu.user",
map[string]string{"host": "webserver01", "cpu": "0"},
map[string]interface{}{"value": 42.5},
time.Unix(1356998400, 0),
),
}) {
t.Fatalf("unexpected points: %#v", points)
}
return nil
}
// Open connection to the service.
conn, err := net.Dial("tcp", s.Addr().String())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
// Write telnet data and close.
if _, err := conn.Write([]byte("put sys.cpu.user 1356998400 42.5 host=webserver01 cpu=0")); err != nil {
t.Fatal(err)
}
if err := conn.Close(); err != nil {
t.Fatal(err)
}
tick := time.Tick(10 * time.Millisecond)
timeout := time.After(10 * time.Second)
for {
select {
case <-tick:
// Verify that the writer was called.
if atomic.LoadInt32(&called) > 0 {
return
}
case <-timeout:
t.Fatal("points writer not called")
}
}
}
// Ensure a point can be written via the HTTP protocol.
func TestService_HTTP(t *testing.T) {
t.Parallel()
s := NewService("db0")
if err := s.Open(); err != nil {
t.Fatal(err)
}
defer s.Close()
// Mock points writer.
var called bool
s.PointsWriter.WritePointsFn = func(database, retentionPolicy string, consistencyLevel models.ConsistencyLevel, points []models.Point) error {
called = true
if database != "db0" {
t.Fatalf("unexpected database: %s", database)
} else if retentionPolicy != "" {
t.Fatalf("unexpected retention policy: %s", retentionPolicy)
} else if !reflect.DeepEqual(points, []models.Point{
models.MustNewPoint(
"sys.cpu.nice",
map[string]string{"dc": "lga", "host": "web01"},
map[string]interface{}{"value": 18.0},
time.Unix(1346846400, 0),
),
}) {
spew.Dump(points)
t.Fatalf("unexpected points: %#v", points)
}
return nil
}
// Write HTTP request to server.
resp, err := http.Post("http://"+s.Addr().String()+"/api/put", "application/json", strings.NewReader(`{"metric":"sys.cpu.nice", "timestamp":1346846400, "value":18, "tags":{"host":"web01", "dc":"lga"}}`))
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
// Verify status and body.
if resp.StatusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: %d", resp.StatusCode)
}
// Verify that the writer was called.
if !called {
t.Fatal("points writer not called")
}
}
type Service struct {
*opentsdb.Service
PointsWriter PointsWriter
}
// NewService returns a new instance of Service.
func NewService(database string) *Service {
srv, _ := opentsdb.NewService(opentsdb.Config{
BindAddress: "127.0.0.1:0",
Database: database,
ConsistencyLevel: "one",
})
s := &Service{Service: srv}
s.Service.PointsWriter = &s.PointsWriter
s.Service.MetaClient = &DatabaseCreator{}
if !testing.Verbose() {
s.Logger = log.New(ioutil.Discard, "", log.LstdFlags)
}
return s
}
// PointsWriter represents a mock impl of PointsWriter.
type PointsWriter struct {
WritePointsFn func(database, retentionPolicy string, consistencyLevel models.ConsistencyLevel, points []models.Point) error
}
func (w *PointsWriter) WritePoints(database, retentionPolicy string, consistencyLevel models.ConsistencyLevel, points []models.Point) error {
return w.WritePointsFn(database, retentionPolicy, consistencyLevel, points)
}
type DatabaseCreator struct {
}
func (d *DatabaseCreator) CreateDatabase(name string) (*meta.DatabaseInfo, error) {
return nil, nil
}
|