diff --git a/config/mappings/agmarknet/mandi-price.select.yaml b/config/mappings/agmarknet/mandi-price.select.yaml new file mode 100644 index 00000000..d10a0ce8 --- /dev/null +++ b/config/mappings/agmarknet/mandi-price.select.yaml @@ -0,0 +1,366 @@ +# Agmarknet Vistaar, openagrinet:MandiPrice, select. Both directions, one file. +# +# One file per binding-action rather than one per direction, because both legs of +# an exchange are one contract: the response has to answer the request that was +# sent, and splitting them lets one change without the other. +# +# The upstream is Agmarknet's Vistaar select. It takes governed codes -- state, +# district, market, commodity -- plus a date range, and every one of them is in +# the payload, so nothing here needs resolving before the call. That is why the +# mandi plugin has no prerequisites: a MandiPrice select names the market it +# wants rather than a point to search from. +# +# THE ANSWER IS WITHIN THE PACK; THE REQUEST IS NOT, AND CANNOT BE YET. +# +# Every field the response sets is declared by openagrinet:MandiPrice v0.1 and +# correctly placed, inside the closed market and prices property sets. Where +# the upstream reports something the pack has no home for, it is dropped rather +# than invented. +# +# The REQUEST half is a different matter, and an earlier version of this note +# claimed otherwise. It reads market and validity, and the pack's OnDemand +# branch EXCLUDES both -- see the note on the guards below. That is a gap in +# the pack rather than a mistake here: OnDemand describes what a provider can +# obtain and Direct describes an obtained reading, and a request is neither. + +# What this capability cannot serve, refused before the provider is called. +# +# THE PACK AND THESE GUARDS DISAGREE TODAY, and an earlier version of this note +# said the pack leaves market and validity OPTIONAL. It does not: the OnDemand +# branch EXCLUDES them, alongside source, commodity, commodityGroup, grade, +# variety, arrivalDate, prices and generatedAt. So a payload that satisfies +# these guards cannot validate against the pack, and one that validates cannot +# satisfy them. +# +# Nothing breaks in practice: the exclusion sits under if/then, which the +# validator parses and never evaluates. That is an accident to rely on rather +# than a design, and it is written down here so it is not mistaken for one. +# +# Resolving it is a pack change. For the market, coverageAreas is the likely +# home -- inherited from AgricultureResource, not excluded in OnDemand, and it +# accepts an AdministrativeAreaReference. For the date range there is no +# candidate at all: historyPeriod and updateFrequency are durations describing +# provider capability, not a window a caller asks for. +# +# What IS true, and is the reason this block exists: the pack defines +# market.district and market.state as "name or governed code", so a payload can +# be perfectly valid and still be unanswerable by this upstream, which wants +# codes and a date range. +# +# Refusing here names what is missing. Sending it anyway earns a 400 from +# Agmarknet, or worse an empty result that reads as "no prices". +required: + - check: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + /* Exactly one. Everything downstream reads supportedCommodities[0]: + this guard, the outbound query, and the commodity stamped on each + resource. So a caller sending three commodities passed validation, was + queried for the first, and got a confident signed answer to a third of + what it asked -- the same failure oanbinding refuses at the commitment + level, where guessing would silently serve part of a request. */ + $exists($ra.supportedCommodities[0].code) + and $count($ra.supportedCommodities) = 1 + ) + message: "this capability needs exactly one commodity code in supportedCommodities; send several requests rather than have all but the first dropped" + - check: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + /* Shape, not just presence. The pack describes district and state as + "name or governed code", so the names form is pack-legal -- and the + existence-only check passed it, after which it went to Agmarknet + verbatim as codes. Agmarknet answered nothing, and the caller got a + signed, spec-valid "no prices" for a market that had prices. + + Agmarknet wants a numeric district and a short alphabetic state (96, + CG), so a district that is not all digits or a state longer than four + letters is a name and is refused with the reason. */ + $match($ra.market.district, /^[0-9]+$/) + and $match($ra.market.state, /^[A-Za-z]{2,4}$/) + ) + message: "this capability needs Agmarknet's codes in market: a numeric district and a short alphabetic state, not their names" + - check: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + $exists($ra.validity.startsAt) and $exists($ra.validity.endsAt) + ) + message: "this capability needs a validity window; it reports prices over a date range" + +# The upstream is a GET, so this object becomes the query string. The token is +# not here and must never be: it comes from the adapter's authScheme query, +# whose value is read from an environment variable. This file is published. +# +# marketcode is sent when the payload names one and omitted otherwise, which is +# what the upstream expects: without it the query widens from one market to the +# whole district. +request: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + + /* The upstream wants dd-MM-yyyy; the pack's validity is an ISO date. A + substring reorder rather than a date library, because these are dates + with no time and no zone, and $fromMillis would invent both. */ + $ddmmyyyy := function($iso) { + $substring($iso, 8, 2) & "-" & $substring($iso, 5, 2) & "-" & $substring($iso, 0, 4) + }; + + $base := { + "statecode": $ra.market.state, + "districtcode": $ra.market.district, + "commoditycode": $ra.supportedCommodities[0].code, + "from_date": $ddmmyyyy($ra.validity.startsAt), + "to_date": $ddmmyyyy($ra.validity.endsAt) + }; + + $exists($ra.market.marketCode) + ? $merge([$base, {"marketcode": $ra.market.marketCode}]) + : $base + ) + +# One resource per price record, in Direct mode. +# +# Direct is what the pack requires of an answer: the resource now CARRIES the +# observation rather than advertising that it could obtain one. It requires +# source, commodity, market, arrivalDate, prices and generatedAt, and all six +# are set below. +# +# The upstream's records use Title Case keys WITH SPACES -- `Modal Price`, not +# modalPrice -- so they need backticks, and its prices are STRINGS, so they need +# $number() before they satisfy the pack's numeric types. +response: | + ( + $records := $type(response) = "array" ? response + : $exists(response.data) ? response.data + : $exists(response.records) ? response.records + : []; + + $selected := beckn.message.contract.commitments[0]; + + /* The caller's own @context, echoed back rather than restated here. A + mapping that hardcodes it has to be reissued whenever the pack URL + moves, and it can disagree with what the request actually declared. + Backticks because @ is an operator in JSONata. */ + $ctx := beckn.message.contract.commitments[0].resources[0].resourceAttributes.`@context`; + $ra := $selected.resources[0].resourceAttributes; + + /* dd-MM-yyyy back to ISO, so the answer speaks the pack's date format + rather than the upstream's. */ + $iso := function($ddmmyyyy) { + $substring($ddmmyyyy, 6, 4) & "-" & $substring($ddmmyyyy, 3, 2) & "-" & $substring($ddmmyyyy, 0, 2) + }; + + /* The market code is optional: without it the query widened to the whole + district, so the district code is what identifies the scope. */ + $scope := $exists($ra.market.marketCode) ? $ra.market.marketCode : $ra.market.district; + /* SLUG for the parts that are names rather than codes, so an id has no + spaces and no case surprises. */ + $slug := function($v) { $exists($v) ? $replace($lowercase($v), " ", "-") }; + + /* Bound once because it is used twice -- for a resource's own id and for + the offer's reference to it. Two copies of one expression is how a + dangling reference gets reintroduced. + + Built from CODES, not the names the upstream reports. A market name + carries spaces and a commodity name carries brackets -- "Kasdol APMC", + "Paddy(Common)" -- and an identifier that a consumer may put in a URL or + a filter should not. The codes are already in the payload, so they cost + nothing, and they are stable where a display name is not. + + Every field the upstream distinguishes rows by is in the id. + + It was scope:commodity:date only, and Agmarknet routinely reports + several rows for the same market, commodity and date differing by + Variety and Grade -- the fixture in this package is exactly that pair. + Two distinct resources then shared one id, and the offer referenced it + twice, so a consumer resolving resourceIds could not tell which price + it had. Market is in it too: district-wide $scope is the district, and + the rows come from several markets inside it. + + Built with $join over a list, because an absent Variety or Grade drops + out of an array literal rather than leaving an empty segment. */ + $resourceId := function($r) { + "res:agmarknet:" & $join([ + $scope, + $ra.supportedCommodities[0].code, + $iso($r.`Arrival Date`), + $slug($r.Market), + $slug($r.Variety), + $slug($r.Grade) + ], ":") + }; + + /* Absent rather than present-and-empty: a consumer must be able to tell + "the market reported no minimum" from "the minimum was zero". + + GUARDS SHAPE, NOT JUST PRESENCE. Agmarknet writes an unreported price as + a marker -- "NR", "-", "" -- and $exists() is true for all of them, so + an existence-only guard handed them to $number() and it threw D3030. + That failed the WHOLE response: one unreported cell in one row turned a + good multi-row answer into an adapter error, which is the opposite of + the distinction this function exists to preserve. + + The number branch is not redundant. $match() throws T0410 on a + non-string, so testing the regex first would break the day this + upstream sends a real number instead of a quoted one. */ + /* The price fields the caller asked for. Absent means all three: the + pack requires supportedPriceFields on an OnDemand request, so absence + is a payload this mapping does not have to serve well -- but it should + not silently return nothing either. */ + $fields := $exists($ra.supportedPriceFields) + ? $ra.supportedPriceFields + : ["Minimum", "Maximum", "Modal"]; + + $priced := function($value) { + $type($value) = "number" + ? $value + : ($type($value) = "string" and $match($value, /^[0-9]+(\.[0-9]+)?$/) + ? $number($value)) + }; + + /* EMIT ONLY RECORDS THAT CAN PRODUCE A CONFORMANT RESOURCE. + Absent is honest; present-and-degenerate is a lie in the shape of an + answer, and it is worse here than elsewhere because the answer is + signed. Three ways a record cannot be made conformant: + + no Arrival Date $iso is substring-and-concatenate, and JSONata casts + undefined to "" -- so an absent date became the + string "--", which the pack refuses (format: date) + and which degraded the resource id along with it. + arrivalDate is on the Direct required list, so it + cannot be omitted either. + + no Price Unit prices.required is [currency, unit]. JSONata drops + an absent key rather than emitting null, so the unit + silently vanished and took the whole resource's + validity with it. Not defaulted: Rs./Qtl and Rs./Kg + are both real, so inventing one would misreport a + price by a factor of a hundred. + + no usable price prices carries anyOf [minimum, maximum, modal], so a + row whose three prices are all unreported markers + has nothing to report and cannot satisfy it. + + Dropped rather than refused: the other rows in the same answer are good, + and failing the request would discard them too -- which is the mistake + the price guard above just fixed. */ + $conformant := function($r) { + /* $count(...) > 0 rather than $exists($match(...)): this engine returns + an EMPTY ARRAY from $match when nothing matches, and $exists([]) is + true -- so the $exists form silently accepted every record and the + filter did nothing. Caught by the test, not by reading. */ + $count($match($iso($r.`Arrival Date`), /^[0-9]{4}-[0-9]{2}-[0-9]{2}$/)) > 0 + and $exists($r.`Price Unit`) + /* Judged over the fields the caller asked for, not all three. A row + whose Modal is a marker cannot serve a request for Modal alone, + even though its Minimum is fine. */ + and (("Minimum" in $fields and $exists($priced($r.`Min Price`))) + or ("Maximum" in $fields and $exists($priced($r.`Max Price`))) + or ("Modal" in $fields and $exists($priced($r.`Modal Price`)))) + }; + $usable := $filter($records, $conformant); + + { + "context": { + "version": beckn.context.version, + "action": "on_select", + "networkId": beckn.context.networkId, + "transactionId": beckn.context.transactionId, + "messageId": beckn.context.messageId, + "timestamp": $now() + }, + "message": { + "contract": { + "commitments": [ + { + "status": { + "descriptor": { "code": "DRAFT", "name": "Draft" } + }, + /* The offer is echoed, but its references are not: the request + named an abstract price enquiry and the answer returns the + concrete observations. Leaving resourceIds as they arrived + would point the offer at an id appearing nowhere here. */ + "offer": $merge([ + $selected.offer, + { "resourceIds": [$map($usable, function($r) { $resourceId($r) })] } + ]), + /* Wrapped: JSONata collapses a one-element sequence to a bare + value, so a single-record answer would return an object where + every other count returns a list. */ + "resources": [$map($usable, function($r) { + { + "id": $resourceId($r), + /* Required by Commitment.resources in the Beckn v2 spec, + which defines no quantity property and carries no Quantity + schema at all -- a defect upstream. One resource is one + market's observation for one day, so one. */ + "quantity": 1, + "resourceAttributes": { + "@context": $ctx, + "@type": "openagrinet:MandiPrice", + "informationMode": "Direct", + /* Stated, not echoed. subjectCategories is a closed enum on + AgricultureResource, and "Market" is what a MandiPrice resource + IS -- the pack states it in both of its own examples. Echoing the + request meant a caller sending ["Weather"] on a MandiPrice select + got it faithfully republished over this adapter's signature, and + nothing caught it: the value is enum-legal, so it validates. The + sibling weather mapping has always stated ["Weather"]. */ + "subjectCategories": ["Market"], + "source": { + "sourceId": "agmarknet", + "sourceName": "Agmarknet Vistaar" + }, + "commodity": { + "code": $ra.supportedCommodities[0].code, + "name": $r.Commodity + }, + "commodityGroup": $r.Group, + "variety": $r.Variety, + "grade": $r.Grade, + /* Every member comes from the RECORD, so the answer describes what + the provider reported rather than what was asked for. + + marketCode is absent, and that is not an omission. This upstream + TAKES a market code as a query parameter and reports none back -- + there is no such field anywhere in its response. Restating the + requested code against a returned row would assert something + unverified, and district-wide there is no requested code at all + while the rows come from several markets, so one code would have + been wrong for most of them. The pack requires only marketName + and calls marketCode "when available"; here it is not. */ + "market": { + "marketName": $r.Market, + "district": $r.District, + "state": $r.State + }, + "arrivalDate": $iso($r.`Arrival Date`), + "prices": { + /* Only what was asked for. supportedPriceFields was validated on the + way in and then ignored on the way out, so a caller asking for Modal + alone got all three. A ternary with no else yields nothing and the key + drops, which is the same mechanism an unreported price already uses. */ + "minimum": "Minimum" in $fields ? $priced($r.`Min Price`), + "maximum": "Maximum" in $fields ? $priced($r.`Max Price`), + "modal": "Modal" in $fields ? $priced($r.`Modal Price`), + /* INR is a fact about THIS UPSTREAM, not about the pack, and it is + stated rather than read because Agmarknet reports no currency field + to read -- unlike the unit on the next line, which it does report. + The pack requires currency, so it cannot be omitted either. + + Worth being deliberate about the difference: on_select, DRAFT and the + @type values are PACK constants, correctly stated. INR and the + agmarknet sourceId are PROVIDER constants -- they change if this file + is pointed at another upstream, and the pack ones do not. */ + "currency": "INR", + "unit": $r.`Price Unit` + }, + "generatedAt": $now() + } + } + })] + } + ] + } + } + } + ) diff --git a/config/oan-provider-adapter.yaml b/config/oan-provider-adapter.yaml index 2bba72e1..34730d21 100644 --- a/config/oan-provider-adapter.yaml +++ b/config/oan-provider-adapter.yaml @@ -4,6 +4,11 @@ # capability's call plan from the registry, calls the provider, and answers with # the mapped result. There is no callback -- the answer is the HTTP response. # +# Two capabilities are served here, weather and mandi, by two domain packages +# in one pipeline. They share this module, the registry client and the mapper, +# and share nothing else: which one answers is decided by the payload, not by +# the URL, the domain or the order they appear in. +# # Adding a provider is three things, and none of them is a Go change here: # 1. a registry row binding "|" to a call plan # 2. one mapping file per action, published at the URL that row names @@ -270,6 +275,43 @@ modules: # # maxResponseBytes: 4194304 # default: 4194304 (4 MiB) + # A second capability in the same pipeline, from a different domain + # package. Nothing about it is weather's business: a different + # upstream, a different mapping, a different set of prerequisites -- + # and the same two registry rows. This entry, plus "mandi" in steps + # below, is the entire cost of adding it. + - id: mandi + config: + bindingKeys: "agmarknet|openagrinet:MandiPrice" + + # Agmarknet's Vistaar API takes its token as a QUERY parameter, + # which is what authScheme query is for. The adapter holds the + # parameter's NAME and the name of the variable carrying the + # value -- never the value -- and redacts it from the URL it + # logs, so a token cannot reach the log by way of the request. + authScheme: query + queryName: token + queryValueEnv: MANDI_TOKEN + + # MANDI_TOKEN MUST BE SET IN THE ADAPTER'S ENVIRONMENT. Without it + # every mandi call fails immediately and permanently: the step + # refuses to call an upstream unauthenticated, and marks the failure + # as one no retry can fix. + # + # There is no default and there cannot be one -- a credential is not + # something this file may hold. So the failure mode is a plugin that + # loads cleanly, registers cleanly, passes startup validation, and + # then fails 100% of requests. Worth knowing before deploying: + # + # docker: -e MANDI_TOKEN=... or an env_file + # kubernetes: a Secret, mounted as an environment variable + # local: MANDI_TOKEN=... ./server --config=... + # + # The error names the SCHEME rather than the variable, deliberately: + # this runs behind a signed network call and a peer has no business + # learning which variables this deployment reads. The variable name + # is in the adapter's own log at error level. + # Declaring a step above is not enough: THIS list is what runs. A step # that appears under providerSteps but not here never executes, and the # request falls through to the 404 above -- which looks like a registry @@ -277,7 +319,8 @@ modules: steps: - validateSign # the sender's key, from the registry - validateSchema # the pinned Beckn v2 spec - - weather # resolve, map out, call, map back + - weather # openagrinet:WeatherObservation, or pass through + - mandi # openagrinet:MandiPrice, or pass through - signAck # signs whatever the step answered with # ---------------------------------------------------------------------------- diff --git a/install/build-plugins.sh b/install/build-plugins.sh index 3bf07a01..8ac0d713 100755 --- a/install/build-plugins.sh +++ b/install/build-plugins.sh @@ -33,6 +33,7 @@ plugins=( "oanregistry" "jsonmapper" "weather" + "mandi" "manifestloader" "reqpreprocessor" "otelsetup" diff --git a/pkg/plugin/implementation/mandi/cmd/plugin.go b/pkg/plugin/implementation/mandi/cmd/plugin.go new file mode 100644 index 00000000..5b4fbb70 --- /dev/null +++ b/pkg/plugin/implementation/mandi/cmd/plugin.go @@ -0,0 +1,104 @@ +// Command plugin builds the mandi provider step as a loadable plugin. +// +// The filename of the built .so is the id a deployment names in providerSteps, +// so this package is mandi's whole public surface: a config map in, a step out. +package main + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + + "github.com/beckn-one/beckn-onix/pkg/log" + "github.com/beckn-one/beckn-onix/pkg/plugin/definition" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/mandi" +) + +// mandiProvider implements definition.ProviderStepProvider. +type mandiProvider struct{} + +// newStepFunc creates a new step. Indirected for tests. +var newStepFunc = mandi.New + +// parseConfig turns the plugin config map into a typed Config. Anything absent +// is left zero: mandi.New applies the defaults and validates the auth scheme, +// so those rules live in one place. +func (p mandiProvider) parseConfig(config map[string]string) (*mandi.Config, error) { + cfg := &mandi.Config{ + BindingKeys: splitList(config["bindingKeys"]), + // Absent means the Beckn v2 convention. See upstream.Config for why + // this is a default rather than something to set. + ProviderIDAt: config["providerIdAt"], + CapabilityCodeAt: config["capabilityCodeAt"], + AuthScheme: config["authScheme"], + UsernameEnv: config["usernameEnv"], + PasswordEnv: config["passwordEnv"], + HeaderName: config["headerName"], + HeaderValueEnv: config["headerValueEnv"], + QueryName: config["queryName"], + QueryValueEnv: config["queryValueEnv"], + } + + if raw, exists := config["maxResponseBytes"]; exists && raw != "" { + value, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return nil, fmt.Errorf("invalid maxResponseBytes value '%s': %w", raw, err) + } + if value <= 0 { + return nil, fmt.Errorf("maxResponseBytes must be positive, got %d", value) + } + cfg.MaxResponseBytes = value + } + + return cfg, nil +} + +// New creates a new mandi provider step instance. +func (p mandiProvider) New(ctx context.Context, registry definition.ProviderRecordLookup, mapper definition.Mapper, config map[string]string) (definition.Step, func() error, error) { + if ctx == nil { + return nil, nil, errors.New("context cannot be nil") + } + + cfg, err := p.parseConfig(config) + if err != nil { + log.Errorf(ctx, err, "Failed to parse mandi configuration") + return nil, nil, fmt.Errorf("failed to parse mandi configuration: %w", err) + } + + step, closer, err := newStepFunc(ctx, registry, mapper, cfg) + if err != nil { + log.Errorf(ctx, err, "Failed to create mandi step") + return nil, nil, err + } + + log.Infof(ctx, "Mandi step created successfully") + return step, closer, nil +} + +// splitList reads a comma-separated config value, which is how a list reaches a +// plugin -- the config is map[string]string. Blanks are dropped and spaces +// trimmed, so a trailing comma or a wrapped line is not a config error. +// +// A comma is unambiguous here: a binding key separates its own halves with a +// pipe. +func splitList(raw string) []string { + if strings.TrimSpace(raw) == "" { + return nil + } + var out []string + for _, part := range strings.Split(raw, ",") { + if trimmed := strings.TrimSpace(part); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} + +// Provider is the exported plugin instance. +var Provider = mandiProvider{} + +// Compile-time proof the provider satisfies the interface the manager asserts +// against. A mismatch is otherwise a runtime cast failure at startup. +var _ definition.ProviderStepProvider = Provider diff --git a/pkg/plugin/implementation/mandi/cmd/plugin_test.go b/pkg/plugin/implementation/mandi/cmd/plugin_test.go new file mode 100644 index 00000000..262f73a6 --- /dev/null +++ b/pkg/plugin/implementation/mandi/cmd/plugin_test.go @@ -0,0 +1,282 @@ +package main + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" + + "github.com/beckn-one/beckn-onix/pkg/model" + "github.com/beckn-one/beckn-onix/pkg/plugin/definition" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/mandi" +) + +type stubRegistry struct{} + +func (stubRegistry) ProviderRecord(context.Context, string) (*model.ProviderRecord, error) { + return nil, nil +} + +type stubMapper struct{} + +func (stubMapper) Verify(context.Context, string, any) error { return nil } + +func (stubMapper) Transform(context.Context, string, definition.Direction, any) ([]byte, error) { + return nil, nil +} + +func TestParseConfig(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + config map[string]string + expected *mandi.Config + expectedErr string + }{ + { + // Everything absent is left zero: mandi.New defaults it, so the + // rules are defined in exactly one place. + name: "leaves everything unset for New to default", + config: map[string]string{}, + expected: &mandi.Config{}, + }, + { + // Query auth is why this capability has its own entry rather than + // sharing weather's: Agmarknet's Vistaar API takes its token as a + // QUERY parameter, and these two keys are the only place that is + // expressible. + name: "reads the query auth scheme this capability needs", + config: map[string]string{ + "bindingKeys": "agmarknet|openagrinet:MandiPrice", + "authScheme": "query", + "queryName": "api-key", + "queryValueEnv": "MANDI_TOKEN", + }, + expected: &mandi.Config{ + BindingKeys: []string{"agmarknet|openagrinet:MandiPrice"}, + AuthScheme: "query", + QueryName: "api-key", + QueryValueEnv: "MANDI_TOKEN", + }, + }, + { + name: "reads every supported setting", + config: map[string]string{ + "bindingKeys": "other|capability", + "authScheme": "basic", + "usernameEnv": "U", + "passwordEnv": "P", + "headerName": "X-Key", + "headerValueEnv": "V", + "queryName": "q", + "queryValueEnv": "Q", + "maxResponseBytes": "2048", + }, + expected: &mandi.Config{ + BindingKeys: []string{"other|capability"}, + AuthScheme: "basic", + UsernameEnv: "U", + PasswordEnv: "P", + HeaderName: "X-Key", + HeaderValueEnv: "V", + QueryName: "q", + QueryValueEnv: "Q", + MaxResponseBytes: 2048, + }, + }, + { + name: "rejects a malformed response cap", + config: map[string]string{"maxResponseBytes": "lots"}, + expectedErr: "invalid maxResponseBytes value 'lots'", + }, + { + name: "rejects a non-positive response cap", + config: map[string]string{"maxResponseBytes": "0"}, + expectedErr: "maxResponseBytes must be positive", + }, + { + // Present but empty is not the same as malformed. A rendered + // config with an unset variable produces this, and it should read + // as "unset" rather than failing startup. + name: "treats an empty response cap as unset", + config: map[string]string{"maxResponseBytes": ""}, + expected: &mandi.Config{}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := mandiProvider{}.parseConfig(tc.config) + + if tc.expectedErr != "" { + if err == nil { + t.Fatalf("expected error %q but got none", tc.expectedErr) + } + if !strings.Contains(err.Error(), tc.expectedErr) { + t.Errorf("expected error containing %q, got %q", tc.expectedErr, err.Error()) + } + return + } + + if err != nil { + t.Fatalf("expected no error, got: %v", err) + } + if !reflect.DeepEqual(got, tc.expected) { + t.Errorf("expected config %+v, got %+v", tc.expected, got) + } + }) + } +} + +// A plugin config is map[string]string, so a list arrives comma-separated -- +// the convention reqpreprocessor and schemav2validator already use. Binding keys +// separate their own halves with a pipe, so a comma is unambiguous. +func TestSplitList(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + raw string + want []string + }{ + {name: "empty is nil, not a one-element list of nothing", raw: "", want: nil}, + {name: "whitespace only is nil", raw: " ", want: nil}, + {name: "one value", raw: "a|openagrinet:One", want: []string{"a|openagrinet:One"}}, + { + name: "several, with blanks and spacing a wrapped config line produces", + raw: "a|openagrinet:One, b|openagrinet:Two ,, ", + want: []string{"a|openagrinet:One", "b|openagrinet:Two"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + if got := splitList(tc.raw); !reflect.DeepEqual(got, tc.want) { + t.Errorf("splitList(%q) = %v, want %v", tc.raw, got, tc.want) + } + }) + } +} + +// The override is two keys, both or neither. Absent leaves the step on the +// Beckn v2 convention, which is what every deployment should be running. +func TestParseConfigReadsTheBindingKeyOverride(t *testing.T) { + t.Parallel() + + cfg, err := mandiProvider{}.parseConfig(map[string]string{ + "bindingKeys": "a|openagrinet:One", + "providerIdAt": "who.provider", + "capabilityCodeAt": "what[].type", + }) + if err != nil { + t.Fatalf("parseConfig() returned an unexpected error: %v", err) + } + if cfg.ProviderIDAt != "who.provider" || cfg.CapabilityCodeAt != "what[].type" { + t.Errorf("override = %q / %q, want the configured paths", cfg.ProviderIDAt, cfg.CapabilityCodeAt) + } +} + +// Absent leaves them empty, and upstream reads that as "use the convention". +func TestParseConfigLeavesTheOverrideUnsetByDefault(t *testing.T) { + t.Parallel() + + cfg, err := mandiProvider{}.parseConfig(map[string]string{"bindingKeys": "a|openagrinet:One"}) + if err != nil { + t.Fatalf("parseConfig() returned an unexpected error: %v", err) + } + if cfg.ProviderIDAt != "" || cfg.CapabilityCodeAt != "" { + t.Errorf("override = %q / %q, want both empty", cfg.ProviderIDAt, cfg.CapabilityCodeAt) + } +} + +func TestNew(t *testing.T) { + t.Parallel() + + t.Run("rejects a nil context", func(t *testing.T) { + t.Parallel() + + //nolint:staticcheck // deliberately passing a nil context to assert the guard. + _, _, err := mandiProvider{}.New(nil, stubRegistry{}, stubMapper{}, map[string]string{}) + if err == nil { + t.Fatal("expected an error for a nil context, got none") + } + }) + + t.Run("rejects an unparseable config", func(t *testing.T) { + t.Parallel() + + _, _, err := mandiProvider{}.New(context.Background(), stubRegistry{}, stubMapper{}, + map[string]string{"maxResponseBytes": "lots"}) + if err == nil { + t.Fatal("expected an error for an invalid cap, got none") + } + }) + + t.Run("propagates an invalid auth scheme from New", func(t *testing.T) { + t.Parallel() + + _, _, err := mandiProvider{}.New(context.Background(), stubRegistry{}, stubMapper{}, + map[string]string{"authScheme": "oauth"}) + if err == nil { + t.Fatal("expected an unknown auth scheme to be refused") + } + }) + + // A domain plugin serves a family of capabilities, so it cannot guess which + // of them a deployment has providers for. Refused at startup rather than + // answering to nothing. + t.Run("refuses a config naming no capability", func(t *testing.T) { + t.Parallel() + + if _, _, err := (mandiProvider{}).New(context.Background(), stubRegistry{}, stubMapper{}, + map[string]string{}); err == nil { + t.Fatal("expected a config with no bindingKeys to be refused") + } + }) + + t.Run("returns the step and its closer on a config that loads", func(t *testing.T) { + // Not parallel: it swaps the package-level newStepFunc. + closed := false + original := newStepFunc + newStepFunc = func(context.Context, definition.ProviderRecordLookup, definition.Mapper, + *mandi.Config) (definition.Step, func() error, error) { + return nil, func() error { closed = true; return nil }, nil + } + defer func() { newStepFunc = original }() + + _, closer, err := mandiProvider{}.New(context.Background(), stubRegistry{}, stubMapper{}, + map[string]string{"bindingKeys": "agmarknet|openagrinet:MandiPrice"}) + if err != nil { + t.Fatalf("New() returned an unexpected error: %v", err) + } + if closer == nil { + t.Fatal("New() returned no closer, so nothing can release the step") + } + if err := closer(); err != nil { + t.Errorf("closer() returned %v, want nil", err) + } + if !closed { + t.Error("closer() did not reach the step's own closer") + } + }) + + t.Run("propagates a failure from the step constructor", func(t *testing.T) { + original := newStepFunc + wanted := errors.New("upstream refused the config") + newStepFunc = func(context.Context, definition.ProviderRecordLookup, definition.Mapper, + *mandi.Config) (definition.Step, func() error, error) { + return nil, nil, wanted + } + defer func() { newStepFunc = original }() + + _, _, err := mandiProvider{}.New(context.Background(), stubRegistry{}, stubMapper{}, + map[string]string{"bindingKeys": "agmarknet|openagrinet:MandiPrice"}) + if !errors.Is(err, wanted) { + t.Errorf("New() error = %v, want it to wrap %v", err, wanted) + } + }) +} diff --git a/pkg/plugin/implementation/mandi/mandi.go b/pkg/plugin/implementation/mandi/mandi.go new file mode 100644 index 00000000..26e8b94b --- /dev/null +++ b/pkg/plugin/implementation/mandi/mandi.go @@ -0,0 +1,38 @@ +// Package mandi serves the network's market price capabilities. +// +// One package per schema pack family, so which plugin owns a capability is +// readable from its binding key: openagrinet:MandiPrice is mandi's, +// openagrinet:WeatherObservation is weather's. +// +// Almost nothing lives here, and that is the point. Recognising a capability, +// resolving the call plan, authenticating, calling with the registry's budget +// and translating in both directions are all internal/upstream's, because none +// of them differ by domain. What this package owns is its name, and +// prerequisites -- the work a mapping cannot express, which is domain knowledge +// by definition. +// +// The upstream this was written against is Agmarknet's Vistaar API, whose +// select takes governed codes for state, district, market and commodity plus a +// date range, all of which a MandiPrice payload carries. So the package is a +// name and nothing else: see prerequisites.go for why that is worth stating. +package mandi + +import ( + "context" + + "github.com/beckn-one/beckn-onix/pkg/plugin/definition" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/upstream" +) + +// Config is upstream's, unchanged. Aliased here so a domain plugin's cmd package +// need not know where the machinery lives. +type Config = upstream.Config + +// New creates the mandi step. +// +// Which capabilities it answers to is configuration, with no default: a package +// serving a family cannot guess which of them a deployment has providers for. +func New(ctx context.Context, registry definition.ProviderRecordLookup, mapper definition.Mapper, + cfg *Config) (definition.Step, func() error, error) { + return upstream.New(ctx, registry, mapper, prerequisites, cfg) +} diff --git a/pkg/plugin/implementation/mandi/mappings_test.go b/pkg/plugin/implementation/mandi/mappings_test.go new file mode 100644 index 00000000..6be53f46 --- /dev/null +++ b/pkg/plugin/implementation/mandi/mappings_test.go @@ -0,0 +1,844 @@ +package mandi_test + +// mappings_test.go runs the shipped mandi mapping through the real mapper and +// the real provider step. It is the only test that proves the three pieces fit: +// a mapping is JSONata inside YAML fetched over HTTP, and nothing but running +// it establishes that what is published actually produces valid Beckn. +// +// An external test package on purpose -- it uses the plugins exactly as the +// adapter does, through their exported surface and nothing else. + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "slices" + "strings" + "testing" + + "github.com/beckn-one/beckn-onix/pkg/model" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/jsonmapper" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/mandi" +) + +// mappingsDir is where the shipped mappings live, relative to this package. +const mappingsDir = "../../../../config/mappings/agmarknet" + +// shippedMapping is the file this binding-action publishes: one file, both +// directions. The action segment of the name must match the action the registry +// entry declares -- a mismatch would apply a correct mapping to the wrong call, +// silently. +const shippedMapping = "mandi-price.select.yaml" + +// shippedCapability is what the pack calls this capability, and the second half +// of the binding key the registry indexes the provider's record by. +const shippedCapability = "openagrinet:MandiPrice" + +const shippedBindingKey = "agmarknet|" + shippedCapability + +// selectRequest is a MandiPrice select in OnDemand mode: it names the market and +// commodity it wants prices for, and carries no prices of its own -- the pack +// forbids that combination. +const selectRequest = `{ + "context": { + "version": "2.0.0", + "action": "select", + "networkId": "oan-dev", + "transactionId": "9f2c1a8e-4b70-4d31-9c55-6f2e0b1d7a44", + "messageId": "7d41b9e0-52a6-4c18-8b73-1e9f0a4c6d22", + "timestamp": "2026-09-03T06:12:01.330Z" + }, + "message": { + "contract": { + "commitments": [ + { + "status": { "descriptor": { "code": "DRAFT", "name": "Draft" } }, + "resources": [ + { + "id": "res:agmarknet:price-enquiry", + "quantity": 1, + "resourceAttributes": { + "@context": "https://schemas.openagrinet.global/schema/MandiPrice/v0.1/context.jsonld", + "@type": "openagrinet:MandiPrice", + "informationMode": "OnDemand", + "subjectCategories": ["Market"], + "supportedCommodities": [{ "code": "2", "name": "Paddy(Common)" }], + "supportedPriceFields": ["Minimum", "Maximum", "Modal"], + "market": { + "marketName": "Kasdol APMC", + "marketCode": "2056", + "district": "96", + "state": "CG" + }, + "validity": { "startsAt": "2025-08-20", "endsAt": "2025-08-21" } + } + } + ], + "offer": { + "id": "offer:agmarknet:open-data", + "resourceIds": ["res:agmarknet:price-enquiry"], + "provider": { + "id": "agmarknet", + "descriptor": { "code": "AGMARKNET-01", "name": "Agmarknet Vistaar" } + } + } + } + ] + } + } +}` + +// providerResponse is a verbatim Agmarknet Vistaar answer, taken from the +// working example in the provider backend's own documentation. Two records, so +// the mapping is exercised on a list rather than a single object. +// +// Note what it is: Title Case keys WITH SPACES, and prices as STRINGS. Both are +// the reason the mapping needs backticks and $number, and pinning a real +// capture here is what keeps that honest. +const providerResponse = `[ + { + "Grade": "Non-FAQ", + "Group": "Cereals", + "State": "Chattisgarh", + "Market": "Kasdol APMC", + "Variety": "D.B.", + "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Max Price": "2100", + "Min Price": "1900", + "Price Unit": "Rs./Qtl", + "Modal Price": "2000", + "Arrival Date": "20-08-2025" + }, + { + "Grade": "FAQ", + "Group": "Cereals", + "State": "Chattisgarh", + "Market": "Kasdol APMC", + "Variety": "Common", + "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Price Unit": "Rs./Qtl", + "Modal Price": "2050", + "Arrival Date": "21-08-2025" + } +]` + +// serveMappings publishes the shipped mapping files over HTTP, which is how the +// mapper fetches them in production. +func serveMappings(t *testing.T) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := os.ReadFile(filepath.Join(mappingsDir, filepath.Base(r.URL.Path))) + if err != nil { + t.Errorf("could not read the mapping %q: %v", r.URL.Path, err) + w.WriteHeader(http.StatusNotFound) + return + } + fmt.Fprint(w, string(body)) + })) +} + +// stubRegistry answers with the call plan the live registry holds for this +// capability. +type stubRegistry struct{ plan *model.ProviderRecord } + +func (s *stubRegistry) ProviderRecord(context.Context, string) (*model.ProviderRecord, error) { + return s.plan, nil +} + +// runShipped drives the real step over the real mapping and returns the query +// the provider saw and the answer produced. +func runShipped(t *testing.T, request string) (url.Values, map[string]any) { + return runShippedWith(t, request, providerResponse) +} + +// runShippedWith is runShipped with the upstream's answer under the test's +// control, for the cases where what Agmarknet returns is the thing being +// exercised rather than the request that fetched it. +func runShippedWith(t *testing.T, request, providerBody string) (url.Values, map[string]any) { + t.Helper() + + mappings := serveMappings(t) + defer mappings.Close() + + var gotQuery url.Values + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotQuery = r.URL.Query() + fmt.Fprint(w, providerBody) + })) + defer upstream.Close() + + mapper, closeMapper, err := jsonmapper.New(context.Background(), &jsonmapper.Config{}) + if err != nil { + t.Fatalf("failed to build the mapper: %v", err) + } + defer closeMapper() + + registry := &stubRegistry{plan: &model.ProviderRecord{ + BindingKey: shippedBindingKey, + ParticipantID: "agmarknet", + CapabilityCode: shippedCapability, + BaseURL: upstream.URL, + Actions: map[string]model.ActionPlan{ + "select": {Method: http.MethodGet, Path: "/v1/fetch-agmarknet-vistaar", + Mappings: mappings.URL + "/" + shippedMapping, TimeoutMs: 30000, RetryMax: 3}, + }, + }} + + step, closeStep, err := mandi.New(context.Background(), registry, mapper, + &mandi.Config{BindingKeys: []string{shippedBindingKey}}) + if err != nil { + t.Fatalf("failed to build the step: %v", err) + } + defer closeStep() + + stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)} + if err := step.Run(stepCtx); err != nil { + t.Fatalf("Run() returned an unexpected error: %v", err) + } + if len(stepCtx.ResponseBody) == 0 { + t.Fatal("the step produced no answer") + } + var answer map[string]any + if err := json.Unmarshal(stepCtx.ResponseBody, &answer); err != nil { + t.Fatalf("the answer is not JSON: %v\n%s", err, stepCtx.ResponseBody) + } + return gotQuery, answer +} + +func TestShippedMappingServesARealSelect(t *testing.T) { + gotQuery, answer := runShipped(t, selectRequest) + + // --- the request reached the provider as Agmarknet expects -------------- + // Every one of these comes off the payload. Nothing was resolved before the + // call, which is the whole claim of this plugin having no prerequisites. + for param, want := range map[string]string{ + "statecode": "CG", + "districtcode": "96", + "marketcode": "2056", + "commoditycode": "2", + // dd-MM-yyyy, not the ISO the payload carried. + "from_date": "20-08-2025", + "to_date": "21-08-2025", + } { + if got := gotQuery.Get(param); got != want { + t.Errorf("upstream query %s = %q, want %q", param, got, want) + } + } + // The credential is the adapter's business, never the mapping's. + if gotQuery.Has("token") { + t.Error("the mapping must not put a token in the query; authScheme does that") + } + + // --- the answer is Beckn ----------------------------------------------- + beckncontext, _ := answer["context"].(map[string]any) + if beckncontext["action"] != "on_select" { + t.Errorf("action = %v, want on_select", beckncontext["action"]) + } + if beckncontext["transactionId"] != "9f2c1a8e-4b70-4d31-9c55-6f2e0b1d7a44" { + t.Errorf("transactionId = %v, want the one from the request", beckncontext["transactionId"]) + } + // A mapping transforms a payload; it does not assert who anyone is. + for _, field := range []string{"bapId", "bapUri", "bppId", "bppUri"} { + if _, present := beckncontext[field]; present { + t.Errorf("response context carries %q; a mapping must not assert identity", field) + } + } + + // Written out so the answer can be validated against the Beckn v2 spec and + // the MandiPrice pack by tooling outside Go. Skipped unless asked for. + if path := os.Getenv("MANDI_DUMP_ANSWER"); path != "" { + raw, _ := json.MarshalIndent(answer, "", " ") + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatalf("could not write the answer: %v", err) + } + } + + commitment := firstCommitment(t, answer) + if status := commitment["status"].(map[string]any)["descriptor"].(map[string]any); status["code"] != "DRAFT" { + t.Errorf("status = %v, want DRAFT -- the spec's enum is DRAFT, ACTIVE, CLOSED", status["code"]) + } + + // --- one resource per price record -------------------------------------- + resources, _ := commitment["resources"].([]any) + if len(resources) != 2 { + t.Fatalf("got %d resources, want 2 -- one per record the provider answered with", len(resources)) + } + + returned := make([]string, 0, len(resources)) + for _, entry := range resources { + resource, _ := entry.(map[string]any) + id, _ := resource["id"].(string) + // Codes, not names: no spaces or brackets in an identifier. + if !strings.HasPrefix(id, "res:agmarknet:2056:2:") { + t.Errorf("resource id = %q, want one built from the market and commodity codes", id) + } + if strings.ContainsAny(id, " ()") { + t.Errorf("resource id %q contains a space or bracket; use codes, not display names", id) + } + // Required by Commitment.resources in the spec even though the spec + // defines no quantity property. + if _, present := resource["quantity"]; !present { + t.Errorf("resource %s carries no quantity", id) + } + returned = append(returned, id) + } + + // The offer must reference what was actually returned, not what was asked + // for. This is the assertion that fails the moment the offer is echoed. + offer, _ := commitment["offer"].(map[string]any) + referenced, _ := offer["resourceIds"].([]any) + if len(referenced) != len(returned) { + t.Fatalf("offer references %d resources, want %d", len(referenced), len(returned)) + } + for _, reference := range referenced { + if !slices.Contains(returned, reference.(string)) { + t.Errorf("offer references %v, which is not among the resources returned", reference) + } + } + if offer["id"] != "offer:agmarknet:open-data" { + t.Errorf("offer id = %v, want the one the request offered", offer["id"]) + } + + // --- the MandiPrice pack, Direct mode ----------------------------------- + first, _ := resources[0].(map[string]any) + attributes, _ := first["resourceAttributes"].(map[string]any) + for _, f := range []struct{ key, want string }{ + {"@type", "openagrinet:MandiPrice"}, + {"informationMode", "Direct"}, + } { + if attributes[f.key] != f.want { + t.Errorf("%s = %v, want %v", f.key, attributes[f.key], f.want) + } + } + // Direct requires all six of these. + for _, required := range []string{"source", "commodity", "market", "arrivalDate", "prices", "generatedAt"} { + if attributes[required] == nil { + t.Errorf("resourceAttributes carries no %q", required) + } + } + // OnDemand's fields must NOT appear: the pack forbids prices alongside + // them, and an answer advertising a capability is a category error. + for _, absent := range []string{"supportedCommodities", "supportedPriceFields"} { + if _, present := attributes[absent]; present { + t.Errorf("a Direct answer must not carry %q", absent) + } + } + + // --- the prices, converted from strings --------------------------------- + prices, _ := attributes["prices"].(map[string]any) + for field, want := range map[string]float64{"minimum": 1900, "maximum": 2100, "modal": 2000} { + got, ok := prices[field].(float64) + if !ok { + t.Errorf("prices.%s = %#v, want a number -- the upstream sends strings", field, prices[field]) + continue + } + if got != want { + t.Errorf("prices.%s = %v, want %v", field, got, want) + } + } + if prices["currency"] != "INR" || prices["unit"] != "Rs./Qtl" { + t.Errorf("prices currency/unit = %v/%v, want INR/Rs./Qtl", prices["currency"], prices["unit"]) + } + + // arrivalDate is ISO in the answer, though the upstream reported dd-MM-yyyy. + if attributes["arrivalDate"] != "2025-08-20" { + t.Errorf("arrivalDate = %v, want 2025-08-20 in ISO", attributes["arrivalDate"]) + } + + // The pack's enum is Crop, Livestock, Weather, Market, Scheme, Knowledge, + // Service -- so "Market", not "MarketPrice". Echoed from the request, which + // is why getting it wrong there would produce an invalid answer here. + categories, _ := attributes["subjectCategories"].([]any) + if len(categories) != 1 || categories[0] != "Market" { + t.Errorf("subjectCategories = %v, want [Market] from the pack's enum", categories) + } + + market, _ := attributes["market"].(map[string]any) + if market["marketName"] != "Kasdol APMC" || market["state"] != "Chattisgarh" { + t.Errorf("market = %v, want the names the provider reported", market) + } + + // --- a record the market reported only partially ------------------------- + // The second record has no Min or Max Price. Those must be absent, not zero: + // a consumer must be able to tell "not reported" from "reported as zero". + second, _ := resources[1].(map[string]any) + secondPrices, _ := second["resourceAttributes"].(map[string]any)["prices"].(map[string]any) + for _, absent := range []string{"minimum", "maximum"} { + if _, present := secondPrices[absent]; present { + t.Errorf("prices.%s is present for a record that did not report it", absent) + } + } + if secondPrices["modal"] != float64(2050) { + t.Errorf("the second record's modal price = %v, want 2050", secondPrices["modal"]) + } +} + +// The pack leaves every field this upstream needs optional, so a spec-valid +// select can still be unanswerable. The mapping refuses those before the +// provider is called, with its own message. +func TestShippedMappingRefusesWhatItCannotServe(t *testing.T) { + for _, tc := range []struct{ name, drop, expect string }{ + {"no commodity code", "supportedCommodities", "commodity code"}, + {"no market codes", "market", "codes in market"}, + {"no validity window", "validity", "validity window"}, + } { + t.Run(tc.name, func(t *testing.T) { + // Built by deleting a key from the decoded fixture rather than by + // editing its text: removing the last member of an object leaves a + // trailing comma, and the resulting parse error would look like a + // mapping failure. + var payload map[string]any + if err := json.Unmarshal([]byte(selectRequest), &payload); err != nil { + t.Fatalf("the fixture is not JSON: %v", err) + } + attributes := payload["message"].(map[string]any)["contract"].(map[string]any)["commitments"].([]any)[0].(map[string]any)["resources"].([]any)[0].(map[string]any)["resourceAttributes"].(map[string]any) + if _, present := attributes[tc.drop]; !present { + t.Fatalf("the fixture has no %q, so this case tests nothing", tc.drop) + } + delete(attributes, tc.drop) + body, err := json.Marshal(payload) + if err != nil { + t.Fatalf("could not rebuild the payload: %v", err) + } + + mappings := serveMappings(t) + defer mappings.Close() + + called := false + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + fmt.Fprint(w, providerResponse) + })) + defer upstream.Close() + + mapper, closeMapper, err := jsonmapper.New(context.Background(), &jsonmapper.Config{}) + if err != nil { + t.Fatalf("failed to build the mapper: %v", err) + } + defer closeMapper() + + registry := &stubRegistry{plan: &model.ProviderRecord{ + BindingKey: shippedBindingKey, BaseURL: upstream.URL, + Actions: map[string]model.ActionPlan{ + "select": {Method: http.MethodGet, Path: "/v1/fetch-agmarknet-vistaar", + Mappings: mappings.URL + "/" + shippedMapping, TimeoutMs: 30000}, + }, + }} + step, closeStep, err := mandi.New(context.Background(), registry, mapper, + &mandi.Config{BindingKeys: []string{shippedBindingKey}}) + if err != nil { + t.Fatalf("failed to build the step: %v", err) + } + defer closeStep() + + stepCtx := &model.StepContext{Context: t.Context(), Body: body} + if err := step.Run(stepCtx); err == nil { + t.Fatal("expected an unserviceable payload to be refused") + } else if !strings.Contains(err.Error(), tc.expect) { + t.Errorf("error %q should carry the mapping's own message about %q", err, tc.expect) + } + if called { + t.Error("the provider was called for a payload the mapping refuses") + } + }) + } +} + +// firstCommitment reaches the one commitment an answer carries. +func firstCommitment(t *testing.T, answer map[string]any) map[string]any { + t.Helper() + message, _ := answer["message"].(map[string]any) + contract, _ := message["contract"].(map[string]any) + commitments, _ := contract["commitments"].([]any) + if len(commitments) != 1 { + t.Fatalf("got %d commitments, want 1", len(commitments)) + } + commitment, _ := commitments[0].(map[string]any) + return commitment +} + +// resourcesOf returns the answer's resources as maps, so a test can look at +// each record the provider's rows became. +func resourcesOf(t *testing.T, answer map[string]any) []map[string]any { + t.Helper() + + raw, _ := firstCommitment(t, answer)["resources"].([]any) + out := make([]map[string]any, 0, len(raw)) + for _, r := range raw { + m, ok := r.(map[string]any) + if !ok { + t.Fatalf("resource is not an object: %#v", r) + } + out = append(out, m) + } + return out +} + +// Agmarknet writes an unreported price as a marker rather than omitting the +// field -- "NR", "-", "". $exists() is true for all of them, so an +// existence-only guard handed them to $number() and it threw D3030, which +// failed the WHOLE response: one unreported cell in one row turned a good +// multi-row answer into an adapter error. +func TestShippedMappingSurvivesUnreportedPrices(t *testing.T) { + t.Parallel() + + // Row one has a marker in TWO price fields and a real modal, so it stays + // and its markers must come back absent. Row two is ordinary. Row three + // has markers in all three, so it has nothing to report and is dropped. + const withMarkers = `[ + { + "Grade": "Non-FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "D.B.", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "NR", "Max Price": "-", "Modal Price": "2000", + "Price Unit": "Rs./Qtl", "Arrival Date": "20-08-2025" + }, + { + "Grade": "FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "Common", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "1900", "Max Price": "2100", "Modal Price": "2050", + "Price Unit": "Rs./Qtl", "Arrival Date": "21-08-2025" + }, + { + "Grade": "FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "Common", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "NR", "Max Price": "NR", "Modal Price": "NR", + "Price Unit": "Rs./Qtl", "Arrival Date": "22-08-2025" + } + ]` + + _, answer := runShippedWith(t, selectRequest, withMarkers) + + byDate := map[string]map[string]any{} + for _, r := range resourcesOf(t, answer) { + ra := r["resourceAttributes"].(map[string]any) + byDate[ra["arrivalDate"].(string)] = ra + } + + // The row with a real modal survives -- one unreported cell must not + // discard it, and must not discard the rows beside it either. + partial, ok := byDate["2025-08-20"] + if !ok { + t.Fatal("the partially-priced row is missing; an unreported cell discarded it") + } + prices := partial["prices"].(map[string]any) + if _, isNum := prices["modal"].(float64); !isNum { + t.Errorf("prices.modal = %#v, want the reported number", prices["modal"]) + } + // Absent, not zero: a consumer must tell "no minimum reported" from + // "the minimum was zero". + for _, field := range []string{"minimum", "maximum"} { + if v, present := prices[field]; present { + t.Errorf("prices.%s = %#v for an unreported price; absent is honest, zero is a lie", field, v) + } + } + + if _, ok := byDate["2025-08-21"]; !ok { + t.Error("the fully-priced row is missing from the answer") + } + + // Nothing to report at all: the pack's prices.anyOf cannot be satisfied, + // so the row is dropped rather than emitted as an invalid resource. + if _, ok := byDate["2025-08-22"]; ok { + t.Error("a row whose every price is unreported was emitted; it cannot satisfy prices.anyOf") + } +} + +// A record that cannot produce a conformant resource is dropped, not emitted +// with a degenerate value. Absent is honest; present-and-wrong is a lie in the +// shape of an answer, and it is signed. +func TestShippedMappingDropsRecordsItCannotMakeConformant(t *testing.T) { + t.Parallel() + + // One good row, then one for each way a record fails the pack. + const mixed = `[ + { + "Grade": "FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "Common", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "1900", "Max Price": "2100", "Modal Price": "2000", + "Price Unit": "Rs./Qtl", "Arrival Date": "20-08-2025" + }, + { + "Grade": "FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "Common", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "1800", "Max Price": "2000", "Modal Price": "1900", + "Price Unit": "Rs./Qtl" + }, + { + "Grade": "FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "Common", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "1700", "Max Price": "1900", "Modal Price": "1800", + "Arrival Date": "23-08-2025" + } + ]` + + _, answer := runShippedWith(t, selectRequest, mixed) + + resources := resourcesOf(t, answer) + if len(resources) != 1 { + var dates []string + for _, r := range resources { + ra := r["resourceAttributes"].(map[string]any) + dates = append(dates, fmt.Sprint(ra["arrivalDate"])) + } + t.Fatalf("got %d resources with dates %v, want only the conformant one", len(resources), dates) + } + + ra := resources[0]["resourceAttributes"].(map[string]any) + if ra["arrivalDate"] != "2025-08-20" { + t.Errorf("arrivalDate = %v, want the one good record", ra["arrivalDate"]) + } + // The degenerate date the old mapping produced, specifically. + if ra["arrivalDate"] == "--" { + t.Error(`arrivalDate = "--", which the pack refuses as format: date`) + } + if _, ok := ra["prices"].(map[string]any)["unit"]; !ok { + t.Error("prices.unit is missing, which the pack requires") + } + + // The offer must not reference a resource the filter removed -- dropping a + // record and leaving its id in resourceIds would trade an invalid resource + // for a dangling reference. + ids := map[string]bool{} + for _, r := range resources { + ids[r["id"].(string)] = true + } + offer, _ := firstCommitment(t, answer)["offer"].(map[string]any) + referenced, _ := offer["resourceIds"].([]any) + if len(referenced) != len(resources) { + t.Errorf("offer references %d resources, want %d", len(referenced), len(resources)) + } + for _, ref := range referenced { + if !ids[fmt.Sprint(ref)] { + t.Errorf("the offer references %q, which is not among the answer's resources", ref) + } + } +} + +// The answer must assert what it knows, not repeat what it was told. Both of +// these were echoes of the request, and an echo is a claim this adapter signs. +func TestShippedMappingStatesWhatItKnowsRatherThanEchoing(t *testing.T) { + t.Parallel() + + // A caller asking a MandiPrice question with the WRONG category. It is + // enum-legal, so nothing downstream would reject it -- which is exactly + // why echoing it was dangerous. + wrongCategory := strings.Replace(selectRequest, + `"subjectCategories": ["Market"]`, `"subjectCategories": ["Weather"]`, 1) + if wrongCategory == selectRequest { + t.Fatal("the fixture no longer states subjectCategories; this test needs updating") + } + + _, answer := runShippedWith(t, wrongCategory, providerResponse) + + for _, r := range resourcesOf(t, answer) { + ra := r["resourceAttributes"].(map[string]any) + + // Stated from the pack, not taken from the caller. + cats, _ := ra["subjectCategories"].([]any) + if len(cats) != 1 || cats[0] != "Market" { + t.Errorf(`subjectCategories = %#v, want ["Market"] regardless of what the request said`, ra["subjectCategories"]) + } + + market, _ := ra["market"].(map[string]any) + // The upstream reports no market code, so the answer must not claim one. + if code, present := market["marketCode"]; present { + t.Errorf("market.marketCode = %#v; this upstream reports no code, so asserting one is unfounded", code) + } + // And what is there comes from the record. + if market["marketName"] != "Kasdol APMC" || market["state"] != "Chattisgarh" { + t.Errorf("market = %#v, want the values the provider reported", market) + } + // marketName is the one member the pack requires. + if _, ok := market["marketName"]; !ok { + t.Error("market.marketName is missing, which the pack requires") + } + } +} + +// The guards now check what they claim to. Both of these payloads are legal +// against the pack and unanswerable by this upstream, which is the gap between +// "valid" and "serviceable" the required block exists to close. +func TestShippedMappingRefusesPayloadsItCannotAnswer(t *testing.T) { + for _, tc := range []struct { + name string + mutate func(map[string]any) + expect string + }{ + { + // The pack describes district and state as "name or governed + // code", so this validates -- and went to Agmarknet verbatim as + // codes, which answered nothing. The caller then received a + // signed, spec-valid "no prices" for a market that had prices. + name: "names where the upstream wants codes", + mutate: func(ra map[string]any) { + ra["market"] = map[string]any{ + "marketName": "Kasdol APMC", + "district": "Balodabazar", + "state": "Chattisgarh", + } + }, + expect: "codes in market", + }, + { + // Everything downstream reads supportedCommodities[0], so this + // used to be queried for Paddy alone and answered confidently. + name: "more commodities than one request can serve", + mutate: func(ra map[string]any) { + ra["supportedCommodities"] = []any{ + map[string]any{"code": "2", "name": "Paddy(Common)"}, + map[string]any{"code": "3", "name": "Wheat"}, + } + }, + expect: "exactly one commodity", + }, + } { + t.Run(tc.name, func(t *testing.T) { + var payload map[string]any + if err := json.Unmarshal([]byte(selectRequest), &payload); err != nil { + t.Fatalf("the fixture is not JSON: %v", err) + } + ra := payload["message"].(map[string]any)["contract"].(map[string]any)["commitments"].([]any)[0].(map[string]any)["resources"].([]any)[0].(map[string]any)["resourceAttributes"].(map[string]any) + tc.mutate(ra) + body, err := json.Marshal(payload) + if err != nil { + t.Fatalf("could not rebuild the payload: %v", err) + } + + mappings := serveMappings(t) + defer mappings.Close() + + called := false + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + called = true + fmt.Fprint(w, providerResponse) + })) + defer upstream.Close() + + mapper, closeMapper, err := jsonmapper.New(context.Background(), &jsonmapper.Config{}) + if err != nil { + t.Fatalf("failed to build the mapper: %v", err) + } + defer closeMapper() + + registry := &stubRegistry{plan: &model.ProviderRecord{ + BindingKey: shippedBindingKey, BaseURL: upstream.URL, + Actions: map[string]model.ActionPlan{ + "select": {Method: http.MethodGet, Path: "/v1/fetch-agmarknet-vistaar", + Mappings: mappings.URL + "/" + shippedMapping, TimeoutMs: 30000}, + }, + }} + step, closeStep, err := mandi.New(context.Background(), registry, mapper, + &mandi.Config{BindingKeys: []string{shippedBindingKey}}) + if err != nil { + t.Fatalf("failed to build the step: %v", err) + } + defer closeStep() + + stepCtx := &model.StepContext{Context: t.Context(), Body: body} + if err := step.Run(stepCtx); err == nil { + t.Fatal("expected a payload this upstream cannot answer to be refused") + } else if !strings.Contains(err.Error(), tc.expect) { + t.Errorf("error %q should explain the refusal in terms of %q", err, tc.expect) + } + // The point of refusing early: the upstream is never troubled with + // a request that cannot produce an answer. + if called { + t.Error("the provider was called for a payload the mapping refuses") + } + }) + } +} + +// Agmarknet routinely reports several rows for the same market, commodity and +// date differing only by Variety and Grade. The id was built from +// scope:commodity:date, so those rows collided: two distinct resources under +// one id, referenced twice by the offer, and a consumer resolving resourceIds +// could not tell which price it had. +func TestShippedMappingGivesCollidingRowsDistinctIDs(t *testing.T) { + t.Parallel() + + // The same pair as this package's own fixture -- FAQ/Common and + // Non-FAQ/D.B. -- but sharing an arrival date, which is what the fixture + // was accidentally saved by not doing. + const sameDay = `[ + { + "Grade": "Non-FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "D.B.", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "1900", "Max Price": "2100", "Modal Price": "2000", + "Price Unit": "Rs./Qtl", "Arrival Date": "20-08-2025" + }, + { + "Grade": "FAQ", "Group": "Cereals", "State": "Chattisgarh", + "Market": "Kasdol APMC", "Variety": "Common", "District": "Balodabazar", + "Commodity": "Paddy(Common)", + "Min Price": "1600", "Max Price": "1800", "Modal Price": "1700", + "Price Unit": "Rs./Qtl", "Arrival Date": "20-08-2025" + } + ]` + + _, answer := runShippedWith(t, selectRequest, sameDay) + + resources := resourcesOf(t, answer) + if len(resources) != 2 { + t.Fatalf("got %d resources, want both rows", len(resources)) + } + seen := map[string]int{} + for _, r := range resources { + seen[r["id"].(string)]++ + } + for id, n := range seen { + if n > 1 { + t.Errorf("id %q is shared by %d resources; the offer cannot reference one of them unambiguously", id, n) + } + } + if len(seen) != 2 { + t.Errorf("got %d distinct ids for 2 rows: %v", len(seen), seen) + } +} + +// supportedPriceFields was validated on the way in and ignored on the way out, +// so a caller asking for one field got all three. +func TestShippedMappingHonoursTheRequestedPriceFields(t *testing.T) { + t.Parallel() + + modalOnly := strings.Replace(selectRequest, + `"supportedPriceFields": ["Minimum", "Maximum", "Modal"]`, + `"supportedPriceFields": ["Modal"]`, 1) + if modalOnly == selectRequest { + t.Fatal("the fixture no longer lists all three price fields; this test needs updating") + } + + _, answer := runShippedWith(t, modalOnly, providerResponse) + + for _, r := range resourcesOf(t, answer) { + prices := r["resourceAttributes"].(map[string]any)["prices"].(map[string]any) + if _, ok := prices["modal"]; !ok { + t.Error("prices.modal is missing, and it is the field that was asked for") + } + for _, unasked := range []string{"minimum", "maximum"} { + if v, present := prices[unasked]; present { + t.Errorf("prices.%s = %#v was returned though the request did not ask for it", unasked, v) + } + } + // The pack requires these whatever was asked for. + for _, required := range []string{"currency", "unit"} { + if _, ok := prices[required]; !ok { + t.Errorf("prices.%s is missing, which the pack requires", required) + } + } + } +} diff --git a/pkg/plugin/implementation/mandi/prerequisites.go b/pkg/plugin/implementation/mandi/prerequisites.go new file mode 100644 index 00000000..2027248b --- /dev/null +++ b/pkg/plugin/implementation/mandi/prerequisites.go @@ -0,0 +1,20 @@ +package mandi + +import "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/upstream" + +// prerequisites is what a mandi capability needs that its payload does not +// carry, keyed by binding key. +// +// Empty, and for a better reason than weather's: a market price select names +// the market it wants. The MandiPrice pack has no top-level location -- only +// market.marketCode, market.district and market.state -- so there is nothing +// to resolve. Agmarknet's Vistaar select takes exactly those codes, and the +// mapping reads them straight off the payload. +// +// An entry would be needed only for real I/O: a commodity name to resolve to a +// code, a token to exchange, a point to turn into a market. Each of those is a +// different upstream than the one this was written against, and each would +// bring the question of where the provider-to-function binding belongs -- see +// the note in weather/prerequisites.go and prefer keeping the payload explicit +// over adding an entry here. +var prerequisites = upstream.Prerequisites{}