diff --git a/config/mappings/pmfby/grievance.status.yaml b/config/mappings/pmfby/grievance.status.yaml new file mode 100644 index 000000000..9f6edffb1 --- /dev/null +++ b/config/mappings/pmfby/grievance.status.yaml @@ -0,0 +1,117 @@ +# PMFBY status -> POST /krphapi/FGMS/GetGrievenceTicketsStatus. +# Reads a grievance back by ticket number and phone. Only listed fields are +# returned; the farmer's personal details are dropped. +# Required fields and formats are checked by the Grievance pack. + +request: | + ( + $attributes := beckn.message.contract.commitments[0].commitmentAttributes; + { + "requestorMobileNo": $attributes.applicantMobile, + "GrievenceSupportTicketNo": $attributes.case.ticketNo + } + ) + +# Empty PMFBY fields are dropped. No record, or one for another ticket, is a +# 400 in our words; any other failure is a 502. PMFBY's message is logged, +# never returned. +response: | + ( + /* Constants */ + $PMFBY_SUCCESS_CODE := "1"; + $SCHEME := { "code": "PMFBY", "name": "Pradhan Mantri Fasal Bima Yojana" }; + $INFORMATION_MODE := "Direct"; + $COMMITMENT_STATUS_ACTIVE := "ACTIVE"; + $DEFAULT_CASE_STATUS := "UnderReview"; + $CASE_STATUS_CODES := { "open": "Registered", "resolved": "Resolved" }; + $SEASON_NAMES := { "1": "Kharif", "2": "Rabi", "3": "Zaid" }; + $TIMESTAMP_FORMAT := "[Y0001]-[M01]-[D01]T[H01]:[m01]:[s01]Z"; + + /* e.g. "2026-10-12T11:02:49.000Z" */ + $RFC3339_PATTERN := /^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:[0-9]{2})$/; + $DIGITS_PATTERN := /^[0-9]+$/; + $YEAR_PATTERN := /^[0-9]{4}$/; + + $MSG_NOT_FOUND := "No grievance was found for this ticket number and phone number."; + $MSG_UNAVAILABLE := "PMFBY could not return the grievance status right now. Please try again later."; + + $ERROR_REFUSED := { "status": 400, "code": "BIZ_GENERIC_ERROR" }; + $ERROR_UNAVAILABLE := { "status": 502, "code": "NET_DOWNSTREAM_UNAVAILABLE" }; + /* PMFBY fails everything with responseCode "0", so its wording (lower case) + is the only way to tell a caller mistake from a PMFBY fault. */ + $KNOWN_REFUSALS := { "no record found": $MSG_NOT_FOUND }; + + /* Helpers */ + /* PMFBY sends null or "" for missing values. */ + $isText := function($value) { $type($value) = "string" and $value != "" }; + $asText := function($value) { $isText($value) ? $value }; + /* Ids come as numbers or digit strings, and 0 means none. */ + $asId := function($value) { + ($type($value) = "number" or $type($value) = "string") + and $contains($string($value), $DIGITS_PATTERN) and $number($string($value)) != 0 + ? $string($value) + }; + /* Drops an object with no fields instead of sending {}. */ + $omitIfEmpty := function($object) { $exists($keys($object)[0]) ? $object }; + $error := function($kind, $message, $cause) { + { "_error": $merge([$kind, { "message": $message, "cause": $cause }]) } + }; + + /* Inputs */ + $commitment := beckn.message.contract.commitments[0]; + $attributes := $commitment.commitmentAttributes; + /* [0] reads a bare record or a list of one alike. */ + $record := response.responseDynamic[0]; + $pmfbyMessage := $type(response.responseMessage) = "string" ? $trim(response.responseMessage); + $refusalMessage := $pmfbyMessage ? $lookup($KNOWN_REFUSALS, $lowercase($pmfbyMessage)); + $isAccepted := $string(response.responseCode) = $PMFBY_SUCCESS_CODE; + /* PMFBY reads by phone alone, so a record for another ticket is no record. */ + $isOtherTicket := $exists($record.GrievenceSupportTicketNo) + and $string($record.GrievenceSupportTicketNo) != $string($attributes.case.ticketNo); + $isFound := $isAccepted and ($isText($record.ApplicationNo) or $isText($record.TicketStatus)) and $not($isOtherTicket); + $mappedStatus := $isText($record.TicketStatus) ? $lookup($CASE_STATUS_CODES, $lowercase($trim($record.TicketStatus))); + $caseStatus := $exists($mappedStatus) ? $mappedStatus : $DEFAULT_CASE_STATUS; + $requestYear := $asId($record.RequestYear); + $cropYear := $requestYear and $contains($requestYear, $YEAR_PATTERN) ? $requestYear; + $filedOn := $isText($record.InsertDateTime) and $contains($record.InsertDateTime, $RFC3339_PATTERN) + ? $fromMillis($toMillis($record.InsertDateTime), $TIMESTAMP_FORMAT); + /* The officer's answer; "" until there is one. */ + $reply := $asText($isText($record.latestRemark) ? $trim($record.latestRemark)); + + /* Output */ + $isFound + ? { + "context": $merge([beckn.context, { "action": "on_status", "timestamp": $now() }]), + "message": { + "contract": $merge([beckn.message.contract, { + "commitments": [$merge([$commitment, { + "status": { "descriptor": { "code": $COMMITMENT_STATUS_ACTIVE } }, + "commitmentAttributes": { + "@context": $attributes.`@context`, + "@type": $attributes.`@type`, + "informationMode": $INFORMATION_MODE, + "scheme": $SCHEME, + "enrolmentId": $asText($record.ApplicationNo), + "cropYear": $cropYear, + "season": $lookup($SEASON_NAMES, $asId($record.RequestSeason)), + "grievance": { + "category": $omitIfEmpty({ "code": $asId($record.TicketCategoryID), "name": $asText($record.TicketCategoryName) }), + "subCategory": $omitIfEmpty({ "code": $asId($record.TicketSubCategoryID), "name": $asText($record.TicketSubCategoryName) }), + "description": $asText($record.GrievenceDescription) + }, + "case": { + "ticketNo": $attributes.case.ticketNo, + "status": { "code": $caseStatus }, + "filedOn": $filedOn, + "reply": $reply, + "cropName": $asText($record.CropName) + } + } + }])] + }]) + } + } + : $isAccepted ? $error($ERROR_REFUSED, $MSG_NOT_FOUND) + : $exists($refusalMessage) ? $error($ERROR_REFUSED, $refusalMessage, $pmfbyMessage) + : $error($ERROR_UNAVAILABLE, $MSG_UNAVAILABLE, $pmfbyMessage) + ) diff --git a/config/mappings/pmfby/grievance.support.yaml b/config/mappings/pmfby/grievance.support.yaml new file mode 100644 index 000000000..f38c15544 --- /dev/null +++ b/config/mappings/pmfby/grievance.support.yaml @@ -0,0 +1,112 @@ +# PMFBY support -> POST /krphapi/FGMS/AddKRPHNCIPGrievenceSupportTicket. +# Files a grievance; no OTP. "Grievence" is PMFBY's spelling. +# Required fields and formats are checked by the Grievance pack. +# The two halves cannot share variables, so the default codes are in both. + +request: | + ( + /* Constants */ + $DEFAULT_CATEGORY_CODE := "3"; + $DEFAULT_SUB_CATEGORY_CODE := "10"; + $DEFAULT_RECEIPT_SOURCE_ID := "134306"; + $SEASON_CODES := { "Kharif": "1", "Rabi": "2", "Zaid": "3" }; + + $DATE_FORMAT := "[Y0001]-[M01]-[D01]"; + $IST_OFFSET := "+0530"; + + /* Helpers */ + $orDefault := function($value, $default) { $exists($value) ? $string($value) : $default }; + + /* Inputs */ + $channel := beckn.message.support.channels[0]; + $grievance := $channel.grievance; + $filedAt := $exists($channel.complaintDate) ? $toMillis($channel.complaintDate) : $millis(); + + /* Output */ + { + "requestorMobileNo": $channel.applicantMobile, + "applicationNo": $channel.enrolmentId, + "requestYear": $string($channel.cropYear), + "requestSeason": $lookup($SEASON_CODES, $channel.season), + "ticketCategoryID": $orDefault($grievance.category.code, $DEFAULT_CATEGORY_CODE), + "ticketSubCategoryID": $orDefault($grievance.subCategory.code, $DEFAULT_SUB_CATEGORY_CODE), + "grievenceDescription": $trim($grievance.description), + "complaintDate": $fromMillis($filedAt, $DATE_FORMAT, $IST_OFFSET), + "receiptSourceID": $orDefault($channel.receiptSourceId, $DEFAULT_RECEIPT_SOURCE_ID) + } + ) + +# Echoes the request and adds the ticket. A known PMFBY refusal is a 400 in our +# words; any other failure is a 502. PMFBY's message is logged, never returned. +response: | + ( + /* Constants */ + $DEFAULT_CATEGORY_CODE := "3"; + $DEFAULT_SUB_CATEGORY_CODE := "10"; + $PMFBY_SUCCESS_CODE := "1"; + $PROVIDER_NAME := "PMFBY Grievance Portal"; + $SCHEME := { "code": "PMFBY", "name": "Pradhan Mantri Fasal Bima Yojana" }; + $INFORMATION_MODE := "Direct"; + $CASE_STATUS_REGISTERED := "Registered"; + $TIMESTAMP_FORMAT := "[Y0001]-[M01]-[D01]T[H01]:[m01]:[s01]Z"; + + $MSG_PHONE_MISMATCH := "The phone number does not match the one registered for this application number."; + $MSG_APPLICATION_NOT_FOUND := "No PMFBY application was found for this application number."; + $MSG_UNAVAILABLE := "PMFBY could not register the grievance right now. Please try again later."; + + $ERROR_REFUSED := { "status": 400, "code": "BIZ_GENERIC_ERROR" }; + $ERROR_UNAVAILABLE := { "status": 502, "code": "NET_DOWNSTREAM_UNAVAILABLE" }; + /* PMFBY fails everything with responseCode "0", so its wording (lower case) + is the only way to tell a caller mistake from a PMFBY fault. */ + $KNOWN_REFUSALS := { + "mobile number not matched": $MSG_PHONE_MISMATCH, + "application details not exists": $MSG_APPLICATION_NOT_FOUND + }; + + /* Helpers */ + $orDefault := function($value, $default) { $exists($value) ? $string($value) : $default }; + $error := function($kind, $message, $cause) { + { "_error": $merge([$kind, { "message": $message, "cause": $cause }]) } + }; + + /* Inputs */ + $channel := beckn.message.support.channels[0]; + $grievance := $channel.grievance; + $ticketNo := response.responseDynamic.GrievenceSupportTicketNo; + $pmfbyMessage := $type(response.responseMessage) = "string" ? $trim(response.responseMessage); + $refusalMessage := $pmfbyMessage ? $lookup($KNOWN_REFUSALS, $lowercase($pmfbyMessage)); + $hasTicket := $type($ticketNo) = "number" or ($type($ticketNo) = "string" and $ticketNo != ""); + /* "Ticket Already Raised" also comes back as success with a ticket. */ + $isLodged := $string(response.responseCode) = $PMFBY_SUCCESS_CODE and $hasTicket; + + /* Output */ + $isLodged + ? { + "context": $merge([beckn.context, { "action": "on_support", "timestamp": $now() }]), + "message": { + "support": { + "channels": [{ + "@context": $channel.`@context`, + "@type": $channel.`@type`, + "informationMode": $INFORMATION_MODE, + "provider": { "id": $channel.provider.id, "descriptor": { "name": $PROVIDER_NAME } }, + "scheme": $SCHEME, + "enrolmentId": $channel.enrolmentId, + "grievance": { + "category": { "code": $orDefault($grievance.category.code, $DEFAULT_CATEGORY_CODE) }, + "subCategory": { "code": $orDefault($grievance.subCategory.code, $DEFAULT_SUB_CATEGORY_CODE) }, + "description": $trim($grievance.description) + }, + "case": { + "ticketNo": $string($ticketNo), + "status": { "code": $CASE_STATUS_REGISTERED }, + /* Lodge time, so it agrees with what a status read reports. */ + "filedOn": $fromMillis($millis(), $TIMESTAMP_FORMAT) + } + }] + } + } + } + : $exists($refusalMessage) ? $error($ERROR_REFUSED, $refusalMessage, $pmfbyMessage) + : $error($ERROR_UNAVAILABLE, $MSG_UNAVAILABLE, $pmfbyMessage) + ) diff --git a/config/provider-adapter.yaml b/config/provider-adapter.yaml index e2684cb31..0c28a3815 100644 --- a/config/provider-adapter.yaml +++ b/config/provider-adapter.yaml @@ -236,6 +236,11 @@ modules: extendedSchema_maxCacheSize: "100" extendedSchema_downloadTimeout: "30" + # Pack @types whose errors keep only the first reason, without the + # schema dump, the caller's value or the other branches' reasons. + # Unlisted types keep the full message. + extendedSchema_conciseErrorTypes: "openagrinet:Grievance" + # ------------------------------------------------------------------ # jsonmapper -- the JSONata mapper. # @@ -505,6 +510,32 @@ modules: # concurrency for this capability. searchConcurrency: 1 + # ------------------------------------------------------------------ + # Grievance -- support files a grievance, status reads it back. + # ------------------------------------------------------------------ + - id: Grievance + config: + bindingKeys: "pmfby|openagrinet:Grievance" + + # Paired by position, tried in order: status names the binding on + # its contract, support on its channel. + providerIdAt: "message.contract.commitments[].offer.provider.id, message.support.channels[].provider.id" + capabilityCodeAt: "message.contract.commitments[].commitmentAttributes.@type, message.support.channels[].@type" + + # Logs in at PMFBY's NICUsersLogin and sends the token bare in + # Authorization, held until the login reply's validTo. + # tokenUrl is set per deployment. + pmfby: + authScheme: tokenHeader + tokenUrl: <> + tokenUserField: appAccessUID + tokenUserEnv: PMFBY_ACCESS_UID + tokenSecretField: appAccessPWD + tokenSecretEnv: PMFBY_ACCESS_PWD + tokenResponseField: responseDynamic.token.Token + tokenExpiryField: responseDynamic.token.validTo + headerName: Authorization + # 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 @@ -516,6 +547,7 @@ modules: - MandiPrice # its binding key, or pass through - KnowledgeAdvisory # its binding key, or pass through - AgricultureFacility # its binding key, or pass through + - Grievance # its binding key, or pass through - signAck # signs whatever the step answered with # ---------------------------------------------------------------------------- diff --git a/install/build-plugins.sh b/install/build-plugins.sh index 4e505cc05..519a22822 100755 --- a/install/build-plugins.sh +++ b/install/build-plugins.sh @@ -34,6 +34,7 @@ plugins=( "jsonmapper" "WeatherObservation" "MandiPrice" + "Grievance" "KnowledgeAdvisory" "AgricultureFacility" "manifestloader" diff --git a/pkg/plugin/implementation/Grievance/Grievance.go b/pkg/plugin/implementation/Grievance/Grievance.go new file mode 100644 index 000000000..7073e617d --- /dev/null +++ b/pkg/plugin/implementation/Grievance/Grievance.go @@ -0,0 +1,20 @@ +// Package Grievance serves the network's grievance capabilities: filing a +// grievance against a government scheme and checking on it. Provider details +// live in the registry, the mappings and the adapter config, not here. +package Grievance + +import ( + "context" + + "github.com/beckn-one/beckn-onix/pkg/plugin/definition" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common" +) + +// Config is the shared step config. +type Config = common.Config + +// New creates the grievance step. It has no prerequisites. +func New(ctx context.Context, registry definition.ProviderRecordLookup, mapper definition.Mapper, + cfg *Config) (definition.Step, func() error, error) { + return common.New(ctx, registry, mapper, nil, cfg) +} diff --git a/pkg/plugin/implementation/Grievance/cmd/plugin.go b/pkg/plugin/implementation/Grievance/cmd/plugin.go new file mode 100644 index 000000000..dae3fc93a --- /dev/null +++ b/pkg/plugin/implementation/Grievance/cmd/plugin.go @@ -0,0 +1,92 @@ +// Command plugin builds the grievance provider step as a loadable plugin. The +// .so's filename is the id a deployment names in providerSteps. +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/Grievance" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common" +) + +// grievanceProvider implements definition.ProviderStepProvider. +type grievanceProvider struct{} + +// newStepFunc creates a new step. Indirected for tests. +var newStepFunc = Grievance.New + +// parseConfig turns the plugin config map into a typed Config. Defaults and +// validation are left to Grievance.New. +func (p grievanceProvider) parseConfig(config map[string]string) (*Grievance.Config, error) { + cfg := &Grievance.Config{ + BindingKeys: splitList(config["bindingKeys"]), + ProviderIDAt: config["providerIdAt"], + CapabilityCodeAt: config["capabilityCodeAt"], + } + + 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 + } + + // One credential profile per provider. + auth, err := common.ParseProviderAuth(config) + if err != nil { + return nil, err + } + cfg.AuthByProvider = auth + + return cfg, nil +} + +// New creates a new grievance provider step instance. +func (p grievanceProvider) 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 grievance configuration") + return nil, nil, fmt.Errorf("failed to parse grievance configuration: %w", err) + } + + step, closer, err := newStepFunc(ctx, registry, mapper, cfg) + if err != nil { + log.Errorf(ctx, err, "Failed to create grievance step") + return nil, nil, err + } + + log.Infof(ctx, "Grievance step created successfully") + return step, closer, nil +} + +// splitList reads a comma-separated config value, trimming spaces and dropping +// blanks. +func splitList(raw string) []string { + 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 = grievanceProvider{} + +// Compile-time check that Provider satisfies the interface. +var _ definition.ProviderStepProvider = Provider diff --git a/pkg/plugin/implementation/Grievance/cmd/plugin_test.go b/pkg/plugin/implementation/Grievance/cmd/plugin_test.go new file mode 100644 index 000000000..dfbbc4b9c --- /dev/null +++ b/pkg/plugin/implementation/Grievance/cmd/plugin_test.go @@ -0,0 +1,173 @@ +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/Grievance" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common" +) + +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 +} + +const ( + pmfbyBindingKey = "pmfby|openagrinet:Grievance" + providerIDAt = "message.contract.commitments[].offer.provider.id, message.support.channels[].provider.id" + capabilityCodeAt = "message.contract.commitments[].commitmentAttributes.@type, message.support.channels[].@type" +) + +// pmfbyConfig is the block config/provider-adapter.yaml ships, as the plugin +// receives it: flattened. +func pmfbyConfig() map[string]string { + return map[string]string{ + "bindingKeys": pmfbyBindingKey, + "providerIdAt": providerIDAt, + "capabilityCodeAt": capabilityCodeAt, + "authScheme-pmfby": "tokenHeader", + "tokenUrl-pmfby": "https://pmfby.example/krphapi/FGMS/NICUsersLogin", + "tokenUserField-pmfby": "appAccessUID", + "tokenUserEnv-pmfby": "PMFBY_USER", + "tokenSecretField-pmfby": "appAccessPWD", + "tokenSecretEnv-pmfby": "PMFBY_PASSWORD", + "tokenResponseField-pmfby": "token", + "tokenTtl-pmfby": "10m", + "headerName-pmfby": "Authorization", + } +} + +func TestParseConfig_PMFBYBlock_ReadsEverySetting(t *testing.T) { + got, err := grievanceProvider{}.parseConfig(pmfbyConfig()) + if err != nil { + t.Fatal(err) + } + want := &Grievance.Config{ + BindingKeys: []string{pmfbyBindingKey}, + ProviderIDAt: providerIDAt, + CapabilityCodeAt: capabilityCodeAt, + AuthByProvider: map[string]*common.AuthProfile{"pmfby": { + Provider: "pmfby", Scheme: "tokenHeader", TokenURL: "https://pmfby.example/krphapi/FGMS/NICUsersLogin", + TokenUserField: "appAccessUID", TokenUserEnv: "PMFBY_USER", + TokenSecretName: "appAccessPWD", TokenSecretEnv: "PMFBY_PASSWORD", + TokenResponseField: "token", TokenTTLRaw: "10m", HeaderName: "Authorization", + }}, + } + if !reflect.DeepEqual(got, want) { + t.Errorf("parseConfig() = %+v, want %+v", got, want) + } +} + +func TestParseConfig_ResponseCap_ValidatesValue(t *testing.T) { + for raw, wantErr := range map[string]string{ + "lots": "invalid maxResponseBytes value 'lots'", + "0": "maxResponseBytes must be positive", + "-1": "maxResponseBytes must be positive", + "": "", + "2048": "", + } { + t.Run(raw, func(t *testing.T) { + _, err := grievanceProvider{}.parseConfig(map[string]string{"maxResponseBytes": raw}) + if wantErr == "" && err != nil { + t.Errorf("parseConfig() = %v, want no error", err) + } + if wantErr != "" && (err == nil || !strings.Contains(err.Error(), wantErr)) { + t.Errorf("parseConfig() = %v, want %q", err, wantErr) + } + }) + } +} + +func TestParseConfig_MisspelledAuthSetting_Refused(t *testing.T) { + config := pmfbyConfig() + config["tokenUrll-pmfby"] = "https://typo.example" + if _, err := (grievanceProvider{}).parseConfig(config); err == nil { + t.Error("parseConfig() accepted a misspelled credential setting") + } +} + +func TestSplitList_CommaSeparated_TrimsAndDropsBlanks(t *testing.T) { + for raw, want := range map[string][]string{ + "": nil, + " ": nil, + pmfbyBindingKey + ", b|openagrinet:Two ,,": {pmfbyBindingKey, "b|openagrinet:Two"}, + } { + if got := splitList(raw); !reflect.DeepEqual(got, want) { + t.Errorf("splitList(%q) = %v, want %v", raw, got, want) + } + } +} + +func TestNew_ShippedPMFBYConfig_BuildsStep(t *testing.T) { + step, closer, err := grievanceProvider{}.New(context.Background(), stubRegistry{}, stubMapper{}, pmfbyConfig()) + if err != nil { + t.Fatalf("New() = %v, want the shipped config to load", err) + } + if step == nil || closer == nil { + t.Fatal("New() returned no step or no closer") + } + if err := closer(); err != nil { + t.Errorf("closer() = %v", err) + } +} + +func TestNew_InvalidConfig_Refused(t *testing.T) { + for name, edit := range map[string]func(map[string]string){ + "no binding keys": func(c map[string]string) { delete(c, "bindingKeys") }, + "providerIdAt alone": func(c map[string]string) { delete(c, "capabilityCodeAt") }, + "capabilityCodeAt alone": func(c map[string]string) { delete(c, "providerIdAt") }, + "unpaired path lists": func(c map[string]string) { c["capabilityCodeAt"] = "a.@type" }, + "blank listed path": func(c map[string]string) { c["providerIdAt"] = "a.id, " }, + "no auth block for pmfby": func(c map[string]string) { delete(c, "authScheme-pmfby") }, + "unknown auth scheme": func(c map[string]string) { c["authScheme-pmfby"] = "loginHeader" }, + "tokenHeader without ttl": func(c map[string]string) { delete(c, "tokenTtl-pmfby") }, + "tokenHeader without login": func(c map[string]string) { delete(c, "tokenUrl-pmfby") }, + "tokenHeader without header": func(c map[string]string) { delete(c, "headerName-pmfby") }, + "auth for unserved provider": func(c map[string]string) { c["authScheme-pmkisan"] = "none" }, + "bad response cap": func(c map[string]string) { c["maxResponseBytes"] = "lots" }, + } { + t.Run(name, func(t *testing.T) { + config := pmfbyConfig() + edit(config) + if _, _, err := (grievanceProvider{}).New(context.Background(), stubRegistry{}, stubMapper{}, config); err == nil { + t.Error("New() accepted an invalid config") + } + }) + } +} + +func TestNew_NilContext_Refused(t *testing.T) { + //nolint:staticcheck // deliberately passing a nil context to assert the guard. + if _, _, err := (grievanceProvider{}).New(nil, stubRegistry{}, stubMapper{}, pmfbyConfig()); err == nil { + t.Error("New() accepted a nil context") + } +} + +func TestNew_StepConstructorFails_PropagatesError(t *testing.T) { + original := newStepFunc + wanted := errors.New("upstream refused the config") + newStepFunc = func(context.Context, definition.ProviderRecordLookup, definition.Mapper, + *Grievance.Config) (definition.Step, func() error, error) { + return nil, nil, wanted + } + defer func() { newStepFunc = original }() + + if _, _, err := (grievanceProvider{}).New(context.Background(), stubRegistry{}, stubMapper{}, pmfbyConfig()); !errors.Is(err, wanted) { + t.Errorf("New() = %v, want it to wrap %v", err, wanted) + } +} diff --git a/pkg/plugin/implementation/Grievance/mappings_test.go b/pkg/plugin/implementation/Grievance/mappings_test.go new file mode 100644 index 000000000..725d20da7 --- /dev/null +++ b/pkg/plugin/implementation/Grievance/mappings_test.go @@ -0,0 +1,958 @@ +package Grievance_test + +// mappings_test.go runs the shipped PMFBY mappings through the real step, +// against a stub PMFBY that logs in, files a ticket and reads it back. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + "time" + + "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/Grievance" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common/util" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/jsonmapper" +) + +const ( + mappingsDir = "../../../../config/mappings/pmfby" + shippedBindingKey = "pmfby|openagrinet:Grievance" + stubToken = "stub-login-token" + knownTicket = "100626000099001" + phone = "9876543210" + insertPath = "/krphapi/FGMS/AddKRPHNCIPGrievenceSupportTicket" + statusPath = "/krphapi/FGMS/GetGrievenceTicketsStatus" + loginPath = "/krphapi/FGMS/NICUsersLogin" +) + +// knownTicketRecord is PMFBY's record for knownTicket, personal data included. +// It carries the category by name only. +const knownTicketRecord = `{"responseCode":"1","responseMessage":"Fetched","recordCount":1,"responseDynamic":{ + "GrievenceSupportTicketID":109301,"TicketStatus":"Under Review", + "ComplaintDate":"2026-10-04","InsertDateTime":"2026-10-04T10:00:00.000Z","ApplicationNo":"040108251010160770605", + "GrievenceDescription":"Claim not received", + "TicketCategoryName":"Claim","TicketSubCategoryName":"Claim not received","CropName":"Paddy", + "FarmerName":"Ramesh","StateMasterName":"Karnataka","DistrictMasterName":"Mysuru", + "InsuranceCompany":"Insurer"}}` + +// pmfbyStub stands in for PMFBY's FGMS. Zero fields give the ordinary answer. +type pmfbyStub struct { + down bool // 503 on every call but the login + loginDown bool // 401 on the login + answers map[string]string // overrides the answer for a path + statusFor map[string]int // answers a path with this status and no body + calls map[string]int // per path, login included + bodies map[string]map[string]any +} + +func (p *pmfbyStub) ServeHTTP(w http.ResponseWriter, r *http.Request) { + p.calls[r.URL.Path]++ + if r.URL.Path == loginPath { + var login map[string]string + _ = json.NewDecoder(r.Body).Decode(&login) + if p.loginDown || login["appAccessUID"] != "user" || login["appAccessPWD"] != "secret" { + w.WriteHeader(http.StatusUnauthorized) + return + } + fmt.Fprintf(w, `{"responseCode":"1","responseDynamic":{"token":{"Token":%q}}}`, stubToken) + return + } + if p.down { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + if r.Header.Get("Authorization") != stubToken { + w.WriteHeader(http.StatusUnauthorized) + return + } + if status, set := p.statusFor[r.URL.Path]; set { + w.WriteHeader(status) + return + } + raw, _ := io.ReadAll(r.Body) + var body map[string]any + _ = json.Unmarshal(raw, &body) + p.bodies[r.URL.Path] = body + + if answer, overridden := p.answers[r.URL.Path]; overridden { + fmt.Fprint(w, answer) + return + } + switch r.URL.Path { + case insertPath: + fmt.Fprintf(w, `{"responseCode":"1","responseMessage":"Ticket created","responseDynamic":{"GrievenceSupportTicketNo":%q,"GrievenceSupportTicketID":8842317}}`, knownTicket) + case statusPath: + if body["GrievenceSupportTicketNo"] != knownTicket { + fmt.Fprint(w, `{"responseCode":"0","responseMessage":"No record found","responseDynamic":[]}`) + return + } + fmt.Fprint(w, knownTicketRecord) + default: + w.WriteHeader(http.StatusNotFound) + } +} + +type stubRegistry struct{ plan *model.ProviderRecord } + +func (s *stubRegistry) ProviderRecord(context.Context, string) (*model.ProviderRecord, error) { + return s.plan, nil +} + +// harness is the real step over the shipped mappings, wired to a stub PMFBY. +type harness struct { + pmfby *pmfbyStub + step definition.Step +} + +// newHarness builds the step. statusRetries is status's retryMax (support +// gets 0); tweak edits the config first. +func newHarness(t *testing.T, pmfby *pmfbyStub, statusRetries int, tweak ...func(*Grievance.Config)) *harness { + t.Helper() + t.Setenv("PMFBY_USER", "user") + t.Setenv("PMFBY_PASSWORD", "secret") + pmfby.calls = map[string]int{} + pmfby.bodies = map[string]map[string]any{} + + mappings := httptest.NewServer(http.FileServer(http.Dir(mappingsDir))) + t.Cleanup(mappings.Close) + upstream := httptest.NewServer(pmfby) + t.Cleanup(upstream.Close) + + mapper, closeMapper, err := jsonmapper.New(context.Background(), &jsonmapper.Config{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = closeMapper() }) + + action := func(path, file string, retries int) model.ActionPlan { + return model.ActionPlan{Method: http.MethodPost, Path: path, + Mappings: mappings.URL + "/" + file, RetryMax: retries} + } + registry := &stubRegistry{plan: &model.ProviderRecord{ + BindingKey: shippedBindingKey, ParticipantID: "pmfby", CapabilityCode: "openagrinet:Grievance", + BaseURL: upstream.URL, + Actions: map[string]model.ActionPlan{ + "support": action(insertPath, "grievance.support.yaml", 0), + "status": action(statusPath, "grievance.status.yaml", statusRetries), + }, + }} + + cfg := &Grievance.Config{ + BindingKeys: []string{shippedBindingKey}, + ProviderIDAt: "message.contract.commitments[].offer.provider.id, message.support.channels[].provider.id", + CapabilityCodeAt: "message.contract.commitments[].commitmentAttributes.@type, message.support.channels[].@type", + AuthByProvider: map[string]*common.AuthProfile{"pmfby": { + Scheme: util.AuthSchemeTokenHeader, TokenURL: upstream.URL + loginPath, + TokenUserField: "appAccessUID", TokenUserEnv: "PMFBY_USER", + TokenSecretName: "appAccessPWD", TokenSecretEnv: "PMFBY_PASSWORD", + TokenResponseField: "responseDynamic.token.Token", TokenTTLRaw: "10m", HeaderName: "Authorization", + }}, + } + for _, apply := range tweak { + apply(cfg) + } + step, closeStep, err := Grievance.New(context.Background(), registry, mapper, cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = closeStep() }) + return &harness{pmfby: pmfby, step: step} +} + +// answer is the part of a Beckn reply these tests read. +type answer struct { + Context map[string]any `json:"context"` + Message struct { + Contract struct { + ID string `json:"id"` + Commitments []struct { + Status struct { + Descriptor struct { + Code string `json:"code"` + } `json:"descriptor"` + } `json:"status"` + Offer map[string]any `json:"offer"` + CommitmentAttributes map[string]any `json:"commitmentAttributes"` + } `json:"commitments"` + } `json:"contract"` + Support struct { + Channels []map[string]any `json:"channels"` + } `json:"support"` + } `json:"message"` +} + +// channel is the case record on an on_support. +func (a *answer) channel() map[string]any { + return a.Message.Support.Channels[0] +} + +func (a *answer) attributes() map[string]any { + return a.Message.Contract.Commitments[0].CommitmentAttributes +} + +func (a *answer) status() string { + return a.Message.Contract.Commitments[0].Status.Descriptor.Code +} + +// band reads a nested object: the grievance, the case, or its status. +func band(of map[string]any, name string) map[string]any { + inner, _ := of[name].(map[string]any) + return inner +} + +// send runs body through the step. A nil answer means the step passed it on. +func (h *harness) send(t *testing.T, body []byte) (*answer, error) { + t.Helper() + stepCtx := &model.StepContext{Context: t.Context(), Body: body} + if err := h.step.Run(stepCtx); err != nil { + return nil, err + } + if len(stepCtx.ResponseBody) == 0 { + return nil, nil + } + var got answer + if err := json.Unmarshal(stepCtx.ResponseBody, &got); err != nil { + t.Fatalf("answer is not JSON: %v\n%s", err, stepCtx.ResponseBody) + } + return &got, nil +} + +// mustSend is send for a request that has to succeed. +func (h *harness) mustSend(t *testing.T, body []byte) *answer { + t.Helper() + got, err := h.send(t, body) + if err != nil { + t.Fatalf("Run() = %v, want success", err) + } + if got == nil { + t.Fatal("the step passed the request on instead of answering it") + } + return got +} + +const packContext = "https://openagrinet.github.io/network-specs/api-schemas/Grievance/v0.1/context.jsonld" + +// scheme is what every PMFBY payload names. +var scheme = map[string]any{"code": "PMFBY", "name": "Pradhan Mantri Fasal Bima Yojana"} + +// provider is the portal a support channel names. +var provider = map[string]any{"id": "pmfby", "descriptor": map[string]any{"name": "PMFBY Grievance Portal"}} + +func becknContext(action string) map[string]any { + return map[string]any{"version": "2.0.0", "action": action, "transactionId": "txn-1", "messageId": "msg-1"} +} + +// request builds a status payload whose one commitment carries attributes. +func request(action string, attributes map[string]any) []byte { + attributes["@context"] = packContext + attributes["@type"] = "openagrinet:Grievance" + attributes["informationMode"] = "OnDemand" + if _, named := attributes["scheme"]; !named { + attributes["scheme"] = scheme + } + body, _ := json.Marshal(map[string]any{ + "context": becknContext(action), + "message": map[string]any{"contract": map[string]any{ + "id": "contract-1", + "commitments": []any{map[string]any{ + "status": map[string]any{"descriptor": map[string]any{"code": "ACTIVE"}}, + "resources": []any{map[string]any{"id": "res:pmfby:grievance"}}, + "offer": map[string]any{"id": "off:pmfby:grievance", "provider": map[string]any{"id": "pmfby"}}, + "commitmentAttributes": attributes, + }}, + }}, + }) + return body +} + +// support builds a support request from flat fields: categoryCode, +// subCategoryCode and description under grievance, the rest on channels[0]. +func support(fields map[string]any) []byte { + channel := map[string]any{"@context": packContext, "@type": "openagrinet:Grievance", + "informationMode": "OnDemand", "scheme": scheme, "provider": provider} + body := map[string]any{"channels": []any{channel}} + grievance := map[string]any{} + for field, value := range fields { + switch field { + case "categoryCode": + grievance["category"] = map[string]any{"code": value, "name": "Claim"} + case "subCategoryCode": + grievance["subCategory"] = map[string]any{"code": value, "name": "Claim not received"} + case "description": + grievance["description"] = value + default: + channel[field] = value + } + } + if len(grievance) > 0 { + channel["grievance"] = grievance + } + raw, _ := json.Marshal(map[string]any{"context": becknContext("support"), "message": map[string]any{"support": body}}) + return raw +} + +func supportFields() map[string]any { + return map[string]any{ + "enrolmentId": "040108251010160770605", "categoryCode": "3", "subCategoryCode": "10", + "description": " Claim not received ", "applicantMobile": phone, + "cropYear": "2025", "season": "Kharif", + } +} + +func statusAttributes() map[string]any { + return map[string]any{"applicantMobile": phone, "case": map[string]any{"ticketNo": knownTicket}} +} + +// refusal is the HTTP status and Beckn error a failed request is answered with. +func refusal(t *testing.T, err error) (int, *model.Error) { + t.Helper() + var invalid *model.SchemaValidationErr + var coded *model.CodedErr + switch { + case errors.As(err, &invalid): + return http.StatusBadRequest, invalid.BecknError() + case errors.As(err, &coded): + return coded.HTTPStatus(), coded.BecknError() + } + t.Fatalf("error = %v (%T), want a refusal", err, err) + return 0, nil +} + +// assertCoded fails unless err is answered with this status and code. +func assertCoded(t *testing.T, err error, status int, code string) { + t.Helper() + if got, refused := refusal(t, err); got != status || refused.Code != code { + t.Errorf("error = %d %s (%v), want %d %s", got, refused.Code, err, status, code) + } +} + +// assertRefused fails unless err is a 400 BIZ_GENERIC_ERROR carrying message. +func assertRefused(t *testing.T, err error, message string) { + t.Helper() + if status, refused := refusal(t, err); status != http.StatusBadRequest || + refused.Code != "BIZ_GENERIC_ERROR" || refused.Message != message { + t.Errorf("error = %d %+v, want 400 BIZ_GENERIC_ERROR %q", status, refused, message) + } +} + +// assertFailedUpstream fails unless err is a 502 NET_DOWNSTREAM_UNAVAILABLE carrying message. +func assertFailedUpstream(t *testing.T, err error, message string) { + t.Helper() + if status, refused := refusal(t, err); status != http.StatusBadGateway || + refused.Code != util.CodeUpstreamUnavailable || refused.Message != message { + t.Errorf("error = %d %+v, want 502 %s %q", status, refused, util.CodeUpstreamUnavailable, message) + } +} + +// assertJustNowUTC fails unless value is an RFC 3339 UTC instant within a minute of now. +func assertJustNowUTC(t *testing.T, value any) { + t.Helper() + text, _ := value.(string) + at, err := time.Parse("2006-01-02T15:04:05Z", text) + if err != nil || time.Since(at).Abs() > time.Minute { + t.Errorf("filedOn = %v, want this moment as YYYY-MM-DDTHH:MM:SSZ", value) + } +} + +// todayIST is the date PMFBY expects complaintDate in. +func todayIST() string { + return time.Now().In(time.FixedZone("IST", 5*3600+1800)).Format("2006-01-02") +} + +// --- support --------------------------------------------------------------- + +func TestSupport_Lodged_SendsPMFBYFieldsAndReturnsTicket(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + got := h.mustSend(t, support(supportFields())) + + // Every value goes as a string, the season as its code. + want := map[string]any{ + "requestorMobileNo": phone, "applicationNo": "040108251010160770605", + "requestYear": "2025", "requestSeason": "1", "ticketCategoryID": "3", "ticketSubCategoryID": "10", + "grievenceDescription": "Claim not received", "complaintDate": todayIST(), "receiptSourceID": "134306", + } + sent := h.pmfby.bodies[insertPath] + for field, value := range want { + if sent[field] != value { + t.Errorf("PMFBY %s = %v, want %v", field, sent[field], value) + } + } + if len(sent) != len(want) { + t.Errorf("PMFBY was sent %d fields (%v), want exactly %d", len(sent), sent, len(want)) + } + + if got.Context["action"] != "on_support" || got.Context["transactionId"] != "txn-1" { + t.Errorf("context = %v, want on_support on the request's transaction", got.Context) + } + // enrolmentId and the grievance are echoed; a name beside a code is dropped. + channel := got.channel() + grievance := band(channel, "grievance") + if channel["enrolmentId"] != "040108251010160770605" || len(band(grievance, "category")) != 1 || + band(grievance, "category")["code"] != "3" || band(grievance, "subCategory")["code"] != "10" || + grievance["description"] != "Claim not received" { + t.Errorf("channel = %v, want enrolmentId and grievance echoed, the description trimmed", channel) + } + filed := band(channel, "case") + status := band(filed, "status") + assertJustNowUTC(t, filed["filedOn"]) + if filed["ticketNo"] != knownTicket || len(status) != 1 || status["code"] != "Registered" || + band(channel, "provider")["id"] != "pmfby" || + channel["informationMode"] != "Direct" || channel["scheme"] == nil { + t.Errorf("channel = %v", channel) + } + for _, private := range []string{"applicantMobile", "ticketId"} { + if _, echoed := channel[private]; echoed { + t.Errorf("%s must not be echoed", private) + } + } +} + +// The pack holds ticketNo as a string of digits, whichever PMFBY sends. +func TestSupport_NumericTicketNo_ReturnedAsString(t *testing.T) { + h := newHarness(t, &pmfbyStub{answers: map[string]string{ + insertPath: fmt.Sprintf(`{"responseCode":1,"responseDynamic":{"GrievenceSupportTicketNo":%s}}`, knownTicket)}}, 0) + if got := band(h.mustSend(t, support(supportFields())).channel(), "case")["ticketNo"]; got != knownTicket { + t.Errorf("ticketNo = %#v, want %q", got, knownTicket) + } +} + +func TestSupport_NumericCropYear_SentAsString(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + fields := supportFields() + fields["cropYear"] = 2025 + h.mustSend(t, support(fields)) + if got := h.pmfby.bodies[insertPath]["requestYear"]; got != "2025" { + t.Errorf("requestYear = %#v, want the string \"2025\" PMFBY takes", got) + } +} + +func TestSupport_EachSeason_SendsItsCode(t *testing.T) { + for season, code := range map[string]string{"Kharif": "1", "Rabi": "2", "Zaid": "3"} { + t.Run(season, func(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + fields := supportFields() + fields["season"] = season + h.mustSend(t, support(fields)) + if got := h.pmfby.bodies[insertPath]["requestSeason"]; got != code { + t.Errorf("requestSeason = %v, want %v", got, code) + } + }) + } +} + +func TestSupport_CategoryCodes_SentAsGiven(t *testing.T) { + for codes, want := range map[[2]string][2]string{{"3", "10"}: {"3", "10"}, {"12", "5"}: {"12", "5"}, {"1", "1"}: {"1", "1"}} { + t.Run(codes[0]+"/"+codes[1], func(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + fields := supportFields() + fields["categoryCode"], fields["subCategoryCode"] = codes[0], codes[1] + h.mustSend(t, support(fields)) + sent := h.pmfby.bodies[insertPath] + if sent["ticketCategoryID"] != want[0] || sent["ticketSubCategoryID"] != want[1] { + t.Errorf("sent %v / %v, want %v", sent["ticketCategoryID"], sent["ticketSubCategoryID"], want) + } + }) + } +} + +// Without category codes, PMFBY's published pair is sent and echoed. +func TestSupport_NoCategoryCodes_SendsAndEchoesDefaults(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + fields := supportFields() + delete(fields, "categoryCode") + delete(fields, "subCategoryCode") + got := h.mustSend(t, support(fields)) + sent := h.pmfby.bodies[insertPath] + if sent["ticketCategoryID"] != "3" || sent["ticketSubCategoryID"] != "10" { + t.Errorf("sent %v / %v, want 3 / 10", sent["ticketCategoryID"], sent["ticketSubCategoryID"]) + } + grievance := band(got.channel(), "grievance") + if band(grievance, "category")["code"] != "3" || band(grievance, "subCategory")["code"] != "10" { + t.Errorf("grievance = %v, want the defaults echoed", grievance) + } +} + +// The pack lists provider and scheme.name as coming back; the caller may send +// only the ids. +func TestSupport_BareProviderAndScheme_ReturnsPortalNameAndSchemeName(t *testing.T) { + fields := supportFields() + fields["provider"] = map[string]any{"id": "pmfby", "availableAt": []any{}} + fields["scheme"] = map[string]any{"code": "PMFBY"} + channel := newHarness(t, &pmfbyStub{}, 0).mustSend(t, support(fields)).channel() + if got := band(channel, "provider"); !reflect.DeepEqual(got, provider) { + t.Errorf("provider = %v, want %v", got, provider) + } + if got := band(channel, "scheme"); !reflect.DeepEqual(got, scheme) { + t.Errorf("scheme = %v, want %v", got, scheme) + } +} + +// complaintDate goes to PMFBY as its IST date and comes back as filedOn in UTC. +// complaintDate goes to PMFBY as its IST date; filedOn stays the lodge time, +// so a later status read reports the same moment. +func TestSupport_CallerDateAndSource_SentInsteadOfDefaults(t *testing.T) { + for given, want := range map[string]string{ + "2026-09-20": "2026-09-20", + "2026-09-20T04:30:00Z": "2026-09-20", + "2026-09-21T01:00:00+05:30": "2026-09-21", + } { + t.Run(given, func(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + fields := supportFields() + fields["complaintDate"], fields["receiptSourceId"] = given, "200001" + got := h.mustSend(t, support(fields)) + sent := h.pmfby.bodies[insertPath] + if sent["complaintDate"] != want || sent["receiptSourceID"] != "200001" { + t.Errorf("PMFBY was sent %v / %v, want %s and the caller's source", sent["complaintDate"], sent["receiptSourceID"], want) + } + assertJustNowUTC(t, band(got.channel(), "case")["filedOn"]) + }) + } +} + +// "Ticket Already Raised" is a success: the existing ticket comes back. +func TestSupport_TicketAlreadyRaised_ReturnsTheExistingTicket(t *testing.T) { + h := newHarness(t, &pmfbyStub{answers: map[string]string{insertPath: `{"responseCode":"1","responseMessage":"Ticket Already Raised for !", + "responseDynamic":{"GrievenceSupportTicketNo":"121026000099003","GrievenceSupportTicketID":13}}`}}, 0) + filed := band(h.mustSend(t, support(supportFields())).channel(), "case") + if filed["ticketNo"] != "121026000099003" || band(filed, "status")["code"] != "Registered" { + t.Errorf("case = %v, want the existing ticket, Registered", filed) + } + assertJustNowUTC(t, filed["filedOn"]) +} + +// A refusal PMFBY is known to give is a 400 in our words, never PMFBY's. +func TestSupport_KnownRefusal_Returns400InOurWords(t *testing.T) { + for name, tc := range map[string]struct{ reply, message string }{ + "phone not matched": {`{"responseObject":null,"responseDynamic":null,"responseCode":"0","responseMessage":"Mobile Number Not Matched","jsonString":null,"recordCount":0}`, + "The phone number does not match the one registered for this application number."}, + "no application": {`{"responseCode":"0","responseMessage":"Application Details not Exists","responseDynamic":null}`, + "No PMFBY application was found for this application number."}, + "case and spaces": {`{"responseCode":"0","responseMessage":" mobile number NOT matched ","responseDynamic":null}`, + "The phone number does not match the one registered for this application number."}, + } { + t.Run(name, func(t *testing.T) { + _, err := newHarness(t, &pmfbyStub{answers: map[string]string{insertPath: tc.reply}}, 0).send(t, support(supportFields())) + assertRefused(t, err, tc.message) + }) + } +} + +// Any other failure is PMFBY's own: a 502 in our words, PMFBY's text withheld. +func TestSupport_PortalFails_Returns502WithoutItsText(t *testing.T) { + for name, reply := range map[string]string{ + "success, no ticket": `{"responseCode":"1","responseDynamic":{}}`, + "success, blank ticket": `{"responseCode":"1","responseDynamic":{"GrievenceSupportTicketNo":""}}`, + "no message": `{"responseCode":"0","responseMessage":null}`, + "internal error": `{"responseCode":"0","responseMessage":"AxiosError: Request failed with status code 403","responseDynamic":null}`, + "token rejected": `{"responseCode":"0","responseMessage":"Token required!","responseDynamic":null}`, + "unknown wording": `{"responseCode":"0","responseMessage":"Invalid","responseDynamic":{"GrievenceSupportTicketNo":"100626000099001"}}`, + } { + t.Run(name, func(t *testing.T) { + _, err := newHarness(t, &pmfbyStub{answers: map[string]string{insertPath: reply}}, 0).send(t, support(supportFields())) + assertFailedUpstream(t, err, "PMFBY could not register the grievance right now. Please try again later.") + }) + } +} + +// A 4xx from PMFBY is its refusal. +func TestSupport_PortalAnswers4xx_Returns502(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusNotFound, http.StatusConflict, http.StatusUnprocessableEntity} { + t.Run(http.StatusText(status), func(t *testing.T) { + h := newHarness(t, &pmfbyStub{statusFor: map[string]int{insertPath: status}}, 0) + _, err := h.send(t, support(supportFields())) + assertCoded(t, err, http.StatusBadGateway, util.CodeUpstreamUnavailable) + if got := h.pmfby.calls[insertPath]; got != 1 { + t.Errorf("the lodge call was made %d times, want 1", got) + } + }) + } +} + +func TestStatus_PortalAnswers400_Returns502(t *testing.T) { + h := newHarness(t, &pmfbyStub{statusFor: map[string]int{statusPath: http.StatusBadRequest}}, 0) + _, err := h.send(t, request("status", statusAttributes())) + assertCoded(t, err, http.StatusBadGateway, util.CodeUpstreamUnavailable) +} + +func TestSupport_PortalDown_Returns502AfterOneAttempt(t *testing.T) { + h := newHarness(t, &pmfbyStub{down: true}, 0) + _, err := h.send(t, support(supportFields())) + assertCoded(t, err, http.StatusBadGateway, util.CodeUpstreamUnavailable) + if got := h.pmfby.calls[insertPath]; got != 1 { + t.Errorf("the lodge call was made %d times, want 1: a retry files a duplicate", got) + } +} + +func TestSupport_SecondChannel_Returns400WithoutCallingPMFBY(t *testing.T) { + var payload map[string]any + _ = json.Unmarshal(support(supportFields()), &payload) + body := payload["message"].(map[string]any)["support"].(map[string]any) + body["channels"] = append(body["channels"].([]any), body["channels"].([]any)[0]) + raw, _ := json.Marshal(payload) + + h := newHarness(t, &pmfbyStub{}, 0) + _, err := h.send(t, raw) + assertCoded(t, err, http.StatusBadRequest, "SCH_INVALID_FORMAT") + if h.pmfby.calls[insertPath] != 0 { + t.Error("PMFBY was called for a request naming two channels") + } +} + +// Without the channel paths a support request names no binding the step reads. +func TestSupport_ChannelPathsNotConfigured_PassesThroughUntouched(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0, func(c *Grievance.Config) { + c.ProviderIDAt = "message.contract.commitments[].offer.provider.id" + c.CapabilityCodeAt = "message.contract.commitments[].commitmentAttributes.@type" + }) + got, err := h.send(t, support(supportFields())) + if err != nil || got != nil { + t.Errorf("Run() = %v, %v; want the support request passed on untouched", got, err) + } + if len(h.pmfby.calls) != 0 { + t.Errorf("PMFBY was called %v without a way to recognise the request", h.pmfby.calls) + } +} + +// Two channels are refused only when they name this step's binding. +func TestSupport_OtherProviderTwoChannels_PassesThrough(t *testing.T) { + var payload map[string]any + _ = json.Unmarshal(support(supportFields()), &payload) + body := payload["message"].(map[string]any)["support"].(map[string]any) + channel := body["channels"].([]any)[0].(map[string]any) + channel["provider"] = map[string]any{"id": "someone-else"} + body["channels"] = append(body["channels"].([]any), channel) + raw, _ := json.Marshal(payload) + + h := newHarness(t, &pmfbyStub{}, 0) + got, err := h.send(t, raw) + if err != nil || got != nil { + t.Errorf("Run() = %v, %v; want another provider's request passed on", got, err) + } +} + +func TestSupport_NoProvider_PassesThrough(t *testing.T) { + var payload map[string]any + _ = json.Unmarshal(support(supportFields()), &payload) + channels := payload["message"].(map[string]any)["support"].(map[string]any)["channels"].([]any) + delete(channels[0].(map[string]any), "provider") + body, _ := json.Marshal(payload) + h := newHarness(t, &pmfbyStub{}, 0) + got, err := h.send(t, body) + if err != nil || got != nil { + t.Errorf("Run() = %v, %v; want a support request naming no provider passed on", got, err) + } +} + +// --- status ---------------------------------------------------------------- + +func TestStatus_KnownTicket_ReturnsCaseRecordAndComplaint(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + got := h.mustSend(t, request("status", statusAttributes())) + + sent := h.pmfby.bodies[statusPath] + if len(sent) != 2 || sent["GrievenceSupportTicketNo"] != knownTicket || sent["requestorMobileNo"] != phone { + t.Errorf("PMFBY was sent %v, want the ticket and the phone alone", sent) + } + attributes := got.attributes() + // The category comes back by name, with no code. + grievance := band(attributes, "grievance") + category, subCategory := band(grievance, "category"), band(grievance, "subCategory") + if len(category) != 1 || category["name"] != "Claim" || len(subCategory) != 1 || + subCategory["name"] != "Claim not received" || grievance["description"] != "Claim not received" { + t.Errorf("grievance = %v, want the complaint as PMFBY holds it", grievance) + } + filed := band(attributes, "case") + if attributes["enrolmentId"] != "040108251010160770605" || attributes["informationMode"] != "Direct" || + filed["ticketNo"] != knownTicket || filed["filedOn"] != "2026-10-04T10:00:00Z" || filed["cropName"] != "Paddy" { + t.Errorf("commitmentAttributes = %v", attributes) + } + // "Under Review" is not a phrase the lookup lists, so it is UnderReview. + if status := band(filed, "status"); len(status) != 1 || status["code"] != "UnderReview" { + t.Errorf("case.status = %v, want code UnderReview alone", status) + } + if got.Context["action"] != "on_status" || got.status() != "ACTIVE" { + t.Errorf("action %v, status %v; want on_status, ACTIVE", got.Context["action"], got.status()) + } +} + +// The live record carries ids, crop year and season as numbers, and the +// officer's reply in latestRemark. +func TestStatus_FullRecord_ReturnsCodesCropYearSeasonAndReply(t *testing.T) { + record := strings.Replace(knownTicketRecord, `"responseDynamic":{`, `"responseDynamic":{ + "TicketCategoryID":3,"TicketSubCategoryID":10,"RequestYear":2025,"RequestSeason":2, + "latestRemark":" Claim credited on 12 Oct ",`, 1) + attributes := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())).attributes() + + grievance := band(attributes, "grievance") + if category, subCategory := band(grievance, "category"), band(grievance, "subCategory"); category["code"] != "3" || + category["name"] != "Claim" || subCategory["code"] != "10" || subCategory["name"] != "Claim not received" { + t.Errorf("grievance = %v, want code and name on both levels", grievance) + } + if attributes["cropYear"] != "2025" || attributes["season"] != "Rabi" { + t.Errorf("cropYear, season = %v, %v; want 2025, Rabi", attributes["cropYear"], attributes["season"]) + } + if reply := band(attributes, "case")["reply"]; reply != "Claim credited on 12 Oct" { + t.Errorf("case.reply = %q, want the trimmed remark", reply) + } +} + +// Values PMFBY uses for "none", or that fit no pack field, are dropped. +func TestStatus_UnusableCodes_Dropped(t *testing.T) { + for name, fields := range map[string]string{ + "zeros": `"TicketCategoryID":0,"TicketSubCategoryID":"0","RequestYear":0,"RequestSeason":0,"latestRemark":"",`, + "out of range": `"RequestYear":25,"RequestSeason":4,"latestRemark":" ",`, + "not a number": `"TicketCategoryID":"x","RequestYear":"20x5","RequestSeason":"Kharif",`, + } { + t.Run(name, func(t *testing.T) { + record := strings.Replace(knownTicketRecord, `"responseDynamic":{`, `"responseDynamic":{`+fields, 1) + attributes := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())).attributes() + _, year := attributes["cropYear"] + _, season := attributes["season"] + _, reply := band(attributes, "case")["reply"] + if category := band(band(attributes, "grievance"), "category"); year || season || reply || category["code"] != nil { + t.Errorf("attributes = %v, want no cropYear, season, reply or category code", attributes) + } + }) + } +} + +func TestStatus_KnownTicket_DropsPersonalAndInternalFields(t *testing.T) { + got := newHarness(t, &pmfbyStub{}, 0).mustSend(t, request("status", statusAttributes())) + allowed := map[string]map[string]bool{ + "": {"@context": true, "@type": true, "informationMode": true, "scheme": true, + "enrolmentId": true, "grievance": true, "case": true}, + "grievance": {"category": true, "subCategory": true, "description": true}, + "case": {"ticketNo": true, "status": true, "filedOn": true, "cropName": true}, + } + attributes := got.attributes() + for name, fields := range allowed { + in := attributes + if name != "" { + in = band(attributes, name) + } + for field := range in { + if !fields[field] { + t.Errorf("%s %s is mapped; the response is an allow-list", name, field) + } + } + } +} + +// InsertDateTime comes back as filedOn in UTC, whatever offset PMFBY states. +func TestStatus_InsertDateTimeWithOffset_FiledOnInUTC(t *testing.T) { + record := strings.Replace(knownTicketRecord, `"2026-10-04T10:00:00.000Z"`, `"2026-10-04T15:30:00+05:30"`, 1) + got := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())) + if filed := band(got.attributes(), "case")["filedOn"]; filed != "2026-10-04T10:00:00Z" { + t.Errorf("filedOn = %v, want 2026-10-04T10:00:00Z", filed) + } +} + +func TestStatus_BareScheme_ReturnsSchemeName(t *testing.T) { + attributes := statusAttributes() + attributes["scheme"] = map[string]any{"code": "PMFBY"} + got := newHarness(t, &pmfbyStub{}, 0).mustSend(t, request("status", attributes)) + if got := band(got.attributes(), "scheme"); !reflect.DeepEqual(got, scheme) { + t.Errorf("scheme = %v, want %v", got, scheme) + } +} + +// A missing phrase still maps to UnderReview. +func TestStatus_NoStatusPhrase_UnderReviewWithoutName(t *testing.T) { + for name, phrase := range map[string]string{"null": `null`, "empty": `""`} { + t.Run(name, func(t *testing.T) { + record := strings.Replace(knownTicketRecord, `"TicketStatus":"Under Review"`, `"TicketStatus":`+phrase, 1) + got := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())) + status := band(band(got.attributes(), "case"), "status") + if len(status) != 1 || status["code"] != "UnderReview" { + t.Errorf("case.status = %v, want code UnderReview alone", status) + } + }) + } +} + +func TestStatus_NullFields_DroppedNotMappedOrFailed(t *testing.T) { + for name, tc := range map[string]struct { + from, to string + band, field string + }{ + "null category name": {`"TicketCategoryName":"Claim"`, `"TicketCategoryName":null`, "grievance", "category"}, + "empty sub-category": {`"TicketSubCategoryName":"Claim not received"`, `"TicketSubCategoryName":""`, "grievance", "subCategory"}, + "null description": {`"GrievenceDescription":"Claim not received"`, `"GrievenceDescription":null`, "grievance", "description"}, + "null insert time": {`"InsertDateTime":"2026-10-04T10:00:00.000Z"`, `"InsertDateTime":null`, "case", "filedOn"}, + "date-only insert": {`"InsertDateTime":"2026-10-04T10:00:00.000Z"`, `"InsertDateTime":"2026-10-04"`, "case", "filedOn"}, + "null application no": {`"ApplicationNo":"040108251010160770605"`, `"ApplicationNo":null`, "", "enrolmentId"}, + "empty crop": {`"CropName":"Paddy"`, `"CropName":""`, "case", "cropName"}, + } { + t.Run(name, func(t *testing.T) { + record := strings.Replace(knownTicketRecord, tc.from, tc.to, 1) + got := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())) + in := got.attributes() + if tc.band != "" { + in = band(in, tc.band) + } + if value, present := in[tc.field]; present { + t.Errorf("%s = %v, want it dropped", tc.field, value) + } + if band(got.attributes(), "case")["ticketNo"] != knownTicket { + t.Errorf("case = %v, want the rest of the answer intact", band(got.attributes(), "case")) + } + }) + } +} + +// PMFBY's TicketStatus maps to a CaseStatusCode; an unlisted phrase is UnderReview. +func TestStatus_TicketStatus_MapsToCaseStatusCode(t *testing.T) { + for phrase, want := range map[string]string{"Resolved": "Resolved", " resolved ": "Resolved", + "Open": "Registered", "Under Review": "UnderReview", "Escalated": "UnderReview"} { + t.Run(phrase, func(t *testing.T) { + record := strings.Replace(knownTicketRecord, `"TicketStatus":"Under Review"`, `"TicketStatus":"`+phrase+`"`, 1) + got := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())) + if status := band(band(got.attributes(), "case"), "status"); len(status) != 1 || status["code"] != want { + t.Errorf("case.status = %v, want code %s alone", status, want) + } + }) + } +} + +// No record, or one for another ticket (PMFBY reads by phone alone), is a 400 +// in our words. +func TestStatus_NoMatchingRecord_Returns400(t *testing.T) { + another := strings.Replace(knownTicketRecord, `"responseDynamic":{`, `"responseDynamic":{"GrievenceSupportTicketNo":"100626000099002",`, 1) + for name, reply := range map[string]string{ + "no record": `{"responseCode":"0","responseMessage":"No record found","responseDynamic":[]}`, + "null": `{"responseCode":"1","responseDynamic":null}`, + "empty object": `{"responseCode":"1","responseDynamic":{}}`, + "empty in list": `{"responseCode":"1","responseDynamic":[{}]}`, + "all fields null": `{"responseCode":"1","responseDynamic":{"ApplicationNo":null,"TicketStatus":null}}`, + "another ticket": another, + } { + t.Run(name, func(t *testing.T) { + _, err := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: reply}}, 0). + send(t, request("status", statusAttributes())) + assertRefused(t, err, "No grievance was found for this ticket number and phone number.") + }) + } +} + +// Any other failure is PMFBY's own: a 502 in our words, PMFBY's text withheld. +func TestStatus_PortalFails_Returns502WithoutItsText(t *testing.T) { + for name, reply := range map[string]string{ + "no envelope": `{"message":"An error has occurred."}`, + "no message": `{"responseCode":"0","responseMessage":"","responseDynamic":null}`, + "internal error": `{"responseCode":"0","responseMessage":"AxiosError: Request failed with status code 403","responseDynamic":null}`, + "token rejected": `{"responseCode":"0","responseMessage":"Token required!","responseDynamic":null}`, + } { + t.Run(name, func(t *testing.T) { + _, err := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: reply}}, 0). + send(t, request("status", statusAttributes())) + assertFailedUpstream(t, err, "PMFBY could not return the grievance status right now. Please try again later.") + }) + } +} + +func TestStatus_RecordNamingTheTicket_ReturnsIt(t *testing.T) { + record := strings.Replace(knownTicketRecord, `"responseDynamic":{`, `"responseDynamic":{"GrievenceSupportTicketNo":"`+knownTicket+`",`, 1) + got := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())) + if no := band(got.attributes(), "case")["ticketNo"]; no != knownTicket { + t.Errorf("ticketNo = %v, want %s", no, knownTicket) + } +} + +func TestStatus_RecordInAList_ReadAsTheRecord(t *testing.T) { + record := strings.Replace(strings.Replace(knownTicketRecord, `"responseDynamic":{`, `"responseDynamic":[{`, 1), `}}`, `}]}`, 1) + got := newHarness(t, &pmfbyStub{answers: map[string]string{statusPath: record}}, 0). + mustSend(t, request("status", statusAttributes())) + if crop := band(got.attributes(), "case")["cropName"]; crop != "Paddy" { + t.Errorf("case = %v, want the record read from the list", band(got.attributes(), "case")) + } +} + +func TestStatus_PortalDown_RetriedAsTheRegistrySays(t *testing.T) { + h := newHarness(t, &pmfbyStub{down: true}, 2) + _, err := h.send(t, request("status", statusAttributes())) + assertCoded(t, err, http.StatusBadGateway, util.CodeUpstreamUnavailable) + if got := h.pmfby.calls[statusPath]; got != 3 { + t.Errorf("status called %d times, want 3: a read is safe to retry", got) + } +} + +// --- across the flow ------------------------------------------------------- + +func TestFlow_SupportThenStatus_LogsInOnce(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + ticket := band(h.mustSend(t, support(supportFields())).channel(), "case")["ticketNo"] + got := h.mustSend(t, request("status", map[string]any{"applicantMobile": phone, "case": map[string]any{"ticketNo": ticket}})) + + if status := band(band(got.attributes(), "case"), "status"); status["code"] != "UnderReview" { + t.Errorf("status of the filed ticket = %v", got.attributes()) + } + if logins := h.pmfby.calls[loginPath]; logins != 1 { + t.Errorf("logged in %d times over two calls, want 1", logins) + } +} + +func TestFlow_AnswerNotJSON_ReturnsErrorWithoutMapping(t *testing.T) { + h := newHarness(t, &pmfbyStub{answers: map[string]string{insertPath: `maintenance`}}, 0) + got, err := h.send(t, support(supportFields())) + if err == nil || got != nil { + t.Errorf("Run() = %v, %v; want an error and no answer", got, err) + } +} + +// init included: PMFBY has no OTP, so there is no init. +func TestFlow_UnservedAction_Returns400WithoutCallingPMFBY(t *testing.T) { + for _, action := range []string{"init", "select"} { + t.Run(action, func(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + _, err := h.send(t, request(action, statusAttributes())) + assertCoded(t, err, http.StatusBadRequest, "SCH_INVALID_FORMAT") + if len(h.pmfby.calls) != 0 { + t.Errorf("PMFBY was called %v for an action it does not serve", h.pmfby.calls) + } + }) + } +} + +// PM-KISAN shares the Grievance type; only the provider tells the desks apart. +func TestFlow_OtherSchemeSameType_PassesThroughUntouched(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + body := strings.Replace(string(request("status", statusAttributes())), `"id":"pmfby"`, `"id":"pmkisan"`, 1) + if !strings.Contains(body, `"id":"pmkisan"`) { + t.Fatal("the request names no pmfby provider to replace") + } + got, err := h.send(t, []byte(body)) + if err != nil || got != nil || len(h.pmfby.calls) != 0 { + t.Errorf("Run() = %v, %v, PMFBY calls %v; want PM-KISAN's request passed on untouched", got, err, h.pmfby.calls) + } +} + +func TestFlow_OtherCapability_PassesThroughUntouched(t *testing.T) { + h := newHarness(t, &pmfbyStub{}, 0) + body := strings.Replace(string(request("status", statusAttributes())), + "openagrinet:Grievance", "openagrinet:SchemeApplication", 1) + got, err := h.send(t, []byte(body)) + if err != nil || got != nil { + t.Errorf("Run() = %v, %v; want another capability passed on untouched", got, err) + } + if len(h.pmfby.calls) != 0 { + t.Errorf("PMFBY was called %v for another capability", h.pmfby.calls) + } +} diff --git a/pkg/plugin/implementation/internal/common/auth.go b/pkg/plugin/implementation/internal/common/auth.go index 4c780112a..a2afd67f2 100644 --- a/pkg/plugin/implementation/internal/common/auth.go +++ b/pkg/plugin/implementation/internal/common/auth.go @@ -27,7 +27,7 @@ type AuthProfile struct { // operator guessing which block to look at. Provider string - // One of none, basic, header, query, oauth2 or tokenQuery. + // One of none, basic, header, query, oauth2, tokenQuery or tokenHeader. Scheme string // Variable NAMES, never the values. @@ -60,18 +60,18 @@ type AuthProfile struct { TokenSecretName string TokenSecretEnv string - // Where the token sits in the response body, as a top-level key. + // Where the token sits in the response body, as a dotted path. TokenResponseField string - // How long a token may be held, as written in config and as parsed. - // - // Configured because the response carries no expiry at all -- nothing in it - // says how long the token lives, so an operator states what they believe - // and this refuses to guess. Set it too long and calls fail with the - // provider's own rejection until it lapses; the cache is dropped on such a - // rejection so the next call re-exchanges rather than waiting that out. + // How long a token may be held, for a response that states no expiry. Too + // long costs one rejected call: the rejection drops the token. TokenTTLRaw string TokenTTL time.Duration + + // Where the response states the token's expiry, as a dotted path to Unix + // seconds or an RFC 3339 time. Set this or tokenTtl: the issuer's own + // expiry beats an estimate. + TokenExpiryField string } // authenticator is one provider's profile plus the token it holds. @@ -116,21 +116,34 @@ func (a *AuthProfile) validate() error { "%s: authScheme oauth2 requires tokenUrl, clientIdEnv and clientSecretEnv", a.Provider) } - case util.AuthSchemeTokenQuery: + case util.AuthSchemeTokenQuery, util.AuthSchemeTokenHeader: // Every one of these is required, and none can be defaulted: the JSON - // keys, the response key and the parameter name are all this + // keys, the response key and the parameter or header name are all this // provider's spelling, and guessing any of them sends a malformed // request whose rejection says nothing about the cause. if a.TokenURL == "" || a.TokenUserField == "" || a.TokenUserEnv == "" || a.TokenSecretName == "" || a.TokenSecretEnv == "" || - a.TokenResponseField == "" || a.QueryName == "" { - return fmt.Errorf("%s: authScheme tokenQuery requires tokenUrl, "+ - "tokenUserField, tokenUserEnv, tokenSecretField, tokenSecretEnv, "+ - "tokenResponseField and queryName", a.Provider) + a.TokenResponseField == "" { + return fmt.Errorf("%s: authScheme %s requires tokenUrl, "+ + "tokenUserField, tokenUserEnv, tokenSecretField, tokenSecretEnv "+ + "and tokenResponseField", a.Provider, a.Scheme) + } + if a.Scheme == util.AuthSchemeTokenQuery && a.QueryName == "" { + return fmt.Errorf("%s: authScheme tokenQuery requires queryName", a.Provider) + } + if a.Scheme == util.AuthSchemeTokenHeader && a.HeaderName == "" { + return fmt.Errorf("%s: authScheme tokenHeader requires headerName", a.Provider) + } + if a.TokenExpiryField != "" { + // Two lifetimes would leave an operator guessing which one holds. + if a.TokenTTLRaw != "" { + return fmt.Errorf("%s: set tokenTtl or tokenExpiryField, not both", a.Provider) + } + break } if a.TokenTTLRaw == "" { - return fmt.Errorf("%s: authScheme tokenQuery requires tokenTtl, "+ - "because the token response carries no expiry to read", a.Provider) + return fmt.Errorf("%s: authScheme %s requires tokenTtl or tokenExpiryField, "+ + "or the token's lifetime is unknown", a.Provider, a.Scheme) } ttl, err := time.ParseDuration(a.TokenTTLRaw) if err != nil { @@ -152,7 +165,7 @@ func (a *AuthProfile) validate() error { default: return fmt.Errorf( "%s: unknown authScheme %q: must be none, basic, header, query, "+ - "oauth2 or tokenQuery", a.Provider, a.Scheme) + "oauth2, tokenQuery or tokenHeader", a.Provider, a.Scheme) } return nil } @@ -245,7 +258,8 @@ var authFields = map[string]func(*AuthProfile, string){ // Kept as written and turned into a duration by validate, so a bad value is // refused at startup naming the provider rather than read as zero -- which // would mean re-exchanging a token on every single call. - "tokenTtl": func(p *AuthProfile, v string) { p.TokenTTLRaw = v }, + "tokenTtl": func(p *AuthProfile, v string) { p.TokenTTLRaw = v }, + "tokenExpiryField": func(p *AuthProfile, v string) { p.TokenExpiryField = v }, } // providerIDFrom returns the provider half of "|". @@ -307,6 +321,14 @@ func (s *Step) authenticate(auth *authenticator, req *http.Request) error { } req.Header.Set("Authorization", "Bearer "+token) + case util.AuthSchemeTokenHeader: + // Exchanged like tokenQuery, sent bare in the header. + token, err := s.providerToken(req.Context(), auth) + if err != nil { + return err + } + req.Header.Set(cfg.HeaderName, token) + case util.AuthSchemeTokenQuery: // Exchanged like oauth2, placed like query: the same held token, put // where this provider reads it from. diff --git a/pkg/plugin/implementation/internal/common/common.go b/pkg/plugin/implementation/internal/common/common.go index 9390a9373..3012eb414 100644 --- a/pkg/plugin/implementation/internal/common/common.go +++ b/pkg/plugin/implementation/internal/common/common.go @@ -52,6 +52,9 @@ type Config struct { // A network convention, not a preference: every participant must agree or // requests silently fail to match. Configurable only so a spec change can // be tracked without waiting for a release. Both or neither. + // + // Either may list several comma-separated paths, paired by position and + // tried in order until one finds a binding. ProviderIDAt string `yaml:"providerIdAt" json:"providerIdAt"` CapabilityCodeAt string `yaml:"capabilityCodeAt" json:"capabilityCodeAt"` @@ -79,7 +82,7 @@ type Config struct { // safe for concurrent use. type Step struct { config *Config - paths Paths + paths []Paths prerequisites Prerequisites registry definition.ProviderRecordLookup mapper definition.Mapper @@ -106,14 +109,14 @@ func New(ctx context.Context, registry definition.ProviderRecordLookup, mapper d return nil, nil, err } - paths, err := BindingPaths(cfg) + pathList, err := bindingPathList(cfg) if err != nil { return nil, nil, err } step := &Step{ config: cfg, - paths: paths, + paths: pathList, prerequisites: prerequisites, registry: registry, mapper: mapper, @@ -144,21 +147,45 @@ func New(ctx context.Context, registry definition.ProviderRecordLookup, mapper d // same way this step does -- AgricultureFacility, whose search.go fans one // request out over several capability types. The alternative is reading the // same two config fields a second time, which is how the two drift apart. +// +// A path list is refused: such a caller reads only one pair. func BindingPaths(cfg *Config) (Paths, error) { + pathList, err := bindingPathList(cfg) + if err != nil { + return Paths{}, err + } + if len(pathList) > 1 { + return Paths{}, errors.New("providerIdAt and capabilityCodeAt list several paths, which only the step itself reads") + } + return pathList[0], nil +} + +// bindingPathList returns the path pairs to try, in order: the comma-separated +// halves paired by position. +func bindingPathList(cfg *Config) ([]Paths, error) { if cfg.ProviderIDAt == "" && cfg.CapabilityCodeAt == "" { - return BecknV2, nil + return []Paths{BecknV2}, nil } if cfg.ProviderIDAt == "" { - return Paths{}, errors.New("capabilityCodeAt is set without providerIdAt") + return nil, errors.New("capabilityCodeAt is set without providerIdAt") } if cfg.CapabilityCodeAt == "" { - return Paths{}, errors.New("providerIdAt is set without capabilityCodeAt") - } - paths := Paths{ProviderID: cfg.ProviderIDAt, CapabilityCode: cfg.CapabilityCodeAt} - if err := paths.Validate(); err != nil { - return Paths{}, err + return nil, errors.New("providerIdAt is set without capabilityCodeAt") + } + providers := strings.Split(cfg.ProviderIDAt, ",") + capabilities := strings.Split(cfg.CapabilityCodeAt, ",") + if len(providers) != len(capabilities) { + return nil, fmt.Errorf("providerIdAt lists %d paths and capabilityCodeAt %d; they pair by position", + len(providers), len(capabilities)) + } + pairs := make([]Paths, len(providers)) + for i := range providers { + pairs[i] = Paths{ProviderID: strings.TrimSpace(providers[i]), CapabilityCode: strings.TrimSpace(capabilities[i])} + if err := pairs[i].Validate(); err != nil { + return nil, err + } } - return paths, nil + return pairs, nil } // applyDefaults fills in what was left out and rejects what cannot be defaulted. diff --git a/pkg/plugin/implementation/internal/common/common_test.go b/pkg/plugin/implementation/internal/common/common_test.go index 2a18fbfe6..e7c9609ef 100644 --- a/pkg/plugin/implementation/internal/common/common_test.go +++ b/pkg/plugin/implementation/internal/common/common_test.go @@ -1288,7 +1288,7 @@ func TestNewUsesTheBecknConventionByDefault(t *testing.T) { t.Parallel() step := newStep(t, &stubRegistry{}, &stubMapper{}) - if step.paths != BecknV2 { + if !slices.Equal(step.paths, []Paths{BecknV2}) { t.Errorf("paths = %+v, want the Beckn v2 convention", step.paths) } } @@ -2558,3 +2558,135 @@ func TestOAuth2DoesNotRetryAMissingCredential(t *testing.T) { t.Errorf("token endpoint called %d times, want 0", got) } } + +// refusalStep is a step over mapper whose provider answers {"ok":true}, +// counting calls. +func refusalStep(t *testing.T, mapper *stubMapper, calls *int) *Step { + t.Helper() + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + *calls++ + fmt.Fprint(w, `{"ok":true}`) + })) + t.Cleanup(provider.Close) + return newStep(t, &stubRegistry{plan: testPlan(provider.URL, http.MethodPost)}, mapper) +} + +// refusedAs reads the HTTP status and Beckn error a refusal is answered with, +// whichever of the two step error types carries it. +func refusedAs(t *testing.T, err error) (int, *model.Error) { + t.Helper() + var invalid *model.SchemaValidationErr + var coded *model.CodedErr + switch { + case errors.As(err, &invalid): + return http.StatusBadRequest, invalid.BecknError() + case errors.As(err, &coded): + return coded.HTTPStatus(), coded.BecknError() + } + t.Fatalf("error = %v (%T), want a refusal", err, err) + return 0, nil +} + +func TestRefusal_RequestHalf_RefusesWithoutCallingProvider(t *testing.T) { + var calls int + mapper := &stubMapper{ + requestResult: []byte(`{"_error":{"status":400,"code":"SCH_REQUIRED_FIELD_MISSING", + "message":"phone is required","path":"$.message.phone"}}`), + responseResult: []byte(`{"a":1}`), + } + _, err := runStep(t, refusalStep(t, mapper, &calls), selectBody) + + status, refused := refusedAs(t, err) + if status != http.StatusBadRequest || refused.Code != "SCH_REQUIRED_FIELD_MISSING" || + refused.Message != "phone is required" || refused.Details == nil || refused.Details.Path != "$.message.phone" { + t.Fatalf("answer = %d %+v, want 400 SCH_REQUIRED_FIELD_MISSING at $.message.phone", status, refused) + } + if calls != 0 { + t.Errorf("provider called %d times for a refused request, want 0", calls) + } +} + +func TestRefusal_ResponseHalf_ReturnsItsStatusAndCode(t *testing.T) { + for name, tc := range map[string]struct { + refusal string + status int + code string + }{ + "explicit status": {`{"status":502,"code":"NET_DOWNSTREAM_UNAVAILABLE","message":"refused"}`, http.StatusBadGateway, "NET_DOWNSTREAM_UNAVAILABLE"}, + "status omitted": {`{"code":"BIZ_GENERIC_ERROR","message":"refused"}`, http.StatusBadRequest, "BIZ_GENERIC_ERROR"}, + } { + t.Run(name, func(t *testing.T) { + var calls int + mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{"_error":` + tc.refusal + `}`)} + stepCtx, err := runStep(t, refusalStep(t, mapper, &calls), selectBody) + + if status, refused := refusedAs(t, err); status != tc.status || refused.Code != tc.code { + t.Fatalf("answer = %d %+v, want %d %s", status, refused, tc.status, tc.code) + } + if len(stepCtx.ResponseBody) != 0 { + t.Errorf("a refusal produced a response body: %s", stepCtx.ResponseBody) + } + }) + } +} + +func TestRefusal_UnreadableError_FailsRatherThanAnswering(t *testing.T) { + for name, refusal := range map[string]string{ + "no message": `{"_error":{"status":502,"code":"X"}}`, + "not object": `{"_error":"refused"}`, + } { + t.Run(name, func(t *testing.T) { + var calls int + mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(refusal)} + stepCtx, err := runStep(t, refusalStep(t, mapper, &calls), selectBody) + if err == nil || len(stepCtx.ResponseBody) != 0 { + t.Errorf("Run() = %v with body %s, want an error and no answer", err, stepCtx.ResponseBody) + } + }) + } +} + +// cause is the provider's own wording, kept for the log: the caller never sees it. +func TestRefusal_Cause_NotSentToCaller(t *testing.T) { + var calls int + mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte( + `{"_error":{"status":502,"code":"NET_DOWNSTREAM_UNAVAILABLE","message":"try again later","cause":"AxiosError: 403"}}`)} + _, err := runStep(t, refusalStep(t, mapper, &calls), selectBody) + + _, refused := refusedAs(t, err) + if refused.Message != "try again later" || strings.Contains(err.Error(), "AxiosError") { + t.Errorf("answer = %+v (%v), want the mapping's message without the cause", refused, err) + } +} + +// A mapping that writes "_error": null means no error, not an unreadable one. +func TestRefusal_NullErrorField_AnswersAsBefore(t *testing.T) { + var calls int + mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{"_error":null,"message":{"ok":true}}`)} + stepCtx, err := runStep(t, refusalStep(t, mapper, &calls), selectBody) + if err != nil || calls != 1 || !strings.Contains(string(stepCtx.ResponseBody), `"ok":true`) { + t.Errorf("Run() = %v, body %s, %d calls; want the mapped answer after one call", err, stepCtx.ResponseBody, calls) + } +} + +// A caller of BindingPaths reads one pair, so a list is refused rather than +// handed over as one path that matches nothing. +func TestBindingPaths_PathList_Refused(t *testing.T) { + cfg := &Config{ProviderIDAt: "a.id, b.id", CapabilityCodeAt: "a.@type, b.@type"} + if _, err := BindingPaths(cfg); err == nil { + t.Error("BindingPaths() accepted a path list") + } + cfg = &Config{ProviderIDAt: "a.id", CapabilityCodeAt: "a.@type"} + if got, err := BindingPaths(cfg); err != nil || got != (Paths{ProviderID: "a.id", CapabilityCode: "a.@type"}) { + t.Errorf("BindingPaths() = %+v, %v; want the single pair", got, err) + } +} + +func TestRefusal_NoErrorField_AnswersAsBefore(t *testing.T) { + var calls int + mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{"message":{"ok":true}}`)} + stepCtx, err := runStep(t, refusalStep(t, mapper, &calls), selectBody) + if err != nil || string(stepCtx.ResponseBody) != `{"message":{"ok":true}}` || calls != 1 { + t.Errorf("Run() = %v, body %s, %d calls; want the mapped answer after one call", err, stepCtx.ResponseBody, calls) + } +} diff --git a/pkg/plugin/implementation/internal/common/redact.go b/pkg/plugin/implementation/internal/common/redact.go index 1faa93be1..000c62f5b 100644 --- a/pkg/plugin/implementation/internal/common/redact.go +++ b/pkg/plugin/implementation/internal/common/redact.go @@ -138,7 +138,7 @@ func (a *authenticator) secretForms() []string { forms = append(forms, encoded) } return forms - case util.AuthSchemeTokenQuery: + case util.AuthSchemeTokenQuery, util.AuthSchemeTokenHeader: // oauth2's two halves, with query's escaping: the secret goes to the // token endpoint, and the token it returns travels in a QUERY STRING -- // which is the exposed placement, so its escaped form has to be diff --git a/pkg/plugin/implementation/internal/common/serve.go b/pkg/plugin/implementation/internal/common/serve.go index 6d34eca71..1671df2c0 100644 --- a/pkg/plugin/implementation/internal/common/serve.go +++ b/pkg/plugin/implementation/internal/common/serve.go @@ -3,10 +3,12 @@ package common import ( + "bytes" "context" "encoding/json" "errors" "fmt" + "net/http" "slices" "strings" @@ -22,7 +24,20 @@ import ( // recognises its own work, so adding a provider is one config entry rather than // a routing-table change. func (s *Step) Run(ctx *model.StepContext) error { - binding, err := BindingFrom(s.paths, ctx.Body) + var binding Binding + err := errNoBinding + for i, paths := range s.paths { + binding, err = BindingFrom(paths, ctx.Body) + if errors.Is(err, errNoBinding) { + continue + } + // A later pair may read a shared container (support channels), so an + // unresolvable payload is refused only if it names this step's binding. + if err != nil && i > 0 && !s.namesServed(paths, ctx.Body) { + return nil + } + break + } if errors.Is(err, errNoBinding) { return nil } @@ -82,6 +97,22 @@ func (s *Step) serves(key string) bool { return slices.Contains(s.config.BindingKeys, key) } +// namesServed reports whether the payload names a binding this step serves. +func (s *Step) namesServed(paths Paths, body []byte) bool { + var payload any + if json.Unmarshal(body, &payload) != nil { + return false + } + for _, provider := range ValuesAt(payload, paths.ProviderID) { + for _, capability := range ValuesAt(payload, paths.CapabilityCode) { + if s.serves(Binding{ParticipantID: provider, CapabilityCode: capability}.Key()) { + return true + } + } + } + return false +} + // serve runs the exchange this step exists for: resolve, map out, call, map back. func (s *Step) serve(ctx *model.StepContext, plan *model.ProviderRecord) error { action := extractAction(ctx.Body) @@ -145,6 +176,11 @@ func (s *Step) serve(ctx *model.StepContext, plan *model.ProviderRecord) error { if err != nil { return err } + // The provider body is not logged: it can carry the farmer's details. + if cause, err := refusalIn(becknResponse); err != nil { + log.Warnf(ctx, "%s refused the request: %v (provider said: %q)", plan.BindingKey, err, s.redactString(cause)) + return err + } if len(becknResponse) == 0 { // No response half, or its transform matched nothing. Either way there // is no Beckn response, and returning the provider's own shape would @@ -172,12 +208,47 @@ func (s *Step) buildRequest(ctx context.Context, call model.ActionPlan, beckn an if err != nil { return nil, err } + if _, err := refusalIn(mapped); err != nil { + return nil, err + } if len(mapped) == 0 { log.Debugf(ctx, "the request half of %s produced nothing; sending an empty request", call.Mappings) } return mapped, nil } +// refusalIn returns the error a mapping asks for under the reserved _error +// field {status, code, message, path, cause}, or nil if there is none. Status +// 0 or 400 is a 400 with details.path. cause is the provider's own wording, +// returned for the log and never sent to the caller. +func refusalIn(mapped []byte) (string, error) { + // Cheap text check first: almost no answer carries the field. + if !bytes.Contains(mapped, []byte(`"_error"`)) { + return "", nil + } + var document map[string]json.RawMessage + if json.Unmarshal(mapped, &document) != nil || document["_error"] == nil || string(document["_error"]) == "null" { + return "", nil + } + var r struct { + Status int `json:"status"` + Code string `json:"code"` + Message string `json:"message"` + Path string `json:"path"` + Cause string `json:"cause"` + } + if json.Unmarshal(document["_error"], &r) != nil || r.Message == "" { + return "", errors.New("a mapping refused the call with an unreadable _error; it needs status, code and message") + } + switch r.Status { + case 0, http.StatusBadRequest: + // CodedErr drops details, so the path needs SchemaValidationErr. + return r.Cause, &model.SchemaValidationErr{Errors: []model.Error{*model.NewCodedErrorWithCause(r.Code, r.Message, r.Path, nil)}} + default: + return r.Cause, model.NewCodedErr(r.Status, r.Code, errors.New(r.Message)) + } +} + // extractAction reads the Beckn action a request is for. func extractAction(body []byte) string { var payload struct { diff --git a/pkg/plugin/implementation/internal/common/token.go b/pkg/plugin/implementation/internal/common/token.go index 88b09f5d0..1a4766c90 100644 --- a/pkg/plugin/implementation/internal/common/token.go +++ b/pkg/plugin/implementation/internal/common/token.go @@ -87,7 +87,7 @@ type tokenResponse struct { // issuer that failed. Unclassified they would surface as 500, telling a peer // this adapter broke when it did not. func (s *Step) exchangeToken(ctx context.Context, auth *authenticator) (string, time.Duration, error) { - if auth.cfg.Scheme == util.AuthSchemeTokenQuery { + if auth.cfg.Scheme == util.AuthSchemeTokenQuery || auth.cfg.Scheme == util.AuthSchemeTokenHeader { return s.exchangeQueryToken(ctx, auth) } return s.exchangeOAuth2Token(ctx, auth) @@ -96,14 +96,13 @@ func (s *Step) exchangeToken(ctx context.Context, auth *authenticator) (string, // exchangeQueryToken POSTs a JSON body of configured field names and reads the // token out of a configured response field. // -// The lifetime comes from config, not from the response: this endpoint says -// nothing about how long its token lives. That is the whole reason tokenTtl is -// required rather than defaulted -- see AuthProfile. +// The lifetime is read from tokenExpiryField when the endpoint states one, and +// taken from tokenTtl when it does not -- see AuthProfile. func (s *Step) exchangeQueryToken(ctx context.Context, auth *authenticator) (string, time.Duration, error) { cfg := auth.cfg user, secret := os.Getenv(cfg.TokenUserEnv), os.Getenv(cfg.TokenSecretEnv) if user == "" || secret == "" { - return "", 0, s.missingCredential(ctx, cfg.Provider, util.AuthSchemeTokenQuery, + return "", 0, s.missingCredential(ctx, cfg.Provider, cfg.Scheme, cfg.TokenUserEnv+" and "+cfg.TokenSecretEnv) } @@ -123,18 +122,29 @@ func (s *Step) exchangeQueryToken(ctx context.Context, auth *authenticator) (str return "", 0, err } - // Decoded into a map because the field name is configured: a struct tag - // cannot be written for a key that is not known until config is read. - var fields map[string]any + // Decoded generically because the token's path is configured. + var fields any if err := json.Unmarshal(body, &fields); err != nil { return "", 0, s.permanentTokenErr(fmt.Errorf("token response from %s is not JSON: %w", cfg.TokenURL, err)) } - token, ok := fields[cfg.TokenResponseField].(string) - if !ok || token == "" { + tokens := ValuesAt(fields, cfg.TokenResponseField) + if len(tokens) == 0 || tokens[0] == "" { return "", 0, s.permanentTokenErr(fmt.Errorf( "token response from %s carries no %s", cfg.TokenURL, cfg.TokenResponseField)) } + token := tokens[0] + + if cfg.TokenExpiryField != "" { + lifetime, ok := expiryLifetime(fields, cfg.TokenExpiryField) + if !ok { + // Not cached for a guessed period, the same rule as oauth2. + return "", 0, s.permanentTokenErr(fmt.Errorf( + "token response from %s carries no future %s, so its lifetime is unknown", + cfg.TokenURL, cfg.TokenExpiryField)) + } + return token, lifetime, nil + } // The skew is subtracted here for the same reason oauth2 subtracts it from // expires_in: a request that passed the expiry check must not arrive after @@ -246,6 +256,30 @@ func (s *Step) permanentTokenErr(err error) error { return util.DoNotRetry(s.tokenErr(err)) } +// expiryLifetime reads an expiry stated as Unix seconds or an RFC 3339 time and +// returns how long the token may be held, false when it is absent, unreadable, +// or already past. +func expiryLifetime(fields any, path string) (time.Duration, bool) { + leaves := LeavesAt(fields, path) + if len(leaves) == 0 { + return 0, false + } + var expiry time.Time + switch value := leaves[0].(type) { + case float64: + expiry = time.Unix(int64(value), 0) + case string: + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + return 0, false + } + expiry = parsed + default: + return 0, false + } + return tokenLifetime(int(time.Until(expiry)/time.Second), util.TokenRefreshSkew) +} + // tokenLifetime turns expires_in into how long the token may be held, or // reports that it may not be held at all. // diff --git a/pkg/plugin/implementation/internal/common/tokenheader_test.go b/pkg/plugin/implementation/internal/common/tokenheader_test.go new file mode 100644 index 000000000..01236b4d5 --- /dev/null +++ b/pkg/plugin/implementation/internal/common/tokenheader_test.go @@ -0,0 +1,405 @@ +// tokenHeader: tokenQuery's exchange, the bare token sent in a configured header. +package common + +import ( + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/beckn-one/beckn-onix/pkg/model" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common/util" +) + +// assertStatus fails unless err is a CodedErr carrying status. +func assertStatus(t *testing.T, err error, status int) { + t.Helper() + var coded *model.CodedErr + if !errors.As(err, &coded) || coded.HTTPStatus() != status { + t.Fatalf("error = %v, want a coded %d", err, status) + } +} + +// tokenHeaderProfile is tokenQuery's profile under the header scheme. queryName +// is cleared to prove it is not needed. +func tokenHeaderProfile(tokenURL string) AuthProfile { + profile := tokenQueryProfile(tokenURL) + profile.Scheme = util.AuthSchemeTokenHeader + profile.QueryName = "" + profile.HeaderName = "Authorization" + return profile +} + +// tokenHeaderStep records the Authorization header and query each provider call +// arrived with. status, when non-zero, is what the provider answers. +func tokenHeaderStep(t *testing.T, tokenURL string, headers, queries *[]string, status *int) *Step { + t.Helper() + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + *headers = append(*headers, r.Header.Get("Authorization")) + *queries = append(*queries, r.URL.RawQuery) + if status != nil && *status != 0 { + w.WriteHeader(*status) + return + } + fmt.Fprint(w, `{"ok":true}`) + })) + t.Cleanup(provider.Close) + + return newStep(t, &stubRegistry{plan: testPlan(provider.URL, http.MethodPost)}, + &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{"a":1}`)}, + func(c *Config) { + profile := tokenHeaderProfile(tokenURL) + if err := profile.validate(); err != nil { + t.Fatalf("profile does not validate: %v", err) + } + c.setProviderAuth(profile) + }) +} + +func TestTokenHeader_ValidLogin_SendsBareTokenInHeaderNotQuery(t *testing.T) { + t.Setenv("TEST_TQ_USER", "pmfby-user") + t.Setenv("TEST_TQ_SECRET", "pmfby-password") + + ts := &queryTokenServer{token: "the-token"} + var headers, queries []string + step := tokenHeaderStep(t, ts.start(t), &headers, &queries, nil) + + if _, err := runStep(t, step, selectBody); err != nil { + t.Fatalf("Run() returned an unexpected error: %v", err) + } + if len(headers) != 1 || headers[0] != "the-token" { + t.Errorf("provider saw Authorization %q, want the bare token, no Bearer prefix", headers) + } + if strings.Contains(queries[0], "the-token") { + t.Errorf("provider saw query %q; the token belongs in the header only", queries[0]) + } + if ts.gotUser != "pmfby-user" || ts.gotSecret != "pmfby-password" || ts.gotType != "application/json" { + t.Errorf("login saw %q/%q as %q, want the configured pair as JSON", ts.gotUser, ts.gotSecret, ts.gotType) + } +} + +// PMFBY's login wraps the token in its reply envelope, so the field is a path. +func TestTokenHeader_NestedTokenField_ReadsTheDottedPath(t *testing.T) { + t.Setenv("TEST_TQ_USER", "user") + t.Setenv("TEST_TQ_SECRET", "secret") + + login := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprint(w, `{"responseCode":"1","responseDynamic":{"token":{"Token":"nested-token"}}}`) + })) + t.Cleanup(login.Close) + var authorization string + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authorization = r.Header.Get("Authorization") + fmt.Fprint(w, `{"ok":true}`) + })) + t.Cleanup(provider.Close) + + step := newStep(t, &stubRegistry{plan: testPlan(provider.URL, http.MethodPost)}, + &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{"a":1}`)}, + func(c *Config) { + profile := tokenHeaderProfile(login.URL) + profile.TokenResponseField = "responseDynamic.token.Token" + if err := profile.validate(); err != nil { + t.Fatalf("profile does not validate: %v", err) + } + c.setProviderAuth(profile) + }) + + if _, err := runStep(t, step, selectBody); err != nil { + t.Fatalf("Run() returned an unexpected error: %v", err) + } + if authorization != "nested-token" { + t.Errorf("provider saw Authorization %q, want the token read from the nested path", authorization) + } +} + +func TestTokenHeader_RepeatedCalls_LogsInOnce(t *testing.T) { + t.Setenv("TEST_TQ_USER", "user") + t.Setenv("TEST_TQ_SECRET", "secret") + + ts := &queryTokenServer{} + var headers, queries []string + step := tokenHeaderStep(t, ts.start(t), &headers, &queries, nil) + + for i := 0; i < 3; i++ { + if _, err := runStep(t, step, selectBody); err != nil { + t.Fatalf("request %d: %v", i, err) + } + } + if got := ts.calls.Load(); got != 1 { + t.Errorf("login called %d times for 3 requests, want 1", got) + } +} + +func TestTokenHeader_ProviderRejectsToken_LogsInAgainNextCall(t *testing.T) { + t.Setenv("TEST_TQ_USER", "user") + t.Setenv("TEST_TQ_SECRET", "secret") + + ts := &queryTokenServer{} + var headers, queries []string + status := http.StatusUnauthorized + step := tokenHeaderStep(t, ts.start(t), &headers, &queries, &status) + + if _, err := runStep(t, step, selectBody); err == nil { + t.Fatal("Run() succeeded against a 401, want an error") + } + status = 0 + if _, err := runStep(t, step, selectBody); err != nil { + t.Fatalf("second Run() = %v, want it to recover with a fresh token", err) + } + if got := ts.calls.Load(); got != 2 || headers[1] != "tok-2" { + t.Errorf("login called %d times, second call sent %q; want 2 and tok-2", got, headers[1]) + } +} + +func TestTokenHeader_LoginFails_Returns502WithoutCallingProvider(t *testing.T) { + t.Setenv("TEST_TQ_USER", "user") + t.Setenv("TEST_TQ_SECRET", "wrong") + + ts := &queryTokenServer{status: http.StatusUnauthorized, body: `{"message":"invalid credentials"}`} + var headers, queries []string + step := tokenHeaderStep(t, ts.start(t), &headers, &queries, nil) + + _, err := runStep(t, step, selectBody) + assertStatus(t, err, http.StatusBadGateway) + if len(headers) != 0 { + t.Errorf("provider was called %d times without a token, want 0", len(headers)) + } +} + +func TestTokenHeader_LoginResponseWithoutToken_Returns502(t *testing.T) { + t.Setenv("TEST_TQ_USER", "user") + t.Setenv("TEST_TQ_SECRET", "secret") + + ts := &queryTokenServer{field: "accessToken"} + var headers, queries []string + step := tokenHeaderStep(t, ts.start(t), &headers, &queries, nil) + + _, err := runStep(t, step, selectBody) + assertStatus(t, err, http.StatusBadGateway) +} + +func TestTokenHeader_CredentialUnset_FailsPermanentlyWithoutLogin(t *testing.T) { + t.Setenv("TEST_TQ_USER", "") + t.Setenv("TEST_TQ_SECRET", "") + + ts := &queryTokenServer{} + var headers, queries []string + step := tokenHeaderStep(t, ts.start(t), &headers, &queries, nil) + + _, err := runStep(t, step, selectBody) + if err == nil || ts.calls.Load() != 0 || len(headers) != 0 { + t.Fatalf("err %v, %d logins, %d provider calls; want an error and no calls", err, ts.calls.Load(), len(headers)) + } + if strings.Contains(err.Error(), "TEST_TQ_USER") { + t.Errorf("error %q names the environment variable", err) + } + if !strings.Contains(err.Error(), util.AuthSchemeTokenHeader) { + t.Errorf("error %q should name the scheme", err) + } +} + +func TestTokenHeader_AfterCall_RedactsSecretAndToken(t *testing.T) { + t.Setenv("TEST_TQ_USER", "an-identifier") + t.Setenv("TEST_TQ_SECRET", "the-secret-value") + + ts := &queryTokenServer{token: "the-token-value"} + var headers, queries []string + step := tokenHeaderStep(t, ts.start(t), &headers, &queries, nil) + if _, err := runStep(t, step, selectBody); err != nil { + t.Fatal(err) + } + + redacted := step.redactString("secret the-secret-value token the-token-value user an-identifier") + if strings.Contains(redacted, "the-secret-value") || strings.Contains(redacted, "the-token-value") { + t.Errorf("redacted %q still carries a credential", redacted) + } + if !strings.Contains(redacted, "an-identifier") { + t.Errorf("redacted %q hides the identifier, which is not a secret", redacted) + } +} + +func TestTokenHeader_IncompleteProfile_RefusedAtStartup(t *testing.T) { + for name, blank := range map[string]func(*AuthProfile){ + "tokenUrl": func(p *AuthProfile) { p.TokenURL = "" }, + "tokenUserField": func(p *AuthProfile) { p.TokenUserField = "" }, + "tokenUserEnv": func(p *AuthProfile) { p.TokenUserEnv = "" }, + "tokenSecretField": func(p *AuthProfile) { p.TokenSecretName = "" }, + "tokenSecretEnv": func(p *AuthProfile) { p.TokenSecretEnv = "" }, + "tokenResponseField": func(p *AuthProfile) { p.TokenResponseField = "" }, + "tokenTtl": func(p *AuthProfile) { p.TokenTTLRaw = "" }, + "headerName": func(p *AuthProfile) { p.HeaderName = "" }, + "tokenTtl not a duration": func(p *AuthProfile) { p.TokenTTLRaw = "ten minutes" }, + "tokenTtl within skew": func(p *AuthProfile) { p.TokenTTLRaw = "30s" }, + } { + t.Run(name, func(t *testing.T) { + profile := tokenHeaderProfile("https://login.example") + profile.Provider = "pmfby" + blank(&profile) + err := profile.validate() + if err == nil { + t.Fatal("validate() accepted an incomplete tokenHeader profile") + } + if !strings.Contains(err.Error(), "pmfby") { + t.Errorf("error %q should name the provider", err) + } + }) + } +} + +func TestTokenQuery_WithoutQueryName_StillRefused(t *testing.T) { + profile := tokenQueryProfile("https://login.example") + profile.QueryName = "" + if err := profile.validate(); err == nil || !strings.Contains(err.Error(), "queryName") { + t.Errorf("validate() = %v, want tokenQuery refused without queryName", err) + } +} + +func TestParseProviderAuth_TokenHeaderBlock_BuildsProfile(t *testing.T) { + profiles, err := ParseProviderAuth(map[string]string{ + "authScheme-pmfby": "tokenHeader", + "tokenUrl-pmfby": "https://login.example", + "tokenUserField-pmfby": "userName", + "tokenUserEnv-pmfby": "PMFBY_USER", + "tokenSecretField-pmfby": "password", + "tokenSecretEnv-pmfby": "PMFBY_PASSWORD", + "tokenResponseField-pmfby": "token", + "tokenTtl-pmfby": "10m", + "headerName-pmfby": "Authorization", + }) + if err != nil { + t.Fatal(err) + } + profile := profiles["pmfby"] + if profile == nil || profile.Scheme != util.AuthSchemeTokenHeader { + t.Fatalf("profiles = %+v, want a tokenHeader profile for pmfby", profiles) + } + if err := profile.validate(); err != nil { + t.Errorf("validate() = %v, want the parsed profile accepted", err) + } +} + +// expiryLoginStep logs in at a server whose reply states the token's expiry, as +// PMFBY's does, and counts the logins. +func expiryLoginStep(t *testing.T, reply string, logins *atomic.Int32) *Step { + t.Helper() + login := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + logins.Add(1) + fmt.Fprint(w, reply) + })) + t.Cleanup(login.Close) + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprint(w, `{"ok":true}`) + })) + t.Cleanup(provider.Close) + + return newStep(t, &stubRegistry{plan: testPlan(provider.URL, http.MethodPost)}, + &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{"a":1}`)}, + func(c *Config) { + profile := tokenHeaderProfile(login.URL) + profile.TokenResponseField = "responseDynamic.token.Token" + profile.TokenTTLRaw = "" + profile.TokenExpiryField = "responseDynamic.token.validTo" + if err := profile.validate(); err != nil { + t.Fatalf("profile does not validate: %v", err) + } + c.setProviderAuth(profile) + }) +} + +// pmfbyLoginReply is PMFBY's login reply. Its expirationTime is the moment of +// issue, not the expiry; validTo is the expiry. +func pmfbyLoginReply(validTo string) string { + return `{"responseCode":"1","responseDynamic":{"token":{"Token":"tok",` + + fmt.Sprintf(`"expirationTime":%d,"validTo":%s}}}`, time.Now().Unix(), validTo) +} + +func TestTokenHeader_ExpiryInReply_HeldWithoutTokenTtl(t *testing.T) { + t.Setenv("TEST_TQ_USER", "user") + t.Setenv("TEST_TQ_SECRET", "secret") + + var logins atomic.Int32 + inHalfAnHour := `"` + time.Now().Add(30*time.Minute).UTC().Format("2006-01-02T15:04:05.000Z") + `"` + step := expiryLoginStep(t, pmfbyLoginReply(inHalfAnHour), &logins) + + for i := 0; i < 3; i++ { + if _, err := runStep(t, step, selectBody); err != nil { + t.Fatalf("request %d: %v", i, err) + } + } + if got := logins.Load(); got != 1 { + t.Errorf("login called %d times for 3 requests, want 1", got) + } +} + +func TestTokenHeader_ExpiryMissingOrPast_Returns502AndCachesNothing(t *testing.T) { + for name, validTo := range map[string]string{ + "missing": "null", + "not a time": `"soon"`, + "past": `"` + time.Now().Add(-time.Minute).UTC().Format(time.RFC3339) + `"`, + } { + t.Run(name, func(t *testing.T) { + t.Setenv("TEST_TQ_USER", "user") + t.Setenv("TEST_TQ_SECRET", "secret") + + var logins atomic.Int32 + step := expiryLoginStep(t, pmfbyLoginReply(validTo), &logins) + + _, err := runStep(t, step, selectBody) + assertStatus(t, err, http.StatusBadGateway) + _, _ = runStep(t, step, selectBody) + if got := logins.Load(); got != 2 { + t.Errorf("login called %d times for 2 requests, want 2: nothing is cached", got) + } + }) + } +} + +func TestExpiryLifetime_FutureExpiry_ReturnsTimeLeftLessSkew(t *testing.T) { + expiry := time.Now().Add(30 * time.Minute) + for name, stated := range map[string]any{ + "unix seconds": float64(expiry.Unix()), + "rfc 3339": expiry.UTC().Format(time.RFC3339Nano), + } { + t.Run(name, func(t *testing.T) { + lifetime, ok := expiryLifetime(map[string]any{"token": map[string]any{"expiry": stated}}, "token.expiry") + want := 30*time.Minute - util.TokenRefreshSkew + if !ok || lifetime > want || lifetime < want-2*time.Second { + t.Errorf("expiryLifetime() = %s, %v; want about %s", lifetime, ok, want) + } + }) + } +} + +func TestTokenHeader_TtlAndExpiryField_RefusedAtStartup(t *testing.T) { + profile := tokenHeaderProfile("https://login.example") + profile.Provider = "pmfby" + profile.TokenExpiryField = "responseDynamic.token.validTo" + if err := profile.validate(); err == nil || !strings.Contains(err.Error(), "not both") { + t.Errorf("validate() = %v, want tokenTtl and tokenExpiryField refused together", err) + } +} + +func TestParseProviderAuth_TokenExpiryField_BuildsProfileWithoutTtl(t *testing.T) { + profiles, err := ParseProviderAuth(map[string]string{ + "authScheme-pmfby": "tokenHeader", + "tokenUrl-pmfby": "https://login.example", + "tokenUserField-pmfby": "userName", + "tokenUserEnv-pmfby": "PMFBY_USER", + "tokenSecretField-pmfby": "password", + "tokenSecretEnv-pmfby": "PMFBY_PASSWORD", + "tokenResponseField-pmfby": "responseDynamic.token.Token", + "tokenExpiryField-pmfby": "responseDynamic.token.validTo", + "headerName-pmfby": "Authorization", + }) + if err != nil { + t.Fatal(err) + } + if err := profiles["pmfby"].validate(); err != nil { + t.Errorf("validate() = %v, want the profile accepted without tokenTtl", err) + } +} diff --git a/pkg/plugin/implementation/internal/common/util/constant.go b/pkg/plugin/implementation/internal/common/util/constant.go index df4cb1d12..5d305d6e8 100644 --- a/pkg/plugin/implementation/internal/common/util/constant.go +++ b/pkg/plugin/implementation/internal/common/util/constant.go @@ -80,6 +80,10 @@ const ( // Inherits query's exposure -- the token reaches proxy logs and transport // errors -- so it is redacted the same way. See redact. AuthSchemeTokenQuery = "tokenQuery" + + // tokenQuery's login exchange, but the token is sent bare (no "Bearer") + // in the configured header. + AuthSchemeTokenHeader = "tokenHeader" ) // CodeUpstreamUnavailable: the provider could not be reached or failed. Not diff --git a/pkg/plugin/implementation/schemav2validator/README.md b/pkg/plugin/implementation/schemav2validator/README.md index 6ba0c7d61..98f54fa2c 100644 --- a/pkg/plugin/implementation/schemav2validator/README.md +++ b/pkg/plugin/implementation/schemav2validator/README.md @@ -42,6 +42,7 @@ schemaValidator: | `extendedSchema_maxCacheSize` | string | No | `"100"` | Maximum number of cached domain schemas | | `extendedSchema_downloadTimeout` | string | No | `"30"` | Timeout for downloading domain schemas | | `extendedSchema_allowedDomains` | string | No | `""` | Comma-separated domain whitelist (empty = all allowed) | +| `extendedSchema_conciseErrorTypes` | string | No | `""` | Comma-separated `@type`s whose schema errors are shortened to the first reason with the field named, without the schema or value dump (empty = every message in full) | | `extendedSchema_localSchemaPath` | string | No | `""` | Path to local schema directory; schemas preloaded at startup, network used as fallback | ### Auxiliary Specs diff --git a/pkg/plugin/implementation/schemav2validator/cmd/plugin.go b/pkg/plugin/implementation/schemav2validator/cmd/plugin.go index 9db7f49db..ff09e984a 100644 --- a/pkg/plugin/implementation/schemav2validator/cmd/plugin.go +++ b/pkg/plugin/implementation/schemav2validator/cmd/plugin.go @@ -104,6 +104,9 @@ func (vp schemav2ValidatorProvider) New(ctx context.Context, config map[string]s if v, ok := config["extendedSchema_localSchemaPath"]; ok && v != "" { cfg.ExtendedSchemaConfig.LocalSchemaPath = v } + if v, ok := config["extendedSchema_conciseErrorTypes"]; ok && v != "" { + cfg.ExtendedSchemaConfig.ConciseErrorTypes = strings.FieldsFunc(v, func(r rune) bool { return r == ',' || r == ' ' }) + } } diff --git a/pkg/plugin/implementation/schemav2validator/extended_schema.go b/pkg/plugin/implementation/schemav2validator/extended_schema.go index 1c3d7169f..64824406a 100644 --- a/pkg/plugin/implementation/schemav2validator/extended_schema.go +++ b/pkg/plugin/implementation/schemav2validator/extended_schema.go @@ -4,11 +4,13 @@ import ( "context" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io/fs" "net/url" "os" "path/filepath" + "regexp" "slices" "strings" "sync" @@ -72,6 +74,8 @@ type ExtendedSchemaConfig struct { DownloadTimeout int // seconds, default 30 AllowedDomains []string // whitelist (empty = all allowed) LocalSchemaPath string // when non-empty, preload schemas from this directory; network is fallback + // @types whose errors keep only the first reason. Empty: every message as before. + ConciseErrorTypes []string } // referencedObject represents a domain-specific object with @context. @@ -216,6 +220,15 @@ func (v *schemav2Validator) validateExtendedSchemas(ctx context.Context, body in var schemaErrors []model.Error v.extractSchemaErrors(err, &schemaErrors) + // Only a failed schema rule is shortened. An error this validator + // already worded (bad @context, unknown @type) is left as it is. + var validatorError *model.Error + if slices.Contains(refConfig.ConciseErrorTypes, obj.Type) && !errors.As(err, &validatorError) { + for i := range schemaErrors { + schemaErrors[i].Message = conciseMessage(schemaErrors[i].Details, schemaErrors[i].Message) + } + } + // After this, a path is no longer relative to the object. prefixSchemaErrorPaths(schemaErrors, obj.Path) return &model.SchemaValidationErr{Errors: schemaErrors} @@ -225,6 +238,37 @@ func (v *schemav2Validator) validateExtendedSchemas(ctx context.Context, body in return nil } +// How kin-openapi words a failed schema rule. +const ( + errorDumpMarker = "\nSchema:" // starts the schema and value dump + nestedReasonPrefix = "doesn't match schema due to: " + missingPropertyPrefix = "property " + patternReasonPrefix = "string doesn't match the regular expression" + forbiddenValueReason = `Doesn't match schema "not"` +) + +// wholeObjectPath matches a path that ends in a combinator: the object as a +// whole failed a rule, not one field. +var wholeObjectPath = regexp.MustCompile(`(^|\.)(allOf|anyOf|oneOf)$`) + +// conciseMessage keeps kin-openapi's first reason and names the field it is +// about. It drops the schema and value dumps (the value is the caller's own +// input) and the reasons of the other branches. +func conciseMessage(details *model.ErrorDetails, message string) string { + message, _, _ = strings.Cut(strings.ReplaceAll(message, nestedReasonPrefix, ""), errorDumpMarker) + if details == nil || strings.HasPrefix(message, missingPropertyPrefix) { + return message + } + field := strings.ReplaceAll(details.Path, "/", ".") + switch { + case field == "" || wholeObjectPath.MatchString(field): + return message + case strings.HasPrefix(message, patternReasonPrefix) || message == forbiddenValueReason: + return field + " is not valid" + } + return field + ": " + message +} + // prefixSchemaErrorPaths prefixes objPath onto each error's Details.Path, // updating it in place when Details already exists so any other field (e.g. // a chained Cause) set upstream is preserved rather than discarded. diff --git a/pkg/plugin/implementation/schemav2validator/extended_schema_test.go b/pkg/plugin/implementation/schemav2validator/extended_schema_test.go index 9facca442..e9d8acfad 100644 --- a/pkg/plugin/implementation/schemav2validator/extended_schema_test.go +++ b/pkg/plugin/implementation/schemav2validator/extended_schema_test.go @@ -1987,3 +1987,78 @@ func TestPayloadDirectedReader(t *testing.T) { }) } } + +func TestValidateExtendedSchemas_ConciseTypes_NamesTheFieldAndDropsTheDump(t *testing.T) { + for name, c := range map[string]struct { + listed []string + want string + }{ + "listed": {[]string{"Season"}, `season: value is not one of the allowed values ["Kharif","Rabi","Zaid"]`}, + "unlisted": {[]string{"Other"}, `value is not one of the allowed values ["Kharif","Rabi","Zaid"]`}, + "none": {nil, `value is not one of the allowed values ["Kharif","Rabi","Zaid"]`}, + } { + t.Run(name, func(t *testing.T) { + v := &schemav2Validator{ + config: &Config{EnableExtendedSchema: true, ExtendedSchemaConfig: ExtendedSchemaConfig{LocalSchemaPath: "preloaded", ConciseErrorTypes: c.listed}}, + schemaCache: newSchemaCache(10), + } + v.schemaCache.rawSchemas["Season/attributes.yaml"] = []byte(`openapi: 3.0.3 +info: {title: Season, version: 1.0.0} +components: + schemas: + Season: + type: object + properties: + season: {type: string, enum: [Kharif, Rabi, Zaid]}`) + body := map[string]interface{}{"message": map[string]interface{}{"channel": map[string]interface{}{ + "@context": "https://example.org/Season/v1.0/context.jsonld", "@type": "Season", "season": "Khariff"}}} + var schemaErr *model.SchemaValidationErr + if err := v.validateExtendedSchemas(context.Background(), body); !errors.As(err, &schemaErr) { + t.Fatalf("err = %v, want a schema error", err) + } + if got := schemaErr.Errors[0].Message; got != c.want { + t.Errorf("Message = %q, want %q", got, c.want) + } + }) + } +} + +func TestConciseMessage(t *testing.T) { + for _, c := range []struct{ path, in, want string }{ + {"cropYear", `string doesn't match the regular expression "^[0-9]{4}$"` + "\nSchema:\n {}\n\nValue:\n \"x\"", "cropYear is not valid"}, + {"description", "minimum string length is 10\nSchema:\n {}", "description: minimum string length is 10"}, + {"case/ticketNo", "string doesn't match the regular expression \"^[0-9]+$\"\nSchema:", "case.ticketNo is not valid"}, + {"description", `property "description" is missing` + "\nSchema:", `property "description" is missing`}, + {"allOf", "doesn't match schema due to: doesn't match any schema from \"anyOf\"\nSchema:\n {}\n Or doesn't match", `doesn't match any schema from "anyOf"`}, + {"allOf", "doesn't match schema due to: Doesn't match schema \"not\"\nSchema:\n {}", `Doesn't match schema "not"`}, + {"enrolmentId", "Doesn't match schema \"not\"\nSchema:\n {}\n\nValue:\n \"123456789012\"", "enrolmentId is not valid"}, + {"grievance/oneOf", "value doesn't match any schema from \"oneOf\"", `value doesn't match any schema from "oneOf"`}, + {"", "some reason", "some reason"}, + } { + if got := conciseMessage(&model.ErrorDetails{Path: c.path}, c.in); got != c.want { + t.Errorf("conciseMessage(%q) = %q, want %q", c.path, got, c.want) + } + } +} + +func TestValidateExtendedSchemas_ConciseTypes_LeavesTheValidatorsOwnErrorsAlone(t *testing.T) { + v := &schemav2Validator{ + config: &Config{EnableExtendedSchema: true, ExtendedSchemaConfig: ExtendedSchemaConfig{LocalSchemaPath: "preloaded", ConciseErrorTypes: []string{"Season"}}}, + schemaCache: newSchemaCache(10), + } + // The document holds no schema named Season, so the type cannot be found. + v.schemaCache.rawSchemas["Season/attributes.yaml"] = []byte(`openapi: 3.0.3 +info: {title: Other, version: 1.0.0} +components: + schemas: + Other: {type: object}`) + body := map[string]interface{}{"message": map[string]interface{}{"channel": map[string]interface{}{ + "@context": "https://example.org/Season/v1.0/context.jsonld", "@type": "Season"}}} + var schemaErr *model.SchemaValidationErr + if err := v.validateExtendedSchemas(context.Background(), body); !errors.As(err, &schemaErr) { + t.Fatalf("err = %v, want a schema error", err) + } + if got := schemaErr.Errors[0]; got.Code != "SCH_INVALID_ENTITY_TYPE" || strings.HasPrefix(got.Message, "message.channel") { + t.Errorf("error = %+v, want the validator's own message without a field prefix", got) + } +}