Fix #897 and shuffle directory around (#1054)

* Fix #897 and shuffle directory around

* Update find-lint

* goimports

Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>

5 files changed, 581 insertions, 0 deletions

diff --git a/internal/test/client.go b/internal/test/client.go
new file mode 100644
index 00000000..a38540ac
--- /dev/null
+++ b/internal/test/client.go
@@ -0,0 +1,158 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// 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 test
+
+import (
+	"crypto/tls"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+// Request contains the information necessary to issue a request and test its result
+type Request struct {
+	Req              *http.Request
+	WantedBody       string
+	WantedStatusCode int
+	LastErr          *LastRequestErr
+}
+
+// LastRequestErr is a synchronised error wrapper
+// Useful for obtaining the last error from a set of requests
+type LastRequestErr struct {
+	sync.Mutex
+	Err error
+}
+
+// Set sets the error
+func (r *LastRequestErr) Set(err error) {
+	r.Lock()
+	defer r.Unlock()
+	r.Err = err
+}
+
+// Get gets the error
+func (r *LastRequestErr) Get() error {
+	r.Lock()
+	defer r.Unlock()
+	return r.Err
+}
+
+// CanonicalJSONInput canonicalises a slice of JSON strings
+// Useful for test input
+func CanonicalJSONInput(jsonData []string) []string {
+	for i := range jsonData {
+		jsonBytes, err := gomatrixserverlib.CanonicalJSON([]byte(jsonData[i]))
+		if err != nil && err != io.EOF {
+			panic(err)
+		}
+		jsonData[i] = string(jsonBytes)
+	}
+	return jsonData
+}
+
+// Do issues a request and checks the status code and body of the response
+func (r *Request) Do() (err error) {
+	client := &http.Client{
+		Timeout: 5 * time.Second,
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: true,
+			},
+		},
+	}
+	res, err := client.Do(r.Req)
+	if err != nil {
+		return err
+	}
+	defer (func() { err = res.Body.Close() })()
+
+	if res.StatusCode != r.WantedStatusCode {
+		return fmt.Errorf("incorrect status code. Expected: %d  Got: %d", r.WantedStatusCode, res.StatusCode)
+	}
+
+	if r.WantedBody != "" {
+		resBytes, err := ioutil.ReadAll(res.Body)
+		if err != nil {
+			return err
+		}
+		jsonBytes, err := gomatrixserverlib.CanonicalJSON(resBytes)
+		if err != nil {
+			return err
+		}
+		if string(jsonBytes) != r.WantedBody {
+			return fmt.Errorf("returned wrong bytes. Expected:\n%s\n\nGot:\n%s", r.WantedBody, string(jsonBytes))
+		}
+	}
+
+	return nil
+}
+
+// DoUntilSuccess blocks and repeats the same request until the response returns the desired status code and body.
+// It then closes the given channel and returns.
+func (r *Request) DoUntilSuccess(done chan error) {
+	r.LastErr = &LastRequestErr{}
+	for {
+		if err := r.Do(); err != nil {
+			r.LastErr.Set(err)
+			time.Sleep(1 * time.Second) // don't tightloop
+			continue
+		}
+		close(done)
+		return
+	}
+}
+
+// Run repeatedly issues a request until success, error or a timeout is reached
+func (r *Request) Run(label string, timeout time.Duration, serverCmdChan chan error) {
+	fmt.Printf("==TESTING== %v (timeout: %v)\n", label, timeout)
+	done := make(chan error, 1)
+
+	// We need to wait for the server to:
+	// - have connected to the database
+	// - have created the tables
+	// - be listening on the given port
+	go r.DoUntilSuccess(done)
+
+	// wait for one of:
+	// - the test to pass (done channel is closed)
+	// - the server to exit with an error (error sent on serverCmdChan)
+	// - our test timeout to expire
+	// We don't need to clean up since the main() function handles that in the event we panic
+	select {
+	case <-time.After(timeout):
+		fmt.Printf("==TESTING== %v TIMEOUT\n", label)
+		if reqErr := r.LastErr.Get(); reqErr != nil {
+			fmt.Println("Last /sync request error:")
+			fmt.Println(reqErr)
+		}
+		panic(fmt.Sprintf("%v server timed out", label))
+	case err := <-serverCmdChan:
+		if err != nil {
+			fmt.Println("=============================================================================================")
+			fmt.Printf("%v server failed to run. If failing with 'pq: password authentication failed for user' try:", label)
+			fmt.Println("    export PGHOST=/var/run/postgresql")
+			fmt.Println("=============================================================================================")
+			panic(err)
+		}
+	case <-done:
+		fmt.Printf("==TESTING== %v PASSED\n", label)
+	}
+}
diff --git a/internal/test/config.go b/internal/test/config.go
new file mode 100644
index 00000000..06510c8b
--- /dev/null
+++ b/internal/test/config.go
@@ -0,0 +1,208 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// 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 test
+
+import (
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/pem"
+	"fmt"
+	"io/ioutil"
+	"math/big"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/matrix-org/dendrite/internal/config"
+	"github.com/matrix-org/gomatrixserverlib"
+	"gopkg.in/yaml.v2"
+)
+
+const (
+	// ConfigFile is the name of the config file for a server.
+	ConfigFile = "dendrite.yaml"
+	// ServerKeyFile is the name of the file holding the matrix server private key.
+	ServerKeyFile = "server_key.pem"
+	// TLSCertFile is the name of the file holding the TLS certificate used for federation.
+	TLSCertFile = "tls_cert.pem"
+	// TLSKeyFile is the name of the file holding the TLS key used for federation.
+	TLSKeyFile = "tls_key.pem"
+	// MediaDir is the name of the directory used to store media.
+	MediaDir = "media"
+)
+
+// MakeConfig makes a config suitable for running integration tests.
+// Generates new matrix and TLS keys for the server.
+func MakeConfig(configDir, kafkaURI, database, host string, startPort int) (*config.Dendrite, int, error) {
+	var cfg config.Dendrite
+
+	port := startPort
+	assignAddress := func() config.Address {
+		result := config.Address(fmt.Sprintf("%s:%d", host, port))
+		port++
+		return result
+	}
+
+	serverKeyPath := filepath.Join(configDir, ServerKeyFile)
+	tlsCertPath := filepath.Join(configDir, TLSKeyFile)
+	tlsKeyPath := filepath.Join(configDir, TLSCertFile)
+	mediaBasePath := filepath.Join(configDir, MediaDir)
+
+	if err := NewMatrixKey(serverKeyPath); err != nil {
+		return nil, 0, err
+	}
+
+	if err := NewTLSKey(tlsKeyPath, tlsCertPath); err != nil {
+		return nil, 0, err
+	}
+
+	cfg.Version = config.Version
+
+	cfg.Matrix.ServerName = gomatrixserverlib.ServerName(assignAddress())
+	cfg.Matrix.PrivateKeyPath = config.Path(serverKeyPath)
+	cfg.Matrix.FederationCertificatePaths = []config.Path{config.Path(tlsCertPath)}
+
+	cfg.Media.BasePath = config.Path(mediaBasePath)
+
+	cfg.Kafka.Addresses = []string{kafkaURI}
+	// TODO: Different servers should be using different topics.
+	// Make this configurable somehow?
+	cfg.Kafka.Topics.OutputRoomEvent = "test.room.output"
+	cfg.Kafka.Topics.OutputClientData = "test.clientapi.output"
+	cfg.Kafka.Topics.OutputTypingEvent = "test.typing.output"
+	cfg.Kafka.Topics.UserUpdates = "test.user.output"
+
+	// TODO: Use different databases for the different schemas.
+	// Using the same database for every schema currently works because
+	// the table names are globally unique. But we might not want to
+	// rely on that in the future.
+	cfg.Database.Account = config.DataSource(database)
+	cfg.Database.AppService = config.DataSource(database)
+	cfg.Database.Device = config.DataSource(database)
+	cfg.Database.MediaAPI = config.DataSource(database)
+	cfg.Database.RoomServer = config.DataSource(database)
+	cfg.Database.ServerKey = config.DataSource(database)
+	cfg.Database.SyncAPI = config.DataSource(database)
+	cfg.Database.PublicRoomsAPI = config.DataSource(database)
+
+	cfg.Listen.ClientAPI = assignAddress()
+	cfg.Listen.AppServiceAPI = assignAddress()
+	cfg.Listen.FederationAPI = assignAddress()
+	cfg.Listen.MediaAPI = assignAddress()
+	cfg.Listen.RoomServer = assignAddress()
+	cfg.Listen.SyncAPI = assignAddress()
+	cfg.Listen.PublicRoomsAPI = assignAddress()
+	cfg.Listen.EDUServer = assignAddress()
+
+	// Bind to the same address as the listen address
+	// All microservices are run on the same host in testing
+	cfg.Bind.ClientAPI = cfg.Listen.ClientAPI
+	cfg.Bind.AppServiceAPI = cfg.Listen.AppServiceAPI
+	cfg.Bind.FederationAPI = cfg.Listen.FederationAPI
+	cfg.Bind.MediaAPI = cfg.Listen.MediaAPI
+	cfg.Bind.RoomServer = cfg.Listen.RoomServer
+	cfg.Bind.SyncAPI = cfg.Listen.SyncAPI
+	cfg.Bind.PublicRoomsAPI = cfg.Listen.PublicRoomsAPI
+	cfg.Bind.EDUServer = cfg.Listen.EDUServer
+
+	return &cfg, port, nil
+}
+
+// WriteConfig writes the config file to the directory.
+func WriteConfig(cfg *config.Dendrite, configDir string) error {
+	data, err := yaml.Marshal(cfg)
+	if err != nil {
+		return err
+	}
+	return ioutil.WriteFile(filepath.Join(configDir, ConfigFile), data, 0666)
+}
+
+// NewMatrixKey generates a new ed25519 matrix server key and writes it to a file.
+func NewMatrixKey(matrixKeyPath string) (err error) {
+	var data [35]byte
+	_, err = rand.Read(data[:])
+	if err != nil {
+		return err
+	}
+	keyOut, err := os.OpenFile(matrixKeyPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
+	if err != nil {
+		return err
+	}
+
+	defer (func() {
+		err = keyOut.Close()
+	})()
+
+	err = pem.Encode(keyOut, &pem.Block{
+		Type: "MATRIX PRIVATE KEY",
+		Headers: map[string]string{
+			"Key-ID": "ed25519:" + base64.RawStdEncoding.EncodeToString(data[:3]),
+		},
+		Bytes: data[3:],
+	})
+	return err
+}
+
+const certificateDuration = time.Hour * 24 * 365 * 10
+
+// NewTLSKey generates a new RSA TLS key and certificate and writes it to a file.
+func NewTLSKey(tlsKeyPath, tlsCertPath string) error {
+	priv, err := rsa.GenerateKey(rand.Reader, 4096)
+	if err != nil {
+		return err
+	}
+
+	notBefore := time.Now()
+	notAfter := notBefore.Add(certificateDuration)
+	serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128)
+	serialNumber, err := rand.Int(rand.Reader, serialNumberLimit)
+	if err != nil {
+		return err
+	}
+
+	template := x509.Certificate{
+		SerialNumber:          serialNumber,
+		NotBefore:             notBefore,
+		NotAfter:              notAfter,
+		KeyUsage:              x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+	}
+	derBytes, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv)
+	if err != nil {
+		return err
+	}
+	certOut, err := os.Create(tlsCertPath)
+	if err != nil {
+		return err
+	}
+	defer certOut.Close() // nolint: errcheck
+	if err = pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: derBytes}); err != nil {
+		return err
+	}
+
+	keyOut, err := os.OpenFile(tlsKeyPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
+	if err != nil {
+		return err
+	}
+	defer keyOut.Close() // nolint: errcheck
+	err = pem.Encode(keyOut, &pem.Block{
+		Type:  "RSA PRIVATE KEY",
+		Bytes: x509.MarshalPKCS1PrivateKey(priv),
+	})
+	return err
+}
diff --git a/internal/test/kafka.go b/internal/test/kafka.go
new file mode 100644
index 00000000..cbf24630
--- /dev/null
+++ b/internal/test/kafka.go
@@ -0,0 +1,76 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// 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 test
+
+import (
+	"io"
+	"os/exec"
+	"path/filepath"
+	"strings"
+)
+
+// KafkaExecutor executes kafka scripts.
+type KafkaExecutor struct {
+	// The location of Zookeeper. Typically this is `localhost:2181`.
+	ZookeeperURI string
+	// The directory where Kafka is installed to. Used to locate kafka scripts.
+	KafkaDirectory string
+	// The location of the Kafka logs. Typically this is `localhost:9092`.
+	KafkaURI string
+	// Where stdout and stderr should be written to. Typically this is `os.Stderr`.
+	OutputWriter io.Writer
+}
+
+// CreateTopic creates a new kafka topic. This is created with a single partition.
+func (e *KafkaExecutor) CreateTopic(topic string) error {
+	cmd := exec.Command(
+		filepath.Join(e.KafkaDirectory, "bin", "kafka-topics.sh"),
+		"--create",
+		"--zookeeper", e.ZookeeperURI,
+		"--replication-factor", "1",
+		"--partitions", "1",
+		"--topic", topic,
+	)
+	cmd.Stdout = e.OutputWriter
+	cmd.Stderr = e.OutputWriter
+	return cmd.Run()
+}
+
+// WriteToTopic writes data to a kafka topic.
+func (e *KafkaExecutor) WriteToTopic(topic string, data []string) error {
+	cmd := exec.Command(
+		filepath.Join(e.KafkaDirectory, "bin", "kafka-console-producer.sh"),
+		"--broker-list", e.KafkaURI,
+		"--topic", topic,
+	)
+	cmd.Stdout = e.OutputWriter
+	cmd.Stderr = e.OutputWriter
+	cmd.Stdin = strings.NewReader(strings.Join(data, "\n"))
+	return cmd.Run()
+}
+
+// DeleteTopic deletes a given kafka topic if it exists.
+func (e *KafkaExecutor) DeleteTopic(topic string) error {
+	cmd := exec.Command(
+		filepath.Join(e.KafkaDirectory, "bin", "kafka-topics.sh"),
+		"--delete",
+		"--if-exists",
+		"--zookeeper", e.ZookeeperURI,
+		"--topic", topic,
+	)
+	cmd.Stderr = e.OutputWriter
+	cmd.Stdout = e.OutputWriter
+	return cmd.Run()
+}
diff --git a/internal/test/server.go b/internal/test/server.go
new file mode 100644
index 00000000..1493dac6
--- /dev/null
+++ b/internal/test/server.go
@@ -0,0 +1,105 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// 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 test
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+
+	"github.com/matrix-org/dendrite/internal/config"
+)
+
+// Defaulting allows assignment of string variables with a fallback default value
+// Useful for use with os.Getenv() for example
+func Defaulting(value, defaultValue string) string {
+	if value == "" {
+		value = defaultValue
+	}
+	return value
+}
+
+// CreateDatabase creates a new database, dropping it first if it exists
+func CreateDatabase(command string, args []string, database string) error {
+	cmd := exec.Command(command, args...)
+	cmd.Stdin = strings.NewReader(
+		fmt.Sprintf("DROP DATABASE IF EXISTS %s; CREATE DATABASE %s;", database, database),
+	)
+	// Send stdout and stderr to our stderr so that we see error messages from
+	// the psql process
+	cmd.Stdout = os.Stderr
+	cmd.Stderr = os.Stderr
+	return cmd.Run()
+}
+
+// CreateBackgroundCommand creates an executable command
+// The Cmd being executed is returned. A channel is also returned,
+// which will have any termination errors sent down it, followed immediately by the channel being closed.
+func CreateBackgroundCommand(command string, args []string) (*exec.Cmd, chan error) {
+	cmd := exec.Command(command, args...)
+	cmd.Stderr = os.Stderr
+	cmd.Stdout = os.Stderr
+
+	if err := cmd.Start(); err != nil {
+		panic("failed to start server: " + err.Error())
+	}
+	cmdChan := make(chan error, 1)
+	go func() {
+		cmdChan <- cmd.Wait()
+		close(cmdChan)
+	}()
+	return cmd, cmdChan
+}
+
+// InitDatabase creates the database and config file needed for the server to run
+func InitDatabase(postgresDatabase, postgresContainerName string, databases []string) {
+	if len(databases) > 0 {
+		var dbCmd string
+		var dbArgs []string
+		if postgresContainerName == "" {
+			dbCmd = "psql"
+			dbArgs = []string{postgresDatabase}
+		} else {
+			dbCmd = "docker"
+			dbArgs = []string{
+				"exec", "-i", postgresContainerName, "psql", "-U", "postgres", postgresDatabase,
+			}
+		}
+		for _, database := range databases {
+			if err := CreateDatabase(dbCmd, dbArgs, database); err != nil {
+				panic(err)
+			}
+		}
+	}
+}
+
+// StartProxy creates a reverse proxy
+func StartProxy(bindAddr string, cfg *config.Dendrite) (*exec.Cmd, chan error) {
+	proxyArgs := []string{
+		"--bind-address", bindAddr,
+		"--sync-api-server-url", "http://" + string(cfg.Listen.SyncAPI),
+		"--client-api-server-url", "http://" + string(cfg.Listen.ClientAPI),
+		"--media-api-server-url", "http://" + string(cfg.Listen.MediaAPI),
+		"--public-rooms-api-server-url", "http://" + string(cfg.Listen.PublicRoomsAPI),
+		"--tls-cert", "server.crt",
+		"--tls-key", "server.key",
+	}
+	return CreateBackgroundCommand(
+		filepath.Join(filepath.Dir(os.Args[0]), "client-api-proxy"),
+		proxyArgs,
+	)
+}
diff --git a/internal/test/slice.go b/internal/test/slice.go
new file mode 100644
index 00000000..00c740db
--- /dev/null
+++ b/internal/test/slice.go
@@ -0,0 +1,34 @@
+// 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 test
+
+import "sort"
+
+// UnsortedStringSliceEqual returns true if the slices have same length & elements.
+// Does not modify the given slice.
+func UnsortedStringSliceEqual(first, second []string) bool {
+	if len(first) != len(second) {
+		return false
+	}
+
+	a, b := first[:], second[:]
+	sort.Strings(a)
+	sort.Strings(b)
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+
+	return true
+}