Merge pull request 'Count a module's own stream events beside its emits in the catalogue agreement (issue 1249)' (#395) from fix/1249-stream-events-agree into main
This commit was merged in pull request #395.
This commit is contained in:
@@ -97,27 +97,17 @@ func theCataloguesEvents(t *testing.T) ([]AnEmitter, []AConsumer, []DeclaredSeat
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
var m struct {
|
emitter, consumer, declared, err := eventsOfManifest(raw)
|
||||||
Module string `json:"module"`
|
if err != nil {
|
||||||
Emits []string `json:"emits"`
|
|
||||||
Consumes []string `json:"consumes"`
|
|
||||||
Seats []struct {
|
|
||||||
Name string `json:"name"`
|
|
||||||
Emits []string `json:"emits"`
|
|
||||||
} `json:"seats"`
|
|
||||||
}
|
|
||||||
if err := json.Unmarshal(raw, &m); err != nil {
|
|
||||||
t.Fatalf("%s: %v", e.Name(), err)
|
t.Fatalf("%s: %v", e.Name(), err)
|
||||||
}
|
}
|
||||||
if len(m.Emits) > 0 {
|
if emitter != nil {
|
||||||
emitters = append(emitters, AnEmitter{Module: m.Module, Emits: m.Emits})
|
emitters = append(emitters, *emitter)
|
||||||
}
|
}
|
||||||
if len(m.Consumes) > 0 {
|
if consumer != nil {
|
||||||
consumers = append(consumers, AConsumer{Module: m.Module, Consumes: m.Consumes})
|
consumers = append(consumers, *consumer)
|
||||||
}
|
|
||||||
for _, s := range m.Seats {
|
|
||||||
seats = append(seats, DeclaredSeat{Name: s.Name, Emits: s.Emits})
|
|
||||||
}
|
}
|
||||||
|
seats = append(seats, declared...)
|
||||||
}
|
}
|
||||||
if len(emitters) == 0 {
|
if len(emitters) == 0 {
|
||||||
t.Fatalf("no module under %s emits anything, so this proved nothing", root)
|
t.Fatalf("no module under %s emits anything, so this proved nothing", root)
|
||||||
@@ -245,3 +235,41 @@ func TestTheBrokerAndTheCatalogueAgreeOnTheRuntimeModule(t *testing.T) {
|
|||||||
t.Fatalf("the broker calls the runtime %q and the catalogue %q", RuntimeModule, catalogue.RuntimeModule)
|
t.Fatalf("the broker calls the runtime %q and the catalogue %q", RuntimeModule, catalogue.RuntimeModule)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// eventsOfManifest is what one manifest says it emits and consumes for the agreement: its `emits`, and the events of
|
||||||
|
// its own streams, which a consumer names as <module>.<event> as it names the others (novox/hq issue 1249: once
|
||||||
|
// mesh-mail declared emits beside its stream, the stream's events were no longer counted).
|
||||||
|
func eventsOfManifest(raw []byte) (*AnEmitter, *AConsumer, []DeclaredSeat, error) {
|
||||||
|
var m struct {
|
||||||
|
Module string `json:"module"`
|
||||||
|
Emits []string `json:"emits"`
|
||||||
|
Consumes []string `json:"consumes"`
|
||||||
|
Streams []struct {
|
||||||
|
Events []string `json:"events"`
|
||||||
|
} `json:"streams"`
|
||||||
|
Seats []struct {
|
||||||
|
Name string `json:"name"`
|
||||||
|
Emits []string `json:"emits"`
|
||||||
|
} `json:"seats"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(raw, &m); err != nil {
|
||||||
|
return nil, nil, nil, err
|
||||||
|
}
|
||||||
|
emits := append([]string{}, m.Emits...)
|
||||||
|
for _, s := range m.Streams {
|
||||||
|
emits = append(emits, s.Events...)
|
||||||
|
}
|
||||||
|
var emitter *AnEmitter
|
||||||
|
if len(emits) > 0 {
|
||||||
|
emitter = &AnEmitter{Module: m.Module, Emits: emits}
|
||||||
|
}
|
||||||
|
var consumer *AConsumer
|
||||||
|
if len(m.Consumes) > 0 {
|
||||||
|
consumer = &AConsumer{Module: m.Module, Consumes: m.Consumes}
|
||||||
|
}
|
||||||
|
var seats []DeclaredSeat
|
||||||
|
for _, s := range m.Seats {
|
||||||
|
seats = append(seats, DeclaredSeat{Name: s.Name, Emits: s.Emits})
|
||||||
|
}
|
||||||
|
return emitter, consumer, seats, nil
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,38 @@
|
|||||||
|
package broker
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// mailShaped is mesh-mail's manifest as mesh-catalog main (b767aac8) declares its events: emits beside two streams of
|
||||||
|
// its own (modules/mesh-mail/module.json), trimmed to the keys the agreement reads.
|
||||||
|
const mailShaped = `{"module": "mesh-mail",
|
||||||
|
"emits": ["mail.send.sent", "mail.send.failed"],
|
||||||
|
"streams": [
|
||||||
|
{"name": "mail", "events": ["mail.*.message.added", "mail.*.message.changed", "mail.*.message.removed"]},
|
||||||
|
{"name": "mail-sends", "events": ["mail.send.approved"]}]}`
|
||||||
|
|
||||||
|
func TestAModulesStreamEventsCountBesideItsEmits(t *testing.T) { // 1249/AC1
|
||||||
|
emitter, _, _, err := eventsOfManifest([]byte(mailShaped))
|
||||||
|
if err != nil || emitter == nil {
|
||||||
|
t.Fatalf("emitter %v, err %v", emitter, err)
|
||||||
|
}
|
||||||
|
bad := Disagreements([]AnEmitter{*emitter},
|
||||||
|
[]AConsumer{{Module: "mesh-mail-notifier", Consumes: []string{"mesh-mail.mail.*.message.added"}}}, nil)
|
||||||
|
if len(bad) != 0 {
|
||||||
|
t.Fatalf("a consumer of mesh-mail's stream event was reported: %v", bad)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAnEventNeitherEmittedNorOnTheStreamIsStillCaught(t *testing.T) { // 1249/AC2
|
||||||
|
emitter, _, _, err := eventsOfManifest([]byte(mailShaped))
|
||||||
|
if err != nil || emitter == nil {
|
||||||
|
t.Fatalf("emitter %v, err %v", emitter, err)
|
||||||
|
}
|
||||||
|
bad := Disagreements([]AnEmitter{*emitter},
|
||||||
|
[]AConsumer{{Module: "mesh-mail-notifier", Consumes: []string{"mesh-mail.mail.*.message.renamed"}}}, nil)
|
||||||
|
if len(bad) != 1 || !strings.Contains(bad[0], "mail.*.message.added") {
|
||||||
|
t.Fatalf("want one disagreement naming the stream's events among what mesh-mail emits, got %v", bad)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user