diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json b/.github/references/ubuntu_22_04_clang_arm_manifest.json index 53f33f8070..779105a480 100644 --- a/.github/references/ubuntu_22_04_clang_arm_manifest.json +++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json @@ -6570,7 +6570,7 @@ "Namespace index": { "name": "Namespace index", "description": "The index of the namespace.", - "validator": "INTEGER_VALIDATOR", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", "required": "true", "sensitive": "false", "expressionLanguageScope": "NONE", @@ -6746,7 +6746,7 @@ "Namespace index": { "name": "Namespace index", "description": "The index of the namespace.", - "validator": "INTEGER_VALIDATOR", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", "required": "true", "sensitive": "false", "expressionLanguageScope": "NONE", @@ -6779,6 +6779,10 @@ { "value": "String", "displayName": "String" + }, + { + "value": "Guid", + "displayName": "Guid" } ] }, @@ -6903,17 +6907,17 @@ }, "Parent node ID": { "name": "Parent node ID", - "description": "Specifies the ID of the root node to traverse", + "description": "Specifies the ID of the root node to traverse. Only required when a new node is created.", "validator": "VALID", - "required": "true", + "required": "false", "sensitive": "false", "expressionLanguageScope": "NONE" }, "Parent node ID type": { "name": "Parent node ID type", - "description": "Specifies the type of the provided node ID", + "description": "Specifies the type of the provided node ID. Only required when a new node is created.", "validator": "VALID", - "required": "true", + "required": "false", "sensitive": "false", "expressionLanguageScope": "NONE", "allowableValues": [ @@ -6928,14 +6932,18 @@ { "value": "String", "displayName": "String" + }, + { + "value": "Guid", + "displayName": "Guid" } ] }, "Parent node namespace index": { "name": "Parent node namespace index", - "description": "The index of the namespace of the parent node.", - "validator": "INTEGER_VALIDATOR", - "required": "true", + "description": "The index of the namespace of the parent node. Only used when a new node is created.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", "sensitive": "false", "expressionLanguageScope": "NONE", "defaultValue": "0" @@ -6966,7 +6974,7 @@ }, "Target node ID type": { "name": "Target node ID type", - "description": "ID type of target node. Allowed values are: Int, String.", + "description": "ID type of target node. Allowed values are: Int, String, Guid.", "validator": "VALID", "required": "true", "sensitive": "false", diff --git a/PROCESSORS.md b/PROCESSORS.md index 87d3c7aff5..924a08446f 100644 --- a/PROCESSORS.md +++ b/PROCESSORS.md @@ -945,21 +945,21 @@ Fetches OPC-UA node In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. -| Name | Default Value | Allowable Values | Description | -|---------------------------------|---------------|--------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| **OPC server endpoint** | | | Specifies the address, port and relative path of an OPC endpoint | -| Application URI | | | Application URI of the client in the format 'urn:unconfigured:application'. Mandatory, if using Secure Channel and must match the URI included in the certificate's Subject Alternative Names. | -| Username | | | Username to log in with. | -| Password | | | Password to log in with.
**Sensitive Property: true** | -| Certificate path | | | Path to the DER-encoded cert file | -| Key path | | | Path to the DER-encoded key file | -| Trusted server certificate path | | | Comma separated list of paths to the DER-encoded trusted server certificates | -| Path reference types | | | Specify the reference types between nodes in the path if Path Node ID type is used. If not provided, all reference types are assumed to be Organizes. The format is 'referenceType1/referenceType2/.../referenceTypeN' and the supported reference types are Organizes, HasComponent, HasProperty, and HasSubtype. | -| **Node ID type** | | Path
Int
String | Specifies the type of the provided node ID | -| **Node ID** | | | Specifies the ID of the root node to traverse. In case of a Path Node ID Type, the path should be provided in the format of 'path/to/node'. | -| **Namespace index** | 0 | | The index of the namespace. | -| **Max depth** | 0 | | Specifiec the max depth of browsing. 0 means unlimited. | -| **Lazy mode** | Off | On
New Value
Off | Only creates flowfiles from nodes with new timestamp from the server. If set to 'New Value', it will only create flowfiles if the value of the node data has changed since the last fetch, the timestamp is ignored. | +| Name | Default Value | Allowable Values | Description | +|---------------------------------|---------------|----------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| **OPC server endpoint** | | | Specifies the address, port and relative path of an OPC endpoint | +| Application URI | | | Application URI of the client in the format 'urn:unconfigured:application'. Mandatory, if using Secure Channel and must match the URI included in the certificate's Subject Alternative Names. | +| Username | | | Username to log in with. | +| Password | | | Password to log in with.
**Sensitive Property: true** | +| Certificate path | | | Path to the DER-encoded cert file | +| Key path | | | Path to the DER-encoded key file | +| Trusted server certificate path | | | Comma separated list of paths to the DER-encoded trusted server certificates | +| Path reference types | | | Specify the reference types between nodes in the path if Path Node ID type is used. If not provided, all reference types are assumed to be Organizes. The format is 'referenceType1/referenceType2/.../referenceTypeN' and the supported reference types are Organizes, HasComponent, HasProperty, and HasSubtype. | +| **Node ID type** | | Path
Int
String
Guid | Specifies the type of the provided node ID | +| **Node ID** | | | Specifies the ID of the root node to traverse. In case of a Path Node ID Type, the path should be provided in the format of 'path/to/node'. | +| **Namespace index** | 0 | | The index of the namespace. | +| **Max depth** | 0 | | Specifiec the max depth of browsing. 0 means unlimited. | +| **Lazy mode** | Off | On
New Value
Off | Only creates flowfiles from nodes with new timestamp from the server. If set to 'New Value', it will only create flowfiles if the value of the node data has changed since the last fetch, the timestamp is ignored. | ### Relationships @@ -973,6 +973,7 @@ In the list below, the names of required properties appear in bold. Any other pr | Attribute | Relationship | Description | |-----------------|--------------|-----------------------------------------------------------------------------------------| | NodeID | success | ID of the node. | +| Namespace index | success | Namespace index of the node. | | NodeID type | success | Type of the node ID. | | Browsename | success | The browse name of the node. | | Full path | success | The full path of the node. | @@ -2514,11 +2515,11 @@ In the list below, the names of required properties appear in bold. Any other pr | Key path | | | Path to the DER-encoded key file | | Trusted server certificate path | | | Comma separated list of paths to the DER-encoded trusted server certificates | | Path reference types | | | Specify the reference types between nodes in the path if Path Node ID type is used. If not provided, all reference types are assumed to be Organizes. The format is 'referenceType1/referenceType2/.../referenceTypeN' and the supported reference types are Organizes, HasComponent, HasProperty, and HasSubtype. | -| **Parent node ID type** | | Path
Int
String | Specifies the type of the provided node ID | -| **Parent node ID** | | | Specifies the ID of the root node to traverse | -| **Parent node namespace index** | 0 | | The index of the namespace of the parent node. | +| Parent node ID type | | Path
Int
String
Guid | Specifies the type of the provided node ID. Only required when a new node is created. | +| Parent node ID | | | Specifies the ID of the root node to traverse. Only required when a new node is created. | +| Parent node namespace index | 0 | | The index of the namespace of the parent node. Only used when a new node is created. | | **Value type** | | Int64
UInt64
Int32
UInt32
Boolean
Float
Double
String | Set the OPC value type of the created nodes | -| **Target node ID type** | | Int
String | ID type of target node. Allowed values are: Int, String.
**Supports Expression Language: true** | +| **Target node ID type** | | | ID type of target node. Allowed values are: Int, String, Guid.
**Supports Expression Language: true** | | **Target node ID** | | | ID of target node.
**Supports Expression Language: true** | | Target node browse name | | | Browse name of target node. Only used when new node is created.
**Supports Expression Language: true** | | **Target node namespace index** | 0 | | The index of the namespace of the target node.
**Supports Expression Language: true** | diff --git a/extensions/opc/include/BaseOPCProcessor.h b/extensions/opc/include/BaseOPCProcessor.h index 9ec4afe00c..00fd4dd565 100644 --- a/extensions/opc/include/BaseOPCProcessor.h +++ b/extensions/opc/include/BaseOPCProcessor.h @@ -80,10 +80,12 @@ class BaseOPCProcessor : public core::ProcessorImpl { protected: virtual bool reconnect(); void readPathReferenceTypes(core::ProcessContext& context, const std::string& node_id); - void parseIdType(core::ProcessContext& context, const core::PropertyReference& prop); + void parseNode(core::ProcessContext& context); std::string node_id_; - int32_t namespace_idx_ = 0; + UA_UInt16 namespace_idx_ = 0; + bool path_node_id_resolved_ = false; + opc::NodeId node_; opc::OPCNodeIDType id_type_{}; opc::ClientPtr connection_; diff --git a/extensions/opc/include/FetchOPCHistory.h b/extensions/opc/include/FetchOPCHistory.h index c15b636936..9e00b68f1b 100644 --- a/extensions/opc/include/FetchOPCHistory.h +++ b/extensions/opc/include/FetchOPCHistory.h @@ -80,7 +80,7 @@ class FetchOPCHistory final : public BaseOPCProcessor { EXTENSIONAPI static constexpr auto NameSpaceIndex = core::PropertyDefinitionBuilder<>::createProperty("Namespace index") .withDescription("The index of the namespace.") - .withValidator(core::StandardPropertyValidators::INTEGER_VALIDATOR) + .withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR) .withDefaultValue("0") .isRequired(true) .build(); @@ -151,8 +151,6 @@ class FetchOPCHistory final : public BaseOPCProcessor { std::optional end_timestamp_; uint64_t batch_size_ = 0; std::shared_ptr record_set_writer_; - opc::NodeId node_; - bool path_node_id_resolved_ = false; }; } // namespace org::apache::nifi::minifi::processors diff --git a/extensions/opc/include/FetchOPCProcessor.h b/extensions/opc/include/FetchOPCProcessor.h index 5ae0d0ba96..8c11ae0ab7 100644 --- a/extensions/opc/include/FetchOPCProcessor.h +++ b/extensions/opc/include/FetchOPCProcessor.h @@ -73,10 +73,10 @@ class FetchOPCProcessor final : public BaseOPCProcessor { EXTENSIONAPI static constexpr const char* Description = "Fetches OPC-UA node"; - EXTENSIONAPI static constexpr auto NodeIDType = core::PropertyDefinitionBuilder<3>::createProperty("Node ID type") + EXTENSIONAPI static constexpr auto NodeIDType = core::PropertyDefinitionBuilder()>::createProperty("Node ID type") .withDescription("Specifies the type of the provided node ID") .isRequired(true) - .withAllowedValues({"Path", "Int", "String"}) + .withAllowedValues(magic_enum::enum_names()) .build(); EXTENSIONAPI static constexpr auto NodeID = core::PropertyDefinitionBuilder<>::createProperty("Node ID") .withDescription("Specifies the ID of the root node to traverse. In case of a Path Node ID Type, the path should be provided in the format of 'path/to/node'.") @@ -84,7 +84,7 @@ class FetchOPCProcessor final : public BaseOPCProcessor { .build(); EXTENSIONAPI static constexpr auto NameSpaceIndex = core::PropertyDefinitionBuilder<>::createProperty("Namespace index") .withDescription("The index of the namespace.") - .withValidator(core::StandardPropertyValidators::INTEGER_VALIDATOR) + .withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR) .withDefaultValue("0") .isRequired(true) .build(); @@ -115,6 +115,7 @@ class FetchOPCProcessor final : public BaseOPCProcessor { EXTENSIONAPI static constexpr auto Relationships = std::array{Success, Failure}; EXTENSIONAPI static constexpr auto NodeIDAttr = core::OutputAttributeDefinition<>{"NodeID", { Success }, "ID of the node."}; + EXTENSIONAPI static constexpr auto NamespaceIndexAttr = core::OutputAttributeDefinition<>{"Namespace index", {Success}, "Namespace index of the node."}; EXTENSIONAPI static constexpr auto NodeIDTypeAttr = core::OutputAttributeDefinition<>{"NodeID type", { Success }, "Type of the node ID."}; EXTENSIONAPI static constexpr auto BrowsenameAttr = core::OutputAttributeDefinition<>{"Browsename", { Success }, "The browse name of the node."}; EXTENSIONAPI static constexpr auto FullPathAttr = core::OutputAttributeDefinition<>{"Full path", { Success }, "The full path of the node."}; @@ -123,8 +124,8 @@ class FetchOPCProcessor final : public BaseOPCProcessor { EXTENSIONAPI static constexpr auto TypenameAttr = core::OutputAttributeDefinition<>{"Typename", { Success }, "The type name of the node data."}; EXTENSIONAPI static constexpr auto DatasizeAttr = core::OutputAttributeDefinition<>{"Datasize", { Success }, "The size of the node data."}; - EXTENSIONAPI static constexpr auto OutputAttributes = std::array {NodeIDAttr, NodeIDTypeAttr, BrowsenameAttr, FullPathAttr, SourcetimestampAttr, - TypenameAttr, DatasizeAttr}; + EXTENSIONAPI static constexpr auto OutputAttributes = std::array {NodeIDAttr, + NamespaceIndexAttr, NodeIDTypeAttr, BrowsenameAttr, FullPathAttr, SourcetimestampAttr, TypenameAttr, DatasizeAttr}; EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false; EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false; @@ -149,7 +150,7 @@ class FetchOPCProcessor final : public BaseOPCProcessor { uint64_t max_depth_ = 0; LazyModeOptions lazy_mode_ = LazyModeOptions::Off; - std::vector translated_node_ids_; // Only used when user provides path, path->nodeid translation is only done once + std::vector translated_node_ids_; // Only used when user provides path; cached translation, re-resolved on (re)connect (see path_node_id_resolved_) }; } // namespace org::apache::nifi::minifi::processors diff --git a/extensions/opc/include/OPCCommon.h b/extensions/opc/include/OPCCommon.h index 3664080f39..3d2c177c5a 100644 --- a/extensions/opc/include/OPCCommon.h +++ b/extensions/opc/include/OPCCommon.h @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -202,5 +203,7 @@ void logFunc(void *context, UA_LogLevel level, UA_LogCategory category, const ch std::optional mapOpcReferenceType(const std::string& ref_type); +std::expected buildNodeId(opc::OPCNodeIDType id_type, UA_UInt16 namespace_idx, const std::string& node_id); + } // namespace org::apache::nifi::minifi::opc diff --git a/extensions/opc/include/PutOPCProcessor.h b/extensions/opc/include/PutOPCProcessor.h index 2ee4bb9d8f..772db43b93 100644 --- a/extensions/opc/include/PutOPCProcessor.h +++ b/extensions/opc/include/PutOPCProcessor.h @@ -41,20 +41,17 @@ class PutOPCProcessor final : public BaseOPCProcessor { public: EXTENSIONAPI static constexpr const char* Description = "Creates/updates OPC nodes"; - EXTENSIONAPI static constexpr auto ParentNodeIDType = core::PropertyDefinitionBuilder<3>::createProperty("Parent node ID type") - .withDescription("Specifies the type of the provided node ID") - .isRequired(true) - .withAllowedValues({"Path", "Int", "String"}) + EXTENSIONAPI static constexpr auto ParentNodeIDType = core::PropertyDefinitionBuilder()>::createProperty("Parent node ID type") + .withDescription("Specifies the type of the provided node ID. Only required when a new node is created.") + .withAllowedValues(magic_enum::enum_names()) .build(); EXTENSIONAPI static constexpr auto ParentNodeID = core::PropertyDefinitionBuilder<>::createProperty("Parent node ID") - .withDescription("Specifies the ID of the root node to traverse") - .isRequired(true) + .withDescription("Specifies the ID of the root node to traverse. Only required when a new node is created.") .build(); EXTENSIONAPI static constexpr auto ParentNameSpaceIndex = core::PropertyDefinitionBuilder<>::createProperty("Parent node namespace index") - .withDescription("The index of the namespace of the parent node.") - .withValidator(core::StandardPropertyValidators::INTEGER_VALIDATOR) + .withDescription("The index of the namespace of the parent node. Only used when a new node is created.") + .withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR) .withDefaultValue("0") - .isRequired(true) .build(); EXTENSIONAPI static constexpr auto ValueType = core::PropertyDefinitionBuilder()>::createProperty("Value type") .withDescription("Set the OPC value type of the created nodes") @@ -62,7 +59,7 @@ class PutOPCProcessor final : public BaseOPCProcessor { .isRequired(true) .build(); EXTENSIONAPI static constexpr auto TargetNodeIDType = core::PropertyDefinitionBuilder<>::createProperty("Target node ID type") - .withDescription("ID type of target node. Allowed values are: Int, String.") + .withDescription("ID type of target node. Allowed values are: Int, String, Guid.") .supportsExpressionLanguage(true) .isRequired(true) .build(); @@ -117,14 +114,13 @@ class PutOPCProcessor final : public BaseOPCProcessor { void initialize() override; private: - bool readParentNodeId(); std::expected, std::string> configureTargetNode(core::ProcessContext& context, core::FlowFile& flow_file) const; void updateNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessSession& session, const std::shared_ptr& flow_file) const; void createNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessContext& context, core::ProcessSession& session, const std::shared_ptr& flow_file) const; - opc::NodeId parent_node_id_; opc::OPCNodeDataType node_data_type_{}; UA_UInt32 create_node_reference_type_ = UA_NS0ID_HASCOMPONENT; + bool parent_node_defined_ = false; }; } // namespace org::apache::nifi::minifi::processors diff --git a/extensions/opc/src/BaseOPCProcessor.cpp b/extensions/opc/src/BaseOPCProcessor.cpp index a39cc7416d..b45e27de4a 100644 --- a/extensions/opc/src/BaseOPCProcessor.cpp +++ b/extensions/opc/src/BaseOPCProcessor.cpp @@ -105,10 +105,15 @@ bool BaseOPCProcessor::reconnect() { return false; } logger_->log_debug("Successfully connected."); + + // A new or re-established connection may be to a restarted server that can reassign node ids, + // so if path is used to identify node id, the cached node id should be invalidated + path_node_id_resolved_ = false; return true; } void BaseOPCProcessor::readPathReferenceTypes(core::ProcessContext& context, const std::string& node_id) { + path_reference_types_.clear(); const auto value = context.getProperty(PathReferenceTypes).value_or(""); if (value.empty()) { return; @@ -126,16 +131,25 @@ void BaseOPCProcessor::readPathReferenceTypes(core::ProcessContext& context, con } } -void BaseOPCProcessor::parseIdType(core::ProcessContext& context, const core::PropertyReference& prop) { - id_type_ = utils::parseEnumProperty(context, prop); - - if (id_type_ == opc::OPCNodeIDType::Int) { - try { - static_cast(std::stoi(node_id_)); - } catch(const std::exception&) { - auto error_msg = utils::string::join_pack(node_id_, " cannot be used as an int type node ID"); - throw Exception(PROCESS_SCHEDULE_EXCEPTION, error_msg); +void BaseOPCProcessor::parseNode(core::ProcessContext& context) { + switch (id_type_) { + case opc::OPCNodeIDType::String: + case opc::OPCNodeIDType::Int: + case opc::OPCNodeIDType::Guid: { + if (auto result = opc::buildNodeId(id_type_, namespace_idx_, node_id_)) { + node_ = std::move(*result); + } else { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, result.error()); + } + break; } + case opc::OPCNodeIDType::Path: + path_reference_types_.clear(); + readPathReferenceTypes(context, node_id_); + path_node_id_resolved_ = false; + break; + default: + throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Unsupported Node ID type: {}", magic_enum::enum_name(id_type_))); } } diff --git a/extensions/opc/src/FetchOPCHistory.cpp b/extensions/opc/src/FetchOPCHistory.cpp index 7ec06944b3..939de7c54a 100644 --- a/extensions/opc/src/FetchOPCHistory.cpp +++ b/extensions/opc/src/FetchOPCHistory.cpp @@ -281,32 +281,9 @@ void FetchOPCHistory::onSchedule(core::ProcessContext& context, core::ProcessSes logger_->log_trace("FetchOPCHistory::onSchedule"); BaseOPCProcessor::onSchedule(context, factory); node_id_ = utils::parseProperty(context, NodeID); - parseIdType(context, NodeIDType); - namespace_idx_ = gsl::narrow(utils::parseI64Property(context, NameSpaceIndex)); - - switch (id_type_) { - case opc::OPCNodeIDType::String: - node_ = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_idx_, node_id_.c_str())}; - break; - case opc::OPCNodeIDType::Int: - node_ = opc::NodeId{UA_NODEID_NUMERIC(namespace_idx_, std::stoi(node_id_))}; - break; - case opc::OPCNodeIDType::Guid: { - UA_Guid guid; - if (UA_Guid_parse(&guid, UA_STRING(const_cast(node_id_.c_str()))) != UA_STATUSCODE_GOOD) { - throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("{} cannot be used as a GUID type node ID", node_id_)); - } - node_ = opc::NodeId{UA_NODEID_GUID(namespace_idx_, guid)}; - break; - } - case opc::OPCNodeIDType::Path: - path_reference_types_.clear(); - readPathReferenceTypes(context, node_id_); - path_node_id_resolved_ = false; - break; - default: - throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Unsupported Node ID type: {}", magic_enum::enum_name(id_type_))); - } + id_type_ = utils::parseEnumProperty(context, NodeIDType); + namespace_idx_ = gsl::narrow(utils::parseU64Property(context, NameSpaceIndex)); + parseNode(context); history_type_ = utils::parseEnumProperty(context, HistoryReadType); auto start_timestamp_str = utils::parseOptionalProperty(context, StartTimestamp); diff --git a/extensions/opc/src/FetchOPCProcessor.cpp b/extensions/opc/src/FetchOPCProcessor.cpp index 4fc4b1caa9..63636a4880 100644 --- a/extensions/opc/src/FetchOPCProcessor.cpp +++ b/extensions/opc/src/FetchOPCProcessor.cpp @@ -44,17 +44,13 @@ void FetchOPCProcessor::onSchedule(core::ProcessContext& context, core::ProcessS BaseOPCProcessor::onSchedule(context, factory); node_id_ = utils::parseProperty(context, NodeID); - max_depth_ = utils::parseU64Property(context, MaxDepth); - - parseIdType(context, NodeIDType); + id_type_ = utils::parseEnumProperty(context, NodeIDType); + namespace_idx_ = gsl::narrow(utils::parseU64Property(context, NameSpaceIndex)); + parseNode(context); - namespace_idx_ = gsl::narrow(utils::parseI64Property(context, NameSpaceIndex)); + max_depth_ = utils::parseU64Property(context, MaxDepth); lazy_mode_ = utils::parseEnumProperty(context, Lazy); - - if (id_type_ == opc::OPCNodeIDType::Path) { - readPathReferenceTypes(context, node_id_); - } } void FetchOPCProcessor::onTrigger(core::ProcessContext& context, core::ProcessSession& session) { @@ -77,26 +73,17 @@ void FetchOPCProcessor::onTrigger(core::ProcessContext& context, core::ProcessSe }; if (id_type_ != opc::OPCNodeIDType::Path) { - const auto namespace_index = gsl::narrow_cast(namespace_idx_); - opc::NodeId my_id; - if (id_type_ == opc::OPCNodeIDType::Int) { - my_id = opc::NodeId{UA_NODEID_NUMERIC(namespace_index, std::stoi(node_id_))}; - } else if (id_type_ == opc::OPCNodeIDType::String) { - my_id = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_index, node_id_.c_str())}; - } else { - logger_->log_error("Unhandled id type: '{}'. No flowfiles are generated.", magic_enum::enum_underlying(id_type_)); - context.yield(); - return; - } - connection_->traverse(my_id, found_cb, "", max_depth_); + connection_->traverse(node_, found_cb, "", max_depth_); } else { - if (translated_node_ids_.empty()) { + if (!path_node_id_resolved_) { + translated_node_ids_.clear(); auto sc = connection_->translateBrowsePathsToNodeIdsRequest(node_id_, translated_node_ids_, namespace_idx_, path_reference_types_, logger_); if (sc != UA_STATUSCODE_GOOD) { logger_->log_error("Failed to translate {} to node id, no flow files will be generated ({})", node_id_.c_str(), UA_StatusCode_name(sc)); context.yield(); return; } + path_node_id_resolved_ = true; } for (auto& node_id : translated_node_ids_) { connection_->traverse(node_id, found_cb, node_id_, max_depth_); diff --git a/extensions/opc/src/OPCCommon.cpp b/extensions/opc/src/OPCCommon.cpp index 2b56cea130..d59e99e2e8 100644 --- a/extensions/opc/src/OPCCommon.cpp +++ b/extensions/opc/src/OPCCommon.cpp @@ -17,6 +17,7 @@ #include "OPCCommon.h" +#include #include #include #include @@ -89,6 +90,16 @@ void add_value_to_variant(UA_Variant *variant, double value) { UA_Variant_setScalarCopy(variant, &ua_value, &UA_TYPES[UA_TYPES_DOUBLE]); } +template +std::string printToString(UA_StatusCode (*print_func)(const T*, UA_String*), const T& data, std::string_view type_name) { + UA_String printed = UA_STRING_NULL; + if (print_func(&data, &printed) != UA_STATUSCODE_GOOD) { + throw OPCException(GENERAL_EXCEPTION, utils::string::join_pack("Failed to convert a ", type_name, " to string")); + } + const auto guard = gsl::finally([&printed]() { UA_String_clear(&printed); }); + return {reinterpret_cast(printed.data), printed.length}; +} + core::logging::LOG_LEVEL MapOPCLogLevel(UA_LogLevel ualvl) { switch (ualvl) { case UA_LOGLEVEL_TRACE: @@ -237,7 +248,20 @@ NodeData Client::getNodeData(const UA_ReferenceDescription *ref, const std::stri } else if (ref->nodeId.nodeId.identifierType == UA_NODEIDTYPE_NUMERIC) { nodedata.attributes["NodeID"] = std::to_string(ref->nodeId.nodeId.identifier.numeric); // NOLINT(cppcoreguidelines-pro-type-union-access) nodedata.attributes["NodeID type"] = "numeric"; + } else if (ref->nodeId.nodeId.identifierType == UA_NODEIDTYPE_GUID) { + const auto& guid = ref->nodeId.nodeId.identifier.guid; // NOLINT(cppcoreguidelines-pro-type-union-access) + std::array guid_str_array{}; + auto res = snprintf(guid_str_array.data(), guid_str_array.size(), "%08x-%04x-%04x-%02x%02x-%02x%02x%02x%02x%02x%02x", + guid.data1, guid.data2, guid.data3, guid.data4[0], guid.data4[1], guid.data4[2], + guid.data4[3], guid.data4[4], guid.data4[5], guid.data4[6], guid.data4[7]); + if (res < 0) { + nodedata.attributes["NodeID"] = ""; + } else { + nodedata.attributes["NodeID"] = std::string(guid_str_array.data()); + } + nodedata.attributes["NodeID type"] = "guid"; } + nodedata.attributes["Namespace index"] = std::to_string(ref->nodeId.nodeId.namespaceIndex); nodedata.attributes["Browsename"] = browsename; auto splitted_base_path = utils::string::splitAndTrimRemovingEmpty(base_path, "/"); @@ -503,10 +527,14 @@ std::string variantToString(const UA_Variant& variant, BinaryEncoding binary_enc } switch (variant.type->typeKind) { case UA_DATATYPEKIND_STRING: - case UA_DATATYPEKIND_LOCALIZEDTEXT: { - const auto *value = static_cast(variant.data); + case UA_DATATYPEKIND_XMLELEMENT: { + const auto* value = static_cast(variant.data); return {reinterpret_cast(value->data), value->length}; } + case UA_DATATYPEKIND_LOCALIZEDTEXT: { + const auto* value = static_cast(variant.data); + return {reinterpret_cast(value->text.data), value->text.length}; + } case UA_DATATYPEKIND_BYTESTRING: { const auto* value = static_cast(variant.data); if (value->data == nullptr || value->length == 0) { @@ -545,6 +573,16 @@ std::string variantToString(const UA_Variant& variant, BinaryEncoding binary_enc throw OPCException(GENERAL_EXCEPTION, "Double is non-standard on this system, OPC data cannot be extracted!"); case UA_DATATYPEKIND_DATETIME: return opc::OPCDateTime2String(*static_cast(variant.data)); + case UA_DATATYPEKIND_NODEID: + return printToString(UA_NodeId_print, *static_cast(variant.data), "node id"); + case UA_DATATYPEKIND_EXPANDEDNODEID: + return printToString(UA_ExpandedNodeId_print, *static_cast(variant.data), "expanded node id"); + case UA_DATATYPEKIND_GUID: + return printToString(UA_Guid_print, *static_cast(variant.data), "GUID"); + case UA_DATATYPEKIND_QUALIFIEDNAME: + return printToString(UA_QualifiedName_print, *static_cast(variant.data), "qualified name"); + case UA_DATATYPEKIND_STATUSCODE: + return UA_StatusCode_name(*static_cast(variant.data)); default: throw OPCException(GENERAL_EXCEPTION, "Data type is not supported: " + std::string(variant.type->typeName)); } @@ -595,4 +633,27 @@ UA_StatusCode Client::readHistory(HistoryReadTypeOption history_type, const UA_N return UA_Client_HistoryRead_raw(client_, &node_id, callback, start_time, end_time, UA_STRING_NULL, false, 0, UA_TIMESTAMPSTORETURN_SOURCE, callback_context); } +std::expected buildNodeId(opc::OPCNodeIDType id_type, UA_UInt16 namespace_idx, const std::string& node_id) { + switch (id_type) { + case opc::OPCNodeIDType::String: + return opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_idx, node_id.c_str())}; + case opc::OPCNodeIDType::Int: + try { + return opc::NodeId{UA_NODEID_NUMERIC(namespace_idx, gsl::narrow(std::stoul(node_id)))}; + } catch(const std::exception&) { + auto error_msg = utils::string::join_pack(node_id, " cannot be used as an int type node ID"); + return std::unexpected{error_msg}; + } + case opc::OPCNodeIDType::Guid: { + UA_Guid guid; + if (UA_Guid_parse(&guid, UA_STRING(const_cast(node_id.c_str()))) != UA_STATUSCODE_GOOD) { + return std::unexpected{fmt::format("{} cannot be used as a GUID type node ID", node_id)}; + } + return opc::NodeId{UA_NODEID_GUID(namespace_idx, guid)}; + } + default: + return std::unexpected{fmt::format("Unsupported Node ID type: {}", magic_enum::enum_name(id_type))}; + } +} + } // namespace org::apache::nifi::minifi::opc diff --git a/extensions/opc/src/PutOPCProcessor.cpp b/extensions/opc/src/PutOPCProcessor.cpp index b4636e0ef0..f01c6263d6 100644 --- a/extensions/opc/src/PutOPCProcessor.cpp +++ b/extensions/opc/src/PutOPCProcessor.cpp @@ -20,12 +20,12 @@ #include #include -#include "minifi-cpp/core/ProcessContext.h" +#include "OPCCommon.h" #include "core/ProcessSession.h" #include "core/Resource.h" -#include "OPCCommon.h" -#include "utils/StringUtils.h" +#include "minifi-cpp/core/ProcessContext.h" #include "utils/ProcessorConfigUtils.h" +#include "utils/StringUtils.h" namespace org::apache::nifi::minifi::processors { @@ -39,17 +39,17 @@ void PutOPCProcessor::onSchedule(core::ProcessContext& context, core::ProcessSes BaseOPCProcessor::onSchedule(context, session_factory); - node_id_ = utils::parseProperty(context, ParentNodeID); - - parseIdType(context, ParentNodeIDType); + parent_node_defined_ = false; + if (const auto parent_node_id = utils::parseOptionalProperty(context, ParentNodeID); parent_node_id && !parent_node_id->empty()) { + node_id_ = *parent_node_id; + id_type_ = utils::parseEnumProperty(context, ParentNodeIDType); + namespace_idx_ = gsl::narrow(utils::parseOptionalU64Property(context, ParentNameSpaceIndex).value_or(0)); + parseNode(context); + parent_node_defined_ = true; + } - namespace_idx_ = gsl::narrow(utils::parseI64Property(context, ParentNameSpaceIndex)); node_data_type_ = utils::parseEnumProperty(context, ValueType); - if (id_type_ == opc::OPCNodeIDType::Path) { - readPathReferenceTypes(context, node_id_); - } - const auto value = context.getProperty(CreateNodeReferenceType).value_or(""); if (auto ref_type = opc::mapOpcReferenceType(value)) { create_node_reference_type_ = ref_type.value(); @@ -58,81 +58,47 @@ void PutOPCProcessor::onSchedule(core::ProcessContext& context, core::ProcessSes } } -bool PutOPCProcessor::readParentNodeId() { - if (id_type_ == opc::OPCNodeIDType::Path) { - std::vector translated_node_ids; - if (connection_->translateBrowsePathsToNodeIdsRequest(node_id_, translated_node_ids, namespace_idx_, path_reference_types_, logger_) != - UA_STATUSCODE_GOOD) { - logger_->log_error("Failed to translate {} to node id, no flow files will be put", node_id_.c_str()); - return false; - } else if (translated_node_ids.size() != 1) { - logger_->log_error("{} was translated to multiple node ids, no flow files will be put", node_id_.c_str()); - return false; - } else { - parent_node_id_ = std::move(translated_node_ids[0]); - } - } else { - const auto namespace_index = gsl::narrow_cast(namespace_idx_); - if (id_type_ == opc::OPCNodeIDType::Int) { - parent_node_id_ = opc::NodeId{UA_NODEID_NUMERIC(namespace_index, std::stoi(node_id_))}; - } else { // idType_ == opc::OPCNodeIDType::String - parent_node_id_ = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_index, node_id_.c_str())}; - } - if (!connection_->exists(parent_node_id_)) { - logger_->log_error("Parent node doesn't exist, no flow files will be put"); - return false; - } - } - return true; -} - -std::expected, std::string> PutOPCProcessor::configureTargetNode(core::ProcessContext& context, core::FlowFile& flow_file) const { - const auto namespaceidx = context.getProperty(TargetNodeNameSpaceIndex, &flow_file).value_or(""); - if (namespaceidx.empty()) { +std::expected, std::string> PutOPCProcessor::configureTargetNode(core::ProcessContext& context, + core::FlowFile& flow_file) const { + const auto namespace_idx_str = context.getProperty(TargetNodeNameSpaceIndex, &flow_file).value_or(""); + if (namespace_idx_str.empty()) { return std::unexpected{fmt::format("Flowfile {} had no target namespace index specified, routing to failure!", flow_file.getUUIDStr())}; } - int32_t nsi = 0; + UA_UInt16 namespace_idx = 0; try { - nsi = std::stoi(namespaceidx); + namespace_idx = gsl::narrow(std::stoul(namespace_idx_str)); } catch (const std::exception&) { - return std::unexpected{fmt::format("Flowfile {} has invalid namespace index ({}), routing to failure!", - flow_file.getUUIDStr(), namespaceidx)}; + return std::unexpected{fmt::format("Flowfile {} has invalid namespace index ({}), routing to failure!", flow_file.getUUIDStr(), namespace_idx_str)}; + } + + const auto target_id_type_str = context.getProperty(TargetNodeIDType, &flow_file).value_or(""); + if (target_id_type_str.empty()) { + return std::unexpected{fmt::format("Flowfile {} has invalid target node id type, routing to failure!", flow_file.getUUIDStr())}; } - const auto target_id_type = context.getProperty(TargetNodeIDType, &flow_file).value_or(""); - if (target_id_type.empty()) { - return std::unexpected{fmt::format("Flowfile {} has invalid target node id type, routing to failure!", - flow_file.getUUIDStr())}; + auto target_id_type = magic_enum::enum_cast(target_id_type_str); + if (!target_id_type || *target_id_type == opc::OPCNodeIDType::Path) { + return std::unexpected{fmt::format("Flowfile {} has invalid target node id type '{}', routing to failure!", flow_file.getUUIDStr(), target_id_type_str)}; } const auto target_id = context.getProperty(TargetNodeID, &flow_file).value_or(""); if (target_id.empty()) { - return std::unexpected{fmt::format("Flowfile {} had target node ID type specified ({}) without ID, routing to failure!", - flow_file.getUUIDStr(), target_id_type)}; + return std::unexpected{ + fmt::format("Flowfile {} had target node ID type specified ({}) without ID, routing to failure!", flow_file.getUUIDStr(), target_id_type_str)}; } - const auto namespace_index = gsl::narrow_cast(nsi); opc::NodeId target_node; - if (target_id_type == "Int") { - int32_t numeric_id = 0; - try { - numeric_id = std::stoi(target_id); - } catch (const std::exception&) { - return std::unexpected{fmt::format("Flowfile {}: target node ID is not a valid integer: {}. Routing to failure!", - flow_file.getUUIDStr(), target_id)}; - } - target_node = opc::NodeId{UA_NODEID_NUMERIC(namespace_index, numeric_id)}; - } else if (target_id_type == "String") { - target_node = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_index, target_id.c_str())}; + if (auto result = opc::buildNodeId(*target_id_type, namespace_idx, target_id)) { + target_node = std::move(*result); } else { - return std::unexpected{fmt::format("Flowfile {}: target node ID type is invalid: {}. Routing to failure!", - flow_file.getUUIDStr(), target_id_type)}; + return std::unexpected{fmt::format("Flowfile {}: {}. Routing to failure!", flow_file.getUUIDStr(), result.error())}; } const bool target_node_exists = connection_->exists(target_node); return std::make_pair(target_node_exists, std::move(target_node)); } -void PutOPCProcessor::updateNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessSession& session, const std::shared_ptr& flow_file) const { +void PutOPCProcessor::updateNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessSession& session, + const std::shared_ptr& flow_file) const { logger_->log_trace("Node exists, trying to update it"); try { UA_StatusCode sc = 0; @@ -197,13 +163,12 @@ void PutOPCProcessor::updateNode(const UA_NodeId& target_node, const std::string } } -void PutOPCProcessor::createNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessContext& context, core::ProcessSession& session, - const std::shared_ptr& flow_file) const { +void PutOPCProcessor::createNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessContext& context, + core::ProcessSession& session, const std::shared_ptr& flow_file) const { logger_->log_trace("Node doesn't exist, trying to create new node"); const auto browse_name = context.getProperty(TargetNodeBrowseName, flow_file.get()).value_or(""); if (browse_name.empty()) { - logger_->log_error("Target node browse name is required for flowfile ({}) as new node is to be created", - flow_file->getUUIDStr()); + logger_->log_error("Target node browse name is required for flowfile ({}) as new node is to be created", flow_file->getUUIDStr()); session.transfer(flow_file, Failure); return; } @@ -214,27 +179,27 @@ void PutOPCProcessor::createNode(const UA_NodeId& target_node, const std::string switch (node_data_type_) { case opc::OPCNodeDataType::Int64: { int64_t value = std::stoll(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::UInt64: { uint64_t value = std::stoull(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::Int32: { int32_t value = std::stoi(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::UInt32: { uint32_t value = std::stoul(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::Boolean: { if (auto contentstr_parsed = utils::string::toBool(contentstr)) { - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, contentstr_parsed.value(), result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, contentstr_parsed.value(), result_node.receive()); } else { throw std::runtime_error("Content cannot be converted to bool"); } @@ -242,16 +207,16 @@ void PutOPCProcessor::createNode(const UA_NodeId& target_node, const std::string } case opc::OPCNodeDataType::Float: { float value = std::stof(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::Double: { double value = std::stod(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::String: { - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, contentstr, result_node.receive()); + sc = connection_->add_node(node_, target_node, create_node_reference_type_, browse_name, contentstr, result_node.receive()); break; } default: @@ -281,9 +246,21 @@ void PutOPCProcessor::onTrigger(core::ProcessContext& context, core::ProcessSess return; } - if (!readParentNodeId()) { - context.yield(); - return; + if (parent_node_defined_ && id_type_ == opc::OPCNodeIDType::Path && !path_node_id_resolved_) { + std::vector translated_node_ids; + auto sc = connection_->translateBrowsePathsToNodeIdsRequest(node_id_, translated_node_ids, namespace_idx_, path_reference_types_, logger_); + if (sc != UA_STATUSCODE_GOOD) { + logger_->log_error("Failed to translate path '{}' to a node id: {}", node_id_, UA_StatusCode_name(sc)); + context.yield(); + return; + } + if (translated_node_ids.size() != 1) { + logger_->log_error("Path '{}' resolved to {} node ids; exactly one target node is required for put", node_id_, translated_node_ids.size()); + context.yield(); + return; + } + node_ = std::move(translated_node_ids[0]); + path_node_id_resolved_ = true; } auto flow_file = session.get(); @@ -302,8 +279,11 @@ void PutOPCProcessor::onTrigger(core::ProcessContext& context, core::ProcessSess const auto contentstr = to_string(session.readBuffer(flow_file)); if (target_node_exists) { updateNode(target_node, contentstr, session, flow_file); - } else { + } else if (parent_node_defined_) { createNode(target_node, contentstr, context, session, flow_file); + } else { + logger_->log_error("Target node does not exist and no parent node is defined to create it under; routing to failure"); + session.transfer(flow_file, Failure); } } diff --git a/extensions/opc/tests/FetchOPCProcessorTests.cpp b/extensions/opc/tests/FetchOPCProcessorTests.cpp index 753ec4688f..a585bfdbd4 100644 --- a/extensions/opc/tests/FetchOPCProcessorTests.cpp +++ b/extensions/opc/tests/FetchOPCProcessorTests.cpp @@ -39,6 +39,7 @@ TEST_CASE("Test fetching using path node id", "[fetchopcprocessor]") { REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 4); for (size_t i = 0; i < 3; i++) { auto flow_file = results.at(processors::FetchOPCProcessor::Success)[i]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); CHECK(flow_file->getAttribute("Browsename") == "INT" + std::to_string(i + 1)); CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT" + std::to_string(i + 1)); @@ -50,6 +51,7 @@ TEST_CASE("Test fetching using path node id", "[fetchopcprocessor]") { } auto flow_file = results.at(processors::FetchOPCProcessor::Success)[3]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); CHECK(flow_file->getAttribute("Browsename") == "INT4"); CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT3/INT4"); @@ -75,6 +77,7 @@ TEST_CASE("Test fetching using custom reference type id path", "[fetchopcprocess REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 2); auto flow_file = results.at(processors::FetchOPCProcessor::Success)[0]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); CHECK(flow_file->getAttribute("Browsename") == "INT3"); CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT3"); @@ -84,6 +87,7 @@ TEST_CASE("Test fetching using custom reference type id path", "[fetchopcprocess CHECK(flow_file->getAttribute("Sourcetimestamp")); CHECK(controller.plan->getContent(flow_file) == "3"); flow_file = results.at(processors::FetchOPCProcessor::Success)[1]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); CHECK(flow_file->getAttribute("Browsename") == "INT4"); CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT3/INT4"); @@ -108,6 +112,7 @@ TEST_CASE("Test fetching using string node id", "[fetchopcprocessor]") { REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 1); auto flow_file = results.at(processors::FetchOPCProcessor::Success)[0]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); CHECK(flow_file->getAttribute("Browsename") == "StringNode"); CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "/StringNode"); @@ -118,6 +123,107 @@ TEST_CASE("Test fetching using string node id", "[fetchopcprocessor]") { CHECK(controller.plan->getContent(flow_file) == "42"); } +TEST_CASE("Test fetching using int node id", "[fetchopcprocessor]") { + OpcUaTestServer server(4841); + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("FetchOPCProcessor")}; + auto fetch_opc_processor = controller.getProcessor(); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4841/")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeIDType.name, "Int")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeID.name, "666")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + + const auto results = controller.trigger(); + REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 1); + auto flow_file = results.at(processors::FetchOPCProcessor::Success)[0]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); + CHECK(flow_file->getAttribute("Browsename") == "666"); + CHECK(flow_file->getAttribute("Datasize") == "4"); + CHECK(flow_file->getAttribute("NodeID") == "666"); + CHECK(flow_file->getAttribute("NodeID type") == "numeric"); + CHECK(flow_file->getAttribute("Typename") == "Int32"); + CHECK(flow_file->getAttribute("Sourcetimestamp")); + CHECK(controller.plan->getContent(flow_file) == "256"); +} + +TEST_CASE("Test fetching using guid node id", "[fetchopcprocessor]") { + OpcUaTestServer server(4841); + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("FetchOPCProcessor")}; + auto fetch_opc_processor = controller.getProcessor(); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4841/")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeIDType.name, "Guid")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeID.name, "72962b91-fa75-4ae6-8d28-b404dc7daf63")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + + const auto results = controller.trigger(); + REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 1); + auto flow_file = results.at(processors::FetchOPCProcessor::Success)[0]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); + CHECK(flow_file->getAttribute("Browsename") == "72962b91-fa75-4ae6-8d28-b404dc7daf63"); + CHECK(flow_file->getAttribute("Datasize") == "4"); + CHECK(flow_file->getAttribute("NodeID") == "72962b91-fa75-4ae6-8d28-b404dc7daf63"); + CHECK(flow_file->getAttribute("NodeID type") == "guid"); + CHECK(flow_file->getAttribute("Typename") == "Int32"); + CHECK(flow_file->getAttribute("Sourcetimestamp")); + CHECK(controller.plan->getContent(flow_file) == "7"); +} + +TEST_CASE("Test fetching with limited max depth", "[fetchopcprocessor]") { + OpcUaTestServer server(4841); + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("FetchOPCProcessor")}; + auto fetch_opc_processor = controller.getProcessor(); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4841/")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeIDType.name, "Path")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeID.name, "Simulator/Default/Device1")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::MaxDepth.name, "2")); + + const auto results = controller.trigger(); + REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 3); + for (size_t i = 0; i < 3; i++) { + auto flow_file = results.at(processors::FetchOPCProcessor::Success)[i]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); + CHECK(flow_file->getAttribute("Browsename") == "INT" + std::to_string(i + 1)); + CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT" + std::to_string(i + 1)); + } +} + +TEST_CASE("Test invalid guid node id", "[fetchopcprocessor]") { + OpcUaTestServer server(4841); + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("FetchOPCProcessor")}; + auto fetch_opc_processor = controller.getProcessor(); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4841/")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeIDType.name, "Guid")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeID.name, "not-a-valid-guid")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + + REQUIRE_THROWS_WITH(controller.trigger(), "Process Schedule Operation: not-a-valid-guid cannot be used as a GUID type node ID"); +} + +TEST_CASE("Test no variables found when fetching a node without variable children", "[fetchopcprocessor]") { + OpcUaTestServer server(4841); + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("FetchOPCProcessor")}; + LogTestController::getInstance().setWarn(); + auto fetch_opc_processor = controller.getProcessor(); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4841/")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeIDType.name, "Path")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NodeID.name, "Simulator/Default/Device2/GuidObject")); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::NameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + REQUIRE(fetch_opc_processor->setProperty(processors::FetchOPCProcessor::PathReferenceTypes.name, "Organizes/Organizes/Organizes")); + + const auto results = controller.trigger(); + REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::FetchOPCProcessor::Success).empty()); + REQUIRE(LogTestController::getInstance().contains("Found no variables when traversing the specified node. No flowfiles are generated. Yielding...")); +} + TEST_CASE("Test missing path reference types", "[fetchopcprocessor]") { SingleProcessorTestController controller{minifi::test::utils::make_processor("FetchOPCProcessor")}; auto fetch_opc_processor = controller.getProcessor(); @@ -247,6 +353,7 @@ TEST_CASE("Test fetch for nodes with changed timestamps with lazy mode", "[fetch REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 1); auto flow_file = results.at(processors::FetchOPCProcessor::Success)[0]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); CHECK(flow_file->getAttribute("Browsename") == "INT3"); } @@ -295,6 +402,7 @@ TEST_CASE("Test fetching new values using lazy new value mode", "[fetchopcproces REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 1); auto flow_file = results.at(processors::FetchOPCProcessor::Success)[0]; + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server.getNamespaceIndex())); CHECK(flow_file->getAttribute("Browsename") == "INT2"); } diff --git a/extensions/opc/tests/OPCCommonTests.cpp b/extensions/opc/tests/OPCCommonTests.cpp index e71c04b716..1dd6da4577 100644 --- a/extensions/opc/tests/OPCCommonTests.cpp +++ b/extensions/opc/tests/OPCCommonTests.cpp @@ -65,6 +65,54 @@ TEST_CASE("The binary encoding does not change the conversion of non-binary valu CHECK(opc::variantToString(variant, binary_encoding) == "-128"); } + + SECTION("LocalizedText") { + UA_LocalizedText value = UA_LOCALIZEDTEXT(const_cast("en-US"), const_cast("some message")); + UA_Variant variant; + UA_Variant_setScalar(&variant, &value, &UA_TYPES[UA_TYPES_LOCALIZEDTEXT]); + + CHECK(opc::variantToString(variant, binary_encoding) == "some message"); + } + + SECTION("NodeId") { + UA_NodeId value = UA_NODEID_NUMERIC(1, 42); + UA_Variant variant; + UA_Variant_setScalar(&variant, &value, &UA_TYPES[UA_TYPES_NODEID]); + + CHECK(opc::variantToString(variant, binary_encoding) == "ns=1;i=42"); + } + + SECTION("ExpandedNodeId") { + UA_ExpandedNodeId value = UA_EXPANDEDNODEID_NUMERIC(1, 42); + UA_Variant variant; + UA_Variant_setScalar(&variant, &value, &UA_TYPES[UA_TYPES_EXPANDEDNODEID]); + + CHECK(opc::variantToString(variant, binary_encoding) == "ns=1;i=42"); + } + + SECTION("Guid") { + UA_Guid value = UA_GUID("72962B91-FA75-4AE6-8D28-B404DC7DAF63"); + UA_Variant variant; + UA_Variant_setScalar(&variant, &value, &UA_TYPES[UA_TYPES_GUID]); + + CHECK(opc::variantToString(variant, binary_encoding) == "72962b91-fa75-4ae6-8d28-b404dc7daf63"); + } + + SECTION("QualifiedName") { + UA_QualifiedName value = UA_QUALIFIEDNAME(1, const_cast("Colour")); + UA_Variant variant; + UA_Variant_setScalar(&variant, &value, &UA_TYPES[UA_TYPES_QUALIFIEDNAME]); + + CHECK(opc::variantToString(variant, binary_encoding) == "1:Colour"); + } + + SECTION("StatusCode") { + UA_StatusCode value = UA_STATUSCODE_BADNODEIDUNKNOWN; + UA_Variant variant; + UA_Variant_setScalar(&variant, &value, &UA_TYPES[UA_TYPES_STATUSCODE]); + + CHECK(opc::variantToString(variant, binary_encoding) == "BadNodeIdUnknown"); + } } TEST_CASE("A byte string variant is converted according to the requested binary encoding", "[opccommon]") { diff --git a/extensions/opc/tests/OpcUaTestServer.h b/extensions/opc/tests/OpcUaTestServer.h index 31ccd098c6..4a31ef7cac 100644 --- a/extensions/opc/tests/OpcUaTestServer.h +++ b/extensions/opc/tests/OpcUaTestServer.h @@ -111,6 +111,11 @@ class OpcUaTestServer { addStringVariable("StringNode", "the.answer.node", UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER), 42); addIntVariable("666", opc::OPCNodeIDType::Int, device2_node, 256); addIntVariable("72962b91-fa75-4ae6-8d28-b404dc7daf63", opc::OPCNodeIDType::Guid, device2_node, 7); + addObject("GuidObject", device2_node, UA_NODEID_GUID(ns_index_, UA_GUID("aabbccdd-1122-3344-5566-778899aabbcc"))); + + UA_NodeId ambiguous_parent_node = addObject("AmbiguousParent", default_node); + addObject("Ambiguous", ambiguous_parent_node); + addObject("Ambiguous", ambiguous_parent_node); setHistory("INT1", {HistoryModificationRecord{.value = 1, @@ -229,9 +234,9 @@ class OpcUaTestServer { } else if (id.identifierType == UA_NODEIDTYPE_NUMERIC) { return std::to_string(id.identifier.numeric); } else if (id.identifierType == UA_NODEIDTYPE_GUID) { - char guid_str[37]; - snprintf(guid_str, - sizeof(guid_str), + std::array guid_str; + snprintf(guid_str.data(), + guid_str.size(), "%08x-%04x-%04x-%04x-%012" PRIx64, id.identifier.guid.data1, id.identifier.guid.data2, @@ -240,7 +245,7 @@ class OpcUaTestServer { (gsl::narrow(id.identifier.guid.data4[2]) << 40) | (gsl::narrow(id.identifier.guid.data4[3]) << 32) | (gsl::narrow(id.identifier.guid.data4[4]) << 24) | (gsl::narrow(id.identifier.guid.data4[5]) << 16) | (gsl::narrow(id.identifier.guid.data4[6]) << 8) | gsl::narrow(id.identifier.guid.data4[7])); - return std::string(guid_str); + return std::string(guid_str.data()); } return {}; } @@ -374,13 +379,13 @@ class OpcUaTestServer { response->responseHeader.serviceResult = UA_STATUSCODE_GOOD; } - UA_NodeId addObject(const char* name, UA_NodeId parent) { + UA_NodeId addObject(const char* name, UA_NodeId parent, UA_NodeId requested_node_id = UA_NODEID_NULL) { UA_NodeId object_id; UA_ObjectAttributes attr = UA_ObjectAttributes_default; attr.displayName = UA_LOCALIZEDTEXT_ALLOC("en-US", name); auto status = UA_Server_addObjectNode(server_, - UA_NODEID_NULL, + requested_node_id, parent, UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES), UA_QUALIFIEDNAME(ns_index_, const_cast(name)), @@ -405,7 +410,7 @@ class OpcUaTestServer { switch (type) { case opc::OPCNodeIDType::Int: - node_id = opc::NodeId{UA_NODEID_NUMERIC(ns_index_, std::stoi(browse_name))}; + node_id = opc::NodeId{UA_NODEID_NUMERIC(ns_index_, gsl::narrow(std::stoul(browse_name)))}; break; case opc::OPCNodeIDType::String: node_id = opc::NodeId{UA_NODEID_STRING_ALLOC(ns_index_, browse_name.c_str())}; diff --git a/extensions/opc/tests/PutOPCProcessorTests.cpp b/extensions/opc/tests/PutOPCProcessorTests.cpp index 7197d05c7a..1ebb77460c 100644 --- a/extensions/opc/tests/PutOPCProcessorTests.cpp +++ b/extensions/opc/tests/PutOPCProcessorTests.cpp @@ -15,6 +15,8 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +#include + #include "unit/TestBase.h" #include "unit/Catch.h" #include "OpcUaTestServer.h" @@ -27,13 +29,14 @@ namespace org::apache::nifi::minifi::test { struct NodeData { - uint8_t data; - uint16_t namespace_index; - uint32_t node_id; + uint8_t data = 0; + uint16_t namespace_index = 0; + uint32_t node_id = 0; std::string browse_name; std::string path; std::string path_reference_types; std::string target_reference_type = "HasComponent"; + std::optional guid_node_id = std::nullopt; }; void verifyCreatedNode(const NodeData& expected_node, SingleProcessorTestController& controller) { @@ -61,8 +64,15 @@ void verifyCreatedNode(const NodeData& expected_node, SingleProcessorTestControl REQUIRE(found_node_ids.size() == 1); REQUIRE(found_node_ids[0].get().namespaceIndex == expected_node.namespace_index); - REQUIRE(found_node_ids[0].get().identifierType == UA_NODEIDTYPE_NUMERIC); - REQUIRE(found_node_ids[0].get().identifier.numeric == expected_node.node_id); // NOLINT(cppcoreguidelines-pro-type-union-access) + if (expected_node.guid_node_id) { + REQUIRE(found_node_ids[0].get().identifierType == UA_NODEIDTYPE_GUID); + UA_Guid expected_guid; + REQUIRE(UA_Guid_parse(&expected_guid, UA_STRING(const_cast(expected_node.guid_node_id->c_str()))) == UA_STATUSCODE_GOOD); + CHECK(UA_Guid_equal(&found_node_ids[0].get().identifier.guid, &expected_guid)); // NOLINT(cppcoreguidelines-pro-type-union-access) + } else { + REQUIRE(found_node_ids[0].get().identifierType == UA_NODEIDTYPE_NUMERIC); + REQUIRE(found_node_ids[0].get().identifier.numeric == expected_node.node_id); // NOLINT(cppcoreguidelines-pro-type-union-access) + } UA_ReferenceDescription ref_desc; ref_desc.isForward = true; @@ -80,6 +90,26 @@ void verifyCreatedNode(const NodeData& expected_node, SingleProcessorTestControl CHECK(data.data[0] == expected_node.data); } +void verifyNodeValue(SingleProcessorTestController& controller, const opc::NodeId& target, uint16_t namespace_index, const std::string& browse_name, uint8_t expected_value) { + auto client = minifi::opc::Client::createClient(controller.getLogger(), "", {}, {}, {}); + REQUIRE(client->connect("opc.tcp://127.0.0.1:4840/") == UA_STATUSCODE_GOOD); + + UA_ReferenceDescription ref_desc; + ref_desc.isForward = true; + ref_desc.referenceTypeId = UA_NODEID_NUMERIC(0, UA_NODEIDTYPE_NUMERIC); + ref_desc.nodeId.nodeId = target.get(); + ref_desc.browseName = UA_QUALIFIEDNAME_ALLOC(namespace_index, browse_name.c_str()); + ref_desc.displayName = UA_LOCALIZEDTEXT_ALLOC("en-US", browse_name.c_str()); + const auto ref_desc_guard = gsl::finally([&ref_desc] { + UA_LocalizedText_clear(&ref_desc.displayName); + UA_QualifiedName_clear(&ref_desc.browseName); + }); + ref_desc.nodeClass = UA_NODECLASS_VARIABLE; + ref_desc.typeDefinition.nodeId = UA_NODEID_NUMERIC(0, UA_NODEIDTYPE_NUMERIC); + auto data = client->getNodeData(&ref_desc, ""); + CHECK(data.data[0] == expected_value); +} + TEST_CASE("Test creating a new node with path node id", "[putopcprocessor]") { OpcUaTestServer server; server.start(); @@ -325,7 +355,7 @@ TEST_CASE("Test invalid parent node id path", "[putopcprocessor]") { const auto results = controller.trigger("42"); REQUIRE(results.at(processors::PutOPCProcessor::Success).empty()); REQUIRE(results.at(processors::PutOPCProcessor::Failure).empty()); - REQUIRE(LogTestController::getInstance().contains("to node id, no flow files will be put")); + REQUIRE(LogTestController::getInstance().contains("Failed to translate path 'Simulator/Default/Device1/INT99' to a node id: BadNoDataAvailable")); } TEST_CASE("Test missing target node id", "[putopcprocessor]") { @@ -371,7 +401,7 @@ TEST_CASE("Test invalid target node id", "[putopcprocessor]") { REQUIRE(results.at(processors::PutOPCProcessor::Failure).size() == 1); auto flow_file = results.at(processors::PutOPCProcessor::Failure)[0]; CHECK(controller.plan->getContent(flow_file) == "42"); - REQUIRE(LogTestController::getInstance().contains("target node ID is not a valid integer: invalid_int. Routing to failure")); + REQUIRE(LogTestController::getInstance().contains("invalid_int cannot be used as an int type node ID. Routing to failure!")); } TEST_CASE("Test missing target node type", "[putopcprocessor]") { @@ -394,7 +424,7 @@ TEST_CASE("Test missing target node type", "[putopcprocessor]") { REQUIRE(results.at(processors::PutOPCProcessor::Failure).size() == 1); auto flow_file = results.at(processors::PutOPCProcessor::Failure)[0]; CHECK(controller.plan->getContent(flow_file) == "42"); - REQUIRE(LogTestController::getInstance().contains("target node ID type is invalid: invalid. Routing to failure!")); + REQUIRE(LogTestController::getInstance().contains("has invalid target node id type 'invalid', routing to failure!")); } TEST_CASE("Test value type mismatch", "[putopcprocessor]") { @@ -420,4 +450,196 @@ TEST_CASE("Test value type mismatch", "[putopcprocessor]") { REQUIRE(LogTestController::getInstance().contains("Failed to convert 42 to data type Boolean")); } +TEST_CASE("Test creating a new node with GUID target node id", "[putopcprocessor]") { + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + + NodeData expected_node{.data = 42, .namespace_index = server.getNamespaceIndex(), .browse_name = "everything", .path = "Simulator/Default/Device1", + .path_reference_types = "", .guid_node_id = "12345678-1234-1234-1234-123456789abc"}; + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeIDType.name, "Path")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeID.name, expected_node.path)); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNameSpaceIndex.name, std::to_string(expected_node.namespace_index))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "Guid")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, *expected_node.guid_node_id)); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(expected_node.namespace_index))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeBrowseName.name, expected_node.browse_name)); + + const auto results = controller.trigger(std::to_string(expected_node.data)); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Success).size() == 1); + auto flow_file = results.at(processors::PutOPCProcessor::Success)[0]; + CHECK(controller.plan->getContent(flow_file) == std::to_string(expected_node.data)); + verifyCreatedNode(expected_node, controller); +} + +TEST_CASE("Test creating a new node under a GUID parent node id", "[putopcprocessor]") { + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + + const std::string parent_guid = "aabbccdd-1122-3344-5566-778899aabbcc"; + NodeData expected_node{.data = 42, .namespace_index = server.getNamespaceIndex(), .node_id = 8888, .browse_name = "everything", + .path = "Simulator/Default/Device2/GuidObject", .path_reference_types = "Organizes/Organizes/Organizes"}; + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeIDType.name, "Guid")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeID.name, parent_guid)); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNameSpaceIndex.name, std::to_string(expected_node.namespace_index))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "Int")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, std::to_string(expected_node.node_id))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(expected_node.namespace_index))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeBrowseName.name, expected_node.browse_name)); + + const auto results = controller.trigger(std::to_string(expected_node.data)); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Success).size() == 1); + auto flow_file = results.at(processors::PutOPCProcessor::Success)[0]; + CHECK(controller.plan->getContent(flow_file) == std::to_string(expected_node.data)); + verifyCreatedNode(expected_node, controller); +} + +TEST_CASE("Test invalid GUID parent node id", "[putopcprocessor]") { + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeIDType.name, "Guid")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeID.name, "not-a-guid")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + + REQUIRE_THROWS_WITH(controller.trigger("42"), "Process Schedule Operation: not-a-guid cannot be used as a GUID type node ID"); +} + +TEST_CASE("Test invalid GUID target node id", "[putopcprocessor]") { + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeIDType.name, "Path")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeID.name, "Simulator/Default/Device1")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "Guid")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, "not-a-guid")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeBrowseName.name, "everything")); + + const auto results = controller.trigger("42"); + REQUIRE(results.at(processors::PutOPCProcessor::Success).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).size() == 1); + REQUIRE(LogTestController::getInstance().contains("not-a-guid cannot be used as a GUID type node ID")); +} + +TEST_CASE("Test updating an existing Int node without a parent node defined", "[putopcprocessor]") { + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "Int")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, "666")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + + const auto results = controller.trigger("123"); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Success).size() == 1); + auto flow_file = results.at(processors::PutOPCProcessor::Success)[0]; + CHECK(controller.plan->getContent(flow_file) == "123"); + verifyNodeValue(controller, opc::NodeId{UA_NODEID_NUMERIC(server.getNamespaceIndex(), 666)}, server.getNamespaceIndex(), "666", 123); +} + +TEST_CASE("Test updating an existing String node without a parent node defined", "[putopcprocessor]") { + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "String")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, "the.answer.node")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + + const auto results = controller.trigger("55"); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Success).size() == 1); + auto flow_file = results.at(processors::PutOPCProcessor::Success)[0]; + CHECK(controller.plan->getContent(flow_file) == "55"); + verifyNodeValue(controller, opc::NodeId{UA_NODEID_STRING_ALLOC(server.getNamespaceIndex(), "the.answer.node")}, server.getNamespaceIndex(), "StringNode", 55); +} + +TEST_CASE("Test updating an existing GUID node without a parent node defined", "[putopcprocessor]") { + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + + const std::string target_guid = "72962b91-fa75-4ae6-8d28-b404dc7daf63"; + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "Guid")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, target_guid)); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + + const auto results = controller.trigger("99"); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Success).size() == 1); + auto flow_file = results.at(processors::PutOPCProcessor::Success)[0]; + CHECK(controller.plan->getContent(flow_file) == "99"); + UA_Guid guid; + REQUIRE(UA_Guid_parse(&guid, UA_STRING(const_cast(target_guid.c_str()))) == UA_STATUSCODE_GOOD); + verifyNodeValue(controller, opc::NodeId{UA_NODEID_GUID(server.getNamespaceIndex(), guid)}, server.getNamespaceIndex(), target_guid, 99); +} + +TEST_CASE("Test creating a new node fails without a parent node defined", "[putopcprocessor]") { + LogTestController::getInstance().setError(); + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "Int")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, "9999")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(server.getNamespaceIndex()))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeBrowseName.name, "everything")); + + const auto results = controller.trigger("42"); + REQUIRE(results.at(processors::PutOPCProcessor::Success).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).size() == 1); + REQUIRE(LogTestController::getInstance().contains("no parent node is defined")); +} + +TEST_CASE("Test multiple nodes are returned for path", "[putopcprocessor]") { + OpcUaTestServer server; + server.start(); + SingleProcessorTestController controller{minifi::test::utils::make_processor("PutOPCProcessor")}; + auto put_opc_processor = controller.getProcessor(); + + NodeData expected_node{42, server.getNamespaceIndex(), 9999, "everything", "Simulator/Default/AmbiguousParent/Ambiguous", "Organizes/Organizes/Organizes"}; + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4840/")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeIDType.name, "Path")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNodeID.name, expected_node.path)); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::PathReferenceTypes.name, expected_node.path_reference_types)); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ParentNameSpaceIndex.name, std::to_string(expected_node.namespace_index))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::ValueType.name, "Int32")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeIDType.name, "Int")); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeID.name, std::to_string(expected_node.node_id))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeNameSpaceIndex.name, std::to_string(expected_node.namespace_index))); + REQUIRE(put_opc_processor->setProperty(processors::PutOPCProcessor::TargetNodeBrowseName.name, expected_node.browse_name)); + + const auto results = controller.trigger(std::to_string(expected_node.data)); + REQUIRE(results.at(processors::PutOPCProcessor::Success).empty()); + REQUIRE(results.at(processors::PutOPCProcessor::Failure).empty()); + REQUIRE(LogTestController::getInstance().contains("exactly one target node is required for put")); +} + } // namespace org::apache::nifi::minifi::test