diff --git a/cmd/root.go b/cmd/root.go index 3ce196c..18560ef 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -16,6 +16,11 @@ const ( serviceVersion = "0.1.0-beta" ) +var ( + vendor = "" + model = "" +) + var rootCmd = &cobra.Command{ Use: "chargeflow", Short: "", @@ -45,9 +50,13 @@ func rootFlags() { // Add flag for OCPP version rootCmd.PersistentFlags().StringP("version", "v", ocpp.V16.String(), "OCPP version to use (1.6, 2.0.1 or 2.1)") rootCmd.PersistentFlags().BoolP("debug", "d", false, "Enable debug mode") + rootCmd.PersistentFlags().StringVarP(&vendor, "vendor", "V", "", "Charging-station vendor for vendor/model-specific schema selection") + rootCmd.PersistentFlags().StringVarP(&model, "model", "m", "", "Charging-station model for vendor/model-specific schema selection") _ = viper.BindPFlag("ocpp.version", rootCmd.PersistentFlags().Lookup("version")) _ = viper.BindPFlag("debug", rootCmd.PersistentFlags().Lookup("debug")) + _ = viper.BindPFlag("vendor", rootCmd.PersistentFlags().Lookup("vendor")) + _ = viper.BindPFlag("model", rootCmd.PersistentFlags().Lookup("model")) } func Execute(ctx context.Context) error { diff --git a/cmd/schema_register.go b/cmd/schema_register.go index efd6f0b..08d04e9 100644 --- a/cmd/schema_register.go +++ b/cmd/schema_register.go @@ -19,6 +19,8 @@ type registerConfig struct { SchemaFile string SchemaDir string Action string + Vendor string + Model string } var registerCfg registerConfig @@ -63,7 +65,11 @@ The schema file names should match the OCPP action names (e.g., "BootNotificatio logger := zap.L() cfg := loadRegisterConfig() - version := ocpp.Version(viper.GetString("ocpp.version")) + octx := ocpp.OcppContext{ + Version: ocpp.Version(viper.GetString("ocpp.version")), + Vendor: cfg.Vendor, + Model: cfg.Model, + } remoteRegistry, err := buildRemoteRegistry(logger) if err != nil { @@ -74,9 +80,9 @@ The schema file names should match the OCPP action names (e.g., "BootNotificatio switch { case cfg.SchemaFile != "": - return registerSingleSchema(ctx, logger, remoteRegistry, version, cfg.SchemaFile, cfg.Action) + return registerSingleSchema(ctx, logger, remoteRegistry, octx, cfg.SchemaFile, cfg.Action) default: - return registerSchemasFromDir(ctx, logger, remoteRegistry, version, cfg.SchemaDir) + return registerSchemasFromDir(ctx, logger, remoteRegistry, octx, cfg.SchemaDir) } }, } @@ -85,26 +91,33 @@ func registerSingleSchema( ctx context.Context, logger *zap.Logger, registry schema_registry.SchemaRegistry, - version ocpp.Version, + octx ocpp.OcppContext, filePath, action string, ) error { - logger.Info("Registering schema", + logger = logger.With( zap.String("file", filePath), zap.String("action", action), - zap.String("version", version.String())) + zap.String("vendor", octx.Vendor), + zap.String("model", octx.Model), + zap.String("version", octx.Version.String()), + ) + + logger.Info("Registering schema") schemaData, err := os.ReadFile(filePath) if err != nil { return errors.Wrapf(err, "failed to read schema file: %s", filePath) } - if err := registry.RegisterSchema(ctx, version, action, schemaData); err != nil { + if err := registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: octx, + Action: action, + Schema: schemaData, + }); err != nil { return errors.Wrapf(err, "failed to register schema for action %s", action) } - logger.Info("Successfully registered schema", - zap.String("action", action), - zap.String("version", version.String())) + logger.Info("Successfully registered schema") return nil } @@ -112,12 +125,17 @@ func registerSchemasFromDir( ctx context.Context, logger *zap.Logger, registry schema_registry.SchemaRegistry, - version ocpp.Version, + octx ocpp.OcppContext, dir string, ) error { - logger.Info("Registering schemas from directory", + logger = logger.With( zap.String("directory", dir), - zap.String("version", version.String())) + zap.String("vendor", octx.Vendor), + zap.String("model", octx.Model), + zap.String("version", octx.Version.String()), + ) + + logger.Info("Registering schemas from directory") entries, err := os.ReadDir(dir) if err != nil { @@ -150,7 +168,11 @@ func registerSchemasFromDir( continue } - if err := registry.RegisterSchema(ctx, version, action, schemaData); err != nil { + if err := registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: octx, + Action: action, + Schema: schemaData, + }); err != nil { logger.Error("Failed to register schema", zap.String("file", schemaPath), zap.String("action", action), @@ -181,6 +203,8 @@ func loadRegisterConfig() registerConfig { SchemaFile: viper.GetString("schema.register.file"), SchemaDir: viper.GetString("schema.register.dir"), Action: viper.GetString("schema.register.action"), + Vendor: viper.GetString("vendor"), + Model: viper.GetString("model"), } } diff --git a/cmd/schema_remove.go b/cmd/schema_remove.go index 876b2fc..06f9cda 100644 --- a/cmd/schema_remove.go +++ b/cmd/schema_remove.go @@ -71,7 +71,11 @@ You can remove a single schema by action name, derive the action from a file nam logger := zap.L() cfg := loadRemoveConfig() - version := ocpp.Version(viper.GetString("ocpp.version")) + octx := ocpp.OcppContext{ + Version: ocpp.Version(viper.GetString("ocpp.version")), + Vendor: vendor, + Model: model, + } remoteRegistry, err := buildRemoteRegistry(logger) if err != nil { @@ -82,12 +86,12 @@ You can remove a single schema by action name, derive the action from a file nam switch { case cfg.Action != "": - return removeSingleSchema(ctx, logger, remoteRegistry, version, cfg.Action) + return removeSingleSchema(ctx, logger, remoteRegistry, octx, cfg.Action) case cfg.SchemaFile != "": action, _ := strings.CutSuffix(filepath.Base(cfg.SchemaFile), ".json") - return removeSingleSchema(ctx, logger, remoteRegistry, version, action) + return removeSingleSchema(ctx, logger, remoteRegistry, octx, action) default: - return removeSchemasFromDir(ctx, logger, remoteRegistry, version, cfg.SchemaDir) + return removeSchemasFromDir(ctx, logger, remoteRegistry, octx, cfg.SchemaDir) } }, } @@ -96,20 +100,25 @@ func removeSingleSchema( ctx context.Context, logger *zap.Logger, registry schema_registry.SchemaRegistry, - version ocpp.Version, + octx ocpp.OcppContext, action string, ) error { - logger.Info("Removing schema", + logger = logger.With( zap.String("action", action), - zap.String("version", version.String())) - - if err := registry.DeleteSchema(ctx, version, action); err != nil { + zap.String("vendor", octx.Vendor), + zap.String("model", octx.Model), + zap.String("version", octx.Version.String()), + ) + logger.Info("Removing schema") + + if err := registry.DeleteSchema(ctx, schema_registry.DeleteSchemaRequest{ + OcppContext: octx, + Action: action, + }); err != nil { return errors.Wrapf(err, "failed to remove schema for action %s", action) } - logger.Info("Successfully removed schema", - zap.String("action", action), - zap.String("version", version.String())) + logger.Info("Successfully removed schema") return nil } @@ -117,12 +126,16 @@ func removeSchemasFromDir( ctx context.Context, logger *zap.Logger, registry schema_registry.SchemaRegistry, - version ocpp.Version, + octx ocpp.OcppContext, dir string, ) error { - logger.Info("Removing schemas matching directory", + logger = logger.With( zap.String("directory", dir), - zap.String("version", version.String())) + zap.String("vendor", octx.Vendor), + zap.String("model", octx.Model), + zap.String("version", octx.Version.String()), + ) + logger.Info("Removing schemas matching directory") entries, err := os.ReadDir(dir) if err != nil { @@ -145,7 +158,10 @@ func removeSchemasFromDir( action, _ := strings.CutSuffix(fileName, ".json") - if err := registry.DeleteSchema(ctx, version, action); err != nil { + if err := registry.DeleteSchema(ctx, schema_registry.DeleteSchemaRequest{ + OcppContext: octx, + Action: action, + }); err != nil { logger.Error("Failed to remove schema", zap.String("action", action), zap.String("file", fileName), diff --git a/cmd/validate.go b/cmd/validate.go index 00dafd4..73c4c0d 100644 --- a/cmd/validate.go +++ b/cmd/validate.go @@ -86,7 +86,11 @@ func registerSchemas( // Also could determine the OCPP version from the schema ID. action, _ := strings.CutSuffix(name, ".json") - err = registry.RegisterSchema(ctx, version, action, schemaData) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: version}, + Action: action, + Schema: schemaData, + }) if err != nil { return errors.Wrapf(err, "unable to register OCPP schema: %s", name) } @@ -154,7 +158,7 @@ var validate = &cobra.Command{ if overwrite { ocppVersion := viper.GetString("ocpp.version") version := ocpp.Version(ocppVersion) - err = registerSchemasFromDir(ctx, logger, registry, version, additionalOcppSchemasFolder) + err = registerSchemasFromDir(ctx, logger, registry, ocpp.OcppContext{Version: version, Vendor: vendor, Model: model}, additionalOcppSchemasFolder) if err != nil { return err } @@ -165,7 +169,7 @@ var validate = &cobra.Command{ RunE: func(cmd *cobra.Command, args []string) error { ocppVersion := viper.GetString("ocpp.version") file := viper.GetString("file") - version := ocpp.Version(ocppVersion) + output := viper.GetString("output") logger := zap.L() logger = logger.WithOptions(zap.WithCaller(false), zap.AddStacktrace(zap.FatalLevel)) @@ -177,49 +181,38 @@ var validate = &cobra.Command{ message = args[0] } - output := viper.GetString("output") - validationOpts := []validation.Option{} + if file == "" && message == "" { + return errors.New("no message provided to validate, please provide a message as a command line argument or use the --file flag to read from a file") + } - // Validate provided output extension if present if output != "" { ext := strings.ToLower(filepath.Ext(output)) if !supportedOutputFormats[ext] { return errors.Errorf("unsupported output format '%s', supported: .json, .csv, .txt", ext) } - - validationOpts = append(validationOpts, validation.WithOutput(output)) } - switch { - case file == "" && message == "": - return errors.New("no message provided to validate, please provide a message as a command line argument or use the --file flag to read from a file") - case message != "": - // The message is expected to be a JSON string in the format: - // '[2, "uniqueId", "BootNotification", {"chargePointVendor": "TestVendor", "chargePointModel": "TestModel"}]' - if output == "" { - return service.ValidateMessage(message, version) - } - - // Validate and write report - r, err := service.ValidateMessageWithReport(message, version) - if err != nil { - return err - } - - return validation.WriteReport(output, r) + req := validation.Request{ + OcppContext: ocpp.OcppContext{ + Version: ocpp.Version(ocppVersion), + Vendor: vendor, + Model: model, + }, + Output: output, + } - case file != "": - // Use the options pattern to write output using registered strategies - // Read the messages from the file - return service.ValidateFile(file, version, validationOpts...) + if message != "" { + req.Messages = []string{message} + } else { + req.File = file } - return nil + _, err := service.Validate(req) + return err }, } func init() { - // Add flags for additional OCPP schemas folder validate.Flags().StringVarP(&additionalOcppSchemasFolder, "schemas", "a", "", "Path to additional OCPP schemas folder") validate.Flags().StringP("response-type", "r", "", "Response type to validate against (e.g. 'BootNotificationResponse'). Currently needed if you want to validate a single response message. ") validate.Flags().StringP("file", "f", "", "Path to a file containing the OCPP message to validate. If this flag is set, the message will be read from the file instead of the command line argument.") diff --git a/cmd/validate_test.go b/cmd/validate_test.go index 38a1d11..02f905f 100644 --- a/cmd/validate_test.go +++ b/cmd/validate_test.go @@ -81,8 +81,7 @@ func Test_registerAdditionalSchemas(t *testing.T) { // Call the function to register additional schemas ctx := context.Background() - version := ocpp.Version(test.defaultOcppVersion) - err = registerSchemasFromDir(ctx, logger, registry, version, tempDir) + err = registerSchemasFromDir(ctx, logger, registry, ocpp.OcppContext{Version: ocpp.Version(test.defaultOcppVersion)}, tempDir) if test.expected != nil { // The shared function returns aggregated errors, so we check that an error occurred assert.Error(t, err) diff --git a/gen/mocks/pkg/schema_registry/schema_registry_mock.go b/gen/mocks/pkg/schema_registry/schema_registry_mock.go index bccff86..5e1425f 100644 --- a/gen/mocks/pkg/schema_registry/schema_registry_mock.go +++ b/gen/mocks/pkg/schema_registry/schema_registry_mock.go @@ -6,9 +6,8 @@ package mock_schema_registry import ( "context" - "encoding/json" - "github.com/ChargePi/chargeflow/pkg/ocpp" + "github.com/ChargePi/chargeflow/pkg/schema_registry" "github.com/kaptinlin/jsonschema" mock "github.com/stretchr/testify/mock" ) @@ -41,16 +40,16 @@ func (_m *MockSchemaRegistry) EXPECT() *MockSchemaRegistry_Expecter { } // DeleteSchema provides a mock function for the type MockSchemaRegistry -func (_mock *MockSchemaRegistry) DeleteSchema(ctx context.Context, ocppVersion ocpp.Version, action string) error { - ret := _mock.Called(ctx, ocppVersion, action) +func (_mock *MockSchemaRegistry) DeleteSchema(ctx context.Context, req schema_registry.DeleteSchemaRequest) error { + ret := _mock.Called(ctx, req) if len(ret) == 0 { panic("no return value specified for DeleteSchema") } var r0 error - if returnFunc, ok := ret.Get(0).(func(context.Context, ocpp.Version, string) error); ok { - r0 = returnFunc(ctx, ocppVersion, action) + if returnFunc, ok := ret.Get(0).(func(context.Context, schema_registry.DeleteSchemaRequest) error); ok { + r0 = returnFunc(ctx, req) } else { r0 = ret.Error(0) } @@ -64,30 +63,24 @@ type MockSchemaRegistry_DeleteSchema_Call struct { // DeleteSchema is a helper method to define mock.On call // - ctx context.Context -// - ocppVersion ocpp.Version -// - action string -func (_e *MockSchemaRegistry_Expecter) DeleteSchema(ctx interface{}, ocppVersion interface{}, action interface{}) *MockSchemaRegistry_DeleteSchema_Call { - return &MockSchemaRegistry_DeleteSchema_Call{Call: _e.mock.On("DeleteSchema", ctx, ocppVersion, action)} +// - req schema_registry.DeleteSchemaRequest +func (_e *MockSchemaRegistry_Expecter) DeleteSchema(ctx interface{}, req interface{}) *MockSchemaRegistry_DeleteSchema_Call { + return &MockSchemaRegistry_DeleteSchema_Call{Call: _e.mock.On("DeleteSchema", ctx, req)} } -func (_c *MockSchemaRegistry_DeleteSchema_Call) Run(run func(ctx context.Context, ocppVersion ocpp.Version, action string)) *MockSchemaRegistry_DeleteSchema_Call { +func (_c *MockSchemaRegistry_DeleteSchema_Call) Run(run func(ctx context.Context, req schema_registry.DeleteSchemaRequest)) *MockSchemaRegistry_DeleteSchema_Call { _c.Call.Run(func(args mock.Arguments) { var arg0 context.Context if args[0] != nil { arg0 = args[0].(context.Context) } - var arg1 ocpp.Version + var arg1 schema_registry.DeleteSchemaRequest if args[1] != nil { - arg1 = args[1].(ocpp.Version) - } - var arg2 string - if args[2] != nil { - arg2 = args[2].(string) + arg1 = args[1].(schema_registry.DeleteSchemaRequest) } run( arg0, arg1, - arg2, ) }) return _c @@ -98,14 +91,14 @@ func (_c *MockSchemaRegistry_DeleteSchema_Call) Return(err error) *MockSchemaReg return _c } -func (_c *MockSchemaRegistry_DeleteSchema_Call) RunAndReturn(run func(ctx context.Context, ocppVersion ocpp.Version, action string) error) *MockSchemaRegistry_DeleteSchema_Call { +func (_c *MockSchemaRegistry_DeleteSchema_Call) RunAndReturn(run func(ctx context.Context, req schema_registry.DeleteSchemaRequest) error) *MockSchemaRegistry_DeleteSchema_Call { _c.Call.Return(run) return _c } // GetSchema provides a mock function for the type MockSchemaRegistry -func (_mock *MockSchemaRegistry) GetSchema(ctx context.Context, ocppVersion ocpp.Version, action string) (*jsonschema.Schema, bool) { - ret := _mock.Called(ctx, ocppVersion, action) +func (_mock *MockSchemaRegistry) GetSchema(ctx context.Context, req schema_registry.GetSchemaRequest) (*jsonschema.Schema, bool) { + ret := _mock.Called(ctx, req) if len(ret) == 0 { panic("no return value specified for GetSchema") @@ -113,18 +106,18 @@ func (_mock *MockSchemaRegistry) GetSchema(ctx context.Context, ocppVersion ocpp var r0 *jsonschema.Schema var r1 bool - if returnFunc, ok := ret.Get(0).(func(context.Context, ocpp.Version, string) (*jsonschema.Schema, bool)); ok { - return returnFunc(ctx, ocppVersion, action) + if returnFunc, ok := ret.Get(0).(func(context.Context, schema_registry.GetSchemaRequest) (*jsonschema.Schema, bool)); ok { + return returnFunc(ctx, req) } - if returnFunc, ok := ret.Get(0).(func(context.Context, ocpp.Version, string) *jsonschema.Schema); ok { - r0 = returnFunc(ctx, ocppVersion, action) + if returnFunc, ok := ret.Get(0).(func(context.Context, schema_registry.GetSchemaRequest) *jsonschema.Schema); ok { + r0 = returnFunc(ctx, req) } else { if ret.Get(0) != nil { r0 = ret.Get(0).(*jsonschema.Schema) } } - if returnFunc, ok := ret.Get(1).(func(context.Context, ocpp.Version, string) bool); ok { - r1 = returnFunc(ctx, ocppVersion, action) + if returnFunc, ok := ret.Get(1).(func(context.Context, schema_registry.GetSchemaRequest) bool); ok { + r1 = returnFunc(ctx, req) } else { r1 = ret.Get(1).(bool) } @@ -138,30 +131,24 @@ type MockSchemaRegistry_GetSchema_Call struct { // GetSchema is a helper method to define mock.On call // - ctx context.Context -// - ocppVersion ocpp.Version -// - action string -func (_e *MockSchemaRegistry_Expecter) GetSchema(ctx interface{}, ocppVersion interface{}, action interface{}) *MockSchemaRegistry_GetSchema_Call { - return &MockSchemaRegistry_GetSchema_Call{Call: _e.mock.On("GetSchema", ctx, ocppVersion, action)} +// - req schema_registry.GetSchemaRequest +func (_e *MockSchemaRegistry_Expecter) GetSchema(ctx interface{}, req interface{}) *MockSchemaRegistry_GetSchema_Call { + return &MockSchemaRegistry_GetSchema_Call{Call: _e.mock.On("GetSchema", ctx, req)} } -func (_c *MockSchemaRegistry_GetSchema_Call) Run(run func(ctx context.Context, ocppVersion ocpp.Version, action string)) *MockSchemaRegistry_GetSchema_Call { +func (_c *MockSchemaRegistry_GetSchema_Call) Run(run func(ctx context.Context, req schema_registry.GetSchemaRequest)) *MockSchemaRegistry_GetSchema_Call { _c.Call.Run(func(args mock.Arguments) { var arg0 context.Context if args[0] != nil { arg0 = args[0].(context.Context) } - var arg1 ocpp.Version + var arg1 schema_registry.GetSchemaRequest if args[1] != nil { - arg1 = args[1].(ocpp.Version) - } - var arg2 string - if args[2] != nil { - arg2 = args[2].(string) + arg1 = args[1].(schema_registry.GetSchemaRequest) } run( arg0, arg1, - arg2, ) }) return _c @@ -172,22 +159,22 @@ func (_c *MockSchemaRegistry_GetSchema_Call) Return(schema *jsonschema.Schema, b return _c } -func (_c *MockSchemaRegistry_GetSchema_Call) RunAndReturn(run func(ctx context.Context, ocppVersion ocpp.Version, action string) (*jsonschema.Schema, bool)) *MockSchemaRegistry_GetSchema_Call { +func (_c *MockSchemaRegistry_GetSchema_Call) RunAndReturn(run func(ctx context.Context, req schema_registry.GetSchemaRequest) (*jsonschema.Schema, bool)) *MockSchemaRegistry_GetSchema_Call { _c.Call.Return(run) return _c } // RegisterSchema provides a mock function for the type MockSchemaRegistry -func (_mock *MockSchemaRegistry) RegisterSchema(ctx context.Context, ocppVersion ocpp.Version, action string, rawSchema json.RawMessage) error { - ret := _mock.Called(ctx, ocppVersion, action, rawSchema) +func (_mock *MockSchemaRegistry) RegisterSchema(ctx context.Context, req schema_registry.CreateSchemaRequest) error { + ret := _mock.Called(ctx, req) if len(ret) == 0 { panic("no return value specified for RegisterSchema") } var r0 error - if returnFunc, ok := ret.Get(0).(func(context.Context, ocpp.Version, string, json.RawMessage) error); ok { - r0 = returnFunc(ctx, ocppVersion, action, rawSchema) + if returnFunc, ok := ret.Get(0).(func(context.Context, schema_registry.CreateSchemaRequest) error); ok { + r0 = returnFunc(ctx, req) } else { r0 = ret.Error(0) } @@ -201,36 +188,24 @@ type MockSchemaRegistry_RegisterSchema_Call struct { // RegisterSchema is a helper method to define mock.On call // - ctx context.Context -// - ocppVersion ocpp.Version -// - action string -// - rawSchema json.RawMessage -func (_e *MockSchemaRegistry_Expecter) RegisterSchema(ctx interface{}, ocppVersion interface{}, action interface{}, rawSchema interface{}) *MockSchemaRegistry_RegisterSchema_Call { - return &MockSchemaRegistry_RegisterSchema_Call{Call: _e.mock.On("RegisterSchema", ctx, ocppVersion, action, rawSchema)} +// - req schema_registry.CreateSchemaRequest +func (_e *MockSchemaRegistry_Expecter) RegisterSchema(ctx interface{}, req interface{}) *MockSchemaRegistry_RegisterSchema_Call { + return &MockSchemaRegistry_RegisterSchema_Call{Call: _e.mock.On("RegisterSchema", ctx, req)} } -func (_c *MockSchemaRegistry_RegisterSchema_Call) Run(run func(ctx context.Context, ocppVersion ocpp.Version, action string, rawSchema json.RawMessage)) *MockSchemaRegistry_RegisterSchema_Call { +func (_c *MockSchemaRegistry_RegisterSchema_Call) Run(run func(ctx context.Context, req schema_registry.CreateSchemaRequest)) *MockSchemaRegistry_RegisterSchema_Call { _c.Call.Run(func(args mock.Arguments) { var arg0 context.Context if args[0] != nil { arg0 = args[0].(context.Context) } - var arg1 ocpp.Version + var arg1 schema_registry.CreateSchemaRequest if args[1] != nil { - arg1 = args[1].(ocpp.Version) - } - var arg2 string - if args[2] != nil { - arg2 = args[2].(string) - } - var arg3 json.RawMessage - if args[3] != nil { - arg3 = args[3].(json.RawMessage) + arg1 = args[1].(schema_registry.CreateSchemaRequest) } run( arg0, arg1, - arg2, - arg3, ) }) return _c @@ -241,7 +216,7 @@ func (_c *MockSchemaRegistry_RegisterSchema_Call) Return(err error) *MockSchemaR return _c } -func (_c *MockSchemaRegistry_RegisterSchema_Call) RunAndReturn(run func(ctx context.Context, ocppVersion ocpp.Version, action string, rawSchema json.RawMessage) error) *MockSchemaRegistry_RegisterSchema_Call { +func (_c *MockSchemaRegistry_RegisterSchema_Call) RunAndReturn(run func(ctx context.Context, req schema_registry.CreateSchemaRequest) error) *MockSchemaRegistry_RegisterSchema_Call { _c.Call.Return(run) return _c } diff --git a/internal/validation/opts.go b/internal/validation/opts.go index c7db087..835d571 100644 --- a/internal/validation/opts.go +++ b/internal/validation/opts.go @@ -1,6 +1,17 @@ package validation -// Option is a functional option for ValidateFile. +import "github.com/ChargePi/chargeflow/pkg/ocpp" + +// Request carries all inputs for a single validation run. +type Request struct { + OcppContext ocpp.OcppContext // OCPP version, vendor, and model for schema selection + Messages []string // inline messages to validate (mutually exclusive with File) + File string // path to a newline-delimited file of messages + Output string // optional path to write the report (.json, .csv, .txt) +} + +// Option is a functional option for ValidateFile (kept for backwards compat with callers +// that still build options separately before constructing a ValidationRequest). type Option func(*options) type options struct { diff --git a/internal/validation/service.go b/internal/validation/service.go index d61aead..3f5f12f 100644 --- a/internal/validation/service.go +++ b/internal/validation/service.go @@ -38,95 +38,42 @@ func NewService( } } -// ValidateMessage validates a single OCPP message against the schema. -func (s *Service) ValidateMessage(message string, ocppVersion ocpp.Version) error { - logger := s.logger.With(zap.String("message", message), zap.String("ocppVersion", ocppVersion.String())) - logger.Info("Validating message") - - validationReport, err := s.parseAndValidate(ocppVersion, []string{message}) - if err != nil { - return errors.Wrap(err, "failed to parse message") +// Validate validates messages and returns the report. When vendor and/or model are set, +// the registry attempts vendor/model-specific schemas before falling back to the base OCPP spec schemas. +func (s *Service) Validate(req Request) (*report.Report, error) { + logger := s.logger.With( + zap.String("ocppVersion", req.OcppContext.Version.String()), + zap.String("vendor", req.OcppContext.Vendor), + zap.String("model", req.OcppContext.Model), + ) + logger.Info("Validating messages") + + msgs := req.Messages + if len(msgs) == 0 && req.File != "" { + var err error + msgs, err = s.getMessagesFromFile(req.File) + if err != nil { + return nil, errors.Wrap(err, "unable to read messages from file") + } } - s.outputValidationErrorToLogs(validationReport) - return nil -} - -// ValidateMessageWithReport validates the message and returns the generated report. -// This is used by the CLI when an output file path is requested. -func (s *Service) ValidateMessageWithReport(message string, ocppVersion ocpp.Version) (*report.Report, error) { - validationReport, err := s.parseAndValidate(ocppVersion, []string{message}) + validationReport, err := s.parseAndValidate(req.OcppContext, msgs) if err != nil { - return nil, errors.Wrap(err, "failed to parse message") + return nil, errors.Wrap(err, "failed to parse and validate messages") } s.outputValidationErrorToLogs(validationReport) - return validationReport, nil -} - -// ValidateFile validates a file containing multiple OCPP messages against the schema. -// It accepts optional Option(s). If an output option is provided, the report -// will be written using a strategy based on the output file extension. -// If no options are provided, behavior is unchanged and results are logged to console. -func (s *Service) ValidateFile(file string, ocppVersion ocpp.Version, opts ...Option) error { - logger := s.logger.With(zap.String("file", file), zap.String("ocppVersion", ocppVersion.String())) - logger.Info("Validating file") - - // apply options - fo := &options{} - for _, opt := range opts { - opt(fo) - } - - msgs, err := s.getMessagesFromFile(file) - if err != nil { - return errors.Wrap(err, "unable to read messages from file") - } - - // Use existing helper to parse and validate and get report - validationReport, err := s.parseAndValidate(ocppVersion, msgs) - if err != nil { - return errors.Wrap(err, "unable to parse messages") - } - - // If no output provided, preserve original behavior: log errors to console - if fo.output == "" { - s.outputValidationErrorToLogs(validationReport) - return nil - } - - // Otherwise create strategy and write to file - strat, err := outputStrategyFactory(fo.output) - if err != nil { - return err - } - - if err := strat.Write(fo.output, validationReport); err != nil { - return errors.Wrap(err, "failed to write validation report") - } - - return nil -} - -// ValidateFileWithReport validates the file and returns the generated report. -// This is used by the CLI when an output file path is requested. -func (s *Service) ValidateFileWithReport(file string, ocppVersion ocpp.Version) (*report.Report, error) { - logger := s.logger.With(zap.String("file", file), zap.String("ocppVersion", ocppVersion.String())) - logger.Info("Validating file") - messages, err := s.getMessagesFromFile(file) - if err != nil { - return nil, errors.Wrap(err, "unable to read messages from file") - } - - logger.Info("✅ Successfully parsed file", zap.Int("messages", len(messages))) - - validationReport, err := s.parseAndValidate(ocppVersion, messages) - if err != nil { - return nil, errors.Wrap(err, "unable to parse messages") + if req.Output != "" { + strat, err := outputStrategyFactory(req.Output) + if err != nil { + return nil, err + } + if err := strat.Write(req.Output, validationReport); err != nil { + return nil, errors.Wrap(err, "failed to write validation report") + } } - s.outputValidationErrorToLogs(validationReport) return validationReport, nil } @@ -137,20 +84,14 @@ func (s *Service) outputValidationErrorToLogs(validationReport *report.Report) { return } - // Log the non-parsable messages first for line, errs := range validationReport.NonParsableMessages { logger := s.logger.With(zap.String("line", line)) logger.Error(fmt.Sprintf("Message could not be parsed at %s:", line)) - if len(errs) == 0 { - continue - } - for _, parseErr := range errs { logger.Error(fmt.Sprintf("👉 %s", parseErr)) } } - // Log any parsing or validation errors for messages for messageId, requestResponse := range validationReport.InvalidMessages { for k, validationErrors := range requestResponse { logger := s.logger.With(zap.String("messageId", messageId)) @@ -161,10 +102,6 @@ func (s *Service) outputValidationErrorToLogs(validationReport *report.Report) { logger.Error(fmt.Sprintf("Response for message %s has the following validation errors:", messageId)) } - if len(validationErrors) == 0 { - continue - } - for _, parseErr := range validationErrors { logger.Error(fmt.Sprintf("👉 %s", parseErr)) } @@ -173,24 +110,20 @@ func (s *Service) outputValidationErrorToLogs(validationReport *report.Report) { } // parseAndValidate parses and validates a list of OCPP messages. -func (s *Service) parseAndValidate(ocppVersion ocpp.Version, messages []string) (*report.Report, error) { - logger := s.logger.With(zap.String("ocppVersion", ocppVersion.String()), zap.Int("messages", len(messages))) +func (s *Service) parseAndValidate(octx ocpp.OcppContext, messages []string) (*report.Report, error) { + logger := s.logger.With(zap.String("ocppVersion", octx.Version.String()), zap.Int("messages", len(messages))) logger.Info("Parsing and validating messages") - // Parse the messages parserResults, nonParsedMessages, err := s.parser.Parse(messages) if err != nil { return nil, errors.Wrap(err, "failed to parse messages") } - // Add non-parsable messages to the aggregator for line, result := range nonParsedMessages { s.aggregator.AddNonParsableMessage(line, result) } - // Add parsed messages to the aggregator for messageId, result := range parserResults { - // Validate the request _, found := result.GetRequest() if found { s.aggregator.AddParserResult(messageId, true, result.Request) @@ -202,48 +135,38 @@ func (s *Service) parseAndValidate(ocppVersion ocpp.Version, messages []string) } } - // Only valid messages should be validated further validMessages := s.filterValidMessages(parserResults) invalidMessagesCount := len(parserResults) - len(validMessages) - logger.Info( - "✅ OCPP messages parsed. Proceeding with validation.", + logger.Info("✅ OCPP messages parsed. Proceeding with validation.", zap.Int("invalid_messages", invalidMessagesCount), zap.Int("unparsable_messages", len(nonParsedMessages)), ) for messageId, parserResult := range validMessages { - // Validate the request request, found := parserResult.GetRequest() if found { - result, err := s.validator.ValidateMessage(ocppVersion, request) + result, err := s.validator.ValidateMessage(octx, request) if err != nil { return nil, errors.Wrap(err, "failed to validate request message") } - - // Store the results in the aggregator s.aggregator.AddValidationResults(messageId, true, *result) } - // Validate the response response, found := parserResult.GetResponse() if found { - result, err := s.validator.ValidateMessage(ocppVersion, response) + result, err := s.validator.ValidateMessage(octx, response) if err != nil { return nil, errors.Wrap(err, "failed to validate response message") } - - // Store the results in the aggregator s.aggregator.AddValidationResults(messageId, false, *result) } responseError, found := parserResult.GetResponseError() if found { - result, err := s.validator.ValidateMessage(ocppVersion, responseError) + result, err := s.validator.ValidateMessage(octx, responseError) if err != nil { return nil, errors.Wrap(err, "failed to validate response error message") } - - // Store the results in the aggregator s.aggregator.AddValidationResults(messageId, false, *result) } } @@ -252,7 +175,7 @@ func (s *Service) parseAndValidate(ocppVersion ocpp.Version, messages []string) return &validationReport, nil } -// getMessagesFromFile reads messages from a file, where each message is separated by a newline character. +// getMessagesFromFile reads newline-delimited OCPP messages from a file. func (s *Service) getMessagesFromFile(file string) ([]string, error) { s.logger.Debug("Reading file", zap.String("file", file)) @@ -266,15 +189,13 @@ func (s *Service) getMessagesFromFile(file string) ([]string, error) { return nil, errors.Wrap(err, "failed to read file content") } - messages := strings.Split(string(content), "\n") - return messages, nil + return strings.Split(string(content), "\n"), nil } -// filterValidMessages filters out invalid messages from the parser results. +// filterValidMessages removes parser results that failed basic structural parsing. func (s *Service) filterValidMessages(parserResults map[string]parser.RequestResponseResult) map[string]parser.RequestResponseResult { - maps.DeleteFunc(parserResults, func(messageUniqueId string, parserResult parser.RequestResponseResult) bool { + maps.DeleteFunc(parserResults, func(_ string, parserResult parser.RequestResponseResult) bool { return !parserResult.IsValid() }) - return parserResults } diff --git a/internal/validation/service_test.go b/internal/validation/service_test.go index 9c1fbc7..64935ec 100644 --- a/internal/validation/service_test.go +++ b/internal/validation/service_test.go @@ -14,6 +14,7 @@ import ( mock_schema_registry "github.com/ChargePi/chargeflow/gen/mocks/pkg/schema_registry" "github.com/ChargePi/chargeflow/pkg/ocpp" + "github.com/ChargePi/chargeflow/pkg/schema_registry" ) var ( @@ -250,70 +251,77 @@ func (s *validationServiceTestSuite) TearDownSuite() { _ = os.RemoveAll(dir) } -func (s *validationServiceTestSuite) TestValidateFile() { +func (s *validationServiceTestSuite) TestValidate_File() { tests := []struct { name string - filepath string - version ocpp.Version + req Request setExpectations func(*mock_schema_registry.MockSchemaRegistry) expectedErr error }{ { - name: "Valid file with version 1.6", - filepath: s.files["ocpp16_all_valid"].path, - version: ocpp.V16, + name: "Valid file with version 1.6", + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, File: s.files["ocpp16_all_valid"].path}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) { compile, err := jsonschema.NewCompiler().Compile(bootNotificationSchema) s.Require().NoError(err) - - registry.EXPECT().Type().Return("file") - registry.EXPECT().GetSchema(mock.Anything, ocpp.V16, "BootNotificationRequest").Return(compile, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationRequest"}).Return(compile, true) compile, err = jsonschema.NewCompiler().Compile(bootNotificationResponseSchema) s.Require().NoError(err) - registry.EXPECT().GetSchema(mock.Anything, ocpp.V16, "BootNotificationResponse").Return(compile, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationResponse"}).Return(compile, true) }, expectedErr: nil, }, { - name: "Valid file with version 2.0", - filepath: s.files["ocpp201_all_valid"].path, - version: ocpp.V20, + name: "Valid file with version 2.0", + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, File: s.files["ocpp201_all_valid"].path}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) { compile, err := jsonschema.NewCompiler().Compile(costUpdatedSchema) s.Require().NoError(err) - - registry.EXPECT().Type().Return("file") - registry.EXPECT().GetSchema(mock.Anything, ocpp.V20, "CostUpdatedRequest").Return(compile, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "CostUpdatedRequest"}).Return(compile, true) compile, err = jsonschema.NewCompiler().Compile(costUpdatedResponseSchema) s.Require().NoError(err) - registry.EXPECT().GetSchema(mock.Anything, ocpp.V20, "CostUpdatedResponse").Return(compile, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "CostUpdatedResponse"}).Return(compile, true) }, expectedErr: nil, }, { - name: "Invalid version", - filepath: s.files["ocpp201_all_valid"].path, - version: "ocpp.V99", + name: "Invalid version", + req: Request{OcppContext: ocpp.OcppContext{Version: "ocpp.V99"}, File: s.files["ocpp201_all_valid"].path}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) { - registry.EXPECT().Type().Return("file") - registry.EXPECT().GetSchema(mock.Anything, ocpp.Version("ocpp.V99"), mock.Anything).Return(nil, false) + registry.EXPECT().GetSchema(mock.Anything, mock.MatchedBy(func(r schema_registry.GetSchemaRequest) bool { + return r.OcppContext.Version == ocpp.Version("ocpp.V99") + })).Return(nil, false) }, expectedErr: errors.New("no schema found for action CostUpdatedRequest in OCPP version ocpp.V99"), }, { name: "Non-existent file", - filepath: "./examples/non_existent_file.txt", - version: ocpp.V16, + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, File: "./examples/non_existent_file.txt"}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) {}, expectedErr: errors.New("failed to open file"), }, { - name: "Invalid messages", - filepath: s.files["empty_file"].path, - version: ocpp.V16, + name: "Empty file", + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, File: s.files["empty_file"].path}, + setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) {}, + expectedErr: nil, + }, + { + name: "Vendor/model-specific validation falls back to base schema", + req: Request{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, + File: s.files["ocpp16_all_valid"].path, + }, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) { + compile, err := jsonschema.NewCompiler().Compile(bootNotificationSchema) + s.Require().NoError(err) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, Action: "BootNotificationRequest"}).Return(compile, true) + + compile, err = jsonschema.NewCompiler().Compile(bootNotificationResponseSchema) + s.Require().NoError(err) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, Action: "BootNotificationResponse"}).Return(compile, true) }, expectedErr: nil, }, @@ -327,7 +335,7 @@ func (s *validationServiceTestSuite) TestValidateFile() { } service := NewService(s.logger, registry) - err := service.ValidateFile(tt.filepath, tt.version) + _, err := service.Validate(tt.req) if tt.expectedErr != nil { s.ErrorContains(err, tt.expectedErr.Error()) } else { @@ -337,58 +345,48 @@ func (s *validationServiceTestSuite) TestValidateFile() { } } -func (s *validationServiceTestSuite) TestValidateMessage() { +func (s *validationServiceTestSuite) TestValidate_Message() { tests := []struct { name string - message string - version ocpp.Version + req Request setExpectations func(*mock_schema_registry.MockSchemaRegistry) expectedErr error }{ { - name: "Valid message with version 1.6", - message: ocpp16validReq, - version: ocpp.V16, + name: "Valid message with version 1.6", + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Messages: []string{ocpp16validReq}}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) { compile, err := jsonschema.NewCompiler().Compile(bootNotificationSchema) s.Require().NoError(err) - registry.EXPECT().Type().Return("file") - registry.EXPECT().GetSchema(mock.Anything, ocpp.V16, "BootNotificationRequest").Return(compile, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationRequest"}).Return(compile, true) }, expectedErr: nil, }, { - name: "Valid message with version 2.0", - message: ocpp201validReq, - version: ocpp.V20, + name: "Valid message with version 2.0", + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Messages: []string{ocpp201validReq}}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) { compile, err := jsonschema.NewCompiler().Compile(costUpdatedSchema) s.Require().NoError(err) - registry.EXPECT().Type().Return("file") - registry.EXPECT().GetSchema(mock.Anything, ocpp.V20, "CostUpdatedRequest").Return(compile, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "CostUpdatedRequest"}).Return(compile, true) }, expectedErr: nil, }, { - name: "Invalid message with version 1.6", - message: ocpp16invalidReq, - version: ocpp.V16, - setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) { - //registry.EXPECT().GetSchema(ocpp.V16, "InvalidRequest").Return(nil, false) - }, - expectedErr: nil, + name: "Invalid message with version 1.6 (parse error)", + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Messages: []string{ocpp16invalidReq}}, + setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) {}, + expectedErr: nil, }, { name: "Unparsable message", - message: "{invalid: json}", - version: ocpp.V20, + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Messages: []string{"{invalid: json}"}}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) {}, expectedErr: nil, }, { - name: "Invalid message", - message: "[5, \"1234\", \"Invalid\", {\"errorCode\": \"GenericError\", \"errorDescription\": \"An error occurred\"}]", - version: ocpp.V16, + name: "CALL_RESULT_ERROR for OCPP 1.6 (unsupported)", + req: Request{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Messages: []string{`[5, "1234", "Invalid", {"errorCode": "GenericError", "errorDescription": "An error occurred"}]`}}, setExpectations: func(registry *mock_schema_registry.MockSchemaRegistry) {}, expectedErr: nil, }, @@ -401,10 +399,8 @@ func (s *validationServiceTestSuite) TestValidateMessage() { tt.setExpectations(registry) } - // todo capture messages from logger and check if they match the expected output service := NewService(s.logger, registry) - - err := service.ValidateMessage(tt.message, tt.version) + _, err := service.Validate(tt.req) if tt.expectedErr != nil { s.ErrorContains(err, tt.expectedErr.Error()) } else { diff --git a/pkg/ocpp/context.go b/pkg/ocpp/context.go new file mode 100644 index 0000000..b301f78 --- /dev/null +++ b/pkg/ocpp/context.go @@ -0,0 +1,9 @@ +package ocpp + +// OcppContext carries the OCPP protocol version together with optional +// vendor and model identifiers used for vendor/model-specific schema selection. +type OcppContext struct { + Version Version // Required + Vendor string // Optional. Must be present if model is set. + Model string // Optional. +} diff --git a/pkg/schema_registry/registries/file_registry/file.go b/pkg/schema_registry/registries/file_registry/file.go index e18dc10..ece4d56 100644 --- a/pkg/schema_registry/registries/file_registry/file.go +++ b/pkg/schema_registry/registries/file_registry/file.go @@ -2,7 +2,6 @@ package file_registry import ( "context" - "encoding/json" "strings" "sync" @@ -11,6 +10,7 @@ import ( "go.uber.org/zap" "github.com/ChargePi/chargeflow/pkg/ocpp" + "github.com/ChargePi/chargeflow/pkg/schema_registry" ) const ( @@ -59,89 +59,121 @@ func NewFileSchemaRegistry(logger *zap.Logger, opts ...RegistryOption) *SchemaRe // // The rawSchema should be a valid JSON schema in raw format. // The action is the name of the OCPP action that this schema applies to. Must be suffixed with either "Request" or "Response". -func (fsr *SchemaRegistry) RegisterSchema(_ context.Context, ocppVersion ocpp.Version, action string, rawSchema json.RawMessage) error { - logger := fsr.logger.With(zap.String("ocppVersion", ocppVersion.String()), zap.String("action", action)) +func (fsr *SchemaRegistry) RegisterSchema(_ context.Context, req schema_registry.CreateSchemaRequest) error { + logger := fsr.logger.With(zap.String("ocppVersion", req.OcppContext.Version.String()), zap.String("action", req.Action)) logger.Debug("Registering schema") // Validate the OCPP version - if !ocpp.IsValidProtocolVersion(ocppVersion) { - return errors.Errorf("invalid OCPP version: %s", ocppVersion) + if !ocpp.IsValidProtocolVersion(req.OcppContext.Version) { + return errors.Errorf("invalid OCPP version: %s", req.OcppContext.Version) } // Must be a valid action name ending with "Request" or "Response" - if !(strings.HasSuffix(action, RequestSuffix) || strings.HasSuffix(action, ResponseSuffix)) { - return errors.Errorf("action must end with 'Request' or 'Response': %s", action) + if !(strings.HasSuffix(req.Action, RequestSuffix) || strings.HasSuffix(req.Action, ResponseSuffix)) { + return errors.Errorf("action must end with 'Request' or 'Response': %s", req.Action) } logger.Debug("Compiling schema") - // Compile the schema using the jsonschema compiler - schema, err := compiler.Compile(rawSchema) + schema, err := compiler.Compile(req.Schema) if err != nil { return errors.Wrap(err, "failed to compile schema") } - // Acquire write lock to modify the schemasPerOcppVersion map fsr.mu.Lock() defer fsr.mu.Unlock() - if _, exists := fsr.schemasPerOcppVersion[ocppVersion]; !exists { - fsr.schemasPerOcppVersion[ocppVersion] = make(map[string]*jsonschema.Schema) + if _, exists := fsr.schemasPerOcppVersion[req.OcppContext.Version]; !exists { + fsr.schemasPerOcppVersion[req.OcppContext.Version] = make(map[string]*jsonschema.Schema) } + key := buildStorageKey(req.OcppContext.Vendor, req.OcppContext.Model, req.Action) + if !fsr.config.overwrite { logger.Debug("Overwriting previous schema") - // Check if the schema already exists for the given action - if _, exists := fsr.schemasPerOcppVersion[ocppVersion][action]; exists { - return errors.Errorf("schema for action %s already exists for OCPP version %s", action, ocppVersion) + if _, exists := fsr.schemasPerOcppVersion[req.OcppContext.Version][key]; exists { + return errors.Errorf("schema for action %s already exists for OCPP version %s", req.Action, req.OcppContext.Version) } } - // Register the schema for the specific action - fsr.schemasPerOcppVersion[ocppVersion][action] = schema + fsr.schemasPerOcppVersion[req.OcppContext.Version][key] = schema return nil } // DeleteSchema removes a schema for a specific OCPP version and action. -func (fsr *SchemaRegistry) DeleteSchema(_ context.Context, ocppVersion ocpp.Version, action string) error { - logger := fsr.logger.With(zap.String("ocppVersion", ocppVersion.String()), zap.String("action", action)) +func (fsr *SchemaRegistry) DeleteSchema(_ context.Context, req schema_registry.DeleteSchemaRequest) error { + logger := fsr.logger.With(zap.String("ocppVersion", req.OcppContext.Version.String()), zap.String("action", req.Action)) logger.Debug("Deleting schema") - if !ocpp.IsValidProtocolVersion(ocppVersion) { - return errors.Errorf("invalid OCPP version: %s", ocppVersion) + if !ocpp.IsValidProtocolVersion(req.OcppContext.Version) { + return errors.Errorf("invalid OCPP version: %s", req.OcppContext.Version) } fsr.mu.Lock() defer fsr.mu.Unlock() - schemas, exists := fsr.schemasPerOcppVersion[ocppVersion] + schemas, exists := fsr.schemasPerOcppVersion[req.OcppContext.Version] if !exists { - return errors.Errorf("no schemas registered for OCPP version %s", ocppVersion) + return errors.Errorf("no schemas registered for OCPP version %s", req.OcppContext.Version) } - if _, exists := schemas[action]; !exists { - return errors.Errorf("schema for action %s not found for OCPP version %s", action, ocppVersion) + key := buildStorageKey(req.OcppContext.Vendor, req.OcppContext.Model, req.Action) + if _, exists := schemas[key]; !exists { + return errors.Errorf("schema for action %s not found for OCPP version %s", req.Action, req.OcppContext.Version) } - delete(schemas, action) + delete(schemas, key) return nil } +// buildStorageKey returns a composite key that incorporates vendor and model when +// provided, keeping vendor/model-specific schemas separate from base schemas. +func buildStorageKey(vendor, model, action string) string { + if vendor == "" && model == "" { + return action + } + parts := make([]string, 0, 3) + if vendor != "" { + parts = append(parts, vendor) + } + if model != "" { + parts = append(parts, model) + } + parts = append(parts, action) + return strings.Join(parts, "|") +} + // GetSchema retrieves a schema for a specific OCPP version and action. -func (fsr *SchemaRegistry) GetSchema(_ context.Context, ocppVersion ocpp.Version, action string) (*jsonschema.Schema, bool) { - fsr.logger.Debug("Getting schema", zap.String("ocppVersion", ocppVersion.String()), zap.String("action", action)) +// When Vendor and/or Model are set it first tries the vendor/model-specific +// schema and falls back to the base OCPP spec schema. +func (fsr *SchemaRegistry) GetSchema(_ context.Context, req schema_registry.GetSchemaRequest) (*jsonschema.Schema, bool) { + fsr.logger.Debug("Getting schema", + zap.String("ocppVersion", req.OcppContext.Version.String()), + zap.String("action", req.Action), + zap.String("vendor", req.OcppContext.Vendor), + zap.String("model", req.OcppContext.Model), + ) fsr.mu.RLock() defer fsr.mu.RUnlock() - // Check if the OCPP version exists in the registry - if schemas, exists := fsr.schemasPerOcppVersion[ocppVersion]; exists { - // Check if the action exists for the given OCPP version - if schema, exists := schemas[action]; exists { + schemas, exists := fsr.schemasPerOcppVersion[req.OcppContext.Version] + if !exists { + return nil, false + } + + // Try vendor/model-specific key first when either field is provided. + if req.OcppContext.Vendor != "" || req.OcppContext.Model != "" { + if schema, ok := schemas[buildStorageKey(req.OcppContext.Vendor, req.OcppContext.Model, req.Action)]; ok { return schema, true } } + // Fall back to the base OCPP spec schema. + if schema, ok := schemas[req.Action]; ok { + return schema, true + } + return nil, false } diff --git a/pkg/schema_registry/registries/file_registry/file_test.go b/pkg/schema_registry/registries/file_registry/file_test.go index c7d3137..95d0f1f 100644 --- a/pkg/schema_registry/registries/file_registry/file_test.go +++ b/pkg/schema_registry/registries/file_registry/file_test.go @@ -10,6 +10,7 @@ import ( "go.uber.org/zap" "github.com/ChargePi/chargeflow/pkg/ocpp" + "github.com/ChargePi/chargeflow/pkg/schema_registry" ) type fileRegistryTestSuite struct { @@ -23,6 +24,9 @@ func (s *fileRegistryTestSuite) SetupSuite() { func (s *fileRegistryTestSuite) TestRegisterSchema() { ctx := context.Background() + + const authorizeSchema = `{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}` + tests := []struct { name string preconfigure func(registry *SchemaRegistry) @@ -37,7 +41,7 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { preconfigure: func(registry *SchemaRegistry) {}, ocppVersion: ocpp.V16, action: "AuthorizeRequest", - schema: json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`), + schema: json.RawMessage(authorizeSchema), opts: nil, expectedErr: nil, }, @@ -46,7 +50,7 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { preconfigure: func(registry *SchemaRegistry) {}, ocppVersion: ocpp.V20, action: "AuthorizeRequest", - schema: json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`), + schema: json.RawMessage(authorizeSchema), opts: nil, expectedErr: nil, }, @@ -54,7 +58,7 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { name: "Unsupported OCPP version", ocppVersion: ocpp.Version("unsupported"), action: "AuthorizeRequest", - schema: json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`), + schema: json.RawMessage(authorizeSchema), opts: nil, expectedErr: errors.New("invalid OCPP version: unsupported"), }, @@ -62,7 +66,7 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { name: "Unsupported action", ocppVersion: ocpp.V20, action: "Authorize", - schema: json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`), + schema: json.RawMessage(authorizeSchema), opts: nil, expectedErr: errors.New("action must end with 'Request' or 'Response': Authorize"), }, @@ -78,10 +82,14 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { name: "Schema already registered, overwrite disabled", ocppVersion: ocpp.V16, preconfigure: func(registry *SchemaRegistry) { - _ = registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`)) + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: json.RawMessage(authorizeSchema), + }) }, action: "AuthorizeRequest", - schema: json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`), + schema: json.RawMessage(authorizeSchema), opts: []RegistryOption{WithOverwrite(false)}, expectedErr: errors.New("schema for action AuthorizeRequest already exists for OCPP version 1.6"), }, @@ -89,10 +97,14 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { name: "Schema already registered, overwrite enabled", ocppVersion: ocpp.V16, preconfigure: func(registry *SchemaRegistry) { - _ = registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`)) + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: json.RawMessage(authorizeSchema), + }) }, action: "AuthorizeRequest", - schema: json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`), + schema: json.RawMessage(authorizeSchema), opts: []RegistryOption{WithOverwrite(true)}, expectedErr: nil, }, @@ -110,7 +122,11 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { tt.preconfigure(registry) } - err := registry.RegisterSchema(ctx, tt.ocppVersion, tt.action, tt.schema) + err := registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: tt.ocppVersion}, + Action: tt.action, + Schema: tt.schema, + }) if tt.expectedErr != nil { s.ErrorContains(err, tt.expectedErr.Error()) } else { @@ -122,41 +138,77 @@ func (s *fileRegistryTestSuite) TestRegisterSchema() { func (s *fileRegistryTestSuite) TestGetSchema() { ctx := context.Background() + + const authorizeSchema = `{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}` + tests := []struct { name string preconfigure func(registry *SchemaRegistry) - ocppVersion ocpp.Version - action string + req schema_registry.GetSchemaRequest expectedFound bool }{ { - name: "Get schema for OCPP 1.6", + name: "Get base schema for OCPP 1.6", preconfigure: func(registry *SchemaRegistry) { - _ = registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`)) + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: json.RawMessage(authorizeSchema), + }) }, - ocppVersion: ocpp.V16, - action: "AuthorizeRequest", + req: schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "AuthorizeRequest"}, expectedFound: true, }, { - name: "Get schema for OCPP 2.0", + name: "Get base schema for OCPP 2.0", preconfigure: func(registry *SchemaRegistry) { - _ = registry.RegisterSchema(ctx, ocpp.V20, "AuthorizeRequest", json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`)) + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V20}, + Action: "AuthorizeRequest", + Schema: json.RawMessage(authorizeSchema), + }) }, - ocppVersion: ocpp.V20, - action: "AuthorizeRequest", + req: schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "AuthorizeRequest"}, expectedFound: true, }, { - name: "Schema not found for OCPP version", - ocppVersion: ocpp.Version("unknown_version"), - action: "BootNotificationRequest", + name: "Schema not found - unknown OCPP version", + req: schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.Version("unknown_version")}, Action: "BootNotificationRequest"}, expectedFound: false, }, { - name: "Schema not found for OCPP version", - ocppVersion: ocpp.V20, - action: "BootNotificationRequest", + name: "Schema not found - action missing", + req: schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "BootNotificationRequest"}, + expectedFound: false, + }, + { + name: "Vendor/model-specific schema returned when registered", + preconfigure: func(registry *SchemaRegistry) { + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, + Action: "AuthorizeRequest", + Schema: json.RawMessage(authorizeSchema), + }) + }, + req: schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, Action: "AuthorizeRequest"}, + expectedFound: true, + }, + { + name: "Vendor/model request falls back to base schema when no specific schema exists", + preconfigure: func(registry *SchemaRegistry) { + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: json.RawMessage(authorizeSchema), + }) + }, + req: schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, Action: "AuthorizeRequest"}, + expectedFound: true, + }, + { + name: "Vendor/model request returns not-found when neither specific nor base schema exists", + preconfigure: func(registry *SchemaRegistry) {}, + req: schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, Action: "AuthorizeRequest"}, expectedFound: false, }, } @@ -169,7 +221,7 @@ func (s *fileRegistryTestSuite) TestGetSchema() { test.preconfigure(registry) } - schema, found := registry.GetSchema(ctx, test.ocppVersion, test.action) + schema, found := registry.GetSchema(ctx, test.req) s.Equal(test.expectedFound, found) if test.expectedFound { s.NotNil(schema) @@ -229,7 +281,11 @@ func (s *fileRegistryTestSuite) TestDeleteSchema() { { name: "Delete existing schema", preconfigure: func(registry *SchemaRegistry) { - _ = registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", schema) + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: schema, + }) }, ocppVersion: ocpp.V16, action: "AuthorizeRequest", @@ -250,7 +306,11 @@ func (s *fileRegistryTestSuite) TestDeleteSchema() { { name: "Schema not found for action", preconfigure: func(registry *SchemaRegistry) { - _ = registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", schema) + _ = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: schema, + }) }, ocppVersion: ocpp.V16, action: "BootNotificationRequest", @@ -265,7 +325,10 @@ func (s *fileRegistryTestSuite) TestDeleteSchema() { tt.preconfigure(registry) } - err := registry.DeleteSchema(ctx, tt.ocppVersion, tt.action) + err := registry.DeleteSchema(ctx, schema_registry.DeleteSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: tt.ocppVersion}, + Action: tt.action, + }) if tt.expectedErr != nil { s.ErrorContains(err, tt.expectedErr.Error()) } else { @@ -280,20 +343,30 @@ func (s *fileRegistryTestSuite) TestRegisterAndDeleteLifecycle() { registry := NewFileSchemaRegistry(s.logger) schema := json.RawMessage(`{ "$schema": "http://json-schema.org/draft-04/schema#", "id": "urn:OCPP:1.6:2019:12:AuthorizeRequest", "title": "AuthorizeRequest", "type": "object", "properties": { "idTag": { "type": "string", "maxLength": 20 } }, "additionalProperties": false, "required": [ "idTag" ]}`) - err := registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", schema) + err := registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: schema, + }) s.Require().NoError(err) - _, found := registry.GetSchema(ctx, ocpp.V16, "AuthorizeRequest") + _, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "AuthorizeRequest"}) s.True(found, "schema should be retrievable after registration") - err = registry.DeleteSchema(ctx, ocpp.V16, "AuthorizeRequest") + err = registry.DeleteSchema(ctx, schema_registry.DeleteSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + }) s.Require().NoError(err) - _, found = registry.GetSchema(ctx, ocpp.V16, "AuthorizeRequest") + _, found = registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "AuthorizeRequest"}) s.False(found, "schema should not be retrievable after deletion") - // Re-registration after deletion must succeed (no duplicate conflict) - err = registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", schema) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: ocpp.V16}, + Action: "AuthorizeRequest", + Schema: schema, + }) s.NoError(err) } diff --git a/pkg/schema_registry/registries/remote_registry/cache.go b/pkg/schema_registry/registries/remote_registry/cache.go index 384de49..20d21b5 100644 --- a/pkg/schema_registry/registries/remote_registry/cache.go +++ b/pkg/schema_registry/registries/remote_registry/cache.go @@ -4,13 +4,11 @@ import ( "context" "github.com/kaptinlin/jsonschema" - - "github.com/ChargePi/chargeflow/pkg/ocpp" ) -// Cache stores compiled schemas fetched from the remote registry. +// Cache stores compiled schemas fetched from the remote registry, keyed by subject name. type Cache interface { - Get(ctx context.Context, version ocpp.Version, action string) (*jsonschema.Schema, bool) - Set(ctx context.Context, version ocpp.Version, action string, schema *jsonschema.Schema) - Delete(ctx context.Context, version ocpp.Version, action string) + Get(ctx context.Context, subject string) (*jsonschema.Schema, bool) + Set(ctx context.Context, subject string, schema *jsonschema.Schema) + Delete(ctx context.Context, subject string) } diff --git a/pkg/schema_registry/registries/remote_registry/memory_cache.go b/pkg/schema_registry/registries/remote_registry/memory_cache.go index 74f5d37..ccaebd4 100644 --- a/pkg/schema_registry/registries/remote_registry/memory_cache.go +++ b/pkg/schema_registry/registries/remote_registry/memory_cache.go @@ -6,8 +6,6 @@ import ( "time" "github.com/kaptinlin/jsonschema" - - "github.com/ChargePi/chargeflow/pkg/ocpp" ) type cachedSchema struct { @@ -15,30 +13,25 @@ type cachedSchema struct { cachedAt time.Time } -// MemoryCache is a thread-safe in-memory Cache with TTL-based expiry. +// MemoryCache is a thread-safe in-memory Cache with TTL-based expiry, keyed by subject name. type MemoryCache struct { mu sync.RWMutex - entries map[ocpp.Version]map[string]*cachedSchema + entries map[string]*cachedSchema ttl time.Duration } func NewMemoryCache(ttl time.Duration) *MemoryCache { return &MemoryCache{ - entries: make(map[ocpp.Version]map[string]*cachedSchema), + entries: make(map[string]*cachedSchema), ttl: ttl, } } -func (m *MemoryCache) Get(_ context.Context, version ocpp.Version, action string) (*jsonschema.Schema, bool) { +func (m *MemoryCache) Get(_ context.Context, subject string) (*jsonschema.Schema, bool) { m.mu.RLock() defer m.mu.RUnlock() - actions, ok := m.entries[version] - if !ok { - return nil, false - } - - entry, ok := actions[action] + entry, ok := m.entries[subject] if !ok { return nil, false } @@ -50,25 +43,19 @@ func (m *MemoryCache) Get(_ context.Context, version ocpp.Version, action string return entry.schema, true } -func (m *MemoryCache) Set(_ context.Context, version ocpp.Version, action string, schema *jsonschema.Schema) { +func (m *MemoryCache) Set(_ context.Context, subject string, schema *jsonschema.Schema) { m.mu.Lock() defer m.mu.Unlock() - if _, ok := m.entries[version]; !ok { - m.entries[version] = make(map[string]*cachedSchema) - } - - m.entries[version][action] = &cachedSchema{ + m.entries[subject] = &cachedSchema{ schema: schema, cachedAt: time.Now(), } } -func (m *MemoryCache) Delete(_ context.Context, version ocpp.Version, action string) { +func (m *MemoryCache) Delete(_ context.Context, subject string) { m.mu.Lock() defer m.mu.Unlock() - if actions, ok := m.entries[version]; ok { - delete(actions, action) - } + delete(m.entries, subject) } diff --git a/pkg/schema_registry/registries/remote_registry/remote.go b/pkg/schema_registry/registries/remote_registry/remote.go index 0cee4ad..6b63b30 100644 --- a/pkg/schema_registry/registries/remote_registry/remote.go +++ b/pkg/schema_registry/registries/remote_registry/remote.go @@ -18,9 +18,9 @@ import ( "github.com/pkg/errors" "go.uber.org/zap" - "github.com/ChargePi/chargeflow/pkg/schema_registry/registries/file_registry" - "github.com/ChargePi/chargeflow/pkg/ocpp" + "github.com/ChargePi/chargeflow/pkg/schema_registry" + "github.com/ChargePi/chargeflow/pkg/schema_registry/registries/file_registry" ) type remoteRegistryConfig struct { @@ -142,12 +142,25 @@ func NewRemoteSchemaRegistry(baseURL string, logger *zap.Logger, opts ...Options return registry, nil } -// buildSubjectName constructs a subject name from OCPP version and action. -// Format: ocpp-{version}-{action} -// Example: ocpp-1.6-BootNotificationRequest -func buildSubjectName(ocppVersion ocpp.Version, action string) string { +// buildSubjectName constructs a subject name from OCPP version, action, and optional vendor/model. +// Base format: ocpp-{version}-{action} +// With vendor/model: {vendor}-{model}-ocpp-{version}-{action} +// Omitted parts are skipped when empty. +func buildSubjectName(ocppVersion ocpp.Version, action, vendor, model string) string { versionStr := strings.ReplaceAll(ocppVersion.String(), ".", "-") - return fmt.Sprintf("ocpp-%s-%s", versionStr, action) + base := fmt.Sprintf("ocpp-%s-%s", versionStr, action) + if vendor == "" && model == "" { + return base + } + parts := make([]string, 0, 3) + if vendor != "" { + parts = append(parts, vendor) + } + if model != "" { + parts = append(parts, model) + } + parts = append(parts, base) + return strings.Join(parts, "-") } // getLatestVersion fetches the latest version number for a subject from the remote registry. @@ -231,33 +244,33 @@ func (r *SchemaRegistry) fetchSchemaFromRemote(ctx context.Context, subject stri } } -type RegisterSchemaRequest struct { +type registerSchemaPayload struct { Schema string `json:"schema"` SchemaType string `json:"schemaType"` } -func (r *SchemaRegistry) RegisterSchema(ctx context.Context, ocppVersion ocpp.Version, action string, rawSchema json.RawMessage) error { - logger := r.logger.With(zap.String("ocppVersion", ocppVersion.String()), zap.String("action", action)) +func (r *SchemaRegistry) RegisterSchema(ctx context.Context, req schema_registry.CreateSchemaRequest) error { + logger := r.logger.With(zap.String("ocppVersion", req.OcppContext.Version.String()), zap.String("action", req.Action)) logger.Debug("Registering schema to remote registry") // Validate the OCPP version - if !ocpp.IsValidProtocolVersion(ocppVersion) { - return errors.Errorf("invalid OCPP version: %s", ocppVersion) + if !ocpp.IsValidProtocolVersion(req.OcppContext.Version) { + return errors.Errorf("invalid OCPP version: %s", req.OcppContext.Version) } // Must be a valid action name ending with "Request" or "Response" - if !(strings.HasSuffix(action, "Request") || strings.HasSuffix(action, "Response")) { - return errors.Errorf("action must end with 'Request' or 'Response': %s", action) + if !(strings.HasSuffix(req.Action, "Request") || strings.HasSuffix(req.Action, "Response")) { + return errors.Errorf("action must end with 'Request' or 'Response': %s", req.Action) } - subject := buildSubjectName(ocppVersion, action) + subject := buildSubjectName(req.OcppContext.Version, req.Action, req.OcppContext.Vendor, req.OcppContext.Model) ctx, cancel := context.WithTimeout(ctx, r.config.timeout) defer cancel() // Validate and normalize the schema before sending // First, try to compile it to ensure it's valid JSON Schema - _, err := r.compiler.Compile(rawSchema) + _, err := r.compiler.Compile(req.Schema) if err != nil { return errors.Wrapf(err, "invalid JSON schema format for subject %s", subject) } @@ -265,7 +278,7 @@ func (r *SchemaRegistry) RegisterSchema(ctx context.Context, ocppVersion ocpp.Ve // Clear any formatting by unmarshaling and re-marshaling as compact JSON // This ensures the schema is normalized without any whitespace/formatting or escaping issues var schemaObj interface{} - if err := json.Unmarshal(rawSchema, &schemaObj); err != nil { + if err := json.Unmarshal(req.Schema, &schemaObj); err != nil { return errors.Wrapf(err, "failed to parse schema JSON for subject %s", subject) } @@ -276,16 +289,10 @@ func (r *SchemaRegistry) RegisterSchema(ctx context.Context, ocppVersion ocpp.Ve return errors.Wrapf(err, "failed to normalize schema JSON for subject %s", subject) } - // The schema must be sent as a JSON string - // Convert normalized bytes to string - this is raw JSON without any escaping - // json.Marshal will properly escape this string when serializing the request body - logger.Debug("Schema string prepared for registration", zap.String("subject", subject)) - // Create the request payload - schemaStr := string(normalizedBytes) - payload := RegisterSchemaRequest{ - Schema: schemaStr, + payload := registerSchemaPayload{ + Schema: string(normalizedBytes), SchemaType: "JSON", } @@ -329,25 +336,25 @@ func (r *SchemaRegistry) RegisterSchema(ctx context.Context, ocppVersion ocpp.Ve } // Invalidate cache for this schema - r.cache.Delete(ctx, ocppVersion, action) + r.cache.Delete(ctx, subject) logger.Debug("Successfully registered schema to remote registry") return nil } -func (r *SchemaRegistry) DeleteSchema(ctx context.Context, ocppVersion ocpp.Version, action string) error { - logger := r.logger.With(zap.String("ocppVersion", ocppVersion.String()), zap.String("action", action)) +func (r *SchemaRegistry) DeleteSchema(ctx context.Context, req schema_registry.DeleteSchemaRequest) error { + logger := r.logger.With(zap.String("ocppVersion", req.OcppContext.Version.String()), zap.String("action", req.Action)) logger.Debug("Deleting schema from remote registry") - if !ocpp.IsValidProtocolVersion(ocppVersion) { - return errors.Errorf("invalid OCPP version: %s", ocppVersion) + if !ocpp.IsValidProtocolVersion(req.OcppContext.Version) { + return errors.Errorf("invalid OCPP version: %s", req.OcppContext.Version) } - if !(strings.HasSuffix(action, file_registry.RequestSuffix) || strings.HasSuffix(action, file_registry.ResponseSuffix)) { - return errors.Errorf("action must end with 'Request' or 'Response': %s", action) + if !(strings.HasSuffix(req.Action, file_registry.RequestSuffix) || strings.HasSuffix(req.Action, file_registry.ResponseSuffix)) { + return errors.Errorf("action must end with 'Request' or 'Response': %s", req.Action) } - subject := buildSubjectName(ocppVersion, action) + subject := buildSubjectName(req.OcppContext.Version, req.Action, req.OcppContext.Vendor, req.OcppContext.Model) ctx, cancel := context.WithTimeout(ctx, r.config.timeout) defer cancel() @@ -383,65 +390,84 @@ func (r *SchemaRegistry) DeleteSchema(ctx context.Context, ocppVersion ocpp.Vers return errors.Errorf("unexpected status code %d when deleting schema for subject %s", resp.StatusCode, subject) } - r.cache.Delete(ctx, ocppVersion, action) + r.cache.Delete(ctx, subject) logger.Debug("Successfully deleted schema from remote registry") return nil } -func (r *SchemaRegistry) GetSchema(ctx context.Context, ocppVersion ocpp.Version, action string) (*jsonschema.Schema, bool) { - logger := r.logger.With(zap.String("ocppVersion", ocppVersion.String()), zap.String("action", action)) - logger.Debug("Getting schema") - - // Validate the OCPP version - if !ocpp.IsValidProtocolVersion(ocppVersion) { - logger.Warn("Invalid OCPP version") +// fetchAndCacheSchema fetches a schema for the given subject from the remote registry, +// compiles it, caches it, and returns it. Returns nil, false if anything fails. +func (r *SchemaRegistry) fetchAndCacheSchema(ctx context.Context, subject string) (*jsonschema.Schema, bool) { + latestVersion, err := r.getLatestVersion(ctx, subject) + if err != nil { + r.logger.Warn("Failed to get latest version", zap.String("subject", subject), zap.Error(err)) return nil, false } - // Must be a valid action name ending with "Request" or "Response" - if !(strings.HasSuffix(action, "Request") || strings.HasSuffix(action, "Response")) { - logger.Warn("Invalid action name") + rawSchema, err := r.fetchSchemaFromRemote(ctx, subject, latestVersion) + if err != nil { + r.logger.Warn("Failed to fetch schema from remote", zap.String("subject", subject), zap.Error(err)) return nil, false } - // Check cache first - if schema, ok := r.cache.Get(ctx, ocppVersion, action); ok { - logger.Debug("Returning schema from cache") - return schema, true + schema, err := r.compiler.Compile(rawSchema) + if err != nil { + r.logger.Warn("Failed to compile schema", zap.String("subject", subject), zap.Error(err)) + return nil, false } - // Cache miss or expired - fetch from remote - subject := buildSubjectName(ocppVersion, action) - ctx, cancel := context.WithTimeout(ctx, r.config.timeout) - defer cancel() + r.cache.Set(ctx, subject, schema) + return schema, true +} - // Get the latest version - latestVersion, err := r.getLatestVersion(ctx, subject) - if err != nil { - logger.Warn("Failed to get latest version", zap.Error(err)) +func (r *SchemaRegistry) GetSchema(ctx context.Context, req schema_registry.GetSchemaRequest) (*jsonschema.Schema, bool) { + logger := r.logger.With( + zap.String("ocppVersion", req.OcppContext.Version.String()), + zap.String("action", req.Action), + zap.String("vendor", req.OcppContext.Vendor), + zap.String("model", req.OcppContext.Model), + ) + logger.Debug("Getting schema") + + if !ocpp.IsValidProtocolVersion(req.OcppContext.Version) { + logger.Warn("Invalid OCPP version") return nil, false } - // Fetch the schema - rawSchema, err := r.fetchSchemaFromRemote(ctx, subject, latestVersion) - if err != nil { - logger.Warn("Failed to fetch schema from remote", zap.Error(err)) + if !(strings.HasSuffix(req.Action, "Request") || strings.HasSuffix(req.Action, "Response")) { + logger.Warn("Invalid action name") return nil, false } - // Compile the schema - schema, err := r.compiler.Compile(rawSchema) - if err != nil { - logger.Warn("Failed to compile schema", zap.Error(err)) - return nil, false + ctx, cancel := context.WithTimeout(ctx, r.config.timeout) + defer cancel() + + // When vendor or model is provided, try the specific subject first. + if req.OcppContext.Vendor != "" || req.OcppContext.Model != "" { + specificSubject := buildSubjectName(req.OcppContext.Version, req.Action, req.OcppContext.Vendor, req.OcppContext.Model) + if schema, ok := r.cache.Get(ctx, specificSubject); ok { + logger.Debug("Returning vendor/model-specific schema from cache") + return schema, true + } + if schema, ok := r.fetchAndCacheSchema(ctx, specificSubject); ok { + logger.Debug("Returning vendor/model-specific schema from remote") + return schema, true + } + logger.Debug("No vendor/model-specific schema found, falling back to base schema") } - // Update cache - r.cache.Set(ctx, ocppVersion, action, schema) + baseSubject := buildSubjectName(req.OcppContext.Version, req.Action, "", "") + if schema, ok := r.cache.Get(ctx, baseSubject); ok { + logger.Debug("Returning base schema from cache") + return schema, true + } - logger.Debug("Successfully fetched and cached schema from remote") - return schema, true + schema, ok := r.fetchAndCacheSchema(ctx, baseSubject) + if ok { + logger.Debug("Successfully fetched and cached base schema from remote") + } + return schema, ok } func (r *SchemaRegistry) Type() string { diff --git a/pkg/schema_registry/registries/remote_registry/remote_test.go b/pkg/schema_registry/registries/remote_registry/remote_test.go index 7aa7de1..8ceabcf 100644 --- a/pkg/schema_registry/registries/remote_registry/remote_test.go +++ b/pkg/schema_registry/registries/remote_registry/remote_test.go @@ -15,6 +15,7 @@ import ( "go.uber.org/zap" "github.com/ChargePi/chargeflow/pkg/ocpp" + "github.com/ChargePi/chargeflow/pkg/schema_registry" ) const bootNotificationSchema = `{ @@ -161,7 +162,11 @@ func (s *remoteRegistryIntegrationTestSuite) TestRegisterSchema() { for _, tt := range tests { s.Run(tt.name, func() { - err := registry.RegisterSchema(ctx, tt.ocppVersion, tt.action, tt.schema) + err := registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: tt.ocppVersion}, + Action: tt.action, + Schema: tt.schema, + }) if tt.expectError { s.Error(err) } else { @@ -183,20 +188,20 @@ func (s *remoteRegistryIntegrationTestSuite) TestGetSchema() { validSchema := json.RawMessage(authorizationReqSchema) // First register a schema - err = registry.RegisterSchema(ctx, ocpp.V16, "AuthorizeRequest", validSchema) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "AuthorizeRequest", Schema: validSchema}) s.Require().NoError(err) // Test getting the schema - schema, found := registry.GetSchema(ctx, ocpp.V16, "AuthorizeRequest") + schema, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "AuthorizeRequest"}) s.True(found, "Schema should be found") s.NotNil(schema, "Schema should not be nil") // Test getting non-existent schema - _, found = registry.GetSchema(ctx, ocpp.V16, "NonExistentRequest") + _, found = registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "NonExistentRequest"}) s.False(found, "Non-existent schema should not be found") // Test getting schema for non-existent OCPP version - _, found = registry.GetSchema(ctx, ocpp.V20, "AuthorizeRequest") + _, found = registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "AuthorizeRequest"}) s.False(found, "Schema for different OCPP version should not be found") } @@ -216,16 +221,16 @@ func (s *remoteRegistryIntegrationTestSuite) TestGetSchema_Caching() { validSchema := json.RawMessage(statusNotificationSchema) // Register the schema - err = registry.RegisterSchema(ctx, ocpp.V16, "StatusNotificationRequest", validSchema) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "StatusNotificationRequest", Schema: validSchema}) s.Require().NoError(err) // First fetch - should fetch from remote - schema1, found := registry.GetSchema(ctx, ocpp.V16, "StatusNotificationRequest") + schema1, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "StatusNotificationRequest"}) s.True(found) s.NotNil(schema1) // Second fetch immediately - should use cache - schema2, found := registry.GetSchema(ctx, ocpp.V16, "StatusNotificationRequest") + schema2, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "StatusNotificationRequest"}) s.True(found) s.NotNil(schema2) s.Equal(schema1, schema2, "Should return the same schema instance from cache") @@ -234,7 +239,7 @@ func (s *remoteRegistryIntegrationTestSuite) TestGetSchema_Caching() { time.Sleep(cacheRefresh + 500*time.Millisecond) // Third fetch after cache expiry - should fetch from remote again - schema3, found := registry.GetSchema(ctx, ocpp.V16, "StatusNotificationRequest") + schema3, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "StatusNotificationRequest"}) s.True(found) s.NotNil(schema3) // Note: schema3 will be a new instance, but should validate the same data @@ -273,15 +278,15 @@ func (s *remoteRegistryIntegrationTestSuite) TestGetSchema_MultipleVersions() { }`) // Register first version - err = registry.RegisterSchema(ctx, ocpp.V16, "HeartbeatRequest", schemaV1) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "HeartbeatRequest", Schema: schemaV1}) s.Require().NoError(err) // Register second version (should create a new version in the registry) - err = registry.RegisterSchema(ctx, ocpp.V16, "HeartbeatRequest", schemaV2) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "HeartbeatRequest", Schema: schemaV2}) s.Require().NoError(err) // GetSchema should return the latest version - schema, found := registry.GetSchema(ctx, ocpp.V16, "HeartbeatRequest") + schema, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "HeartbeatRequest"}) s.True(found) s.NotNil(schema) } @@ -314,7 +319,7 @@ func (s *remoteRegistryIntegrationTestSuite) TestGetSchema_InvalidInputs() { for _, tt := range tests { s.Run(tt.name, func() { - schema, found := registry.GetSchema(ctx, tt.ocppVersion, tt.action) + schema, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: tt.ocppVersion}, Action: tt.action}) s.False(found, "Should not find schema for invalid input") s.Nil(schema, "Schema should be nil for invalid input") }) @@ -359,18 +364,18 @@ func (s *remoteRegistryIntegrationTestSuite) TestGetSchema_DifferentOCPPVersions }`) // Register schemas for different OCPP versions - err = registry.RegisterSchema(ctx, ocpp.V16, "StartTransactionRequest", schema16) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "StartTransactionRequest", Schema: schema16}) s.Require().NoError(err) - err = registry.RegisterSchema(ctx, ocpp.V20, "StartTransactionRequest", schema20) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "StartTransactionRequest", Schema: schema20}) s.Require().NoError(err) // Verify both schemas can be retrieved independently - schema1, found1 := registry.GetSchema(ctx, ocpp.V16, "StartTransactionRequest") + schema1, found1 := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "StartTransactionRequest"}) s.True(found1, "OCPP 1.6 schema should be found") s.NotNil(schema1) - schema2, found2 := registry.GetSchema(ctx, ocpp.V20, "StartTransactionRequest") + schema2, found2 := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "StartTransactionRequest"}) s.True(found2, "OCPP 2.0 schema should be found") s.NotNil(schema2) @@ -410,11 +415,11 @@ func (s *remoteRegistryIntegrationTestSuite) TestGetSchema_ResponseSuffix() { }`) // Register response schema - err = registry.RegisterSchema(ctx, ocpp.V16, "BootNotificationResponse", responseSchema) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationResponse", Schema: responseSchema}) s.Require().NoError(err) // Retrieve response schema - schema, found := registry.GetSchema(ctx, ocpp.V16, "BootNotificationResponse") + schema, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationResponse"}) s.True(found, "Response schema should be found") s.NotNil(schema, "Response schema should not be nil") } @@ -436,7 +441,7 @@ func (s *remoteRegistryIntegrationTestSuite) TestDeleteSchema() { { name: "Delete existing schema", setup: func() { - err := registry.RegisterSchema(ctx, ocpp.V16, "MeterValuesRequest", validSchema) + err := registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "MeterValuesRequest", Schema: validSchema}) s.Require().NoError(err) }, ocppVersion: ocpp.V16, @@ -469,7 +474,10 @@ func (s *remoteRegistryIntegrationTestSuite) TestDeleteSchema() { tt.setup() } - err := registry.DeleteSchema(ctx, tt.ocppVersion, tt.action) + err := registry.DeleteSchema(ctx, schema_registry.DeleteSchemaRequest{ + OcppContext: ocpp.OcppContext{Version: tt.ocppVersion}, + Action: tt.action, + }) if tt.expectError { s.Error(err) } else { @@ -491,17 +499,17 @@ func (s *remoteRegistryIntegrationTestSuite) TestDeleteSchema_CacheInvalidation( validSchema := json.RawMessage(authorizationReqSchema) - err = registry.RegisterSchema(ctx, ocpp.V16, "DataTransferRequest", validSchema) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "DataTransferRequest", Schema: validSchema}) s.Require().NoError(err) - schema, found := registry.GetSchema(ctx, ocpp.V16, "DataTransferRequest") + schema, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "DataTransferRequest"}) s.True(found, "schema should be found after registration") s.NotNil(schema) - err = registry.DeleteSchema(ctx, ocpp.V16, "DataTransferRequest") + err = registry.DeleteSchema(ctx, schema_registry.DeleteSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "DataTransferRequest"}) s.Require().NoError(err) - _, found = registry.GetSchema(ctx, ocpp.V16, "DataTransferRequest") + _, found = registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "DataTransferRequest"}) s.False(found, "schema should not be found after deletion — cache must be invalidated") } @@ -512,21 +520,21 @@ func (s *remoteRegistryIntegrationTestSuite) TestRegisterAndDeleteLifecycle() { validSchema := json.RawMessage(bootNotificationSchema) - err = registry.RegisterSchema(ctx, ocpp.V16, "ClearCacheRequest", validSchema) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "ClearCacheRequest", Schema: validSchema}) s.Require().NoError(err) - schema, found := registry.GetSchema(ctx, ocpp.V16, "ClearCacheRequest") + schema, found := registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "ClearCacheRequest"}) s.True(found, "schema should be retrievable after registration") s.NotNil(schema) - err = registry.DeleteSchema(ctx, ocpp.V16, "ClearCacheRequest") + err = registry.DeleteSchema(ctx, schema_registry.DeleteSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "ClearCacheRequest"}) s.Require().NoError(err) - _, found = registry.GetSchema(ctx, ocpp.V16, "ClearCacheRequest") + _, found = registry.GetSchema(ctx, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "ClearCacheRequest"}) s.False(found, "schema should not be retrievable after deletion") // Re-registration after deletion must succeed - err = registry.RegisterSchema(ctx, ocpp.V16, "ClearCacheRequest", validSchema) + err = registry.RegisterSchema(ctx, schema_registry.CreateSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "ClearCacheRequest", Schema: validSchema}) s.NoError(err) } diff --git a/pkg/schema_registry/registry.go b/pkg/schema_registry/registry.go index 32c4249..f7bc4d9 100644 --- a/pkg/schema_registry/registry.go +++ b/pkg/schema_registry/registry.go @@ -9,9 +9,29 @@ import ( "github.com/ChargePi/chargeflow/pkg/ocpp" ) +type CreateSchemaRequest struct { + OcppContext ocpp.OcppContext + Action string + Schema json.RawMessage +} + +type DeleteSchemaRequest struct { + OcppContext ocpp.OcppContext + Action string +} + +type GetSchemaRequest struct { + OcppContext ocpp.OcppContext + Action string +} + type SchemaRegistry interface { - RegisterSchema(ctx context.Context, ocppVersion ocpp.Version, action string, rawSchema json.RawMessage) error - DeleteSchema(ctx context.Context, ocppVersion ocpp.Version, action string) error - GetSchema(ctx context.Context, ocppVersion ocpp.Version, action string) (*jsonschema.Schema, bool) + RegisterSchema(ctx context.Context, req CreateSchemaRequest) error + DeleteSchema(ctx context.Context, req DeleteSchemaRequest) error + // GetSchema retrieves a compiled schema for the given OCPP version and action. + // When Vendor and/or Model are non-empty the registry first attempts to return a + // vendor/model-specific schema and falls back to the base OCPP spec schema when + // no specific schema is found. + GetSchema(ctx context.Context, req GetSchemaRequest) (*jsonschema.Schema, bool) Type() string } diff --git a/pkg/validator/validator.go b/pkg/validator/validator.go index 43ffb57..1c629cb 100644 --- a/pkg/validator/validator.go +++ b/pkg/validator/validator.go @@ -26,9 +26,13 @@ func NewValidator(logger *zap.Logger, registry schema_registry.SchemaRegistry) * } // ValidateMessage validates the message. It checks if the message has an action, a payload, and a unique ID. -// It also validates the payload against the schema for the given action and OCPP version. -func (v *Validator) ValidateMessage(ocppVersion ocpp.Version, message ocpp.Message) (*ValidationResult, error) { - v.logger.Info("Validating message", zap.String("action", message.GetAction())) +// It also validates the payload against the schema for the given OCPP version. +// When OcppContext.Vendor and/or OcppContext.Model are non-empty the registry attempts a +// vendor/model-specific schema first, falling back to the base OCPP spec schema. +func (v *Validator) ValidateMessage(octx ocpp.OcppContext, message ocpp.Message) (*ValidationResult, error) { + logger := v.logger.With(zap.String("vendor", octx.Vendor), zap.String("model", octx.Model), zap.String("action", message.GetAction())) + logger.Info("Validating message") + result := NewValidationResult() // Check if a message has a unique ID @@ -41,81 +45,66 @@ func (v *Validator) ValidateMessage(ocppVersion ocpp.Version, message ocpp.Messa switch message.GetMessageTypeId() { case ocpp.CALL: - // Check if a message has an action action := message.GetAction() if action == "" { result.AddError(actionEmptyErr) break } - // For CALL messages, the action must end with "Request" - action = action + "Request" - - err := v.validatePayload(ocppVersion, payload, action, result) + err := v.validatePayload(octx, payload, action+"Request", result) if err != nil { return result, errors.Wrap(err, "unable to validate message payload") } case ocpp.SEND: - if ocppVersion != ocpp.V21 { + if octx.Version != ocpp.V21 { result.AddError("SEND messages are only supported in OCPP 2.1") return result, nil } - // Check if a message has an action action := message.GetAction() if action == "" { result.AddError(actionEmptyErr) break } - // For SEND messages, the action must end with "Request" - action = action + "Request" - - err := v.validatePayload(ocppVersion, payload, action, result) + err := v.validatePayload(octx, payload, action+"Request", result) if err != nil { return result, errors.Wrap(err, "unable to validate message payload") } + case ocpp.CALL_RESULT: - // Check if a message has an action action := message.GetAction() if action == "" { result.AddError(actionEmptyErr) } - // For CALL_RESULT messages, the action must end with "Response" - action = action + "Response" - - err := v.validatePayload(ocppVersion, payload, action, result) + err := v.validatePayload(octx, payload, action+"Response", result) if err != nil { return result, errors.Wrap(err, "unable to validate message payload") } + case ocpp.CALL_ERROR: - // errors are not validated against schemas, so we skip validation for CALL_ERROR messages - // We will however validate the contents of the error message callError, ok := message.(*ocpp.CallError) if !ok { return result, ErrCannotCastToCallError } - // Validate the error code if !ocpp.IsErrorCodeValid(callError.ErrorCode) { result.AddError(fmt.Sprintf("invalid error code: %s", callError.ErrorCode)) } + case ocpp.CALL_RESULT_ERROR: - if ocppVersion != ocpp.V21 { + if octx.Version != ocpp.V21 { result.AddError("CALL_RESULT_ERROR messages are only supported in OCPP 2.1") return result, nil } - // errors are not validated against schemas, so we skip validation for CALL_ERROR messages - // We will however validate the contents of the error message callError, ok := message.(*ocpp.CallResultError) if !ok { return result, ErrCannotCastToCallError } - // Validate the error code if !ocpp.IsErrorCodeValid(callError.ErrorCode) { result.AddError(fmt.Sprintf("invalid error code: %s", callError.ErrorCode)) } @@ -124,31 +113,27 @@ func (v *Validator) ValidateMessage(ocppVersion ocpp.Version, message ocpp.Messa return result, nil } -func (v *Validator) validatePayload(ocppVersion ocpp.Version, payload interface{}, action string, validationResults *ValidationResult) error { - // Check if a message has a payload +func (v *Validator) validatePayload( + octx ocpp.OcppContext, + payload interface{}, + action string, + validationResults *ValidationResult, +) error { if payload == nil { validationResults.AddError(payloadEmptyErr) return nil } - switch v.registry.Type() { - case "remote": - // For remote, we can validate the payload against the schema directly - case "file": - - } - - // Get the schema for the action and OCPP version - schema, found := v.registry.GetSchema(context.Background(), ocppVersion, action) + schema, found := v.registry.GetSchema(context.Background(), schema_registry.GetSchemaRequest{ + OcppContext: octx, + Action: action, + }) if !found { - return errors.Errorf("no schema found for action %s in OCPP version %s", action, ocppVersion) + return errors.Errorf("no schema found for action %s in OCPP version %s", action, octx.Version) } - // Validate the payload against the schema evaluationResult := schema.Validate(payload) - if !evaluationResult.IsValid() { - // Append each validation error to the validation results for _, evaluationError := range evaluationResult.Errors { validationResults.AddError(evaluationError.Error()) } diff --git a/pkg/validator/validator_test.go b/pkg/validator/validator_test.go index 459fb6d..540a66f 100644 --- a/pkg/validator/validator_test.go +++ b/pkg/validator/validator_test.go @@ -12,6 +12,7 @@ import ( mock_schema_registry "github.com/ChargePi/chargeflow/gen/mocks/pkg/schema_registry" "github.com/ChargePi/chargeflow/pkg/ocpp" + "github.com/ChargePi/chargeflow/pkg/schema_registry" ) var schema = []byte(`{ @@ -111,7 +112,7 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { tests := []struct { name string setupRegistry func(*mock_schema_registry.MockSchemaRegistry) - ocppVersion ocpp.Version + ocppCtx ocpp.OcppContext message ocpp.Message expected *ValidationResult }{ @@ -120,10 +121,9 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { schemaFromCompiler, err := s.compiler.Compile(schema) s.Require().NoError(err) - registry.EXPECT().GetSchema(mock.Anything, ocpp.V20, "BootNotificationRequest").Return(schemaFromCompiler, true) - registry.EXPECT().Type().Return("file") + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "BootNotificationRequest"}).Return(schemaFromCompiler, true) }, - ocppVersion: ocpp.V20, + ocppCtx: ocpp.OcppContext{Version: ocpp.V20}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL, UniqueId: uuid.NewString(), @@ -137,10 +137,9 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { schemaFromCompiler, err := s.compiler.Compile(schema) s.Require().NoError(err) - registry.EXPECT().GetSchema(mock.Anything, ocpp.V16, "BootNotificationRequest").Return(schemaFromCompiler, true) - registry.EXPECT().Type().Return("file") + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationRequest"}).Return(schemaFromCompiler, true) }, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL, UniqueId: uuid.NewString(), @@ -154,10 +153,9 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { schemaFromCompiler, err := s.compiler.Compile(responseSchema) s.Require().NoError(err) - registry.EXPECT().Type().Return("file") - registry.EXPECT().GetSchema(mock.Anything, ocpp.V20, "BootNotificationResponse").Return(schemaFromCompiler, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V20}, Action: "BootNotificationResponse"}).Return(schemaFromCompiler, true) }, - ocppVersion: ocpp.V20, + ocppCtx: ocpp.OcppContext{Version: ocpp.V20}, message: &ocpp.CallResult{ MessageTypeId: ocpp.CALL_RESULT, UniqueId: uuid.NewString(), @@ -169,7 +167,7 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { { name: "Valid OCPP 1.6 error", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) {}, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.CallError{ MessageTypeId: ocpp.CALL_ERROR, UniqueId: uuid.NewString(), @@ -182,7 +180,7 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { { name: "Valid OCPP 2.1 SEND request", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) {}, - ocppVersion: ocpp.V21, + ocppCtx: ocpp.OcppContext{Version: ocpp.V21}, message: &ocpp.CallError{ MessageTypeId: ocpp.CALL_ERROR, UniqueId: uuid.NewString(), @@ -192,6 +190,54 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { }, expected: NewValidationResult(), }, + { + name: "Valid request with vendor and model", + setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { + schemaFromCompiler, err := s.compiler.Compile(schema) + s.Require().NoError(err) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, Action: "BootNotificationRequest"}).Return(schemaFromCompiler, true) + }, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, + message: &ocpp.Call{ + MessageTypeId: ocpp.CALL, + UniqueId: uuid.NewString(), + Action: "BootNotification", + Payload: []byte("{\"chargePointVendor\":\"Vendor\",\"chargePointModel\":\"Model\"}"), + }, + expected: NewValidationResult(), + }, + { + name: "Valid request with vendor only", + setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { + schemaFromCompiler, err := s.compiler.Compile(schema) + s.Require().NoError(err) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme"}, Action: "BootNotificationRequest"}).Return(schemaFromCompiler, true) + }, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme"}, + message: &ocpp.Call{ + MessageTypeId: ocpp.CALL, + UniqueId: uuid.NewString(), + Action: "BootNotification", + Payload: []byte("{\"chargePointVendor\":\"Vendor\",\"chargePointModel\":\"Model\"}"), + }, + expected: NewValidationResult(), + }, + { + name: "Valid request with model only", + setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { + schemaFromCompiler, err := s.compiler.Compile(schema) + s.Require().NoError(err) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Model: "X1"}, Action: "BootNotificationRequest"}).Return(schemaFromCompiler, true) + }, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16, Model: "X1"}, + message: &ocpp.Call{ + MessageTypeId: ocpp.CALL, + UniqueId: uuid.NewString(), + Action: "BootNotification", + Payload: []byte("{\"chargePointVendor\":\"Vendor\",\"chargePointModel\":\"Model\"}"), + }, + expected: NewValidationResult(), + }, } for _, tt := range tests { @@ -203,7 +249,7 @@ func (s *validatorTestSuite) TestValidateMessage_HappyPath() { validator := NewValidator(s.logger, registry) - result, err := validator.ValidateMessage(tt.ocppVersion, tt.message) + result, err := validator.ValidateMessage(tt.ocppCtx, tt.message) s.NoError(err) s.Emptyf(result.Errors(), "expected no validation errors but got %v", result.Errors()) }) @@ -214,7 +260,7 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { tests := []struct { name string setupRegistry func(*mock_schema_registry.MockSchemaRegistry) - ocppVersion ocpp.Version + ocppCtx ocpp.OcppContext message ocpp.Message expected *ValidationResult expectedErr error @@ -224,10 +270,9 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { schemaFromCompiler, err := s.compiler.Compile(schema) s.Require().NoError(err) - registry.EXPECT().GetSchema(mock.Anything, ocpp.V16, "BootNotificationRequest").Return(schemaFromCompiler, true) - registry.EXPECT().Type().Return("file") + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationRequest"}).Return(schemaFromCompiler, true) }, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL, UniqueId: "", @@ -243,7 +288,7 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { { name: "Invalid request - missing payload", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) {}, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL, UniqueId: uuid.NewString(), @@ -259,7 +304,7 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { { name: "Invalid response - missing payload", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) {}, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.CallResult{ MessageTypeId: ocpp.CALL_RESULT, UniqueId: uuid.NewString(), @@ -272,8 +317,8 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { expectedErr: nil, }, { - name: "Invalid error - invalid error code", - ocppVersion: ocpp.V16, + name: "Invalid error - invalid error code", + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.CallError{ MessageTypeId: ocpp.CALL_ERROR, UniqueId: uuid.NewString(), @@ -287,8 +332,8 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { expectedErr: nil, }, { - name: "Invalid error - cannot cast to CallError", - ocppVersion: ocpp.V16, + name: "Invalid error - cannot cast to CallError", + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL_ERROR, UniqueId: uuid.NewString(), @@ -301,7 +346,7 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { { name: "Multiple errors in message - invalid payload, missing uniqueId, malformed payload", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) {}, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL, UniqueId: "", @@ -317,10 +362,9 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { { name: "No registered schema for request", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { - registry.EXPECT().GetSchema(mock.Anything, ocpp.V16, "BootNotificationRequest").Return(nil, false) - registry.EXPECT().Type().Return("file") + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationRequest"}).Return(nil, false) }, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL, UniqueId: uuid.NewString(), @@ -338,10 +382,9 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { schemaFromCompiler, err := s.compiler.Compile(schema) s.Require().NoError(err) - registry.EXPECT().Type().Return("file") - registry.EXPECT().GetSchema(mock.Anything, ocpp.V16, "BootNotificationRequest").Return(schemaFromCompiler, true) + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16}, Action: "BootNotificationRequest"}).Return(schemaFromCompiler, true) }, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Call{ MessageTypeId: ocpp.CALL, UniqueId: uuid.NewString(), @@ -358,7 +401,7 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { name: "Unsupported message (SEND) for ocpp version", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { }, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.Send{ MessageTypeId: ocpp.SEND, UniqueId: uuid.NewString(), @@ -375,7 +418,7 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { name: "Unsupported message (CALL_RESULT_ERROR) for ocpp version", setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { }, - ocppVersion: ocpp.V16, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16}, message: &ocpp.CallResultError{ MessageTypeId: ocpp.CALL_RESULT_ERROR, UniqueId: uuid.NewString(), @@ -388,6 +431,24 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { }, expectedErr: nil, }, + { + name: "No registered schema for vendor and model specific request", + setupRegistry: func(registry *mock_schema_registry.MockSchemaRegistry) { + registry.EXPECT().GetSchema(mock.Anything, schema_registry.GetSchemaRequest{OcppContext: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, Action: "BootNotificationRequest"}).Return(nil, false) + }, + ocppCtx: ocpp.OcppContext{Version: ocpp.V16, Vendor: "Acme", Model: "X1"}, + message: &ocpp.Call{ + MessageTypeId: ocpp.CALL, + UniqueId: uuid.NewString(), + Action: "BootNotification", + Payload: "{\"chargePointVendor\":\"Vendor\",\"chargePointModel\":\"Model\"}", + }, + expected: &ValidationResult{ + isValid: false, + errors: []string{}, + }, + expectedErr: errors.New("no schema found for action BootNotificationRequest in OCPP version 1.6"), + }, } for _, test := range tests { @@ -399,7 +460,7 @@ func (s *validatorTestSuite) TestValidateMessage_UnhappyPath() { validator := NewValidator(s.logger, registry) - result, err := validator.ValidateMessage(test.ocppVersion, test.message) + result, err := validator.ValidateMessage(test.ocppCtx, test.message) if test.expectedErr != nil { s.ErrorContains(err, test.expectedErr.Error()) } else {