From 3202337d13554146f3d86261e03b8d4ec69a1be0 Mon Sep 17 00:00:00 2001 From: OmarAlJarrah Date: Sun, 9 Aug 2026 04:43:27 +0300 Subject: [PATCH 1/3] feat(compilers/openapi): populate the operation streaming fields --- compilers/openapi/conformance_test.go | 14 + compilers/openapi/helpers_test.go | 2 +- .../openapi/internal/lowering/lowering.go | 35 +- .../internal/lowering/lowering_test.go | 16 +- .../openapi/internal/lowering/streaming.go | 80 ++++ .../internal/lowering/streaming_test.go | 74 +++ .../operation/helpers_internal_test.go | 4 +- .../internal/operation/helpers_test.go | 2 +- .../openapi/internal/operation/operations.go | 11 +- .../openapi/internal/operation/streaming.go | 161 +++++++ .../internal/operation/streaming_test.go | 266 +++++++++++ .../internal/schema/compose_internal_test.go | 4 +- .../internal/schema/helpers_internal_test.go | 4 +- .../openapi/internal/schema/helpers_test.go | 4 +- .../openapi/internal/schema/schema_test.go | 6 +- compilers/openapi/openapi.go | 2 +- compilers/openapi/options.go | 15 + compilers/openapi/streaming_test.go | 168 +++++++ .../openapi/sequential-media.golden.json | 16 + .../openapi/streaming-media-30.golden.json | 171 +++++++ .../openapi/streaming-media-30.yaml | 15 + .../openapi/streaming-media-31.golden.json | 427 ++++++++++++++++++ .../openapi/streaming-media-31.yaml | 39 ++ .../openapi/unwitnessed.golden.txt | 5 - 24 files changed, 1501 insertions(+), 40 deletions(-) create mode 100644 compilers/openapi/internal/lowering/streaming.go create mode 100644 compilers/openapi/internal/lowering/streaming_test.go create mode 100644 compilers/openapi/internal/operation/streaming.go create mode 100644 compilers/openapi/internal/operation/streaming_test.go create mode 100644 compilers/openapi/streaming_test.go create mode 100644 testdata/conformance/openapi/streaming-media-30.golden.json create mode 100644 testdata/conformance/openapi/streaming-media-30.yaml create mode 100644 testdata/conformance/openapi/streaming-media-31.golden.json create mode 100644 testdata/conformance/openapi/streaming-media-31.yaml diff --git a/compilers/openapi/conformance_test.go b/compilers/openapi/conformance_test.go index 17ae7d91..3a038110 100644 --- a/compilers/openapi/conformance_test.go +++ b/compilers/openapi/conformance_test.go @@ -187,6 +187,8 @@ func conformanceCases() []conformanceCase { {"multipart-encoding", assertMultipartEncoding}, {"file-body", assertFileBody}, {"sequential-media", assertSequentialMedia}, + {"streaming-media-30", assertStreamingMedia30}, + {"streaming-media-31", assertStreamingMedia31}, {"per-status-errors", assertPerStatusErrors}, {"response-links", assertResponseLinks}, {"webhooks", assertWebhooks}, @@ -1733,6 +1735,11 @@ func assertPartHeaders(t *testing.T, doc *ir.Document, headers []ir.Property) { // Content.ItemEncoding, while a positional prefixEncoding — which a single // every-item encoding has no ordinals for — takes itself and the tail encoding // beside it into Unmodeled instead. +// +// It also pins what itemSchema says about the operation, which for a long while +// was nothing: the keyword states that the body is a sequence of items, so the +// operation streams, and it says so itself rather than being guessed at from +// the media type — multipart/mixed is in no streaming media-type list. func assertSequentialMedia(t *testing.T, doc *ir.Document, _ []ir.Diagnostic) { events, ok := opByName(doc, "streamEvents") require.True(t, ok) @@ -1743,6 +1750,13 @@ func assertSequentialMedia(t *testing.T, doc *ir.Document, _ []ir.Diagnostic) { assert.True(t, c.ItemEncoding.Multi, "the construct describes a repeated tail") assert.Empty(t, c.Unmodeled, "nothing is left over once it lowers") + assert.Equal(t, ir.StreamingServer, events.Streaming) + require.NotNil(t, events.ResponseStream) + require.NotNil(t, events.ResponseStream.Events) + assert.Empty(t, cmp.Diff(*c.Item, *events.ResponseStream.Events), + "the declared item schema is the stream element type") + assert.Empty(t, events.Provenance.Inferred, "a declared sequence is not a heuristic") + parts, ok := opByName(doc, "streamParts") require.True(t, ok) pc := firstContent(t, parts) diff --git a/compilers/openapi/helpers_test.go b/compilers/openapi/helpers_test.go index f1b32f18..31a1a82e 100644 --- a/compilers/openapi/helpers_test.go +++ b/compilers/openapi/helpers_test.go @@ -155,7 +155,7 @@ func newLowerer(doc *load.Document, opts Options) *lowerer { func newRawLowerer(doc *soa.OpenAPI) *lowerer { rawTypes := compile.NewTypes(0) l := &lowerer{ - ctx: lowering.New(0, doc, ir.SourceInfo{}, "", overlay.Origin{}), + ctx: lowering.New(0, doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: rawTypes.Registry()}, types: rawTypes, operationIDs: make(map[string]string), diff --git a/compilers/openapi/internal/lowering/lowering.go b/compilers/openapi/internal/lowering/lowering.go index df1d0e6e..a41935bd 100644 --- a/compilers/openapi/internal/lowering/lowering.go +++ b/compilers/openapi/internal/lowering/lowering.go @@ -44,9 +44,9 @@ type Ctx struct { // SrcIndex is this source's index within the compile, stamped into every // Provenance. SrcIndex int - // Grouping selects how operations are grouped into OperationGroups. It is the - // only policy the context carries: everything else here is a fact about the - // document, and this is a fact about the caller. + // Grouping selects how operations are grouped into OperationGroups. It is one + // of the two caller policies the context carries; everything else here is a + // fact about the document. // // It arrives as the caller wrote it, normalized or not — the compiler's // Options fills an unset one in before building a context, but nothing here @@ -55,6 +55,14 @@ type Ctx struct { // than a second spelling of the default to keep in step. Grouping GroupingStrategy + // streaming is the media-type streaming policy, normalized into the set + // MediaTypeStreams answers from, and nil when the caller disabled it. + // + // It is the second caller policy, and it is unexported where Grouping is not + // because it holds a map: a struct copy would share it, which is the one + // thing keeping the other maps here unexported is for. + streaming map[string]bool + // schemas is the set of component-schema names the document declares. // // It is unexported and read through DeclaresSchema because a struct copy @@ -99,19 +107,26 @@ type Ctx struct { // document as a valid target. It stays nil for a document that declares no // components, which reads the same as an empty set. // +// The streaming policy is normalized into its lookup set here for a related +// reason: normalizing at each reader would be as many places for the comparison +// to differ as there are readers, and a media type that matched at one of them +// and not another would classify one direction of an operation and not the +// other. +// // The $dynamicAnchor index is deliberately not derived here, though GitHub #172 // asked for it. Building it emits a diagnostic when the walk hits its bounds, so // building it is a lowering action rather than context: done at entry, that // warning would reach documents that never write $dynamicRef, changing what the // compiler reports about them. It stays where it is, built on first use. -func New(srcIndex int, doc *soa.OpenAPI, src ir.SourceInfo, grouping GroupingStrategy, origin overlay.Origin) Ctx { +func New(srcIndex int, doc *soa.OpenAPI, src ir.SourceInfo, grouping GroupingStrategy, streaming StreamingMedia, origin overlay.Origin) Ctx { return Ctx{ - Doc: doc, - Source: src, - SrcIndex: srcIndex, - Grouping: grouping, - schemas: declaredSchemaNames(doc), - overlay: origin, + Doc: doc, + Source: src, + SrcIndex: srcIndex, + Grouping: grouping, + schemas: declaredSchemaNames(doc), + streaming: streamingSet(streaming), + overlay: origin, } } diff --git a/compilers/openapi/internal/lowering/lowering_test.go b/compilers/openapi/internal/lowering/lowering_test.go index 707ac74f..55c12164 100644 --- a/compilers/openapi/internal/lowering/lowering_test.go +++ b/compilers/openapi/internal/lowering/lowering_test.go @@ -94,7 +94,7 @@ func TestNew_DerivesTheDeclaredSchemaNames(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { t.Parallel() - c := lowering.New(0, tc.doc, ir.SourceInfo{}, "", overlay.Origin{}) + c := lowering.New(0, tc.doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}) for _, n := range tc.declares { assert.True(t, c.DeclaresSchema(n), "%q is declared", n) } @@ -123,7 +123,7 @@ func TestNew_KeepsTheDocumentItWasGiven(t *testing.T) { doc := docDeclaring("User") src := ir.SourceInfo{Format: "openapi@3.1", Path: "spec.yaml", Hash: "abc"} - c := lowering.New(7, doc, src, lowering.GroupByPathPrefix, overlay.Origin{}) + c := lowering.New(7, doc, src, lowering.GroupByPathPrefix, lowering.StreamingMedia{}, overlay.Origin{}) assert.Same(t, doc, c.Doc, "the document is referenced, never copied") assert.Equal(t, src, c.Source) @@ -140,7 +140,7 @@ func TestWithAuth_ExtendsACopy(t *testing.T) { t.Parallel() doc := docDeclaring("User") src := ir.SourceInfo{Format: "openapi@3.1", Path: "spec.yaml", Hash: "abc"} - before := lowering.New(7, doc, src, lowering.GroupByPathPrefix, overlay.Origin{}) + before := lowering.New(7, doc, src, lowering.GroupByPathPrefix, lowering.StreamingMedia{}, overlay.Origin{}) schemes := map[ir.AuthID]ir.AuthScheme{"a/apiKey": {ID: "a/apiKey"}} after := before.WithAuth(schemes) @@ -203,7 +203,7 @@ func TestExclusiveBoundIsBoolean_FollowsTheDialect(t *testing.T) { for _, tc := range tests { t.Run(tc.version, func(t *testing.T) { t.Parallel() - c := lowering.New(0, &soa.OpenAPI{OpenAPI: tc.version}, ir.SourceInfo{}, "", overlay.Origin{}) + c := lowering.New(0, &soa.OpenAPI{OpenAPI: tc.version}, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}) assert.Equal(t, tc.want, c.ExclusiveBoundIsBoolean()) }) } @@ -215,7 +215,7 @@ func TestExclusiveBoundIsBoolean_FollowsTheDialect(t *testing.T) { // decides whether an internal pointer names anything. func TestRefScope_IsTheContextSeenAsAScope(t *testing.T) { t.Parallel() - c := lowering.New(0, docDeclaring("User"), ir.SourceInfo{Path: "spec.yaml"}, "", overlay.Origin{}) + c := lowering.New(0, docDeclaring("User"), ir.SourceInfo{Path: "spec.yaml"}, "", lowering.StreamingMedia{}, overlay.Origin{}) scope := c.RefScope() @@ -281,7 +281,7 @@ func TestSources_ListsTheOverlayAfterTheSourceItPatched(t *testing.T) { "overlay: 1.0.0\ninfo: {title: O, version: \"1\"}\nactions:\n"+ " - target: $.info\n update: {description: d}\n") - c := lowering.New(0, docDeclaring(), src, "", origin) + c := lowering.New(0, docDeclaring(), src, "", lowering.StreamingMedia{}, origin) require.Len(t, c.Sources(), 2) assert.Equal(t, src, c.Sources()[0], "the source being lowered comes first") @@ -296,7 +296,7 @@ func TestSources_ListsOnlyTheSourceWhenNoOverlayApplied(t *testing.T) { t.Parallel() src := ir.SourceInfo{Format: "openapi@3.1", Path: "spec.yaml"} - c := lowering.New(0, docDeclaring(), src, "", overlay.Origin{}) + c := lowering.New(0, docDeclaring(), src, "", lowering.StreamingMedia{}, overlay.Origin{}) assert.Equal(t, []ir.SourceInfo{src}, c.Sources()) } @@ -312,7 +312,7 @@ func TestProvenanceAt_NamesTheOverlayForThePositionsItIntroduced(t *testing.T) { "overlay: 1.0.0\ninfo: {title: O, version: \"1\"}\nactions:\n"+ " - target: $.info\n update: {description: d}\n") - c := lowering.New(0, docDeclaring(), ir.SourceInfo{}, "", origin) + c := lowering.New(0, docDeclaring(), ir.SourceInfo{}, "", lowering.StreamingMedia{}, origin) assert.Equal(t, ir.Provenance{Source: 1, Pointer: "/info/description"}, c.ProvenanceAt("/info/description"), "the overlay introduced this position") diff --git a/compilers/openapi/internal/lowering/streaming.go b/compilers/openapi/internal/lowering/streaming.go new file mode 100644 index 00000000..4045b6f9 --- /dev/null +++ b/compilers/openapi/internal/lowering/streaming.go @@ -0,0 +1,80 @@ +package lowering + +import "strings" + +// StreamingMediaTypeHeuristic is the name Provenance.Inferred carries on an +// operation whose streaming was read out of a media type rather than declared. +// It is a constant because the marker is what an auditor greps for, and a +// spelling written at the producing site and again at a reading test can drift. +const StreamingMediaTypeHeuristic = "streaming-media-type" + +// StreamingMedia is the media-type streaming policy: which media types mean +// "this body is a sequence of frames" in a document that declares nothing +// saying so. +// +// It is a policy rather than a table in the lowering because the reading is a +// guess (architecture principle 6). OpenAPI below 3.2 has no keyword for a +// sequential body at all, so an SSE or NDJSON API says what it does only by +// naming a media type — and a media type is a content encoding, not a promise +// about framing. The vocabulary is declared here, below both walks, and +// re-exported by the compiler's public options for the reason GroupingStrategy +// is: one declaration cannot drift from itself. +type StreamingMedia struct { + // Disabled turns the inference off. Off means off: an operation then carries + // the streaming fields a 3.2 itemSchema declares and nothing else, which is + // what a caller who does not want guesses in their IR asked for. + Disabled bool `json:"disabled,omitempty"` + // MediaTypes replaces the default list rather than extending it, so a caller + // who states a list gets exactly that list. Empty means the default. + // + // Entries are matched against the media type alone: the comparison is + // case-insensitive and ignores parameters, because `text/event-stream` and + // `text/event-stream; charset=utf-8` name one type. + MediaTypes []string `json:"mediaTypes,omitempty"` +} + +// DefaultStreamingMediaTypes is the list the policy uses when the caller states +// none. It is a default and not a standard: `text/event-stream` is the only +// registered one of the three, and the two JSON-lines spellings are conventions +// that happen to be what generators in this space already look for. A document +// using a fourth spelling is not wrong — it names its own list. +func DefaultStreamingMediaTypes() []string { + return []string{"application/jsonl", "application/x-ndjson", "text/event-stream"} +} + +// MediaTypeStreams reports whether the policy classifies mediaType as a stream +// of frames. It is a predicate rather than a getter for the reason +// DeclaresSchema is: handing back the set would make it writable through a copy +// of the context. +func (c Ctx) MediaTypeStreams(mediaType string) bool { + return c.streaming[normalizeMediaType(mediaType)] +} + +// streamingSet normalizes a policy into the set MediaTypeStreams answers from, +// or nil when the inference is off — which reads the same as an empty set, so +// no lowering has to ask whether the policy was disabled or merely empty. +func streamingSet(p StreamingMedia) map[string]bool { + if p.Disabled { + return nil + } + types := p.MediaTypes + if len(types) == 0 { + types = DefaultStreamingMediaTypes() + } + set := make(map[string]bool, len(types)) + for _, mt := range types { + if normalized := normalizeMediaType(mt); normalized != "" { + set[normalized] = true + } + } + return set +} + +// normalizeMediaType reduces a media type to the form the policy compares: +// lowercased, with any parameters dropped. +func normalizeMediaType(mediaType string) string { + if i := strings.IndexByte(mediaType, ';'); i >= 0 { + mediaType = mediaType[:i] + } + return strings.ToLower(strings.TrimSpace(mediaType)) +} diff --git a/compilers/openapi/internal/lowering/streaming_test.go b/compilers/openapi/internal/lowering/streaming_test.go new file mode 100644 index 00000000..ca0cf456 --- /dev/null +++ b/compilers/openapi/internal/lowering/streaming_test.go @@ -0,0 +1,74 @@ +package lowering_test + +import ( + "testing" + + soa "github.com/speakeasy-api/openapi/openapi" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/dexpace/morphic/compilers/openapi/internal/lowering" + "github.com/dexpace/morphic/compilers/openapi/internal/overlay" + "github.com/dexpace/morphic/ir" +) + +// streamingCtx builds a context carrying nothing but the streaming policy. +func streamingCtx(policy lowering.StreamingMedia) lowering.Ctx { + return lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", policy, overlay.Origin{}) +} + +// TestMediaTypeStreams_AnswersFromThePolicy pins every answer the policy gives, +// because each is a different decision: the default list, a caller's list +// replacing it rather than extending it, the off switch, and the normalization +// that makes one media type written two ways match once. +func TestMediaTypeStreams_AnswersFromThePolicy(t *testing.T) { + t.Parallel() + tests := []struct { + name string + policy lowering.StreamingMedia + mediaType string + want bool + }{ + {"default list", lowering.StreamingMedia{}, "text/event-stream", true}, + {"default list, ordinary type", lowering.StreamingMedia{}, "application/json", false}, + {"parameters ignored", lowering.StreamingMedia{}, "text/event-stream; charset=utf-8", true}, + {"case ignored", lowering.StreamingMedia{}, "TEXT/Event-Stream", true}, + {"surrounding space ignored", lowering.StreamingMedia{}, " text/event-stream ", true}, + {"disabled", lowering.StreamingMedia{Disabled: true}, "text/event-stream", false}, + { + "caller's list replaces the default", + lowering.StreamingMedia{MediaTypes: []string{"application/vnd.acme.frames"}}, + "text/event-stream", false, + }, + { + "caller's list is honoured", + lowering.StreamingMedia{MediaTypes: []string{"Application/VND.acme.frames"}}, + "application/vnd.acme.frames", true, + }, + { + "a blank entry names no media type", + lowering.StreamingMedia{MediaTypes: []string{" ", "text/event-stream"}}, + "", false, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tc.want, streamingCtx(tc.policy).MediaTypeStreams(tc.mediaType)) + }) + } +} + +// TestDefaultStreamingMediaTypes_AreAllRecognized holds the exported default +// list to the set the policy actually applies. Two transcriptions of one set is +// one of them going stale unnoticed, and the exported one is what a caller +// extending the list starts from. +func TestDefaultStreamingMediaTypes_AreAllRecognized(t *testing.T) { + t.Parallel() + defaults := lowering.DefaultStreamingMediaTypes() + require.NotEmpty(t, defaults, "an empty default list would make every case below vacuous") + c := streamingCtx(lowering.StreamingMedia{}) + for _, mediaType := range defaults { + assert.True(t, c.MediaTypeStreams(mediaType), "default media type %q is not recognized", mediaType) + } +} diff --git a/compilers/openapi/internal/operation/helpers_internal_test.go b/compilers/openapi/internal/operation/helpers_internal_test.go index 75bc475a..abf31cb3 100644 --- a/compilers/openapi/internal/operation/helpers_internal_test.go +++ b/compilers/openapi/internal/operation/helpers_internal_test.go @@ -44,7 +44,7 @@ func loweredFor(t *testing.T, src string) (*lowerer, []ir.Diagnostic) { require.NotNil(t, loadedDoc, "load returned no document: %+v", diags) types := compile.NewTypes(0) return &lowerer{ - ctx: lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, overlay.Origin{}), + ctx: lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: types.Registry()}, types: types, operationIDs: make(map[string]string), @@ -56,7 +56,7 @@ func loweredFor(t *testing.T, src string) (*lowerer, []ir.Diagnostic) { func newRawLowerer(doc *soa.OpenAPI) *lowerer { types := compile.NewTypes(0) return &lowerer{ - ctx: lowering.New(0, doc, ir.SourceInfo{}, "", overlay.Origin{}), + ctx: lowering.New(0, doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: types.Registry()}, types: types, operationIDs: make(map[string]string), diff --git a/compilers/openapi/internal/operation/helpers_test.go b/compilers/openapi/internal/operation/helpers_test.go index b8a0254c..bbb6d900 100644 --- a/compilers/openapi/internal/operation/helpers_test.go +++ b/compilers/openapi/internal/operation/helpers_test.go @@ -200,7 +200,7 @@ func serviceWithGrouping(t *testing.T, src string, grouping lowering.GroupingStr require.NotNil(t, loadedDoc) types := compile.NewTypes(0) - c := lowering.New(0, loadedDoc.Doc, loadedDoc.Source, grouping, overlay.Origin{}) + c := lowering.New(0, loadedDoc.Doc, loadedDoc.Source, grouping, lowering.StreamingMedia{}, overlay.Origin{}) var anchors schema.AnchorIndex var acc compile.Diags acc.AppendAll(schema.LowerComponentSchemas(c, types, &anchors)) diff --git a/compilers/openapi/internal/operation/operations.go b/compilers/openapi/internal/operation/operations.go index 351fb057..fd6f16c8 100644 --- a/compilers/openapi/internal/operation/operations.go +++ b/compilers/openapi/internal/operation/operations.go @@ -246,10 +246,11 @@ type opContext struct { // registered alongside it in the same group (ir-design §7.2, §8.1). func lowerOperation(c lowering.Ctx, ts *compile.Types, anchors *schema.AnchorIndex, operationIDs map[string]string, src *soa.Operation, opCtx opContext) (ir.Operation, []ir.Operation, []ir.Diagnostic) { mount, decl := opCtx.ptrs.mount, opCtx.ptrs.decl - // Built through the context so the source index is spelled in one place, then - // marked inferred — the one provenance in this compiler that is. + // Built through the context so the source index is spelled in one place. Its + // heuristic marker is filled in below, once every lowering that can add one + // has run: the grouping choice is known here, but the streaming reading is + // not known until the payloads are lowered. opProv := c.ProvenanceAt(decl) - opProv.Inferred = opCtx.inferred opAuth, diags := auth.LowerSecurityRequirements(c, src.Security, decl) op := ir.Operation{ ID: ids.Op(mount), @@ -280,6 +281,10 @@ func lowerOperation(c lowering.Ctx, ts *compile.Types, anchors *schema.AnchorInd ParamBindings: bindings, } diags = append(diags, lowerRequestBody(c, ts, anchors, &op, &hb, src, decl)...) + // After both payload lowerings, which are what it reads. + streaming, streamDiags := applyStreaming(c, &op, decl) + diags = append(diags, streamDiags...) + op.Provenance.Inferred = joinInferred(opCtx.inferred, streaming) var extra []ir.Operation if opCtx.withCallbacks { var cbDiags []ir.Diagnostic diff --git a/compilers/openapi/internal/operation/streaming.go b/compilers/openapi/internal/operation/streaming.go new file mode 100644 index 00000000..eb348273 --- /dev/null +++ b/compilers/openapi/internal/operation/streaming.go @@ -0,0 +1,161 @@ +package operation + +import ( + "strings" + + "github.com/dexpace/morphic/compilers/openapi/internal/diag" + "github.com/dexpace/morphic/compilers/openapi/internal/ids" + "github.com/dexpace/morphic/compilers/openapi/internal/lowering" + "github.com/dexpace/morphic/ir" +) + +// streamCandidate is one lowered content that carries a stream, and how the +// compiler came to say so. declared distinguishes a 3.2 itemSchema — which +// states outright that the body is a sequence — from a media type the policy +// recognizes, which is a guess about what the media type implies. +type streamCandidate struct { + events ir.TypeRef + declared bool +} + +// streamDirection is one direction's classification: the detail to write, and +// the two facts about how it was reached that the caller has to act on. +type streamDirection struct { + detail *ir.StreamDetail + // inferred reports that at least one contributing content was recognized by + // media type rather than by a declaration, which is what the operation's + // Provenance.Inferred marker records. + inferred bool + // ambiguous reports that the direction had more than one streaming content, + // so no element type was elected. + ambiguous bool +} + +// applyStreaming writes the operation's streaming summary and per-direction +// details, and returns the heuristic marker its provenance should carry ("" when +// the streams it found were all declared). +// +// It runs over the lowered operation rather than the source, so it reads one +// answer per direction no matter how many places a payload was assembled from, +// and it is the only writer of the three streaming fields. +func applyStreaming(c lowering.Ctx, op *ir.Operation, declPtr string) (string, []ir.Diagnostic) { + request := classifyStream(streamCandidates(c, op.Request)) + response := classifyStream(responseCandidates(c, op.Responses)) + if request.detail == nil && response.detail == nil { + return "", nil + } + op.RequestStream, op.ResponseStream = request.detail, response.detail + op.Streaming = streamingMode(request.detail != nil, response.detail != nil) + + var diags []ir.Diagnostic + if request.ambiguous { + diags = append(diags, unelectedElementDiag(c, declPtr+ids.Ptr("requestBody"), "request")) + } + if response.ambiguous { + diags = append(diags, unelectedElementDiag(c, declPtr+ids.Ptr("responses"), "response")) + } + if request.inferred || response.inferred { + return lowering.StreamingMediaTypeHeuristic, diags + } + return "", diags +} + +// classifyStream folds one direction's candidates into the detail to write. +// +// The one thing it refuses to do is elect an element type from several. +// StreamDetail holds one Events per direction while Payload keeps every media +// type, so naming one of two streaming contents would be exactly the +// primary-content selection a compiler must not make (invariant 2). The +// direction still streams — that much all the candidates agree on — and the +// element is left unnamed for a lowering that has the whole set to choose from. +func classifyStream(candidates []streamCandidate) streamDirection { + if len(candidates) == 0 { + return streamDirection{} + } + out := streamDirection{detail: &ir.StreamDetail{}} + for _, candidate := range candidates { + if !candidate.declared { + out.inferred = true + } + } + if len(candidates) > 1 { + out.ambiguous = true + return out + } + events := candidates[0].events + out.detail.Events = &events + return out +} + +// responseCandidates collects the streaming contents of every success response. +// Error responses are deliberately not read: an operation's ResponseStream +// describes what it streams back, and a 4xx body is what it sends instead of +// streaming. +func responseCandidates(c lowering.Ctx, responses []ir.Response) []streamCandidate { + out := make([]streamCandidate, 0, len(responses)) + for i := range responses { + out = append(out, streamCandidates(c, responses[i].Payload)...) + } + return out +} + +// streamCandidates collects the contents of one payload that carry a stream. A +// declared itemSchema wins over the media-type reading at the same content, so +// the two can never both classify it. +func streamCandidates(c lowering.Ctx, payload *ir.Payload) []streamCandidate { + if payload == nil { + return nil + } + var out []streamCandidate + for _, content := range payload.Contents { + switch { + case content.Item != nil: + out = append(out, streamCandidate{events: *content.Item, declared: true}) + case c.MediaTypeStreams(content.MediaType): + // For a frame format the schema under the media type describes one + // frame, not the whole body — the opposite of how an ordinary content + // is read — so it is the element type as well as the content type. + out = append(out, streamCandidate{events: content.Type}) + } + } + return out +} + +// streamingMode is the summary the two directions derive. It is only ever asked +// when at least one of them streams. +func streamingMode(request, response bool) ir.StreamingMode { + switch { + case request && response: + return ir.StreamingBidi + case request: + return ir.StreamingClient + default: + return ir.StreamingServer + } +} + +// unelectedElementDiag reports a direction whose element type was left unnamed +// because several of its contents stream. Nothing is lost — every content is +// still on the payload — but the IR now says less than the source did, which is +// what makes it a degradation rather than a silent choice. +func unelectedElementDiag(c lowering.Ctx, pointer, direction string) ir.Diagnostic { + return c.DiagAt(ir.SeverityInfo, diag.DegradedConstruct, pointer, + "several %s media types stream, so the stream element type is left unnamed rather than electing one", direction) +} + +// joinInferred names every heuristic that shaped one node, in the order the +// lowering applied them. +// +// Provenance.Inferred holds one string and more than one heuristic can reach a +// single operation — grouping by path prefix and reading a stream out of a media +// type are independent choices. Keeping only the first would make the second +// invisible, which defeats the marker's whole purpose, so they are listed. +func joinInferred(markers ...string) string { + kept := make([]string, 0, len(markers)) + for _, marker := range markers { + if marker != "" { + kept = append(kept, marker) + } + } + return strings.Join(kept, ",") +} diff --git a/compilers/openapi/internal/operation/streaming_test.go b/compilers/openapi/internal/operation/streaming_test.go new file mode 100644 index 00000000..20d9b73b --- /dev/null +++ b/compilers/openapi/internal/operation/streaming_test.go @@ -0,0 +1,266 @@ +package operation_test + +import ( + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/dexpace/morphic/compilers" + "github.com/dexpace/morphic/compilers/openapi" + "github.com/dexpace/morphic/compilers/openapi/internal/diag" + "github.com/dexpace/morphic/ir" +) + +// streamingSpec compiles src under a chosen streaming policy. It routes through +// the compiler's public options because the policy is a caller's choice and the +// projection onto the lowering context is what carries it here. +func streamingSpec(t *testing.T, src string, opts openapi.Options) (*ir.Document, []ir.Diagnostic) { + t.Helper() + doc, diags, err := openapi.New().Compile(t.Context(), []compilers.Source{sourceOf(src)}, + compilers.Options{FormatOptions: opts}) + require.NoError(t, err) + require.NotNil(t, doc) + requireNoErrorDiags(t, diags) + return doc, diags +} + +// eventStreamSpec is one GET whose single 200 response declares mediaType with +// an inline object schema. +func eventStreamSpec(version, mediaType string) string { + return pathsSpecVer(version, ` /events: + get: + operationId: getEvents + responses: + "200": + description: stream + content: + `+mediaType+`: + schema: {type: object, properties: {msg: {type: string}}} +`) +} + +// firstResponseContent returns the operation's single success-response content. +func firstResponseContent(t *testing.T, op ir.Operation) ir.Content { + t.Helper() + require.Len(t, op.Responses, 1) + require.NotNil(t, op.Responses[0].Payload) + require.NotEmpty(t, op.Responses[0].Payload.Contents) + return op.Responses[0].Payload.Contents[0] +} + +// TestStreaming_ResponseMediaTypeImpliesServerStream is the probe from +// GitHub #250: a response whose only media type is one of the streaming +// spellings carries a streaming signal, at every version rather than only at +// 3.2, and the element type is the schema under the media type — for a frame +// format that schema describes one frame, not the whole body. +func TestStreaming_ResponseMediaTypeImpliesServerStream(t *testing.T) { + t.Parallel() + for _, version := range []string{"3.0.3", "3.1.0", "3.2.0"} { + for _, mediaType := range openapi.DefaultStreamingMediaTypes() { + t.Run(version+" "+mediaType, func(t *testing.T) { + t.Parallel() + doc, _ := streamingSpec(t, eventStreamSpec(version, mediaType), openapi.Options{}) + op := findOp(t, doc, "getEvents") + + assert.Equal(t, ir.StreamingServer, op.Streaming) + require.NotNil(t, op.ResponseStream, "a streaming media type populates the response direction") + assert.Nil(t, op.RequestStream, "nothing streams towards the server") + + content := firstResponseContent(t, op) + require.NotNil(t, op.ResponseStream.Events, "the frame schema is the stream element type") + assert.Empty(t, cmp.Diff(content.Type, *op.ResponseStream.Events)) + assert.Equal(t, mediaType, content.MediaType, "the media type is still kept as declared") + assert.Equal(t, "streaming-media-type", op.Provenance.Inferred, + "classifying a media type is a heuristic and says so") + }) + } + } +} + +// TestStreaming_MediaTypeParametersDoNotDefeatTheMatch pins that the match is +// against the media type itself: a charset is a parameter of the same type, and +// a document that writes one still declares a stream. +func TestStreaming_MediaTypeParametersDoNotDefeatTheMatch(t *testing.T) { + t.Parallel() + doc, _ := streamingSpec(t, eventStreamSpec("3.1.0", "text/event-stream; charset=utf-8"), openapi.Options{}) + op := findOp(t, doc, "getEvents") + assert.Equal(t, ir.StreamingServer, op.Streaming) + assert.NotNil(t, op.ResponseStream) +} + +// TestStreaming_RequestMediaTypeImpliesClientStream pins the other direction, +// and the two together pin bidi: the summary is derived from both rather than +// from whichever was noticed first. +func TestStreaming_RequestMediaTypeImpliesClientStream(t *testing.T) { + t.Parallel() + tests := []struct { + name string + responseType string + want ir.StreamingMode + wantResponse bool + }{ + {"client only", "application/json", ir.StreamingClient, false}, + {"both directions", "text/event-stream", ir.StreamingBidi, true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + spec := pathsSpec(` /ingest: + post: + operationId: ingest + requestBody: + content: + application/x-ndjson: + schema: {type: object, properties: {row: {type: string}}} + responses: + "200": + description: ok + content: + ` + tc.responseType + `: + schema: {type: object, properties: {msg: {type: string}}} +`) + doc, _ := streamingSpec(t, spec, openapi.Options{}) + op := findOp(t, doc, "ingest") + assert.Equal(t, tc.want, op.Streaming) + assert.Equal(t, tc.wantResponse, op.ResponseStream != nil, "response direction") + require.NotNil(t, op.RequestStream, "the request body's media type streams") + require.NotNil(t, op.RequestStream.Events) + require.NotNil(t, op.Request) + assert.Empty(t, cmp.Diff(op.Request.Contents[0].Type, *op.RequestStream.Events)) + }) + } +} + +// TestStreaming_ItemSchemaDeclaresTheStream pins that a 3.2 itemSchema is a +// declaration and the media-type list is a guess, so the two cannot both fire: +// the declared element type is the one Events names, and no heuristic is +// stamped on an operation that declared what it does. +func TestStreaming_ItemSchemaDeclaresTheStream(t *testing.T) { + t.Parallel() + spec := pathsSpecVer("3.2.0", ` /events: + get: + operationId: getEvents + responses: + "200": + description: stream + content: + text/event-stream: + schema: {type: object, properties: {envelope: {type: string}}} + itemSchema: {type: object, properties: {msg: {type: string}}} +`) + doc, _ := streamingSpec(t, spec, openapi.Options{}) + op := findOp(t, doc, "getEvents") + assert.Equal(t, ir.StreamingServer, op.Streaming) + require.NotNil(t, op.ResponseStream) + require.NotNil(t, op.ResponseStream.Events) + + content := firstResponseContent(t, op) + require.NotNil(t, content.Item) + assert.Empty(t, cmp.Diff(*content.Item, *op.ResponseStream.Events), + "the declared item schema is the element type, not the media type's own schema") + assert.Empty(t, op.Provenance.Inferred, "a declared stream is not a heuristic") +} + +// TestStreaming_SeveralStreamingContentsLeaveTheElementUnnamed pins the one +// place this classification refuses to answer, in both directions. +// StreamDetail holds one element type per direction while Payload keeps every +// media type, so naming one of two streaming contents would be the +// primary-content selection invariant 2 forbids a compiler to make. The +// direction still streams, and the refusal is reported rather than silent. +func TestStreaming_SeveralStreamingContentsLeaveTheElementUnnamed(t *testing.T) { + t.Parallel() + spec := pathsSpec(` /events: + post: + operationId: exchange + requestBody: + required: true + content: + application/x-ndjson: + schema: {type: object, properties: {a: {type: string}}} + application/jsonl: + schema: {type: object, properties: {b: {type: string}}} + responses: + "200": + description: stream + content: + text/event-stream: + schema: {type: object, properties: {c: {type: string}}} + application/x-ndjson: + schema: {type: object, properties: {d: {type: string}}} +`) + doc, diags := streamingSpec(t, spec, openapi.Options{}) + op := findOp(t, doc, "exchange") + assert.Equal(t, ir.StreamingBidi, op.Streaming) + require.NotNil(t, op.RequestStream) + require.NotNil(t, op.ResponseStream) + assert.Nil(t, op.RequestStream.Events, "two candidate element types, so none is elected") + assert.Nil(t, op.ResponseStream.Events, "two candidate element types, so none is elected") + require.NotNil(t, op.Request) + assert.Len(t, op.Request.Contents, 2, "every content is still kept") + + assert.Equal(t, 2, countDiagsAt(diags, diag.DegradedConstruct, ir.SeverityInfo), + "one report per direction; got %+v", diags) + for _, pointer := range []string{"/paths/~1events/post/requestBody", "/paths/~1events/post/responses"} { + assert.True(t, hasDiagCodeAt(diags, diag.DegradedConstruct, pointer), + "the refusal names the direction it was made in: %s", pointer) + } +} + +// TestStreaming_OrdinaryContentDoesNotStream is the negative half: without a +// streaming media type the operation core keeps its zero values, so every spec +// that does not stream is untouched by this classification. +func TestStreaming_OrdinaryContentDoesNotStream(t *testing.T) { + t.Parallel() + doc, _ := streamingSpec(t, eventStreamSpec("3.1.0", "application/json"), openapi.Options{}) + op := findOp(t, doc, "getEvents") + assert.Empty(t, string(op.Streaming)) + assert.Nil(t, op.RequestStream) + assert.Nil(t, op.ResponseStream) + assert.Empty(t, op.Provenance.Inferred) +} + +// TestStreaming_ContentOrderDoesNotChooseAnElement is the two-order oracle for +// the one place a choice was available. Compiled with the two streaming media +// types declared either way round, the operation's streaming fields have to +// agree — electing the first candidate would pass a single-order test and +// produce a different element type here. +func TestStreaming_ContentOrderDoesNotChooseAnElement(t *testing.T) { + t.Parallel() + spec := func(first, second string) string { + return pathsSpec(` /events: + get: + operationId: getEvents + responses: + "200": + description: stream + content: + ` + first + `: + schema: {type: object, properties: {a: {type: string}}} + ` + second + `: + schema: {type: object, properties: {b: {type: string}}} +`) + } + forward, _ := streamingSpec(t, spec("text/event-stream", "application/x-ndjson"), openapi.Options{}) + reverse, _ := streamingSpec(t, spec("application/x-ndjson", "text/event-stream"), openapi.Options{}) + + first := findOp(t, forward, "getEvents") + second := findOp(t, reverse, "getEvents") + assert.Equal(t, first.Streaming, second.Streaming) + assert.Empty(t, cmp.Diff(first.RequestStream, second.RequestStream)) + assert.Empty(t, cmp.Diff(first.ResponseStream, second.ResponseStream), + "the element type must not depend on which media type was written first") +} + +// TestStreaming_MarkerListsEveryHeuristicThatApplied pins that two heuristics +// reaching one operation both survive. Provenance.Inferred holds one string, so +// the streaming marker overwriting the grouping one — or being dropped by it — +// would leave an audit reading the IR believing only one guess was made. +func TestStreaming_MarkerListsEveryHeuristicThatApplied(t *testing.T) { + t.Parallel() + byPrefix := openapi.Options{Grouping: openapi.GroupByPathPrefix} + doc, _ := streamingSpec(t, eventStreamSpec("3.1.0", "text/event-stream"), byPrefix) + op := findOp(t, doc, "getEvents") + assert.Equal(t, "group-path-prefix,streaming-media-type", op.Provenance.Inferred) +} diff --git a/compilers/openapi/internal/schema/compose_internal_test.go b/compilers/openapi/internal/schema/compose_internal_test.go index 75046e87..a47b6a8b 100644 --- a/compilers/openapi/internal/schema/compose_internal_test.go +++ b/compilers/openapi/internal/schema/compose_internal_test.go @@ -36,7 +36,7 @@ func TestRefLastSegment(t *testing.T) { func TestMappingTargetID(t *testing.T) { t.Parallel() l := &lowerer{ - ctx: lowering.New(0, docDeclaring("Cat", "Dog", "A/B"), ir.SourceInfo{}, "", overlay.Origin{}), + ctx: lowering.New(0, docDeclaring("Cat", "Dog", "A/B"), ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: ir.TypeRegistry{}}, } // A $ref to a declared component. @@ -63,7 +63,7 @@ func TestMappingTargetID(t *testing.T) { // (issue #14, f31). It gets a context of its own rather than being added to the // one above: the declared set is derived from the document now, so saying "and // also this one" means saying it to a document. - empty := lowering.New(0, docDeclaring(""), ir.SourceInfo{}, "", overlay.Origin{}) + empty := lowering.New(0, docDeclaring(""), ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}) id, ok = mappingTargetID(empty, l.types, "") require.True(t, ok) assert.Equal(t, ids.AnonType(ids.Ptr("components", "schemas", "")), id) diff --git a/compilers/openapi/internal/schema/helpers_internal_test.go b/compilers/openapi/internal/schema/helpers_internal_test.go index 8cbc6c17..96ff5cd4 100644 --- a/compilers/openapi/internal/schema/helpers_internal_test.go +++ b/compilers/openapi/internal/schema/helpers_internal_test.go @@ -91,7 +91,7 @@ func loweredFor(t *testing.T, src string) (*lowerer, []ir.Diagnostic) { require.NotNil(t, loadedDoc, "load returned no document: %+v", diags) types := compile.NewTypes(0) return &lowerer{ - ctx: lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, overlay.Origin{}), + ctx: lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: types.Registry()}, types: types, }, diags @@ -111,7 +111,7 @@ func lowerSpec(t *testing.T, src string) (*ir.Document, []ir.Diagnostic) { func newRawLowerer(doc *soa.OpenAPI) *lowerer { types := compile.NewTypes(0) return &lowerer{ - ctx: lowering.New(0, doc, ir.SourceInfo{}, "", overlay.Origin{}), + ctx: lowering.New(0, doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: types.Registry()}, types: types, } diff --git a/compilers/openapi/internal/schema/helpers_test.go b/compilers/openapi/internal/schema/helpers_test.go index 970699ed..2b339de0 100644 --- a/compilers/openapi/internal/schema/helpers_test.go +++ b/compilers/openapi/internal/schema/helpers_test.go @@ -67,7 +67,7 @@ func loweredFor(t *testing.T, src string) (*lowerer, []ir.Diagnostic) { require.NotNil(t, loadedDoc, "load returned no document: %+v", diags) types := compile.NewTypes(0) return &lowerer{ - ctx: lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, overlay.Origin{}), + ctx: lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: types.Registry()}, types: types, }, diags @@ -87,7 +87,7 @@ func lowerSpec(t *testing.T, src string) (*ir.Document, []ir.Diagnostic) { func newRawLowerer(doc *soa.OpenAPI) *lowerer { types := compile.NewTypes(0) return &lowerer{ - ctx: lowering.New(0, doc, ir.SourceInfo{}, "", overlay.Origin{}), + ctx: lowering.New(0, doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}), out: &ir.Document{Types: types.Registry()}, types: types, } diff --git a/compilers/openapi/internal/schema/schema_test.go b/compilers/openapi/internal/schema/schema_test.go index f4b1c910..0499574f 100644 --- a/compilers/openapi/internal/schema/schema_test.go +++ b/compilers/openapi/internal/schema/schema_test.go @@ -3225,7 +3225,7 @@ func TestDynamicRef_NonScalarValueIsKeptNotExpanded(t *testing.T) { // prototype changes, so a site that fills in a name or a description keeps it. func TestAppendExample_ConvertsAndAppends(t *testing.T) { t.Parallel() - c := lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", overlay.Origin{}) + c := lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}) proto := ir.Example{Name: "n", Summary: "s", Description: "d"} out, diags := schema.AppendExample(c, nil, proto, strNode("hello"), "/p", "examples", "n") @@ -3243,7 +3243,7 @@ func TestAppendExample_ConvertsAndAppends(t *testing.T) { // that joins them, so a wrong join shows up nowhere else. func TestAppendExample_UnconvertibleValueIsReported(t *testing.T) { t.Parallel() - c := lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", overlay.Origin{}) + c := lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}) nan := &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!float", Value: ".nan"} out, diags := schema.AppendExample(c, nil, ir.Example{}, nan, "/p", "examples", "n") @@ -3260,7 +3260,7 @@ func TestAppendExample_UnconvertibleValueIsReported(t *testing.T) { // it, not at the position that declared it. func TestStampConstraintDiags_RelocatesEveryDiagnosticToTheReadingPointer(t *testing.T) { t.Parallel() - c := lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", overlay.Origin{}) + c := lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}) in := []ir.Diagnostic{ {Code: diag.DegradedConstruct, Provenance: ir.Provenance{Pointer: "/elsewhere"}}, {Code: diag.NumericPrecision, Provenance: ir.Provenance{Source: 9, Pointer: "/other"}}, diff --git a/compilers/openapi/openapi.go b/compilers/openapi/openapi.go index b1feb110..a4c02d29 100644 --- a/compilers/openapi/openapi.go +++ b/compilers/openapi/openapi.go @@ -158,5 +158,5 @@ func loadOptions(o Options) load.Options { // place the loader's result type meets the lowering, so lowering.New can take // the two facts it needs rather than the loader's struct. func loweringCtx(doc *load.Document, o Options) lowering.Ctx { - return lowering.New(rootSrcIndex, doc.Doc, doc.Source, o.Grouping, doc.Overlay) + return lowering.New(rootSrcIndex, doc.Doc, doc.Source, o.Grouping, o.StreamingMedia, doc.Overlay) } diff --git a/compilers/openapi/options.go b/compilers/openapi/options.go index 35c6b812..4e6adf7b 100644 --- a/compilers/openapi/options.go +++ b/compilers/openapi/options.go @@ -19,6 +19,17 @@ const ( GroupByPathPrefix = lowering.GroupByPathPrefix ) +// StreamingMedia is the media-type streaming policy: which media types imply +// that a body is a sequence of frames when the document declares nothing that +// says so. It is the second injectable-policy seam (architecture principle 6), +// and it is named here rather than restated for the reason GroupingStrategy is. +type StreamingMedia = lowering.StreamingMedia + +// DefaultStreamingMediaTypes returns the media types StreamingMedia classifies +// as streams when the caller names none. It is exported so a caller extending +// the list can start from it rather than transcribe it. +func DefaultStreamingMediaTypes() []string { return lowering.DefaultStreamingMediaTypes() } + // Options configures the OpenAPI compiler. It is the concrete type this // compiler expects in compilers.Options.FormatOptions; the zero value is valid // and normalized by withDefaults. @@ -30,6 +41,10 @@ const ( type Options struct { // Grouping selects the operation-grouping strategy. Grouping GroupingStrategy `json:"grouping,omitempty"` + // StreamingMedia selects which media types imply a stream. The zero value is + // the default list, on; a caller who wants only what a document declares + // disables it. + StreamingMedia StreamingMedia `json:"streamingMedia"` // AllowExternalRefs lets reference resolution leave the source document — // reading files off disk and fetching http(s) URLs. Off by default, because // compilers.Source is the whole input ("the caller loads bytes so compilation diff --git a/compilers/openapi/streaming_test.go b/compilers/openapi/streaming_test.go new file mode 100644 index 00000000..59a98e7f --- /dev/null +++ b/compilers/openapi/streaming_test.go @@ -0,0 +1,168 @@ +// This file holds the corpus rows for media-type streaming and the public +// option surface that switches it, which no single source file owns: the policy +// lives on the compiler's options, the media types are read where content is +// lowered, and the result is written onto the operation core. +package openapi_test // external test package — exercises only the public API + +import ( + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/dexpace/morphic/compilers" + "github.com/dexpace/morphic/compilers/openapi" + "github.com/dexpace/morphic/ir" +) + +// eventStreamSpec is one GET whose single 200 response declares mediaType with +// an inline object schema. +func eventStreamSpec(mediaType string) string { + return `openapi: 3.1.0 +info: {title: T, version: "1"} +paths: + /events: + get: + operationId: getEvents + responses: + "200": + description: stream + content: + ` + mediaType + `: + schema: {type: object, properties: {msg: {type: string}}} +` +} + +// compileStreamingSpec compiles src with opts and requires no error diagnostic. +func compileStreamingSpec(t *testing.T, src string, opts openapi.Options) *ir.Document { + t.Helper() + doc, diags, err := openapi.New().Compile(t.Context(), + []compilers.Source{{Path: "spec.yaml", Data: []byte(src)}}, + compilers.Options{FormatOptions: opts}) + require.NoError(t, err) + require.NotNil(t, doc) + assertNoErrorDiags(t, diags) + return doc +} + +// TestStreaming_PolicyDisabledLeavesOnlyWhatIsDeclared pins the off switch +// invariant 6 requires, through the option a caller actually sets. Disabled +// means the media-type reading stops entirely, while a 3.2 itemSchema — a +// declaration, not a guess — still populates the fields it always did. +func TestStreaming_PolicyDisabledLeavesOnlyWhatIsDeclared(t *testing.T) { + t.Parallel() + off := openapi.Options{StreamingMedia: openapi.StreamingMedia{Disabled: true}} + + doc := compileStreamingSpec(t, eventStreamSpec("text/event-stream"), off) + op, ok := opByName(doc, "getEvents") + require.True(t, ok) + assert.Empty(t, string(op.Streaming), "the media-type reading is off") + assert.Nil(t, op.ResponseStream) + assert.Empty(t, op.Provenance.Inferred) + + declared := `openapi: 3.2.0 +info: {title: T, version: "1"} +paths: + /events: + get: + operationId: getEvents + responses: + "200": + description: stream + content: + text/event-stream: + itemSchema: {type: object, properties: {msg: {type: string}}} +` + doc = compileStreamingSpec(t, declared, off) + op, ok = opByName(doc, "getEvents") + require.True(t, ok) + assert.Equal(t, ir.StreamingServer, op.Streaming, "disabling a heuristic does not disable a declaration") + assert.NotNil(t, op.ResponseStream) +} + +// TestStreaming_PolicyMediaTypesReplaceTheDefaults pins that a stated list is +// the whole list. A caller who names their own spelling gets it and does not +// silently keep the defaults beside it — which is the difference between a +// default and a standard. +func TestStreaming_PolicyMediaTypesReplaceTheDefaults(t *testing.T) { + t.Parallel() + own := openapi.Options{StreamingMedia: openapi.StreamingMedia{ + MediaTypes: []string{"application/vnd.acme.frames"}, + }} + tests := []struct { + mediaType string + want ir.StreamingMode + }{ + {"application/vnd.acme.frames", ir.StreamingServer}, + {"text/event-stream", ""}, + } + for _, tc := range tests { + t.Run(tc.mediaType, func(t *testing.T) { + t.Parallel() + doc := compileStreamingSpec(t, eventStreamSpec(tc.mediaType), own) + op, ok := opByName(doc, "getEvents") + require.True(t, ok) + assert.Equal(t, tc.want, op.Streaming) + }) + } +} + +// TestStreaming_DefaultMediaTypesAreTheOnesApplied holds the list a caller can +// start from to the list the compiler actually applies. It is asserted through +// a compile rather than against the policy's own set, because the exported +// accessor and the set the lowering reads are two things a caller has no way to +// tell apart until one of them goes stale. +func TestStreaming_DefaultMediaTypesAreTheOnesApplied(t *testing.T) { + t.Parallel() + defaults := openapi.DefaultStreamingMediaTypes() + require.NotEmpty(t, defaults, "an empty list would make this vacuous") + for _, mediaType := range defaults { + doc := compileStreamingSpec(t, eventStreamSpec(mediaType), openapi.Options{}) + op, ok := opByName(doc, "getEvents") + require.True(t, ok) + assert.Equal(t, ir.StreamingServer, op.Streaming, "default media type %q does not stream", mediaType) + } +} + +// assertStreamingMedia30 is the corpus row for a 3.0 document that says it +// streams only by naming a media type — the version the IR carried no streaming +// signal for at all (GitHub #250). +func assertStreamingMedia30(t *testing.T, doc *ir.Document, _ []ir.Diagnostic) { + op, ok := opByName(doc, "streamEvents") + require.True(t, ok) + assert.Equal(t, ir.StreamingServer, op.Streaming) + require.NotNil(t, op.ResponseStream) + require.NotNil(t, op.ResponseStream.Events) + content := firstContent(t, op) + assert.Empty(t, cmp.Diff(content.Type, *op.ResponseStream.Events)) + assert.Equal(t, "text/event-stream", content.MediaType, "the media type itself is still kept as declared") + assert.Equal(t, "streaming-media-type", op.Provenance.Inferred) +} + +// assertStreamingMedia31 is the corpus row for the two shapes 3.0 cannot show +// on one operation: a request body that streams as well as its response, and a +// response offering two streaming media types, where the element type is left +// unnamed rather than elected. +func assertStreamingMedia31(t *testing.T, doc *ir.Document, diags []ir.Diagnostic) { + both, ok := opByName(doc, "ingestRows") + require.True(t, ok) + assert.Equal(t, ir.StreamingBidi, both.Streaming) + require.NotNil(t, both.RequestStream) + require.NotNil(t, both.RequestStream.Events) + require.NotNil(t, both.Request) + assert.Empty(t, cmp.Diff(both.Request.Contents[0].Type, *both.RequestStream.Events)) + require.NotNil(t, both.ResponseStream) + assert.NotNil(t, both.ResponseStream.Events, "a charset parameter does not defeat the match") + assert.Equal(t, "streaming-media-type", both.Provenance.Inferred) + + either, ok := opByName(doc, "streamEither") + require.True(t, ok) + assert.Equal(t, ir.StreamingServer, either.Streaming) + require.NotNil(t, either.ResponseStream) + assert.Nil(t, either.ResponseStream.Events, "two streaming contents elect no element type") + require.NotNil(t, either.Responses[0].Payload) + assert.Len(t, either.Responses[0].Payload.Contents, 2, "both contents are still kept") + assert.True(t, hasDiagCode(diags, "openapi/degraded-construct"), + "the unelected element type is reported; got %+v", diags) +} diff --git a/testdata/conformance/openapi/sequential-media.golden.json b/testdata/conformance/openapi/sequential-media.golden.json index 30d7f87f..6870e6c8 100644 --- a/testdata/conformance/openapi/sequential-media.golden.json +++ b/testdata/conformance/openapi/sequential-media.golden.json @@ -66,6 +66,14 @@ } ], "oneWay": false, + "streaming": "server", + "responseStream": { + "events": { + "target": "t/anon/paths/~1events/get/responses/200/content/multipart~1mixed/itemSchema", + "nullable": false + }, + "requiresLength": false + }, "idempotency": {}, "auth": null, "bindings": { @@ -149,6 +157,14 @@ } ], "oneWay": false, + "streaming": "server", + "responseStream": { + "events": { + "target": "t/anon/paths/~1parts/get/responses/200/content/multipart~1mixed/itemSchema", + "nullable": false + }, + "requiresLength": false + }, "idempotency": {}, "auth": null, "bindings": { diff --git a/testdata/conformance/openapi/streaming-media-30.golden.json b/testdata/conformance/openapi/streaming-media-30.golden.json new file mode 100644 index 00000000..7217798f --- /dev/null +++ b/testdata/conformance/openapi/streaming-media-30.golden.json @@ -0,0 +1,171 @@ +{ + "irVersion": "0.3.0", + "name": "StreamingMedia30", + "version": "1.0.0", + "docs": {}, + "services": [ + { + "id": "s/openapi/0", + "name": { + "source": "StreamingMedia30", + "canonical": "streaming_media_30" + }, + "docs": {}, + "groups": [ + { + "name": { + "hint": "default" + }, + "docs": {}, + "operations": [ + { + "id": "op/openapi/paths/~1events/get", + "name": { + "source": "streamEvents", + "canonical": "stream_events" + }, + "docs": {}, + "responses": [ + { + "name": { + "hint": "200" + }, + "conditions": { + "statusCodes": [ + { + "from": 200, + "to": 200 + } + ] + }, + "payload": { + "contents": [ + { + "mediaType": "text/event-stream", + "type": { + "target": "t/anon/paths/~1events/get/responses/200/content/text~1event-stream/schema", + "nullable": false + } + } + ] + }, + "docs": { + "description": "ok" + } + } + ], + "oneWay": false, + "streaming": "server", + "responseStream": { + "events": { + "target": "t/anon/paths/~1events/get/responses/200/content/text~1event-stream/schema", + "nullable": false + }, + "requiresLength": false + }, + "idempotency": {}, + "auth": null, + "bindings": { + "http": [ + { + "method": "GET", + "uriTemplate": "/events", + "sharedRoute": false, + "checksumRequired": false, + "isWebhook": false + } + ] + }, + "provenance": { + "source": 0, + "pointer": "/paths/~1events/get", + "inferred": "streaming-media-type" + } + } + ] + } + ], + "auth": null, + "provenance": { + "source": 0 + } + } + ], + "types": { + "t/anon/paths/~1events/get/responses/200/content/text~1event-stream/schema": { + "kind": "model", + "id": "t/anon/paths/~1events/get/responses/200/content/text~1event-stream/schema", + "name": { + "hint": "response" + }, + "anonymous": true, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0, + "pointer": "/paths/~1events/get/responses/200/content/text~1event-stream/schema" + }, + "properties": [ + { + "id": "p/openapi/paths/~1events/get/responses/200/content/text~1event-stream/schema/properties/message", + "name": { + "source": "message", + "canonical": "message" + }, + "wireName": "message", + "type": { + "target": "t/prim/string", + "nullable": false + }, + "required": false, + "clientOptional": false, + "defaultAdded": false, + "visibility": { + "none": false + }, + "flatten": false, + "eventHeader": false, + "eventPayload": false, + "secret": false, + "docs": {}, + "provenance": { + "source": 0, + "pointer": "/paths/~1events/get/responses/200/content/text~1event-stream/schema/properties/message" + } + } + ], + "abstract": false, + "positional": false, + "inputOnly": false + }, + "t/prim/string": { + "kind": "primitive", + "id": "t/prim/string", + "name": {}, + "anonymous": false, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0 + }, + "prim": "string" + } + }, + "servers": [ + { + "name": { + "hint": "server" + }, + "urlTemplate": "/", + "description": {}, + "auth": null + } + ], + "sources": [ + { + "format": "openapi@3.0", + "path": "streaming-media-30.yaml", + "hash": "ca72c9d26dda23dcad23d4a59d848db345ac0bb8a7300831865899268e998741" + } + ] +} diff --git a/testdata/conformance/openapi/streaming-media-30.yaml b/testdata/conformance/openapi/streaming-media-30.yaml new file mode 100644 index 00000000..7fe354d9 --- /dev/null +++ b/testdata/conformance/openapi/streaming-media-30.yaml @@ -0,0 +1,15 @@ +openapi: 3.0.3 +info: {title: StreamingMedia30, version: "1.0.0"} +paths: + /events: + get: + operationId: streamEvents + responses: + "200": + description: ok + content: + text/event-stream: + schema: + type: object + properties: + message: {type: string} diff --git a/testdata/conformance/openapi/streaming-media-31.golden.json b/testdata/conformance/openapi/streaming-media-31.golden.json new file mode 100644 index 00000000..6471f598 --- /dev/null +++ b/testdata/conformance/openapi/streaming-media-31.golden.json @@ -0,0 +1,427 @@ +{ + "irVersion": "0.3.0", + "name": "StreamingMedia31", + "version": "1.0.0", + "docs": {}, + "services": [ + { + "id": "s/openapi/0", + "name": { + "source": "StreamingMedia31", + "canonical": "streaming_media_31" + }, + "docs": {}, + "groups": [ + { + "name": { + "hint": "default" + }, + "docs": {}, + "operations": [ + { + "id": "op/openapi/paths/~1ingest/post", + "name": { + "source": "ingestRows", + "canonical": "ingest_rows" + }, + "docs": {}, + "request": { + "contents": [ + { + "mediaType": "application/x-ndjson", + "type": { + "target": "t/anon/paths/~1ingest/post/requestBody/content/application~1x-ndjson/schema", + "nullable": false + } + } + ], + "unmodeled": { + "openapi:required": { + "reason": "no_ir_home", + "value": false, + "provenance": { + "source": 0, + "pointer": "/paths/~1ingest/post/requestBody/required" + } + } + } + }, + "responses": [ + { + "name": { + "hint": "200" + }, + "conditions": { + "statusCodes": [ + { + "from": 200, + "to": 200 + } + ] + }, + "payload": { + "contents": [ + { + "mediaType": "text/event-stream; charset=utf-8", + "type": { + "target": "t/anon/paths/~1ingest/post/responses/200/content/text~1event-stream; charset=utf-8/schema", + "nullable": false + } + } + ] + }, + "docs": { + "description": "ok" + } + } + ], + "oneWay": false, + "streaming": "bidi", + "requestStream": { + "events": { + "target": "t/anon/paths/~1ingest/post/requestBody/content/application~1x-ndjson/schema", + "nullable": false + }, + "requiresLength": false + }, + "responseStream": { + "events": { + "target": "t/anon/paths/~1ingest/post/responses/200/content/text~1event-stream; charset=utf-8/schema", + "nullable": false + }, + "requiresLength": false + }, + "idempotency": {}, + "auth": null, + "bindings": { + "http": [ + { + "method": "POST", + "uriTemplate": "/ingest", + "sharedRoute": false, + "requestContentTypes": [ + "application/x-ndjson" + ], + "checksumRequired": false, + "isWebhook": false + } + ] + }, + "provenance": { + "source": 0, + "pointer": "/paths/~1ingest/post", + "inferred": "streaming-media-type" + } + }, + { + "id": "op/openapi/paths/~1mixed/get", + "name": { + "source": "streamEither", + "canonical": "stream_either" + }, + "docs": {}, + "responses": [ + { + "name": { + "hint": "200" + }, + "conditions": { + "statusCodes": [ + { + "from": 200, + "to": 200 + } + ] + }, + "payload": { + "contents": [ + { + "mediaType": "application/x-ndjson", + "type": { + "target": "t/anon/paths/~1mixed/get/responses/200/content/application~1x-ndjson/schema", + "nullable": false + } + }, + { + "mediaType": "application/jsonl", + "type": { + "target": "t/anon/paths/~1mixed/get/responses/200/content/application~1jsonl/schema", + "nullable": false + } + } + ] + }, + "docs": { + "description": "ok" + } + } + ], + "oneWay": false, + "streaming": "server", + "responseStream": { + "requiresLength": false + }, + "idempotency": {}, + "auth": null, + "bindings": { + "http": [ + { + "method": "GET", + "uriTemplate": "/mixed", + "sharedRoute": false, + "checksumRequired": false, + "isWebhook": false + } + ] + }, + "provenance": { + "source": 0, + "pointer": "/paths/~1mixed/get", + "inferred": "streaming-media-type" + } + } + ] + } + ], + "auth": null, + "provenance": { + "source": 0 + } + } + ], + "types": { + "t/anon/paths/~1ingest/post/requestBody/content/application~1x-ndjson/schema": { + "kind": "model", + "id": "t/anon/paths/~1ingest/post/requestBody/content/application~1x-ndjson/schema", + "name": { + "hint": "ingestRows_request" + }, + "anonymous": true, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0, + "pointer": "/paths/~1ingest/post/requestBody/content/application~1x-ndjson/schema" + }, + "properties": [ + { + "id": "p/openapi/paths/~1ingest/post/requestBody/content/application~1x-ndjson/schema/properties/row", + "name": { + "source": "row", + "canonical": "row" + }, + "wireName": "row", + "type": { + "target": "t/prim/string", + "nullable": false + }, + "required": false, + "clientOptional": false, + "defaultAdded": false, + "visibility": { + "none": false + }, + "flatten": false, + "eventHeader": false, + "eventPayload": false, + "secret": false, + "docs": {}, + "provenance": { + "source": 0, + "pointer": "/paths/~1ingest/post/requestBody/content/application~1x-ndjson/schema/properties/row" + } + } + ], + "abstract": false, + "positional": false, + "inputOnly": false + }, + "t/anon/paths/~1ingest/post/responses/200/content/text~1event-stream; charset=utf-8/schema": { + "kind": "model", + "id": "t/anon/paths/~1ingest/post/responses/200/content/text~1event-stream; charset=utf-8/schema", + "name": { + "hint": "response" + }, + "anonymous": true, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0, + "pointer": "/paths/~1ingest/post/responses/200/content/text~1event-stream; charset=utf-8/schema" + }, + "properties": [ + { + "id": "p/openapi/paths/~1ingest/post/responses/200/content/text~1event-stream; charset=utf-8/schema/properties/ack", + "name": { + "source": "ack", + "canonical": "ack" + }, + "wireName": "ack", + "type": { + "target": "t/prim/string", + "nullable": false + }, + "required": false, + "clientOptional": false, + "defaultAdded": false, + "visibility": { + "none": false + }, + "flatten": false, + "eventHeader": false, + "eventPayload": false, + "secret": false, + "docs": {}, + "provenance": { + "source": 0, + "pointer": "/paths/~1ingest/post/responses/200/content/text~1event-stream; charset=utf-8/schema/properties/ack" + } + } + ], + "abstract": false, + "positional": false, + "inputOnly": false + }, + "t/anon/paths/~1mixed/get/responses/200/content/application~1jsonl/schema": { + "kind": "model", + "id": "t/anon/paths/~1mixed/get/responses/200/content/application~1jsonl/schema", + "name": { + "hint": "response" + }, + "anonymous": true, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0, + "pointer": "/paths/~1mixed/get/responses/200/content/application~1jsonl/schema" + }, + "properties": [ + { + "id": "p/openapi/paths/~1mixed/get/responses/200/content/application~1jsonl/schema/properties/line", + "name": { + "source": "line", + "canonical": "line" + }, + "wireName": "line", + "type": { + "target": "t/prim/string", + "nullable": false + }, + "required": false, + "clientOptional": false, + "defaultAdded": false, + "visibility": { + "none": false + }, + "flatten": false, + "eventHeader": false, + "eventPayload": false, + "secret": false, + "docs": {}, + "provenance": { + "source": 0, + "pointer": "/paths/~1mixed/get/responses/200/content/application~1jsonl/schema/properties/line" + } + } + ], + "abstract": false, + "positional": false, + "inputOnly": false + }, + "t/anon/paths/~1mixed/get/responses/200/content/application~1x-ndjson/schema": { + "kind": "model", + "id": "t/anon/paths/~1mixed/get/responses/200/content/application~1x-ndjson/schema", + "name": { + "hint": "response" + }, + "anonymous": true, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0, + "pointer": "/paths/~1mixed/get/responses/200/content/application~1x-ndjson/schema" + }, + "properties": [ + { + "id": "p/openapi/paths/~1mixed/get/responses/200/content/application~1x-ndjson/schema/properties/row", + "name": { + "source": "row", + "canonical": "row" + }, + "wireName": "row", + "type": { + "target": "t/prim/string", + "nullable": false + }, + "required": false, + "clientOptional": false, + "defaultAdded": false, + "visibility": { + "none": false + }, + "flatten": false, + "eventHeader": false, + "eventPayload": false, + "secret": false, + "docs": {}, + "provenance": { + "source": 0, + "pointer": "/paths/~1mixed/get/responses/200/content/application~1x-ndjson/schema/properties/row" + } + } + ], + "abstract": false, + "positional": false, + "inputOnly": false + }, + "t/prim/string": { + "kind": "primitive", + "id": "t/prim/string", + "name": {}, + "anonymous": false, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0 + }, + "prim": "string" + } + }, + "servers": [ + { + "name": { + "hint": "server" + }, + "urlTemplate": "/", + "description": {}, + "auth": null + } + ], + "diagnostics": [ + { + "severity": "info", + "code": "openapi/degraded-construct", + "message": "request body is not required; optionality kept under Unmodeled", + "provenance": { + "source": 0, + "pointer": "/paths/~1ingest/post/requestBody" + } + }, + { + "severity": "info", + "code": "openapi/degraded-construct", + "message": "several response media types stream, so the stream element type is left unnamed rather than electing one", + "provenance": { + "source": 0, + "pointer": "/paths/~1mixed/get/responses" + } + } + ], + "sources": [ + { + "format": "openapi@3.1", + "path": "streaming-media-31.yaml", + "hash": "fd1d9fc1f0a504aa5157c194ee8e7c275966a8bc08ace0348710b5cf89797a39" + } + ] +} diff --git a/testdata/conformance/openapi/streaming-media-31.yaml b/testdata/conformance/openapi/streaming-media-31.yaml new file mode 100644 index 00000000..4c5b1c24 --- /dev/null +++ b/testdata/conformance/openapi/streaming-media-31.yaml @@ -0,0 +1,39 @@ +openapi: 3.1.0 +info: {title: StreamingMedia31, version: "1.0.0"} +paths: + /ingest: + post: + operationId: ingestRows + requestBody: + content: + application/x-ndjson: + schema: + type: object + properties: + row: {type: string} + responses: + "200": + description: ok + content: + text/event-stream; charset=utf-8: + schema: + type: object + properties: + ack: {type: string} + /mixed: + get: + operationId: streamEither + responses: + "200": + description: ok + content: + application/x-ndjson: + schema: + type: object + properties: + row: {type: string} + application/jsonl: + schema: + type: object + properties: + line: {type: string} diff --git a/testdata/conformance/openapi/unwitnessed.golden.txt b/testdata/conformance/openapi/unwitnessed.golden.txt index 0ef3569a..c98cc5c6 100644 --- a/testdata/conformance/openapi/unwitnessed.golden.txt +++ b/testdata/conformance/openapi/unwitnessed.golden.txt @@ -131,10 +131,7 @@ Operation.OneWay Operation.OverloadOf Operation.Pagination Operation.ParameterVisibility -Operation.RequestStream -Operation.ResponseStream Operation.ReturnTypeVisibility -Operation.Streaming OperationGroup.Availability OperationGroup.Groups OperationGroup.Resource @@ -170,7 +167,6 @@ Property.WireID Property.WireNameByFormat ProtocolDecl.Name ProtocolDecl.Options -Provenance.Inferred Provenance.Source RPCBinding.FullMethod RPCBinding.IdempotencyLevel @@ -206,7 +202,6 @@ Service.Renames Service.Servers Service.Unmodeled Service.Version -StreamDetail.Events StreamDetail.Initial StreamDetail.RequiresLength TemplateArg.Type From f1b7274fd1a5f08becaca974eddb6c8404c19d91 Mon Sep 17 00:00:00 2001 From: OmarAlJarrah Date: Tue, 11 Aug 2026 10:58:00 +0300 Subject: [PATCH 2/3] docs(ir-spec-matrix): record that OpenAPI reaches client streaming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reading a request body's media type as a frame format populates Operation.RequestStream, so an NDJSON body now lowers to streaming: bidi. The matrix still said OpenAPI reaches client streaming in no way at all, which the conformance contract enforces from the other side: it refused the fixture that witnesses it, because a format marked absent for a row may have no witness. The cell reads ⚠ rather than ✅, which is the legend's "expressible indirectly": nothing in the document declares the stream. It is read from a media type, stamped Provenance.Inferred, and a caller who turns the policy off loses the capability entirely — none of which ✅ would convey. The two now hold each other up. Reverting the cell and keeping the claim fails with the row marked absent yet witnessed; keeping the cell and dropping the claim fails with the row unwitnessed. Neither can drift without the other saying so. --- compilers/openapi/conformance_test.go | 2 +- docs/ir-spec-matrix.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/compilers/openapi/conformance_test.go b/compilers/openapi/conformance_test.go index a0623a16..c6e9b989 100644 --- a/compilers/openapi/conformance_test.go +++ b/compilers/openapi/conformance_test.go @@ -214,7 +214,7 @@ func conformanceCases() []conformanceCase { {"file-body", assertFileBody, nil}, {"sequential-media", assertSequentialMedia, []string{"streaming-server"}}, {"streaming-media-30", assertStreamingMedia30, []string{"streaming-server"}}, - {"streaming-media-31", assertStreamingMedia31, []string{"streaming-server"}}, + {"streaming-media-31", assertStreamingMedia31, []string{"streaming-server", "streaming-client"}}, {"per-status-errors", assertPerStatusErrors, []string{"per-status-errors"}}, {"response-links", assertResponseLinks, nil}, {"webhooks", assertWebhooks, []string{"events-channels", "server-initiated-messages"}}, diff --git a/docs/ir-spec-matrix.md b/docs/ir-spec-matrix.md index 57c36b2f..685d0c33 100644 --- a/docs/ir-spec-matrix.md +++ b/docs/ir-spec-matrix.md @@ -51,7 +51,7 @@ the ones the next compiler will be first to bind to. | `multipart-encoding` | Multipart/form encoding | ✅ encoding | ✅ formData | ✅ multipart | — | — | — | — | — | | `per-status-errors` | Per-status error types | ✅ responses | ✅ | ✅ @error models | ✅ errors list (client/server fault) | — | — | ⚠ status codes | ⚠ {error, R} variants | | `streaming-server` | Streaming: server (SSE/chunk) | ✅ itemSchema/sequential media types (3.2) | — | ✅ streams | ✅ eventstream | ✅ subscription | ✅ | ✅ stream | ⚠ info streams | -| `streaming-client` | Streaming: client / bidi | — | — | ✅ client · ⚠ bidi | ✅ | — | ✅ | ✅ | ⚠ cast/info flows | +| `streaming-client` | Streaming: client / bidi | ⚠ request media type, inferred | — | ✅ client · ⚠ bidi | ✅ | — | ✅ | ✅ | ⚠ cast/info flows | | `events-channels` | Events / pub-sub channels | ✅ webhooks (3.1) | — | ✅ events/sse | — | ✅ subscriptions | ✅ channels | — | ✅ gen_event/info | | `callbacks` | Callbacks / request-reply | ✅ callbacks | — | — | — | — | ✅ reply (static + dynamic address) | — | ⚠ From-reply | | `pagination` | Pagination (first-class) | ⚠ x-* / links | — | ✅ @list/@pageItems + prev/first/last links | ✅ paginated trait | ⚠ connections | — | ⚠ AIP-158 | — | From 826dde6f7efb3c7b8e5fe895ab4a2b06ad9c8510 Mon Sep 17 00:00:00 2001 From: OmarAlJarrah Date: Tue, 11 Aug 2026 11:17:34 +0300 Subject: [PATCH 3/3] fix(compilers/openapi): elect a stream element its contents agree on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit classifyStream refused to name an element type whenever a direction had more than one streaming content, without ever comparing them. A response offering one frame as both text/event-stream and application/x-ndjson — content negotiation, and the commonest streaming shape there is — therefore reached the IR with ResponseStream.Events unset, although both contents lowered to the same TypeID. Two success responses sharing one media type and one schema did the same. The refusal is right where the contents disagree: StreamDetail holds one Events per direction, so naming one of two differing elements would be the primary-content selection invariant 2 forbids. It does not reach a set that names one element between them — there is nothing to elect, and leaving it unnamed says less than the source did rather than declining to choose. The candidates are compared now; ir.TypeRef is a TypeID and a nullability bit, so == is the whole of what agreement means. The diagnostic counted media types, which was wrong twice over: it fired for contents that agreed, and called one media type on two responses several. It names the disagreement instead. streaming-media-31 gains the negotiated shape beside the one it already had, so the corpus holds both sides of the rule: reverting to the count reddens the new operation and leaves the refusal green. --- .../openapi/internal/operation/streaming.go | 44 ++++-- compilers/openapi/streaming_test.go | 25 +++- .../openapi/streaming-media-31.golden.json | 134 +++++++++++++++++- .../openapi/streaming-media-31.yaml | 21 +++ 4 files changed, 204 insertions(+), 20 deletions(-) diff --git a/compilers/openapi/internal/operation/streaming.go b/compilers/openapi/internal/operation/streaming.go index eb348273..6ee4e376 100644 --- a/compilers/openapi/internal/operation/streaming.go +++ b/compilers/openapi/internal/operation/streaming.go @@ -26,8 +26,8 @@ type streamDirection struct { // media type rather than by a declaration, which is what the operation's // Provenance.Inferred marker records. inferred bool - // ambiguous reports that the direction had more than one streaming content, - // so no element type was elected. + // ambiguous reports that the direction's streaming contents named different + // element types, so none was elected. ambiguous bool } @@ -62,12 +62,19 @@ func applyStreaming(c lowering.Ctx, op *ir.Operation, declPtr string) (string, [ // classifyStream folds one direction's candidates into the detail to write. // -// The one thing it refuses to do is elect an element type from several. -// StreamDetail holds one Events per direction while Payload keeps every media -// type, so naming one of two streaming contents would be exactly the -// primary-content selection a compiler must not make (invariant 2). The +// The one thing it refuses to do is elect an element type from candidates that +// disagree. StreamDetail holds one Events per direction while Payload keeps +// every media type, so naming one of two differing contents would be exactly +// the primary-content selection a compiler must not make (invariant 2). The // direction still streams — that much all the candidates agree on — and the // element is left unnamed for a lowering that has the whole set to choose from. +// +// Candidates naming the same element are not that case, so they are compared +// rather than counted. A response offering one frame as both text/event-stream +// and application/x-ndjson is content negotiation over a single element type, +// and there is nothing to elect between: refusing on the count alone left the +// commonest streaming shape there is with its element unnamed, which says less +// than the source did rather than declining to choose. func classifyStream(candidates []streamCandidate) streamDirection { if len(candidates) == 0 { return streamDirection{} @@ -78,11 +85,16 @@ func classifyStream(candidates []streamCandidate) streamDirection { out.inferred = true } } - if len(candidates) > 1 { - out.ambiguous = true - return out - } + // ir.TypeRef is a TypeID and a nullability bit, so == is the whole of what + // "the same element" means: two contents differing in either name different + // elements. events := candidates[0].events + for _, candidate := range candidates[1:] { + if candidate.events != events { + out.ambiguous = true + return out + } + } out.detail.Events = &events return out } @@ -135,12 +147,16 @@ func streamingMode(request, response bool) ir.StreamingMode { } // unelectedElementDiag reports a direction whose element type was left unnamed -// because several of its contents stream. Nothing is lost — every content is -// still on the payload — but the IR now says less than the source did, which is -// what makes it a degradation rather than a silent choice. +// because its streaming contents name different ones. Nothing is lost — every +// content is still on the payload — but the IR now says less than the source +// did, which is what makes it a degradation rather than a silent choice. +// +// It names the disagreement rather than counting media types: contents that +// agree keep their element, and one media type appearing on two responses is +// not several media types. func unelectedElementDiag(c lowering.Ctx, pointer, direction string) ir.Diagnostic { return c.DiagAt(ir.SeverityInfo, diag.DegradedConstruct, pointer, - "several %s media types stream, so the stream element type is left unnamed rather than electing one", direction) + "the %s streams more than one element type, so it is left unnamed rather than electing one", direction) } // joinInferred names every heuristic that shaped one node, in the order the diff --git a/compilers/openapi/streaming_test.go b/compilers/openapi/streaming_test.go index 81479de4..cf630736 100644 --- a/compilers/openapi/streaming_test.go +++ b/compilers/openapi/streaming_test.go @@ -141,10 +141,15 @@ func assertStreamingMedia30(t *testing.T, doc *ir.Document, _ []ir.Diagnostic) { assert.Equal(t, "streaming-media-type", op.Provenance.Inferred) } -// assertStreamingMedia31 is the corpus row for the two shapes 3.0 cannot show -// on one operation: a request body that streams as well as its response, and a -// response offering two streaming media types, where the element type is left -// unnamed rather than elected. +// assertStreamingMedia31 is the corpus row for the three shapes 3.0 cannot show +// on one operation: a request body that streams as well as its response, a +// response whose two streaming contents name different element types, where +// none is elected, and one whose two name the same, where it is. +// +// The last two are the pair that says what the refusal is for. Refusing on the +// count alone would pass the middle one and fail the last, and content +// negotiation over a single frame — the same schema as text/event-stream and as +// application/x-ndjson — is the commoner shape of the two. func assertStreamingMedia31(t *testing.T, doc *ir.Document, diags []ir.Diagnostic) { both, ok := opByName(doc, "ingestRows") require.True(t, ok) @@ -166,4 +171,16 @@ func assertStreamingMedia31(t *testing.T, doc *ir.Document, diags []ir.Diagnosti assert.Len(t, either.Responses[0].Payload.Contents, 2, "both contents are still kept") assert.True(t, openapitest.HasDiag(diags, "openapi/degraded-construct"), "the unelected element type is reported; got %+v", diags) + + negotiated, ok := opByName(doc, "streamNegotiated") + require.True(t, ok) + assert.Equal(t, ir.StreamingServer, negotiated.Streaming) + require.NotNil(t, negotiated.ResponseStream) + require.NotNil(t, negotiated.ResponseStream.Events, + "two contents naming one element type elect it: there is nothing to choose between") + assert.Equal(t, ir.TypeID("t/openapi/components/schemas/Frame"), + negotiated.ResponseStream.Events.Target) + require.NotNil(t, negotiated.Responses[0].Payload) + assert.Len(t, negotiated.Responses[0].Payload.Contents, 2, + "and both media types are still kept, as they are for the unelected case") } diff --git a/testdata/conformance/openapi/streaming-media-31.golden.json b/testdata/conformance/openapi/streaming-media-31.golden.json index 6471f598..bc274815 100644 --- a/testdata/conformance/openapi/streaming-media-31.golden.json +++ b/testdata/conformance/openapi/streaming-media-31.golden.json @@ -179,6 +179,77 @@ "pointer": "/paths/~1mixed/get", "inferred": "streaming-media-type" } + }, + { + "id": "op/openapi/paths/~1negotiated/get", + "name": { + "source": "streamNegotiated", + "canonical": "stream_negotiated" + }, + "docs": {}, + "responses": [ + { + "name": { + "hint": "200" + }, + "conditions": { + "statusCodes": [ + { + "from": 200, + "to": 200 + } + ] + }, + "payload": { + "contents": [ + { + "mediaType": "text/event-stream", + "type": { + "target": "t/openapi/components/schemas/Frame", + "nullable": false + } + }, + { + "mediaType": "application/x-ndjson", + "type": { + "target": "t/openapi/components/schemas/Frame", + "nullable": false + } + } + ] + }, + "docs": { + "description": "ok" + } + } + ], + "oneWay": false, + "streaming": "server", + "responseStream": { + "events": { + "target": "t/openapi/components/schemas/Frame", + "nullable": false + }, + "requiresLength": false + }, + "idempotency": {}, + "auth": null, + "bindings": { + "http": [ + { + "method": "GET", + "uriTemplate": "/negotiated", + "sharedRoute": false, + "checksumRequired": false, + "isWebhook": false + } + ] + }, + "provenance": { + "source": 0, + "pointer": "/paths/~1negotiated/get", + "inferred": "streaming-media-type" + } } ] } @@ -374,6 +445,65 @@ "positional": false, "inputOnly": false }, + "t/openapi/components/schemas/Frame": { + "kind": "model", + "id": "t/openapi/components/schemas/Frame", + "name": { + "source": "Frame", + "canonical": "frame" + }, + "anonymous": false, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0, + "pointer": "/components/schemas/Frame" + }, + "properties": [ + { + "id": "p/openapi/components/schemas/Frame/properties/seq", + "name": { + "source": "seq", + "canonical": "seq" + }, + "wireName": "seq", + "type": { + "target": "t/prim/integer", + "nullable": false + }, + "required": false, + "clientOptional": false, + "defaultAdded": false, + "visibility": { + "none": false + }, + "flatten": false, + "eventHeader": false, + "eventPayload": false, + "secret": false, + "docs": {}, + "provenance": { + "source": 0, + "pointer": "/components/schemas/Frame/properties/seq" + } + } + ], + "abstract": false, + "positional": false, + "inputOnly": false + }, + "t/prim/integer": { + "kind": "primitive", + "id": "t/prim/integer", + "name": {}, + "anonymous": false, + "docs": {}, + "sensitive": false, + "provenance": { + "source": 0 + }, + "prim": "integer" + }, "t/prim/string": { "kind": "primitive", "id": "t/prim/string", @@ -410,7 +540,7 @@ { "severity": "info", "code": "openapi/degraded-construct", - "message": "several response media types stream, so the stream element type is left unnamed rather than electing one", + "message": "the response streams more than one element type, so it is left unnamed rather than electing one", "provenance": { "source": 0, "pointer": "/paths/~1mixed/get/responses" @@ -421,7 +551,7 @@ { "format": "openapi@3.1", "path": "streaming-media-31.yaml", - "hash": "fd1d9fc1f0a504aa5157c194ee8e7c275966a8bc08ace0348710b5cf89797a39" + "hash": "c7c7a5c5051e94131d339363c28854b82e5a6fda41e8381ee46fbd2bbca4fb1a" } ] } diff --git a/testdata/conformance/openapi/streaming-media-31.yaml b/testdata/conformance/openapi/streaming-media-31.yaml index 4c5b1c24..55639b8c 100644 --- a/testdata/conformance/openapi/streaming-media-31.yaml +++ b/testdata/conformance/openapi/streaming-media-31.yaml @@ -37,3 +37,24 @@ paths: type: object properties: line: {type: string} + # Content negotiation: one element offered under two streaming media types. + # Every candidate names the same type, so there is nothing to elect between + # and the element is named — the refusal above is for contents that disagree, + # not for a direction that happens to have more than one. + /negotiated: + get: + operationId: streamNegotiated + responses: + "200": + description: ok + content: + text/event-stream: + schema: {$ref: '#/components/schemas/Frame'} + application/x-ndjson: + schema: {$ref: '#/components/schemas/Frame'} +components: + schemas: + Frame: + type: object + properties: + seq: {type: integer}