Final review of Influxdb collector

Makefile

@@ -16,8 +16,7 @@ lint:
 
 # Runs spellchecker on the code and comments
 # This requires this tool to be installed from https://github.com/crate-ci/typos?tab=readme-ov-file
-# Example installation:
-# cargo install typos-cli
+# Example installation (if you have rust installed): cargo install typos-cli
 spellcheck:
 	typos .
 

collector/collect_test.go

@@ -14,66 +14,55 @@ import (
 )
 
 type mockTransport struct {
+	oldTrans http.RoundTripper
 	respCode int
 	respBody io.ReadCloser
-
-	// hits     map[string]int
-	// returnError bool
-	// resp        *http.Response
-	// err         error
 }
 
-func newMockTransport() mockTransport {
-	t := mockTransport{
+func newMockTransport() (trans mockTransport, restore func()) {
+	trans = mockTransport{
+		oldTrans: http.DefaultClient.Transport,
 		respCode: 200,
 		respBody: io.NopCloser(strings.NewReader("")),
-
-		// hits: make(map[string]int),
-		// err:         err,
-		// returnError: retErr,
-		// resp:        resp,
+	}
+	restore = func() {
+		// Use this func to restore the default value
+		http.DefaultClient.Transport = trans.oldTrans
 	}
 	// Hijack the default http client so no actual http requests are sent over the network
-	http.DefaultClient.Transport = t
-	return t
+	http.DefaultClient.Transport = trans
+	return
 }
 
 func (t mockTransport) RoundTrip(req *http.Request) (resp *http.Response, err error) {
-	// log.Println("HIJACK:", req.URL.String())
-	// t.hits[req.URL.Hostname()] += 1
-	// if t.err != nil {
-	// 	return nil, t.err
-	// }
-	// if t.returnError != false {
-	// 	req.GetBody = func() (io.ReadCloser, error) {
-	// 		return nil, errHTTP
-	// 	}
-	// }
-	// t.resp.Request = req
-	// return t.resp, nil
-
-	// b, err := io.ReadAll(req.Body)
-	// if err != nil {
-	// 	return
-	// }
-	// fmt.Println(string(b))
-
 	return &http.Response{
 		Request:    req,
 		StatusCode: t.respCode,
 		Body:       t.respBody,
 	}, nil
 }
 
-const mockBodyType string = "application/json"
+////////////////////////////////////////////////////////////////////////////////
 
 var mockStates = map[string]string{
 	"temperature": `{ "value": 0, "unit": "Celcius", "timestamp": "%s", "version": "SignalA_v1.0" }`,
 	"SEKPrice":    `{ "value": 0.10403, "unit": "SEK", "timestamp": "%s", "version": "SignalA_v1.0" }`,
 	"DesiredTemp": `{ "value": 25, "unit": "Celsius", "timestamp": "%s", "version": "SignalA_v1.0" }`,
 	"setpoint":    `{ "value": 20, "unit": "Celsius", "timestamp": "%s", "version": "SignalA_v1.0" }`,
+	"consumption": `{ "value": 32, "unit": "Wh", "timestamp": "%s", "version": "SignalA_v1.0" }`,
+	"state":       `{ "value": 1, "unit": "Binary", "timestamp": "%s", "version": "SignalA_v1.0" }`,
+	"power":       `{ "value": 330, "unit": "Wh", "timestamp": "%s", "version": "SignalA_v1.0" }`,
+	"current":     `{ "value": 9, "unit": "mA", "timestamp": "%s", "version": "SignalA_v1.0" }`,
+	"voltage":     `{ "value": 229, "unit": "V", "timestamp": "%s", "version": "SignalA_v1.0" }`,
 }
 
+const (
+	mockBodyType string = "application/json"
+
+	mockStateIncomplete string = `{ "value": 20, "timestamp": "%s" }`
+	mockStateBadVersion string = `{ "value": false, "timestamp": "%s", "version": "SignalB_v1.0" }`
+)
+
 func mockGetState(c *components.Cervice, s *components.System) (f forms.Form, err error) {
 	if c == nil {
 		err = fmt.Errorf("got empty *Cervice instance")
@@ -93,18 +82,64 @@ func mockGetState(c *components.Cervice, s *components.System) (f forms.Form, er
 }
 
 func TestCollectService(t *testing.T) {
-	newMockTransport()
-	ua := newUnitAsset(*initTemplate(), newSystem(), nil)
+	_, restore := newMockTransport()
+	ua := newUnitAsset(*initTemplate(), newSystem())
+	defer func() {
+		// Make sure to run cleanups! Otherwise you'll get leftover errors from influx
+		ua.cleanup()
+		restore()
+	}()
 	ua.apiGetState = mockGetState
+	sample := Sample{"setpoint", map[string][]string{"Location": {"Kitchen"}}}
+
+	// Good case
+	err := ua.collectService(sample)
+	if err != nil {
+		t.Fatalf("Expected nil error, got: %s", err)
+	}
+	good := mockStates["setpoint"]
 
-	// for _, service := range consumeServices {
-	// 	err := ua.collectService(service)
-	// 	if err != nil {
-	// 		t.Fatalf("Expected nil error while pulling %s, got: %s", service, err)
-	// 	}
-	// }
+	// Bad case: a service returns incomplete data
+	mockStates["setpoint"] = mockStateIncomplete
+	err = ua.collectService(sample)
+	if err == nil {
+		t.Fatalf("Expected error, got nil")
+	}
+
+	// Bad case: a service returns bad form version
+	mockStates["setpoint"] = mockStateBadVersion
+	err = ua.collectService(sample)
+	if err == nil {
+		t.Fatalf("Expected error, got nil")
+	}
+
+	// WARN: Don't forget to restore the mocks!
+	mockStates["setpoint"] = good
+}
+
+func TestCollectAllServices(t *testing.T) {
+	_, restore := newMockTransport()
+	ua := newUnitAsset(*initTemplate(), newSystem())
+	defer func() {
+		ua.cleanup()
+		restore()
+	}()
+	ua.apiGetState = mockGetState
+
+	// Good case
 	err := ua.collectAllServices()
 	if err != nil {
 		t.Fatalf("Expected nil error, got: %s", err)
 	}
+	good := mockStates["setpoint"]
+
+	// Bad case: a service returns incomplete data
+	mockStates["setpoint"] = mockStateIncomplete
+	err = ua.collectAllServices()
+	if err == nil {
+		t.Fatalf("Expected error, got nil")
+	}
+
+	// WARN: Don't forget to restore the mocks!
+	mockStates["setpoint"] = good
 }

collector/startup_test.go

@@ -0,0 +1,130 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+	"github.com/influxdata/influxdb-client-go/v2/api"
+	influxdb2http "github.com/influxdata/influxdb-client-go/v2/api/http"
+	"github.com/influxdata/influxdb-client-go/v2/domain"
+)
+
+var errNotImplemented = fmt.Errorf("method not implemented")
+
+type mockInflux struct {
+	pingErr bool
+	pingRun bool
+	closeCh chan bool
+}
+
+// NOTE: This influxdb2.Client interface is too fatty, must add lot's of methods..
+func (i *mockInflux) Setup(ctx context.Context, username, password, org, bucket string, retentionPeriodHours int) (*domain.OnboardingResponse, error) {
+	return nil, errNotImplemented
+}
+func (i *mockInflux) SetupWithToken(ctx context.Context, username, password, org, bucket string, retentionPeriodHours int, token string) (*domain.OnboardingResponse, error) {
+	return nil, errNotImplemented
+}
+func (i *mockInflux) Ready(ctx context.Context) (*domain.Ready, error) {
+	return nil, errNotImplemented
+}
+func (i *mockInflux) Health(ctx context.Context) (*domain.HealthCheck, error) {
+	return nil, errNotImplemented
+}
+func (i *mockInflux) Ping(ctx context.Context) (bool, error) {
+	switch {
+	case i.pingErr:
+		return false, errNotImplemented
+	case i.pingRun:
+		return false, nil
+	}
+	return true, nil
+}
+func (i *mockInflux) Close() {
+	close(i.closeCh)
+}
+func (i *mockInflux) Options() *influxdb2.Options {
+	return nil
+}
+func (i *mockInflux) ServerURL() string {
+	return errNotImplemented.Error()
+}
+func (i *mockInflux) HTTPService() influxdb2http.Service {
+	return nil
+}
+func (i *mockInflux) WriteAPI(org, bucket string) api.WriteAPI {
+	return nil
+}
+func (i *mockInflux) WriteAPIBlocking(org, bucket string) api.WriteAPIBlocking {
+	return nil
+}
+func (i *mockInflux) QueryAPI(org string) api.QueryAPI {
+	return nil
+}
+func (i *mockInflux) AuthorizationsAPI() api.AuthorizationsAPI {
+	return nil
+}
+func (i *mockInflux) OrganizationsAPI() api.OrganizationsAPI {
+	return nil
+}
+func (i *mockInflux) UsersAPI() api.UsersAPI {
+	return nil
+}
+func (i *mockInflux) DeleteAPI() api.DeleteAPI {
+	return nil
+}
+func (i *mockInflux) BucketsAPI() api.BucketsAPI {
+	return nil
+}
+func (i *mockInflux) LabelsAPI() api.LabelsAPI {
+	return nil
+}
+func (i *mockInflux) TasksAPI() api.TasksAPI {
+	return nil
+}
+func (i *mockInflux) APIClient() *domain.Client {
+	return nil
+}
+
+func TestStartup(t *testing.T) {
+	sys := newSystem() // Needs access to the context cancel'r func
+	ua := newUnitAsset(*initTemplate(), sys)
+
+	// Bad case: too short collection period
+	goodPeriod := ua.CollectionPeriod
+	ua.CollectionPeriod = 0
+	err := ua.startup()
+	if err == nil {
+		t.Fatalf("Expected error, got nil")
+	}
+	ua.CollectionPeriod = goodPeriod
+
+	// Bad case: error while pinging influxdb server
+	ua.influx = &mockInflux{pingErr: true}
+	err = ua.startup()
+	if err == nil {
+		t.Fatalf("Expected error, got nil")
+	}
+
+	// Bad case: influxdb not running when pinging
+	ua.influx = &mockInflux{pingRun: true}
+	err = ua.startup()
+	if err == nil {
+		t.Fatalf("Expected error, got nil")
+	}
+
+	// Good case: startup() enters loop and can be shut down again
+	c := make(chan bool)
+	ua.influx = &mockInflux{closeCh: c}
+	go ua.startup()
+	sys.cancel()
+	// Wait for startup() to quit it's loop and call cleanup(), which in turn
+	// should call influx.Close(). If it times out it failed.
+	select {
+	case <-c:
+	case <-time.After(200 * time.Millisecond):
+		t.Fatalf("Expected startup to quit and call close(), but timed out")
+	}
+}