99 lines
3.6 KiB
Go
99 lines
3.6 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
// 1596/AC6: the tool side and the converter over a real bus (the shape of brother's device_integration_test.go,
|
|
// ADR 0436 §7): a JetStream server of the test's own, the work-queue stream convert-asks and its consumer, the state
|
|
// convert-answers, and a fake document server. The tool side's Convert queues an ask; the converter takes it and
|
|
// answers; Convert returns the answer. Compiled by the merge gate's `go vet -tags integration`; never run on this machine.
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// convertAskedSubject is the wire subject the controller composes for the stream's one event:
|
|
// "mesh.stream." + module + ".event." + event (mesh-controller internal/broker/modstreams.go).
|
|
const convertAskedSubject = "mesh.stream.only-office.event." + EventAsked
|
|
|
|
// natsBus is the tool side's bus: the ask published on the stream, the answer read from the state's bucket.
|
|
type natsBus struct {
|
|
js jetstream.JetStream
|
|
kv jetstream.KeyValue
|
|
}
|
|
|
|
func (b natsBus) Publish(id string, body any) error {
|
|
raw, _ := json.Marshal(body)
|
|
_, err := b.js.Publish(context.Background(), convertAskedSubject, raw, jetstream.WithMsgID(id))
|
|
return err
|
|
}
|
|
|
|
func (b natsBus) Answer(id string) (json.RawMessage, error) {
|
|
e, err := b.kv.Get(context.Background(), id)
|
|
if err != nil {
|
|
return nil, nil
|
|
}
|
|
return e.Value(), nil
|
|
}
|
|
|
|
func TestTheToolSideAndTheConverterConvertOverARealBus(t *testing.T) {
|
|
opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}
|
|
s, err := server.NewServer(opts)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
go s.Start()
|
|
if !s.ReadyForConnections(30 * time.Second) {
|
|
t.Fatal("a bus for this test did not come up within 30s")
|
|
}
|
|
t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
|
|
ctx := context.Background()
|
|
nc, err := nats.Connect(s.ClientURL())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc.Close()
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: AsksStream, Subjects: []string{convertAskedSubject},
|
|
Retention: jetstream.WorkQueuePolicy, Discard: jetstream.DiscardNew, MaxAge: 24 * time.Hour, MaxBytes: 8 << 20}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
account := process.Account{URL: s.ClientURL(), Node: "novox", Module: "only-office", User: "novox.only-office.converter", Password: "pw"}
|
|
if _, err := js.CreateConsumer(ctx, AsksStream, jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
|
|
AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: 3}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: account.Module + "_" + AnswersState, TTL: 10 * time.Minute})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
d, tgt := newFakeDocs(t), newFakeTarget(t)
|
|
w, _ := worker(d)
|
|
runCtx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
go runConverter(runCtx, account, w)
|
|
|
|
svc := &Service{Asks: natsBus{js: js, kv: kv}, Now: time.Now, Wait: 20 * time.Second, Poll: 50 * time.Millisecond}
|
|
got, err := svc.Convert(ctx, "http://novox.internal:20001/b/in.docx?X-Amz-Signature=a", "docx", tgt.srv.URL+"/b/in.pdf?sig=1", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got["size"] != float64(len(d.pdf)) || string(tgt.got) != string(d.pdf) {
|
|
t.Fatalf("answer %v, target got %q", got, tgt.got)
|
|
}
|
|
if len(d.requests) != 1 {
|
|
t.Fatalf("%d conversion requests", len(d.requests))
|
|
}
|
|
}
|