diff --git a/config/mappings/agmarknet/mandi-price.select.yaml b/config/mappings/agmarknet/mandi-price.select.yaml new file mode 100644 index 00000000..5492275d --- /dev/null +++ b/config/mappings/agmarknet/mandi-price.select.yaml @@ -0,0 +1,206 @@ +# 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. +# +# NOTHING HERE IS OUTSIDE THE PACK. openagrinet:MandiPrice v0.1 carries every +# field this answer sets. Where the upstream reports something the pack has no +# home for, it is dropped rather than invented. + +# What this capability cannot serve, refused before the provider is called. +# +# The pack requires none of these: a MandiPrice select is OnDemand, and that +# branch requires only supportedCommodities and supportedPriceFields. It leaves +# market and validity optional, and 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; + $exists($ra.supportedCommodities[0].code) + ) + message: "this capability needs a commodity code in supportedCommodities[0].code" + - check: | + ( + $ra := beckn.message.contract.commitments[0].resources[0].resourceAttributes; + $exists($ra.market.state) and $exists($ra.market.district) + ) + message: "this capability needs governed state and district codes in market" + - 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; + + /* 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. + + The market code is optional: without it the query widened to the whole + district, so the district code is what identifies the scope. */ + $iso := function($ddmmyyyy) { + $substring($ddmmyyyy, 6, 4) & "-" & $substring($ddmmyyyy, 3, 2) & "-" & $substring($ddmmyyyy, 0, 2) + }; + + $scope := $exists($ra.market.marketCode) ? $ra.market.marketCode : $ra.market.district; + $resourceId := function($r) { + "res:agmarknet:" & $scope & ":" & $ra.supportedCommodities[0].code + & ":" & $iso($r.`Arrival Date`) + }; + + /* dd-MM-yyyy back to ISO, so the answer speaks the pack's date format + rather than the upstream's. */ + + /* Absent rather than present-and-empty: a consumer must be able to tell + "the market reported no minimum" from "the minimum was zero". */ + $priced := function($value) { $exists($value) ? $number($value) }; + + { + "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($records, 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($records, 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", + "subjectCategories": $ra.subjectCategories, + "source": { + "sourceId": "agmarknet", + "sourceName": "Agmarknet Vistaar" + }, + "commodity": { + "code": $ra.supportedCommodities[0].code, + "name": $r.Commodity + }, + "commodityGroup": $r.Group, + "variety": $r.Variety, + "grade": $r.Grade, + "market": { + "marketName": $r.Market, + "marketCode": $ra.market.marketCode, + "district": $r.District, + "state": $r.State + }, + "arrivalDate": $iso($r.`Arrival Date`), + "prices": { + "minimum": $priced($r.`Min Price`), + "maximum": $priced($r.`Max Price`), + "modal": $number($r.`Modal Price`), + "currency": "INR", + "unit": $r.`Price Unit` + }, + "generatedAt": $now() + } + } + })] + } + ] + } + } + } + ) diff --git a/config/oan-provider-adapter.yaml b/config/oan-provider-adapter.yaml index 6fdc9b13..18f13dea 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 @@ -233,6 +238,24 @@ 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 + # 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 @@ -240,7 +263,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/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..1eae729f --- /dev/null +++ b/pkg/plugin/implementation/mandi/mappings_test.go @@ -0,0 +1,458 @@ +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) { + 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, 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, + 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", "state and district"}, + {"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 +} 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{}