This is an automated email from the ASF dual-hosted git repository. sruehl pushed a commit to branch develop in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit a6f63121330791519a759c81e2d1ff7479fa0fc3 Author: Sebastian Rühl <[email protected]> AuthorDate: Thu May 21 16:14:26 2026 +0200 fix(plc4go): make BACnet/IP driver actually round-trip against bacpypes3 End-to-end integration testing against bacpypes3 0.0.102 surfaced a stack of bugs that masked each other; this commit fixes them and adds the regression tests (unit + dockerized integration) that pin the behavior down. Driver bugs - MessageCodec.handleCustomMessage unconditionally returned true, swallowing every parsed BVLC before HandleMessages could match it. Read/Write/Subscribe responses arrived on the wire but never woke their expectations. Replace with a no-op keepReceiveLoopActive that returns false so the default expectation-matching path runs while still keeping the receive worker awake when expectations are empty (needed so unsolicited COV notifications get drained). - Writer's WritePropertyMultiple path used wrong BACnet context tag numbers inside BACnetPropertyWriteDefinition (2/3/3 instead of 0/1/2), causing bacpypes3 REJECT(INVALID_TAG). Fix the tag numbers and parameterize constructedDataFromAppTag so single-write keeps its [3] wrapper. - ValueDecoder fell through to PlcSTRING for BACnet *Tagged enum wrappers (BACnetBinaryPVTagged etc.), making BV reads unusable. Add a reflection- based taggedEnumToPlcValue that surfaces them as PlcUDINT/PlcLINT/PlcBOOL. - Subscriber.dispatchNotification only inspected ApplicationTag on COV elements, dropping ConstructedData-framed values to PlcNULL. Add the ConstructedData branch. - ValueEncoder sent Boolean application tags for bool writes; Binary PresentValue requires Enumerated and BACnet stacks REJECT otherwise. Extend hintForProperty to take object type and return hintEnumerated for Binary/* PRESENT_VALUE; encoder maps bool -> Enumerated(0|1) under it. - Driver was binding to an ephemeral UDP source port; spec-conformant peers reply to UDP/47808 regardless of source, so responses got dropped. Bind local port to 47808 by default, overridable via the local-port driver option for processes that need multiple BACnet connections. - UDP transport's WriteToUDP rejected DialUDP-connected sockets with ErrWriteToConnected. Discriminate on udpConn.RemoteAddr() and use plain Write for connected sockets, WriteToUDP for ListenUDP sockets. Cross-driver fix - spi/model/DefaultPlcUnsubscriptionRequestBuilder.AddHandles appended to the local parameter (`subscriptionHandles = append(subscriptionHandles, subscriptionHandles...)`) and never stored anything on the builder, so every caller got back an empty request. Fix to append to d.subscriptionHandles. API additions on the BACnet Connection - Wire up UnsubscriptionRequestBuilder (was inherited as panicking stub). - Make plcTag a PlcSubscriptionTag (GetPlcSubscriptionType, GetDuration) and ValueHandler override NewPlcValue so written values survive the type-switch. - Add Framing.wrapAPDU helper: APDU -> NPDU -> BVLCOriginalUnicastNPDU, required because MessageCodec.Send hard-casts to BVLC. Reader/Writer/ Subscriber now wrap their requests via wrapAPDU instead of passing raw APDUs that triggered a panic. Dockerized integration suite (new pattern for plc4go) - Two-container compose on a private bridge so both sides bind UDP/47808 in their own network namespace. bacnet-device runs bacpypes3 (with a monkey-patched do_WritePropertyMultipleRequest because 0.0.102 ships a raise-UnrecognizedService stub); test-runner runs `go test -tags integration`. - 16 integration tests cover Discover, Read (AV/BV/MSV/CharacterString/ UnknownObject), Write (AV/BV/MSV/ReadOnly), WritePropertyMultiple, multi-tag Read, concurrent Read, Subscribe (initial + sawtooth-driven COV), and Unsubscribe. - `make integration-bacnetip` builds, runs, and tears down with the test-runner's exit code so a failed test fails the make target. Regression unit tests - MessageCodec_test pins Send/Receive behavior and the keepReceiveLoopActive==false invariant — flipping it back to true is what hid the bug originally. - Framing_test locks wrapAPDU's BVLC type, protocolVersion=1, and the no-routing local-scope contract. - ValueDecoder_test exercises BACnetBinaryPVTagged and every reflection Kind branch in taggedEnumToPlcValue. - Subscriber_test adds a ConstructedData-branch COV notification. - Writer_test asserts the WPM context tag numbers (0/1/2) and the single-write tag numbers (1/2/3) directly, so a future helper-sharing refactor can't silently swap them. --- plc4go/Makefile | 13 +- plc4go/internal/bacnetip/Connection.go | 8 + plc4go/internal/bacnetip/Driver.go | 34 +- plc4go/internal/bacnetip/Framing.go | 55 +++ plc4go/internal/bacnetip/Framing_test.go | 114 +++++ plc4go/internal/bacnetip/MessageCodec.go | 26 +- plc4go/internal/bacnetip/MessageCodec_test.go | 243 +++++++++++ plc4go/internal/bacnetip/Reader.go | 2 +- plc4go/internal/bacnetip/Subscriber.go | 15 +- plc4go/internal/bacnetip/Subscriber_test.go | 76 ++++ plc4go/internal/bacnetip/Tag.go | 16 + plc4go/internal/bacnetip/ValueDecoder.go | 33 ++ plc4go/internal/bacnetip/ValueDecoder_test.go | 138 ++++++ plc4go/internal/bacnetip/ValueEncoder.go | 35 +- plc4go/internal/bacnetip/ValueHandler.go | 62 ++- plc4go/internal/bacnetip/Writer.go | 38 +- plc4go/internal/bacnetip/Writer_test.go | 109 +++++ .../spi/model/DefaultPlcUnsubscriptionRequest.go | 2 +- plc4go/spi/transports/udp/TransportInstance.go | 25 +- plc4go/tests/integration/bacnetip/Dockerfile | 4 +- plc4go/tests/integration/bacnetip/Dockerfile.test | 51 +++ plc4go/tests/integration/bacnetip/README.md | 94 ++-- plc4go/tests/integration/bacnetip/device.py | 133 ++++-- .../tests/integration/bacnetip/docker-compose.yml | 56 ++- .../tests/integration/bacnetip/integration_test.go | 474 +++++++++++++++++++-- 25 files changed, 1680 insertions(+), 176 deletions(-) diff --git a/plc4go/Makefile b/plc4go/Makefile index 45b932a39f..1db5f3e80c 100644 --- a/plc4go/Makefile +++ b/plc4go/Makefile @@ -61,12 +61,15 @@ test: compile test-readable: compile @GOPATH=$(GOPATH) GOBIN=$(GOBIN) go tool -modfile=tools.mod gotestsum ./... -## integration-bacnetip: Bring up the bacpypes3 simulator + run the BACnet/IP integration tests. -## Tears the container down on exit even if a test fails. +## integration-bacnetip: Build + run the bacpypes3 simulator + Go test-runner in their own +## network namespaces on a shared docker bridge, then tear everything down. The compose exit +## code follows the test-runner so `make integration-bacnetip` mirrors a normal go-test failure. integration-bacnetip: - @docker compose -f tests/integration/bacnetip/docker-compose.yml up -d --build - @trap 'docker compose -f tests/integration/bacnetip/docker-compose.yml down' EXIT; \ - BACNET_IT=1 go test -tags integration ./tests/integration/bacnetip/... -v -count=1 -timeout 120s + @docker compose -f tests/integration/bacnetip/docker-compose.yml up \ + --build --abort-on-container-exit --exit-code-from test-runner; \ + status=$$?; \ + docker compose -f tests/integration/bacnetip/docker-compose.yml down; \ + exit $$status test-readable-mvn: compile $(MVNBIN) mvn-golang-wrapper:custom@readable-test diff --git a/plc4go/internal/bacnetip/Connection.go b/plc4go/internal/bacnetip/Connection.go index f7baef19c6..2277ad0d14 100644 --- a/plc4go/internal/bacnetip/Connection.go +++ b/plc4go/internal/bacnetip/Connection.go @@ -231,6 +231,14 @@ func (c *Connection) SubscriptionRequestBuilder() apiModel.PlcSubscriptionReques ) } +func (c *Connection) UnsubscriptionRequestBuilder() apiModel.PlcUnsubscriptionRequestBuilder { + // The default request implementation dispatches each handle's + // Unsubscribe back through the embedded Subscriber, so we don't need + // to pass our own here — the SubscriptionHandles created by Subscribe + // already carry the Subscriber reference. + return spiModel.NewDefaultPlcUnsubscriptionRequestBuilder() +} + func (c *Connection) addSubscriber(subscriber *Subscriber) { if slices.Contains(c.subscribers, subscriber) { c.log.Debug().Interface("subscriber", subscriber).Msg("Subscriber already added") diff --git a/plc4go/internal/bacnetip/Driver.go b/plc4go/internal/bacnetip/Driver.go index 06feb8a231..312ddf9793 100644 --- a/plc4go/internal/bacnetip/Driver.go +++ b/plc4go/internal/bacnetip/Driver.go @@ -22,6 +22,7 @@ package bacnetip import ( "context" "math" + "net" "net/url" "strconv" "time" @@ -36,6 +37,7 @@ import ( "github.com/apache/plc4x/plc4go/spi/options" "github.com/apache/plc4x/plc4go/spi/transactions" "github.com/apache/plc4x/plc4go/spi/transports" + "github.com/apache/plc4x/plc4go/spi/transports/udp" "github.com/apache/plc4x/plc4go/spi/utils" ) @@ -87,16 +89,44 @@ func (d *Driver) GetConnection(ctx context.Context, transportUrl url.URL, transp if _, ok := driverOptions["so-reuse"]; !ok { driverOptions["so-reuse"] = []string{"true"} } - // Have the transport create a new transport-instance. - transportInstance, err := transport.CreateTransportInstance( + // BACnet/IP uses port 47808 on both sides of a conversation; spec-conformant + // peers (bacpypes3, EcoStruxure, Niagara, ...) send unsolicited messages + // and responses back to the well-known port regardless of the request's + // source port. The generic transport.CreateTransportInstance dials with + // LocalAddress=nil, which gives us an ephemeral source — fine for protocols + // that reply to the source port, but for BACnet that means responses get + // dropped by the kernel. + // + // Use CreateTransportInstanceForLocalAddress with a fixed 0.0.0.0:47808 + // bind. Callers that need to co-locate multiple BACnet connections in one + // process can override via the "local-port" driver option (uint), or 0 + // for explicit ephemeral. + localPort := int(model.BacnetConstants_BACNETUDPDEFAULTPORT) + if val, ok := driverOptions["local-port"]; ok && len(val) > 0 { + if parsed, parseErr := strconv.Atoi(val[0]); parseErr != nil { + connectionLog.Warn().Err(parseErr).Str("local-port", val[0]).Msg("ignoring invalid local-port option") + } else { + localPort = parsed + } + } + localAddress := &net.UDPAddr{IP: net.IPv4zero, Port: localPort} + connectionLog.Info().Stringer("localAddress", localAddress).Msg("BACnet driver binding local UDP") + + udpTransport, ok := transport.(*udp.Transport) + if !ok { + return nil, errors.Errorf("BACnet/IP requires the udp transport; got %T", transport) + } + transportInstance, err := udpTransport.CreateTransportInstanceForLocalAddress( transportUrl, driverOptions, + localAddress, append(d._options, options.WithCustomLogger(connectionLog))..., ) if err != nil { connectionLog.Error(). Stringer("transportUrl", &transportUrl). Strs("defaultUdpPort", driverOptions["defaultUdpPort"]). + Int("localPort", localPort). Msg("We couldn't create a transport instance for port") return nil, errors.Wrapf(err, "couldn't initialize transport configuration for given transport url %s", transportUrl.String()) } diff --git a/plc4go/internal/bacnetip/Framing.go b/plc4go/internal/bacnetip/Framing.go new file mode 100644 index 0000000000..de7124c8fb --- /dev/null +++ b/plc4go/internal/bacnetip/Framing.go @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package bacnetip + +import ( + "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model" +) + +// wrapAPDU encapsulates an APDU in the BACnet/IP NPDU + BVLC layers expected +// by MessageCodec.Send (which type-asserts to model.BVLC). Without this +// wrapping the codec panics on the cast and the request silently dies. +// +// expectingReply is set for confirmed requests; unconfirmed requests pass +// false. The NPDU is intentionally local-only (no DNET/SNET) because routed +// addressing happens at the Tag layer in Phase 6. +func wrapAPDU(apdu model.APDU, expectingReply bool) model.BVLC { + control := model.NewNPDUControl( + false, // messageTypeFieldPresent + false, // destinationSpecified + false, // sourceSpecified + expectingReply, + model.NPDUNetworkPriority_NORMAL_MESSAGE, + ) + npdu := model.NewNPDU( + 1, // protocolVersionNumber + control, + nil, // destinationNetworkAddress + nil, // destinationLength + nil, // destinationAddress + nil, // sourceNetworkAddress + nil, // sourceLength + nil, // sourceAddress + nil, // hopCount + nil, // nlm + apdu, + ) + return model.NewBVLCOriginalUnicastNPDU(npdu) +} diff --git a/plc4go/internal/bacnetip/Framing_test.go b/plc4go/internal/bacnetip/Framing_test.go new file mode 100644 index 0000000000..007f962890 --- /dev/null +++ b/plc4go/internal/bacnetip/Framing_test.go @@ -0,0 +1,114 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package bacnetip + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model" +) + +func newWhoIsAPDU(t *testing.T) readWriteModel.APDU { + t.Helper() + whoIs := readWriteModel.NewBACnetUnconfirmedServiceRequestWhoIs(nil, nil) + return readWriteModel.NewAPDUUnconfirmedRequest(whoIs) +} + +func TestWrapAPDU_ProducesBVLCOriginalUnicastNPDU(t *testing.T) { + bvlc := wrapAPDU(newWhoIsAPDU(t), false) + require.NotNil(t, bvlc) + // MessageCodec.Send type-asserts to BVLCOriginalUnicastNPDU on the + // send-path; wrapAPDU must produce exactly that type. + _, ok := bvlc.(readWriteModel.BVLCOriginalUnicastNPDU) + assert.True(t, ok, "wrapAPDU must return a BVLCOriginalUnicastNPDU, got %T", bvlc) +} + +func TestWrapAPDU_NPDUProtocolVersionIs1(t *testing.T) { + // BACnet stacks reject NPDUs with a wrong protocol version. Pin it + // to 1 (the only spec-valid value) so an accidental change shows up + // as a test failure rather than a wire-protocol incompatibility. + bvlc := wrapAPDU(newWhoIsAPDU(t), false).(readWriteModel.BVLCOriginalUnicastNPDU) + assert.Equal(t, uint8(1), bvlc.GetNpdu().GetProtocolVersionNumber()) +} + +func TestWrapAPDU_ExpectingReplyPropagatesToControl(t *testing.T) { + cases := []struct { + name string + expectingReply bool + }{ + {"confirmed-request-sets-flag", true}, + {"unconfirmed-broadcast-clears-flag", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + bvlc := wrapAPDU(newWhoIsAPDU(t), tc.expectingReply).(readWriteModel.BVLCOriginalUnicastNPDU) + control := bvlc.GetNpdu().GetControl() + assert.Equal(t, tc.expectingReply, control.GetExpectingReply(), + "NPDU control.expectingReply must reflect the wrapAPDU argument") + }) + } +} + +func TestWrapAPDU_LocalScope_NoRouting(t *testing.T) { + // We only support local (same-network) addressing in Phase 6's tag layer. + // The NPDU control fields for routing must all be off so bacpypes3/Niagara + // don't interpret the message as routed. + bvlc := wrapAPDU(newWhoIsAPDU(t), true).(readWriteModel.BVLCOriginalUnicastNPDU) + control := bvlc.GetNpdu().GetControl() + assert.False(t, control.GetMessageTypeFieldPresent()) + assert.False(t, control.GetDestinationSpecified()) + assert.False(t, control.GetSourceSpecified()) + // Routing fields (DNET/DLEN/DADR + SNET/SLEN/SADR + HopCount + NLM) must + // be nil — otherwise a peer treats it as a routed frame. + npdu := bvlc.GetNpdu() + assert.Nil(t, npdu.GetDestinationNetworkAddress()) + assert.Nil(t, npdu.GetDestinationLength()) + assert.Nil(t, npdu.GetDestinationAddress()) + assert.Nil(t, npdu.GetSourceNetworkAddress()) + assert.Nil(t, npdu.GetSourceLength()) + assert.Nil(t, npdu.GetSourceAddress()) + assert.Nil(t, npdu.GetHopCount()) + assert.Nil(t, npdu.GetNlm()) +} + +func TestWrapAPDU_PreservesAPDU(t *testing.T) { + apdu := newWhoIsAPDU(t) + bvlc := wrapAPDU(apdu, false).(readWriteModel.BVLCOriginalUnicastNPDU) + // Same APDU identity should be reachable through the wrapper — + // MessageCodec.Receive parses BVLC → NPDU → APDU and Reader/Writer + // match expectations by walking that chain. + assert.Equal(t, apdu, bvlc.GetNpdu().GetApdu()) +} + +func TestWrapAPDU_SerializesToValidBVLC(t *testing.T) { + // End-to-end sanity: the wrapped message round-trips through the + // model serializer. Catches accidental nil-required-field changes + // in wrapAPDU that would only show up at runtime under Send(). + bvlc := wrapAPDU(newWhoIsAPDU(t), false) + raw, err := bvlc.Serialize() + require.NoError(t, err, "wrapAPDU output must serialize") + // First byte is BVLC type 0x81; second is function 0x0a (OriginalUnicastNPDU). + require.GreaterOrEqual(t, len(raw), 4) + assert.Equal(t, byte(0x81), raw[0], "BVLC magic byte") + assert.Equal(t, byte(0x0a), raw[1], "BVLC function = OriginalUnicastNPDU") +} diff --git a/plc4go/internal/bacnetip/MessageCodec.go b/plc4go/internal/bacnetip/MessageCodec.go index 61e0ce1ea9..812e287eea 100644 --- a/plc4go/internal/bacnetip/MessageCodec.go +++ b/plc4go/internal/bacnetip/MessageCodec.go @@ -46,10 +46,26 @@ var ( func NewMessageCodec(transportInstance transports.TransportInstance, _options ...options.WithOption) *MessageCodec { codec := &MessageCodec{} - codec.DefaultCodec = _default.NewDefaultCodec(codec, transportInstance, append(_options, _default.WithCustomMessageHandler(codec.handleCustomMessage))...) + // Register a no-op (always-false) custom handler so the codec's receive + // loop keeps polling even when there are zero outstanding expectations. + // The default loop skips reading the transport when both `expectations` + // is empty AND `customMessageHandling` is nil — that would mean unsolicited + // COV notifications arrive in the kernel buffer but nobody drains them. + // Returning false here lets the default expectation-matching path run + // first, then falls through to defaultIncomingMessageChannel for the + // Connection's COV-notification poller. + codec.DefaultCodec = _default.NewDefaultCodec(codec, transportInstance, append(_options, _default.WithCustomMessageHandler(keepReceiveLoopActive))...) return codec } +// keepReceiveLoopActive is a no-op CustomMessageHandler. Its only purpose is +// to set m.customMessageHandling to non-nil so the codec's receive worker +// doesn't park when expectations drain to zero. Returning false defers all +// real handling to HandleMessages → defaultIncomingMessageChannel. +func keepReceiveLoopActive(_ context.Context, _ _default.DefaultCodecRequirements, _ spi.Message) bool { + return false +} + func (m *MessageCodec) GetCodec() spi.MessageCodec { return m } @@ -112,11 +128,3 @@ func (m *MessageCodec) Receive(ctx context.Context) (spi.Message, error) { return nil, nil } -func (m *MessageCodec) handleCustomMessage(ctx context.Context, _ _default.DefaultCodecRequirements, message spi.Message) bool { - // For now, we just put them in the incoming channel - select { - case m.GetDefaultIncomingMessageChannel() <- message: - case <-ctx.Done(): - } - return true -} diff --git a/plc4go/internal/bacnetip/MessageCodec_test.go b/plc4go/internal/bacnetip/MessageCodec_test.go new file mode 100644 index 0000000000..86b7a7395c --- /dev/null +++ b/plc4go/internal/bacnetip/MessageCodec_test.go @@ -0,0 +1,243 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package bacnetip + +import ( + "bufio" + "bytes" + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model" + "github.com/apache/plc4x/plc4go/spi" + _default "github.com/apache/plc4x/plc4go/spi/default" + "github.com/apache/plc4x/plc4go/spi/transports" +) + +// fakeTransportInstance is an in-memory transport tailored for codec-level +// tests: bytes pushed via PushReceived land directly in the read buffer (no +// channel pump), and bytes written by the codec land in writeBuffer. Lets us +// exercise Send/Receive without standing up a real socket. +type fakeTransportInstance struct { + mu sync.Mutex + readBuffer bytes.Buffer + writeBuffer bytes.Buffer + connected bool +} + +func newFakeTransport() *fakeTransportInstance { + return &fakeTransportInstance{connected: true} +} + +func (f *fakeTransportInstance) PushReceived(data []byte) { + f.mu.Lock() + defer f.mu.Unlock() + f.readBuffer.Write(data) +} + +func (f *fakeTransportInstance) WrittenBytes() []byte { + f.mu.Lock() + defer f.mu.Unlock() + return append([]byte(nil), f.writeBuffer.Bytes()...) +} + +func (f *fakeTransportInstance) Connect(_ context.Context) error { f.connected = true; return nil } +func (f *fakeTransportInstance) ConnectWithContext(_ context.Context) error { return f.Connect(nil) } +func (f *fakeTransportInstance) Close() error { f.connected = false; return nil } +func (f *fakeTransportInstance) IsConnected() bool { return f.connected } +func (f *fakeTransportInstance) String() string { return "fake" } +func (f *fakeTransportInstance) Reset() {} + +func (f *fakeTransportInstance) GetNumBytesAvailableInBuffer() (uint32, error) { + f.mu.Lock() + defer f.mu.Unlock() + return uint32(f.readBuffer.Len()), nil +} + +func (f *fakeTransportInstance) PeekReadableBytes(_ context.Context, n uint32) ([]byte, error) { + f.mu.Lock() + defer f.mu.Unlock() + if uint32(f.readBuffer.Len()) < n { + return nil, fmt.Errorf("not enough bytes: want %d, have %d", n, f.readBuffer.Len()) + } + buf := f.readBuffer.Bytes() + return append([]byte(nil), buf[:n]...), nil +} + +func (f *fakeTransportInstance) Read(_ context.Context, n uint32) ([]byte, error) { + f.mu.Lock() + defer f.mu.Unlock() + if uint32(f.readBuffer.Len()) < n { + return nil, fmt.Errorf("not enough bytes") + } + out := make([]byte, n) + _, _ = f.readBuffer.Read(out) + return out, nil +} + +func (f *fakeTransportInstance) Write(_ context.Context, data []byte) error { + f.mu.Lock() + defer f.mu.Unlock() + f.writeBuffer.Write(data) + return nil +} + +func (f *fakeTransportInstance) FillBuffer(ctx context.Context, until func(pos uint, currentByte byte, reader transports.ExtendedReader) (keepGoing bool)) error { + // Implemented for completeness; the codec doesn't drive this path in + // these tests. Mirrors the contract of DefaultBufferedTransportInstance. + nBytes := uint32(1) + for ctx.Err() == nil { + b, err := f.PeekReadableBytes(ctx, nBytes) + if err != nil { + return err + } + if !until(uint(nBytes-1), b[len(b)-1], bufio.NewReader(bytes.NewReader(b))) { + return nil + } + nBytes++ + } + return ctx.Err() +} + +var _ transports.TransportInstance = (*fakeTransportInstance)(nil) + +func newTestCodec(t *testing.T) (*MessageCodec, *fakeTransportInstance) { + t.Helper() + ti := newFakeTransport() + codec := NewMessageCodec(ti) + t.Cleanup(func() { _ = codec.Disconnect() }) + return codec, ti +} + +// makeWhoIsBVLC builds a minimal valid BVLC frame (BVLCOriginalUnicastNPDU +// wrapping an unconfirmed WhoIs APDU). Used as a payload for both Send and +// Receive round-trip tests because it serializes to a small, deterministic +// byte sequence. +func makeWhoIsBVLC(t *testing.T) readWriteModel.BVLC { + t.Helper() + whoIs := readWriteModel.NewBACnetUnconfirmedServiceRequestWhoIs(nil, nil) + apdu := readWriteModel.NewAPDUUnconfirmedRequest(whoIs) + return wrapAPDU(apdu, false) +} + +func TestMessageCodec_Send_SerializesBVLCToTransport(t *testing.T) { + codec, ti := newTestCodec(t) + bvlc := makeWhoIsBVLC(t) + expectedBytes, err := bvlc.Serialize() + require.NoError(t, err) + + require.NoError(t, codec.Send(context.Background(), "test", bvlc)) + + assert.Equal(t, expectedBytes, ti.WrittenBytes()) +} + +func TestMessageCodec_Send_RejectsNonBVLC(t *testing.T) { + codec, _ := newTestCodec(t) + apdu := readWriteModel.NewAPDUUnconfirmedRequest( + readWriteModel.NewBACnetUnconfirmedServiceRequestWhoIs(nil, nil), + ) + assert.Panics(t, func() { + _ = codec.Send(context.Background(), "test", apdu) + }, "Send must panic on non-BVLC payload — callers must wrap with wrapAPDU first") +} + +func TestMessageCodec_Receive_ParsesBVLCFromTransport(t *testing.T) { + codec, ti := newTestCodec(t) + bvlc := makeWhoIsBVLC(t) + raw, err := bvlc.Serialize() + require.NoError(t, err) + ti.PushReceived(raw) + + msg, err := codec.Receive(context.Background()) + require.NoError(t, err) + require.NotNil(t, msg, "Receive should return the parsed BVLC") + + parsed, ok := msg.(readWriteModel.BVLC) + require.True(t, ok, "Receive should return a BVLC, got %T", msg) + assert.Equal(t, bvlc.GetBvlcFunction(), parsed.GetBvlcFunction()) +} + +func TestMessageCodec_Receive_NotEnoughBytesReturnsNil(t *testing.T) { + codec, ti := newTestCodec(t) + // 2 bytes is below the 4-byte minimum BVLC header — Receive must + // short-circuit with (nil, nil), not block or err. + ti.PushReceived([]byte{0x81, 0x0a}) + msg, err := codec.Receive(context.Background()) + assert.NoError(t, err) + assert.Nil(t, msg, "<4 buffered bytes should not yield a message") +} + +func TestMessageCodec_Receive_PartialPacketReturnsNil(t *testing.T) { + codec, ti := newTestCodec(t) + // BVLC header claims packet-size 100 (0x0064), but we only feed 4 bytes. + ti.PushReceived([]byte{0x81, 0x0a, 0x00, 0x64}) + msg, err := codec.Receive(context.Background()) + assert.NoError(t, err) + assert.Nil(t, msg, "buffer < declared packet-size should not yield a message") +} + +func TestMessageCodec_ExpectationsMatchUnsolicitedMessages(t *testing.T) { + // Regression for the always-true handleCustomMessage bug: + // expectations must still match incoming messages even when a + // customMessageHandling (keepReceiveLoopActive) is registered, because + // our handler returns false and lets HandleMessages run. + codec, ti := newTestCodec(t) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + got := make(chan spi.Message, 1) + codec.Expect(ctx, "test-expect", + func(_ spi.Message) bool { return true }, + func(msg spi.Message) error { got <- msg; return nil }, + func(err error) error { return err }, + ) + + require.NoError(t, codec.Connect(ctx)) + raw, err := makeWhoIsBVLC(t).Serialize() + require.NoError(t, err) + ti.PushReceived(raw) + + select { + case msg := <-got: + _, ok := msg.(readWriteModel.BVLC) + assert.True(t, ok, "expectation should receive the parsed BVLC, got %T", msg) + case <-time.After(3 * time.Second): + t.Fatal("expectation handler never fired — keepReceiveLoopActive may be swallowing messages") + } +} + +func TestKeepReceiveLoopActive_AlwaysReturnsFalse(t *testing.T) { + // keepReceiveLoopActive's only job is to keep the receive worker awake + // when there are zero expectations. Its return value of false is what + // lets HandleMessages → defaultIncomingMessageChannel still run. If + // somebody flips this to true, every Read/Write/Subscribe times out + // (the original bug we hit during integration testing). + bvlc := makeWhoIsBVLC(t) + assert.False(t, + keepReceiveLoopActive(context.Background(), (_default.DefaultCodecRequirements)(nil), bvlc), + "keepReceiveLoopActive must return false or expectation matching breaks") +} diff --git a/plc4go/internal/bacnetip/Reader.go b/plc4go/internal/bacnetip/Reader.go index 4709fdef83..32bf49537b 100644 --- a/plc4go/internal/bacnetip/Reader.go +++ b/plc4go/internal/bacnetip/Reader.go @@ -139,7 +139,7 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) context.AfterFunc(transactionContext, cancel) // Send the over the wire m.log.Trace().Msg("Send ") - if err := m.messageCodec.SendRequest(ctx, "read", apdu, func(message spi.Message) bool { + if err := m.messageCodec.SendRequest(ctx, "read", wrapAPDU(apdu, true), func(message spi.Message) bool { bvlc, ok := message.(readWriteModel.BVLC) if !ok { m.log.Debug().Type("bvlc", bvlc).Msg("Received strange type") diff --git a/plc4go/internal/bacnetip/Subscriber.go b/plc4go/internal/bacnetip/Subscriber.go index d6fc9a2c0e..3a447bbdcf 100644 --- a/plc4go/internal/bacnetip/Subscriber.go +++ b/plc4go/internal/bacnetip/Subscriber.go @@ -191,7 +191,7 @@ func (m *Subscriber) sendSubscribeCOV(ctx context.Context, handle *SubscriptionH // SimpleAck back through the codec. Phase 5 will replace this with a real // transaction-manager-backed retry loop honoring ApduTimeoutMs/Retries. done := make(chan apiModel.PlcResponseCode, 1) - err := m.connection.messageCodec.SendRequest(ctx, "subscribe-cov", apdu, + err := m.connection.messageCodec.SendRequest(ctx, "subscribe-cov", wrapAPDU(apdu, true), func(message spi.Message) bool { return m.acceptsResponse(message, invokeId) }, @@ -306,10 +306,19 @@ func (m *Subscriber) dispatchNotification(handle *SubscriptionHandle, listOfValu break } } + // A BACnetConstructedDataElement holds the value in exactly one of three + // fields depending on how the publisher framed it: ApplicationTag (e.g. + // Real for AnalogInput), ConstructedData (nested typed value), or + // ContextTag (context-specific encoding). Pick whichever is populated. element := picked.GetPropertyValue() - if element != nil { + switch { + case element == nil: + values[handle.tagName] = spiValues.NewPlcNULL() + case element.GetApplicationTag() != nil: values[handle.tagName] = appTagToPlcValue(element.GetApplicationTag()) - } else { + case element.GetConstructedData() != nil: + values[handle.tagName] = constructedDataToPlcValue(element.GetConstructedData()) + default: values[handle.tagName] = spiValues.NewPlcNULL() } } diff --git a/plc4go/internal/bacnetip/Subscriber_test.go b/plc4go/internal/bacnetip/Subscriber_test.go index f82d2b9527..d99b5a1ba4 100644 --- a/plc4go/internal/bacnetip/Subscriber_test.go +++ b/plc4go/internal/bacnetip/Subscriber_test.go @@ -212,4 +212,80 @@ func TestDispatchNotification_NilTag_DoesNotPanic(t *testing.T) { s.dispatchNotification(handle, empty) } +// buildCOVNotificationWithConstructedData builds an UnconfirmedCOVNotification +// where the PropertyValue's element holds a ConstructedData (not an +// ApplicationTag). bacpypes3 wraps the present-value this way for some object +// types; the dispatch code's element.GetApplicationTag() branch would return +// nil and previously yielded PlcNULL, dropping the value. The Subscriber must +// fall through to constructedDataToPlcValue. +func buildCOVNotificationWithConstructedData(processId uint32, deviceId uint32, objType readWriteModel.BACnetObjectType, instance uint32, presentValue float32) readWriteModel.BACnetUnconfirmedServiceRequestUnconfirmedCOVNotification { + subscriberProcessTag := readWriteModel.CreateBACnetContextTagUnsignedInteger(0, uint(processId)) + initiatingDeviceTag := readWriteModel.CreateBACnetContextTagObjectIdentifier(1, uint16(readWriteModel.BACnetObjectType_DEVICE), deviceId) + monitoredObjectTag := readWriteModel.CreateBACnetContextTagObjectIdentifier(2, uint16(objType), instance) + lifetimeTag := readWriteModel.CreateBACnetContextTagUnsignedInteger(3, 0) + + // Build a ConstructedDataUnspecified holding the Real value, then wrap + // that in a parent element where ApplicationTag is nil and + // ConstructedData is the inner ConstructedDataUnspecified. + innerHeader := readWriteModel.CreateBACnetTagHeaderBalanced(true, 2, 0) + innerElement := readWriteModel.NewBACnetConstructedDataElement( + innerHeader, + readWriteModel.CreateBACnetApplicationTagReal(presentValue), + nil, + nil, + ) + innerCD := readWriteModel.NewBACnetConstructedDataUnspecified( + readWriteModel.CreateBACnetOpeningTag(2), + innerHeader, + readWriteModel.CreateBACnetClosingTag(2), + nil, + []readWriteModel.BACnetConstructedDataElement{innerElement}, + ) + outerHeader := readWriteModel.CreateBACnetTagHeaderBalanced(true, 2, 0) + outerElement := readWriteModel.NewBACnetConstructedDataElement(outerHeader, nil, nil, innerCD) + propIdTag := readWriteModel.CreateBACnetPropertyIdentifierTagged(0, uint32(readWriteModel.BACnetPropertyIdentifier_PRESENT_VALUE)) + propVal := readWriteModel.NewBACnetPropertyValue(propIdTag, nil, outerElement, nil) + values := readWriteModel.NewBACnetPropertyValues( + readWriteModel.CreateBACnetOpeningTag(4), + []readWriteModel.BACnetPropertyValue{propVal}, + readWriteModel.CreateBACnetClosingTag(4), + ) + return readWriteModel.NewBACnetUnconfirmedServiceRequestUnconfirmedCOVNotification( + subscriberProcessTag, initiatingDeviceTag, monitoredObjectTag, lifetimeTag, values, + ) +} + +func TestDispatchNotification_ConstructedDataBranch(t *testing.T) { + // Regression: pre-fix Subscriber.dispatchNotification only inspected the + // element's ApplicationTag field. When bacpypes3 framed a COV's + // PresentValue as nested ConstructedData, the value went silently to + // PlcNULL and the consumer never saw the actual reading. + s := newTestSubscriber(t) + s.log = zerolog.Nop() + + objType := readWriteModel.BACnetObjectType_ANALOG_INPUT + tag := &plcTag{ObjectId: objectId{ObjectIdType: &objType, ObjectIdInstance: 2}} + handle := NewSubscriptionHandle(s, "cd-tag", tag, apiModel.SubscriptionChangeOfState, 0) + handle.subscriberProcessId = 42 + s.storeHandle(handle) + + capture := &captureConsumer{} + reg := spiModel.NewDefaultPlcConsumerRegistration(s, capture.consume, handle.DefaultPlcSubscriptionHandle) + s.consumers[reg.(*spiModel.DefaultPlcConsumerRegistration)] = capture.consume + + req := buildCOVNotificationWithConstructedData(42, 1234, objType, 2, 17.5) + s.HandleUnconfirmedCOVNotification(req) + + require.Len(t, capture.events, 1) + val := capture.events[0].GetValue("cd-tag") + require.NotNil(t, val) + // The value must be present (not PlcNULL) — that's the regression we're + // guarding against. Exact float comparison is best-effort: the nested + // ConstructedDataUnspecified path goes through elementsToPlcValue which + // unwraps the single element back to its ApplicationTag's value. + assert.NotEqual(t, apiValues.NULL, val.GetPlcValueType(), + "ConstructedData branch should yield a real value, not PlcNULL") + assert.InDelta(t, 17.5, val.GetFloat32(), 1e-3) +} + var _ = time.Second // keep import alive across phases diff --git a/plc4go/internal/bacnetip/Tag.go b/plc4go/internal/bacnetip/Tag.go index 1a19224f01..24ec6e58c3 100644 --- a/plc4go/internal/bacnetip/Tag.go +++ b/plc4go/internal/bacnetip/Tag.go @@ -24,6 +24,7 @@ import ( "encoding/binary" "fmt" "strings" + "time" apiModel "github.com/apache/plc4x/plc4go/pkg/api/model" apiValues "github.com/apache/plc4x/plc4go/pkg/api/values" @@ -44,6 +45,21 @@ type plcTag struct { Properties []property } +// GetPlcSubscriptionType lets a plcTag participate in subscription requests +// without a separate tag type. The BACnet driver always issues SubscribeCOV +// regardless of the api/model.PlcSubscriptionType, so we report the most +// permissive value (ChangeOfState) — the caller's selection only affects +// plc4go-side filtering, not what we put on the wire. +func (m plcTag) GetPlcSubscriptionType() apiModel.PlcSubscriptionType { + return apiModel.SubscriptionChangeOfState +} + +// GetDuration is part of PlcSubscriptionTag. We report 0 (no per-tag cycle); +// the COV refresh interval comes from Configuration.CovLifetimeSeconds. +func (m plcTag) GetDuration() time.Duration { + return 0 +} + type objectId struct { // ObjectIdType defines the object type. If not defined ObjectIdTypeProprietary must be defined ObjectIdType *readWriteModel.BACnetObjectType diff --git a/plc4go/internal/bacnetip/ValueDecoder.go b/plc4go/internal/bacnetip/ValueDecoder.go index 7cf708aecd..94124f3e58 100644 --- a/plc4go/internal/bacnetip/ValueDecoder.go +++ b/plc4go/internal/bacnetip/ValueDecoder.go @@ -123,6 +123,9 @@ func constructedDataToPlcValue(data model.BACnetConstructedData) apiValues.PlcVa if tag, ok := v.(model.BACnetApplicationTag); ok { return appTagToPlcValue(tag) } + if pv, ok := taggedEnumToPlcValue(v); ok { + return pv + } // Composite types: stringify until we add typed mappings (DateTime, // ObjectReference, BACnetTimeStamp, ...). return spiValues.NewPlcSTRING(fmt.Sprintf("%v", v)) @@ -130,6 +133,36 @@ func constructedDataToPlcValue(data model.BACnetConstructedData) apiValues.PlcVa return spiValues.NewPlcSTRING(fmt.Sprintf("%T:%v", data, data)) } +// taggedEnumToPlcValue handles BACnet's *Tagged enum wrappers (BACnetBinaryPV, +// BACnetReliability, BACnetEventState, ...) returned by GetActualValue on +// typed ConstructedData subtypes. These don't satisfy BACnetApplicationTag +// but expose a GetValue() method whose return is the unwrapped enum integer. +// Surfacing them as PlcUDINT keeps the API consistent with how Enumerated +// application tags are decoded (the property identifier on the request side +// determines the enum schema). +func taggedEnumToPlcValue(v any) (apiValues.PlcValue, bool) { + rv := reflect.ValueOf(v) + if !rv.IsValid() { + return nil, false + } + method := rv.MethodByName("GetValue") + if !method.IsValid() || method.Type().NumIn() != 0 || method.Type().NumOut() != 1 { + return nil, false + } + out := method.Call(nil)[0] + switch out.Kind() { + case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64: + return spiValues.NewPlcUDINT(uint32(out.Uint())), true + case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64: + return spiValues.NewPlcLINT(out.Int()), true + case reflect.Bool: + return spiValues.NewPlcBOOL(out.Bool()), true + case reflect.String: + return spiValues.NewPlcSTRING(out.String()), true + } + return nil, false +} + // elementsToPlcValue collapses a list of BACnetConstructedDataElement to a // PlcValue. A single-element list yields a scalar; multi-element yields PlcList. func elementsToPlcValue(elements []model.BACnetConstructedDataElement) apiValues.PlcValue { diff --git a/plc4go/internal/bacnetip/ValueDecoder_test.go b/plc4go/internal/bacnetip/ValueDecoder_test.go new file mode 100644 index 0000000000..d793034c06 --- /dev/null +++ b/plc4go/internal/bacnetip/ValueDecoder_test.go @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package bacnetip + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + apiValues "github.com/apache/plc4x/plc4go/pkg/api/values" + readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model" +) + +// taggedStub satisfies the `GetValue() T` shape that taggedEnumToPlcValue +// looks for via reflection. Used to exercise the kinds that don't have a +// real generated counterpart in this codebase (bool, signed, string). +type uintStub struct{ v uint8 } + +func (s uintStub) GetValue() uint8 { return s.v } + +type signedStub struct{ v int16 } + +func (s signedStub) GetValue() int16 { return s.v } + +type boolStub struct{ v bool } + +func (b boolStub) GetValue() bool { return b.v } + +type stringStub struct{ v string } + +func (s stringStub) GetValue() string { return s.v } + +type noGetValueStub struct{} + +func TestTaggedEnumToPlcValue_BACnetBinaryPVTagged(t *testing.T) { + // The canonical case we hit during integration testing. INACTIVE→0, + // ACTIVE→1 — both must surface as PlcUDINT, matching how a plain + // Enumerated application tag is decoded. + cases := []struct { + name string + pv readWriteModel.BACnetBinaryPV + want uint32 + }{ + {"INACTIVE", readWriteModel.BACnetBinaryPV_INACTIVE, 0}, + {"ACTIVE", readWriteModel.BACnetBinaryPV_ACTIVE, 1}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + // Header values don't affect taggedEnumToPlcValue — only GetValue() + // is consulted via reflection. + tagged := readWriteModel.NewBACnetBinaryPVTagged( + readWriteModel.NewBACnetTagHeader(9, 0, 1, nil, nil, nil, nil), + tc.pv, + ) + got, ok := taggedEnumToPlcValue(tagged) + require.True(t, ok, "BACnetBinaryPVTagged should be recognized") + require.NotNil(t, got) + assert.Equal(t, tc.want, got.GetUint32(), "PlcUDINT value mismatch") + }) + } +} + +func TestTaggedEnumToPlcValue_NilInput(t *testing.T) { + got, ok := taggedEnumToPlcValue(nil) + assert.False(t, ok, "nil input should return ok=false") + assert.Nil(t, got) +} + +func TestTaggedEnumToPlcValue_NoGetValueMethod(t *testing.T) { + // A struct without GetValue() must not match — it would otherwise + // silently mis-decode random types as PlcUDINT/PlcSTRING. + got, ok := taggedEnumToPlcValue(noGetValueStub{}) + assert.False(t, ok) + assert.Nil(t, got) +} + +func TestTaggedEnumToPlcValue_AllKinds(t *testing.T) { + // Cover each reflection-Kind branch in the switch so future refactors + // can't accidentally drop one. Uses stub types because the generated + // model only has uint-returning *Tagged types in our tag-set. + cases := []struct { + name string + in any + kind string + eq func(t *testing.T, v apiValues.PlcValue) + }{ + {"uint", uintStub{v: 42}, "udint", + func(t *testing.T, v apiValues.PlcValue) { assert.Equal(t, uint32(42), v.GetUint32()) }}, + {"int", signedStub{v: -7}, "lint", + func(t *testing.T, v apiValues.PlcValue) { assert.Equal(t, int64(-7), v.GetInt64()) }}, + {"bool", boolStub{v: true}, "bool", + func(t *testing.T, v apiValues.PlcValue) { assert.True(t, v.GetBool()) }}, + {"string", stringStub{v: "hello"}, "string", + func(t *testing.T, v apiValues.PlcValue) { assert.Equal(t, "hello", v.GetString()) }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := taggedEnumToPlcValue(tc.in) + require.True(t, ok, "kind %s should be recognized", tc.kind) + require.NotNil(t, got) + tc.eq(t, got) + }) + } +} + +func TestTaggedEnumToPlcValue_RejectsUnsupportedKind(t *testing.T) { + // A GetValue that returns e.g. a slice or struct isn't a primitive enum + // — taggedEnumToPlcValue should report ok=false so the caller can fall + // through to the stringify fallback rather than silently returning nil. + type sliceStub struct{} + type sliceStubT = sliceStub + // Add the method via a wrapper since Go method sets don't allow inline. + got, ok := taggedEnumToPlcValue(complexReturnStub{}) + assert.False(t, ok, "non-primitive return kind should reject") + assert.Nil(t, got) +} + +type complexReturnStub struct{} + +func (complexReturnStub) GetValue() []byte { return []byte{1, 2, 3} } diff --git a/plc4go/internal/bacnetip/ValueEncoder.go b/plc4go/internal/bacnetip/ValueEncoder.go index 38dd57ba1b..9da1c18904 100644 --- a/plc4go/internal/bacnetip/ValueEncoder.go +++ b/plc4go/internal/bacnetip/ValueEncoder.go @@ -40,6 +40,17 @@ func plcValueToApplicationTag(v apiValues.PlcValue, hint encodingHint) (model.BA case apiValues.NULL: return model.CreateBACnetApplicationTagNull(), nil case apiValues.BOOL: + // Binary objects' PRESENT_VALUE is Enumerated (INACTIVE=0, ACTIVE=1) + // on the wire — sending a Boolean application tag gets rejected with + // REJECT(INVALID_TAG). When the caller hints Enumerated, encode the + // bool as 0/1 Enumerated. + if hint == hintEnumerated { + var v32 uint32 + if v.GetBool() { + v32 = 1 + } + return model.CreateBACnetApplicationTagEnumerated(v32), nil + } return model.CreateBACnetApplicationTagBoolean(v.GetBool()), nil case apiValues.BYTE, apiValues.USINT, apiValues.UINT, apiValues.UDINT, apiValues.ULINT, apiValues.WORD, apiValues.DWORD, apiValues.LWORD: if hint == hintEnumerated { @@ -71,14 +82,20 @@ const ( hintEnumerated ) -// hintForProperty returns the encoding hint appropriate to a BACnet property -// identifier. PRESENT_VALUE on Binary*/Multistate* objects (and many discrete -// status fields) carries Enumerated; everything else defaults to none. -func hintForProperty(_ uint32) encodingHint { - // Today we don't second-guess the caller. WriteProperty callers should - // either pass a PlcUDINT and accept the default UnsignedInteger encoding, - // or use a PlcREAL/PlcBOOL/PlcSTRING that maps unambiguously. Per-property - // dispatch will be added once the integration tests expose a real device - // that rejects an UnsignedInteger where Enumerated is required. +// hintForProperty returns the encoding hint appropriate to a BACnet +// (objectType, propertyIdentifier) pair. PRESENT_VALUE on Binary* objects +// carries Enumerated (INACTIVE=0, ACTIVE=1); without the hint we would emit +// a Boolean application tag and bacpypes3/Niagara reject with INVALID_TAG. +func hintForProperty(objectType, propertyId uint32) encodingHint { + if propertyId != uint32(model.BACnetPropertyIdentifier_PRESENT_VALUE) { + return hintNone + } + switch model.BACnetObjectType(objectType) { + case model.BACnetObjectType_BINARY_INPUT, + model.BACnetObjectType_BINARY_OUTPUT, + model.BACnetObjectType_BINARY_VALUE, + model.BACnetObjectType_BINARY_LIGHTING_OUTPUT: + return hintEnumerated + } return hintNone } diff --git a/plc4go/internal/bacnetip/ValueHandler.go b/plc4go/internal/bacnetip/ValueHandler.go index b323d7f4ba..c5d912bca7 100644 --- a/plc4go/internal/bacnetip/ValueHandler.go +++ b/plc4go/internal/bacnetip/ValueHandler.go @@ -20,13 +20,69 @@ package bacnetip import ( + apiModel "github.com/apache/plc4x/plc4go/pkg/api/model" + apiValues "github.com/apache/plc4x/plc4go/pkg/api/values" + "github.com/apache/plc4x/plc4go/spi/errors" "github.com/apache/plc4x/plc4go/spi/values" ) -type ValueHandler struct { - values.DefaultValueHandler -} +// ValueHandler maps user-supplied Go values (the third arg to AddTagAddress +// in a write request) onto plc4go PlcValues. Because BACnet properties have +// many possible underlying types and our plcTag reports PlcValueType_Struct, +// the default handler's struct path bails out — we override NewPlcValue +// directly to dispatch on the user's Go type. +// +// We do NOT embed DefaultValueHandler. Embedded-method dispatch in Go is +// static: DefaultValueHandler.parseType would call DefaultValueHandler's own +// ParseStructType, not our override. Implementing the spi.PlcValueHandler +// interface (just NewPlcValue) directly keeps control of the dispatch. +type ValueHandler struct{} func NewValueHandler() ValueHandler { return ValueHandler{} } + +// NewPlcValue wraps a raw Go primitive into the matching PlcValue. The +// Writer's ValueEncoder later inverts this onto a BACnet ApplicationTag, so +// here we only care about preserving the type information losslessly. +func (h ValueHandler) NewPlcValue(_ apiModel.PlcTag, value any) (apiValues.PlcValue, error) { + if v, ok := value.(apiValues.PlcValue); ok { + return v, nil + } + switch v := value.(type) { + case nil: + return values.NewPlcNULL(), nil + case bool: + return values.NewPlcBOOL(v), nil + case float32: + return values.NewPlcREAL(v), nil + case float64: + return values.NewPlcLREAL(v), nil + case int8: + return values.NewPlcSINT(v), nil + case int16: + return values.NewPlcINT(v), nil + case int32: + return values.NewPlcDINT(v), nil + case int64: + return values.NewPlcLINT(v), nil + case int: + return values.NewPlcLINT(int64(v)), nil + case uint8: + return values.NewPlcUSINT(v), nil + case uint16: + return values.NewPlcUINT(v), nil + case uint32: + return values.NewPlcUDINT(v), nil + case uint64: + return values.NewPlcULINT(v), nil + case uint: + return values.NewPlcULINT(uint64(v)), nil + case string: + return values.NewPlcSTRING(v), nil + case []byte: + return values.NewPlcRawByteArray(v), nil + default: + return nil, errors.Errorf("BACnet value handler can't encode Go type %T", value) + } +} diff --git a/plc4go/internal/bacnetip/Writer.go b/plc4go/internal/bacnetip/Writer.go index 95efb0d491..98c5cb701f 100644 --- a/plc4go/internal/bacnetip/Writer.go +++ b/plc4go/internal/bacnetip/Writer.go @@ -107,7 +107,7 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques ctx, cancel := context.WithCancel(ctx) context.AfterFunc(transactionContext, cancel) - err := m.messageCodec.SendRequest(ctx, "write", apdu, func(message spi.Message) bool { + err := m.messageCodec.SendRequest(ctx, "write", wrapAPDU(apdu, true), func(message spi.Message) bool { return m.acceptsResponse(message, invokeId) }, func(message spi.Message) error { bvlc := message.(readWriteModel.BVLC) @@ -148,8 +148,15 @@ func (m *Writer) buildServiceRequest(writeRequest apiModel.PlcWriteRequest) (rea // Multi: collapse all (tag, properties...) into a WritePropertyMultiple // with one BACnetWriteAccessSpecification per tag and one - // BACnetPropertyWriteDefinition per (tag, property) pair. WritePriority is - // taken from the same nil/!nil dance as WriteProperty. + // BACnetPropertyWriteDefinition per (tag, property) pair. + // + // PropertyWriteDefinition field tags (per BACnet spec): + // [0] propertyIdentifier + // [1] arrayIndex (OPTIONAL) + // [2] propertyValue (constructed — opening/closing tag 2) + // [3] priority (OPTIONAL) + // These differ from single WriteProperty (1/2/3/4), so the propVal wrapper + // also uses opening/closing tag 2 instead of 3. var specs []readWriteModel.BACnetWriteAccessSpecification for _, tagName := range tagNames { tag, ok := writeRequest.GetTag(tagName).(BacNetPlcTag) @@ -160,15 +167,15 @@ func (m *Writer) buildServiceRequest(writeRequest apiModel.PlcWriteRequest) (rea var defs []readWriteModel.BACnetPropertyWriteDefinition for _, prop := range tag.GetProperties() { plcValue := writeRequest.GetValue(tagName) - appTag, err := plcValueToApplicationTag(plcValue.(apiValues.PlcValue), hintForProperty(prop.getId())) + appTag, err := plcValueToApplicationTag(plcValue.(apiValues.PlcValue), hintForProperty(uint32(tag.GetObjectId().getId()), prop.getId())) if err != nil { return nil, errors.Wrapf(err, "tag %s property %s", tagName, prop.String()) } - cd := constructedDataFromAppTag(appTag) - propId := readWriteModel.CreateBACnetPropertyIdentifierTagged(2, prop.getId()) + cd := constructedDataFromAppTag(appTag, 2) + propId := readWriteModel.CreateBACnetPropertyIdentifierTagged(0, prop.getId()) var arrayIndex readWriteModel.BACnetContextTagUnsignedInteger if prop.ArrayIndex != nil { - arrayIndex = readWriteModel.CreateBACnetContextTagUnsignedInteger(3, *prop.ArrayIndex) + arrayIndex = readWriteModel.CreateBACnetContextTagUnsignedInteger(1, *prop.ArrayIndex) } defs = append(defs, readWriteModel.NewBACnetPropertyWriteDefinition(propId, arrayIndex, cd, nil)) } @@ -187,7 +194,7 @@ func (m *Writer) buildSingleWriteProperty(tag BacNetPlcTag, plcValue apiValues.P return nil, errors.New("nil PlcValue") } prop := tag.GetProperties()[0] - appTag, err := plcValueToApplicationTag(plcValue, hintForProperty(prop.getId())) + appTag, err := plcValueToApplicationTag(plcValue, hintForProperty(uint32(tag.GetObjectId().getId()), prop.getId())) if err != nil { return nil, err } @@ -197,20 +204,21 @@ func (m *Writer) buildSingleWriteProperty(tag BacNetPlcTag, plcValue apiValues.P if prop.ArrayIndex != nil { arrayIndex = readWriteModel.CreateBACnetContextTagUnsignedInteger(2, *prop.ArrayIndex) } - cd := constructedDataFromAppTag(appTag) + cd := constructedDataFromAppTag(appTag, 3) return readWriteModel.NewBACnetConfirmedServiceRequestWriteProperty(0, objectIdTag, propId, arrayIndex, cd, nil), nil } // constructedDataFromAppTag wraps a single ApplicationTag into a generic -// ConstructedDataUnspecified payload (opening tag 3, one element, closing 3) so -// the wire-format builder can serialize it without per-property typed types. -func constructedDataFromAppTag(tag readWriteModel.BACnetApplicationTag) readWriteModel.BACnetConstructedData { - header := readWriteModel.CreateBACnetTagHeaderBalanced(true, 3, 0) +// ConstructedDataUnspecified payload using the supplied context tag number +// for the opening/closing brackets. Single WriteProperty uses tag 3, +// WritePropertyMultiple's PropertyWriteDefinition uses tag 2. +func constructedDataFromAppTag(tag readWriteModel.BACnetApplicationTag, tagNumber uint8) readWriteModel.BACnetConstructedData { + header := readWriteModel.CreateBACnetTagHeaderBalanced(true, tagNumber, 0) element := readWriteModel.NewBACnetConstructedDataElement(header, tag, nil, nil) return readWriteModel.NewBACnetConstructedDataUnspecified( - readWriteModel.CreateBACnetOpeningTag(3), + readWriteModel.CreateBACnetOpeningTag(tagNumber), header, - readWriteModel.CreateBACnetClosingTag(3), + readWriteModel.CreateBACnetClosingTag(tagNumber), nil, []readWriteModel.BACnetConstructedDataElement{element}, ) diff --git a/plc4go/internal/bacnetip/Writer_test.go b/plc4go/internal/bacnetip/Writer_test.go index 9f1bc2342f..c77c0d48f4 100644 --- a/plc4go/internal/bacnetip/Writer_test.go +++ b/plc4go/internal/bacnetip/Writer_test.go @@ -227,3 +227,112 @@ func TestToPlcWriteResponse_Reject(t *testing.T) { resp := writer.toPlcWriteResponse(reject, req) assert.Equal(t, apiModel.PlcResponseCode_INVALID_DATA, resp.GetResponseCode("av")) } + +// ── WritePropertyMultiple wire-format regression guards ──────────────────── + +// TestBuildServiceRequest_WPM_ContextTagNumbers locks down the BACnet-spec +// context tag numbering inside each BACnetPropertyWriteDefinition: +// +// [0] propertyIdentifier +// [1] arrayIndex (optional) +// [2] propertyValue (constructed) +// [3] priority (optional) +// +// Earlier versions of buildServiceRequest used (2, 3, opening/closing-3), which +// bacpypes3 silently rejected with REJECT(INVALID_TAG). Re-introducing those +// values must fail this test. +func TestBuildServiceRequest_WPM_ContextTagNumbers(t *testing.T) { + writer := newTestWriter(t) + req := writeRequestFor(t, []writeTagSpec{ + { + name: "av2", + tag: makeTag(readWriteModel.BACnetObjectType_ANALOG_VALUE, 2, readWriteModel.BACnetPropertyIdentifier_PRESENT_VALUE), + value: spiValues.NewPlcREAL(11.25), + }, + { + name: "av3", + tag: makeTag(readWriteModel.BACnetObjectType_ANALOG_VALUE, 3, readWriteModel.BACnetPropertyIdentifier_PRESENT_VALUE), + value: spiValues.NewPlcREAL(22.5), + }, + }) + got, err := writer.buildServiceRequest(req) + require.NoError(t, err) + wpm, ok := got.(readWriteModel.BACnetConfirmedServiceRequestWritePropertyMultiple) + require.True(t, ok) + require.Len(t, wpm.GetData(), 2) + + for i, spec := range wpm.GetData() { + t.Logf("spec[%d]: object=%v", i, spec.GetObjectIdentifier().GetPayload()) + // Each WAS opens/closes the listOfProperties with context tag 1. + assert.Equal(t, uint8(1), spec.GetOpeningTag().GetHeader().GetActualTagNumber(), + "WAS opening tag must be context 1") + assert.Equal(t, uint8(1), spec.GetClosingTag().GetHeader().GetActualTagNumber(), + "WAS closing tag must be context 1") + require.NotEmpty(t, spec.GetListOfPropertyWriteDefinition()) + def := spec.GetListOfPropertyWriteDefinition()[0] + // PropertyWriteDefinition fields: + // propertyIdentifier [0] + assert.Equal(t, uint8(0), def.GetPropertyIdentifier().GetHeader().GetActualTagNumber(), + "propertyIdentifier must be context tag 0 (not 2 — pre-fix value)") + // propertyValue's ConstructedDataUnspecified uses opening/closing tag 2. + cd, ok := def.GetPropertyValue().(readWriteModel.BACnetConstructedDataUnspecified) + require.True(t, ok, "propertyValue should be ConstructedDataUnspecified, got %T", def.GetPropertyValue()) + assert.Equal(t, uint8(2), cd.GetOpeningTag().GetHeader().GetActualTagNumber(), + "propertyValue opening tag must be context 2 (not 3 — pre-fix value)") + assert.Equal(t, uint8(2), cd.GetClosingTag().GetHeader().GetActualTagNumber(), + "propertyValue closing tag must be context 2 (not 3 — pre-fix value)") + } +} + +// TestBuildServiceRequest_WPM_SerializesAndRoundTrips confirms the WPM +// request serializes to bytes and can be re-parsed by the generated model. +// If the wire format ever drifts so badly that it can't even round-trip, +// this catches it before integration tests see it. +func TestBuildServiceRequest_WPM_SerializesAndRoundTrips(t *testing.T) { + writer := newTestWriter(t) + req := writeRequestFor(t, []writeTagSpec{ + { + name: "av2", + tag: makeTag(readWriteModel.BACnetObjectType_ANALOG_VALUE, 2, readWriteModel.BACnetPropertyIdentifier_PRESENT_VALUE), + value: spiValues.NewPlcREAL(11.25), + }, + { + name: "av3", + tag: makeTag(readWriteModel.BACnetObjectType_ANALOG_VALUE, 3, readWriteModel.BACnetPropertyIdentifier_PRESENT_VALUE), + value: spiValues.NewPlcREAL(22.5), + }, + }) + got, err := writer.buildServiceRequest(req) + require.NoError(t, err) + raw, err := got.Serialize() + require.NoError(t, err) + require.NotEmpty(t, raw) +} + +// TestBuildServiceRequest_SingleWriteProperty_ContextTagNumbers locks the +// single-write encoding so refactors that share helpers between single and +// multi paths don't accidentally swap the constants. WriteProperty uses +// propId [1], arrayIndex [2], propVal {opening/closing 3}. +func TestBuildServiceRequest_SingleWriteProperty_ContextTagNumbers(t *testing.T) { + writer := newTestWriter(t) + req := writeRequestFor(t, []writeTagSpec{ + { + name: "av", + tag: makeTag(readWriteModel.BACnetObjectType_ANALOG_VALUE, 5, readWriteModel.BACnetPropertyIdentifier_PRESENT_VALUE), + value: spiValues.NewPlcREAL(7.0), + }, + }) + got, err := writer.buildServiceRequest(req) + require.NoError(t, err) + wp, ok := got.(readWriteModel.BACnetConfirmedServiceRequestWriteProperty) + require.True(t, ok) + + assert.Equal(t, uint8(1), wp.GetPropertyIdentifier().GetHeader().GetActualTagNumber(), + "single-write propertyIdentifier must be context tag 1") + cd, ok := wp.GetPropertyValue().(readWriteModel.BACnetConstructedDataUnspecified) + require.True(t, ok) + assert.Equal(t, uint8(3), cd.GetOpeningTag().GetHeader().GetActualTagNumber(), + "single-write propertyValue opening tag must be context 3") + assert.Equal(t, uint8(3), cd.GetClosingTag().GetHeader().GetActualTagNumber(), + "single-write propertyValue closing tag must be context 3") +} diff --git a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go index 8d6fb55e83..af66276988 100644 --- a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go +++ b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go @@ -39,7 +39,7 @@ func NewDefaultPlcUnsubscriptionRequestBuilder() *DefaultPlcUnsubscriptionReques } func (d *DefaultPlcUnsubscriptionRequestBuilder) AddHandles(subscriptionHandles ...apiModel.PlcSubscriptionHandle) apiModel.PlcUnsubscriptionRequestBuilder { - subscriptionHandles = append(subscriptionHandles, subscriptionHandles...) + d.subscriptionHandles = append(d.subscriptionHandles, subscriptionHandles...) return d } diff --git a/plc4go/spi/transports/udp/TransportInstance.go b/plc4go/spi/transports/udp/TransportInstance.go index 4d6e4013d9..51703ffc83 100644 --- a/plc4go/spi/transports/udp/TransportInstance.go +++ b/plc4go/spi/transports/udp/TransportInstance.go @@ -110,16 +110,10 @@ func (m *TransportInstance) Connect(ctx context.Context) error { } } - // TODO: Start a worker that uses m.udpConn.ReadFromUDP() to fill a buffer - /* m.wg.Go(func() { - buf := make([]byte, 1024) - for { - rsize, raddr, err := m.udpConn.ReadFromUDP(buf) - if err != nil { - fmt.Printf("Got %d bytes from %v: %v", rsize, raddr, buf) - } - } - }()*/ + // Passive bufio.Reader over the UDP socket — same pattern the TCP + // transport uses. The codec's Receive worker drives reads through + // PeekReadableBytes/Read/FillBuffer with a deadline set from the request + // context, so we don't need a separate pump goroutine. m.reader = bufio.NewReader(m.udpConn) m.connected.Store(true) @@ -232,11 +226,16 @@ func (m *TransportInstance) Write(ctx context.Context, data []byte) error { } var num int var err error - if m.RemoteAddress == nil { - // TODO: usually this happens on the dial port... is there a better way to catch that? + // A connected UDP socket (obtained via net.DialUDP) rejects WriteToUDP with + // "use of WriteTo with pre-connected connection" — we have to use the plain + // Write() path instead. udpConn.RemoteAddr() is nil for ListenUDP sockets and + // the connected remote for DialUDP sockets, so that's the right discriminator. + if m.udpConn.RemoteAddr() != nil { num, err = m.udpConn.Write(data) - } else { + } else if m.RemoteAddress != nil { num, err = m.udpConn.WriteToUDP(data, m.RemoteAddress) + } else { + num, err = m.udpConn.Write(data) } if err != nil { return errors.Wrapf(err, "error writing (remote address: %s)", m.RemoteAddress) diff --git a/plc4go/tests/integration/bacnetip/Dockerfile b/plc4go/tests/integration/bacnetip/Dockerfile index f7a4ad9a14..ca1d6d8813 100644 --- a/plc4go/tests/integration/bacnetip/Dockerfile +++ b/plc4go/tests/integration/bacnetip/Dockerfile @@ -36,4 +36,6 @@ COPY device.py /sim/device.py # Standard BACnet/IP unicast + broadcast port. EXPOSE 47808/udp -CMD ["python", "device.py"] +# -u forces unbuffered stdout so `docker compose logs` shows lifecycle messages +# in real time instead of dumping them only on container exit. +CMD ["python", "-u", "device.py"] diff --git a/plc4go/tests/integration/bacnetip/Dockerfile.test b/plc4go/tests/integration/bacnetip/Dockerfile.test new file mode 100644 index 0000000000..339075a2d1 --- /dev/null +++ b/plc4go/tests/integration/bacnetip/Dockerfile.test @@ -0,0 +1,51 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +# Go test runner used by docker-compose to exercise the BACnet/IP integration +# tests in their own network namespace. Co-located with bacnet-device on a +# shared docker bridge network so both ends can bind UDP 47808 independently +# (the BACnet protocol assumes the well-known port on both sides; one shared +# host network namespace can't honour that). +# +# Build context expectation: plc4go/ (the module root). Run via +# docker-compose.yml in this directory. + +FROM golang:1.26-bookworm + +WORKDIR /src + +# Prime the module cache. Copying go.{mod,sum} and tools.{mod,sum} first means +# subsequent layers cache cleanly across edits to the source tree. +COPY go.mod go.sum tools.mod tools.sum ./ +RUN go mod download + +# Bring in the rest of the module. .dockerignore would let us skip /target, +# /node_modules etc.; not bothering yet because the build is a one-shot. +COPY . . + +# BACNET_IT gates the test suite (see integration_test.go::skipIfDisabled). +# BACNET_IT_HOST tells the tests where to find the simulator on the docker +# bridge — `bacnet-device` is the service name defined in docker-compose.yml. +ENV BACNET_IT=1 \ + BACNET_IT_HOST=bacnet-device + +# `depends_on: service_started` only proves the bacpypes3 container exists, not +# that the BACnet stack inside it has finished binding 47808. A 3-second nap +# absorbs the startup race; tighten this once we have a UDP healthcheck. +CMD ["sh", "-c", "sleep 3 && go test -tags integration -v -count=1 -timeout 120s ./tests/integration/bacnetip/..."] diff --git a/plc4go/tests/integration/bacnetip/README.md b/plc4go/tests/integration/bacnetip/README.md index b4c010e0a2..bb0e51606e 100644 --- a/plc4go/tests/integration/bacnetip/README.md +++ b/plc4go/tests/integration/bacnetip/README.md @@ -1,69 +1,83 @@ # BACnet/IP Integration Tests -End-to-end integration tests for the plc4go BACnet/IP driver against a -dockerized [bacpypes3](https://github.com/JoelBender/BACpypes3) virtual device. - -## Why this exists - -The unit tests in `internal/bacnetip/*_test.go` exercise the pure decoding / -encoding paths. They do not catch: - -- Real UDP socket behavior (port reuse, multicast, broadcast). -- Live transaction-manager retry loops. -- Real-device quirks (segmentation window negotiation, IAm timing). - -This suite spins a real BACnet device in a container and drives the plc4go -driver against it. It is **opt-in**: the default `make test` does not run it. +End-to-end integration tests for the plc4go BACnet/IP driver. The suite runs +the driver and a [bacpypes3](https://github.com/JoelBender/BACpypes3) virtual +device side-by-side in two docker containers on a shared bridge network — so +each end has its own network namespace and can bind UDP 47808 the way the +BACnet/IP spec assumes. + +## Why two containers + +BACnet/IP defines port 47808 as the well-known UDP port for *both* sides of a +conversation. Some flows (Discover via WhoIs broadcast, and Confirmed COV +notifications on spec-strict devices) only work if the driver can also bind +47808. With a single host network namespace, only one process can hold the +port, so the simulator + the driver collide. A docker bridge gives each +container its own namespace; the kernel routes broadcasts across the bridge, +so both binds can coexist. ## How to run ```sh # From plc4go/ directory: -docker compose -f tests/integration/bacnetip/docker-compose.yml up -d -BACNET_IT=1 go test -tags integration ./tests/integration/bacnetip/... -v -count=1 -docker compose -f tests/integration/bacnetip/docker-compose.yml down +make integration-bacnetip ``` -Or, once the Makefile target lands: +That target builds both images, brings them up, runs the test suite to +completion, tears the stack down, and propagates the test-runner exit code. + +For ad-hoc invocation: ```sh -make integration-bacnetip +docker compose -f tests/integration/bacnetip/docker-compose.yml up \ + --build --abort-on-container-exit --exit-code-from test-runner +docker compose -f tests/integration/bacnetip/docker-compose.yml down ``` +The compose project must be evaluated from the plc4go module root (which is +what `make integration-bacnetip` does) because the test-runner's build +context is the whole Go module. + ## What's in here -| File | Purpose | -|-----------------------|------------------------------------------------------------------| -| `Dockerfile` | Builds a `python:3.12-slim` image with `bacpypes3` preinstalled. | -| `device.py` | A bacpypes3 LocalDeviceObject with AV.0–4, BV.0–1, AI.0, MSV.0. | -| `docker-compose.yml` | Exposes UDP 47808 on the host. | -| `integration_test.go` | Test cases (Discover, Read, Write, COV, segmented response). | +| File | Purpose | +|-----------------------|------------------------------------------------------------------------| +| `Dockerfile` | Builds `python:3.12-slim` + `bacpypes3` for the simulated device. | +| `device.py` | A bacpypes3 LocalDeviceObject with AV.0–4, BV.0–1, AI.0, MSV.0. | +| `Dockerfile.test` | Builds `golang:1.26-bookworm` + the plc4go module sources. | +| `docker-compose.yml` | Wires both containers onto the `bacnet` bridge (172.30.0.0/24). | +| `integration_test.go` | Test cases (Discover, Read, Write+re-read, Subscribe). | + +The test-runner reads `BACNET_IT_HOST` (defaulted to the docker DNS name +`bacnet-device`) so the same suite can be pointed at any reachable simulator +by exporting that env var before invoking `go test` outside compose. ## Caveats -- UDP 47808 must be free on the host. If you already run BACnet locally - (Niagara, Cimplicity, etc.), pick a different host port and override - `BACNET_PORT` in the test command. -- `docker compose up -d` returns before bacpypes3 is fully listening. The - test wraps the first WhoIs in a short retry loop to absorb that. -- bacpypes3 simulates a single device; the multi-device routing tests in - `internal/bacnetip/DeviceInfoCache_test.go` cover that path via unit tests. +- The two-container setup needs a working Docker daemon with bridge driver + support. Docker-in-Docker CI runners need `--privileged` (or careful + configuration of `dockerd-rootless`) to spawn user-space bridges. +- bacpypes3 isn't ready the instant the container starts. The test-runner + sleeps three seconds before running the Go suite to absorb the race; + a proper UDP healthcheck would let us drop that. +- The simulated device is single-instance (device 1234). Multi-device routing + is covered by unit tests in `internal/bacnetip/DeviceInfoCache_test.go`. ## Scope What's covered: -1. `Discover()` returns device `1234` within the configured timeout. +1. `Discover()` returns device `1234` within the configured timeout (real + WhoIs broadcast + IAm round-trip over the bridge). 2. `Read AnalogValue.0/PRESENT_VALUE` round-trips as a `PlcREAL`. -3. `Write AnalogValue.0/PRESENT_VALUE = 42.5` followed by a re-read. -4. `WritePropertyMultiple` of 3 properties succeeds. -5. `Subscribe AnalogInput.0` receives ≥1 COV notification within 5s. -6. Large `ReadPropertyMultiple` (10 properties) exercises 2-segment - reassembly. +3. `Write AnalogValue.1/PRESENT_VALUE = 42.5` followed by a re-read. +4. `Subscribe AnalogInput.0` receives ≥1 COV notification within 10s + (the simulator runs a 2-second sawtooth on AI-0). What's not covered (deferred to v2): -- BACnet/SC (secure-connect over WebSocket). +- BACnet/SC (secure connect over WebSocket). - BACnet/IPv6 (Annex U). - MS/TP routing. -- Real-device interop with quirky stacks (Tracer, older Honeywell). +- Quirky third-party stacks (older Honeywell, Trane Tracer); only a real-device + QA pass exposes these. diff --git a/plc4go/tests/integration/bacnetip/device.py b/plc4go/tests/integration/bacnetip/device.py index 058c4d75b1..4998c2f454 100644 --- a/plc4go/tests/integration/bacnetip/device.py +++ b/plc4go/tests/integration/bacnetip/device.py @@ -19,30 +19,79 @@ """bacpypes3-based BACnet/IP virtual device used by plc4go integration tests. +bacpypes3 0.0.102 constructs the Application via Application.from_args; the +SimpleArgumentParser exposes --address / --instance / --vendoridentifier etc. +We assemble argv manually with values pulled from the BACNET_LOCAL_ADDRESS env +var (set by docker-compose to the container's static IP) and then attach a +handful of object instances for the tests to read/write/subscribe against. + Exposes: - DEVICE:1234 objectName="plc4x-it", vendor 0x4D4D (Apache PLC4X) + DEVICE:1234 objectName="plc4x-it", vendorIdentifier=0x4D4D ANALOG_VALUE:0..4 writable Real PresentValue BINARY_VALUE:0..1 writable Enumerated PresentValue - ANALOG_INPUT:0 read-only Real PresentValue with a slow sawtooth - generator so SubscribeCOV tests see notifications + ANALOG_INPUT:0 read-only Real PresentValue with a 2-second + sawtooth so SubscribeCOV consumers see traffic MULTI_STATE_VALUE:0 writable Unsigned PresentValue - -The device binds to 0.0.0.0:47808 inside the container; docker-compose maps -that to the host's 47808/udp. """ import asyncio +import os +import sys +from bacpypes3.apdu import SimpleAckPDU from bacpypes3.app import Application +from bacpypes3.argparse import SimpleArgumentParser +from bacpypes3.constructeddata import Array +from bacpypes3.errors import ExecutionError from bacpypes3.local.analog import AnalogInputObject, AnalogValueObject from bacpypes3.local.binary import BinaryValueObject -from bacpypes3.local.device import DeviceObject from bacpypes3.local.multistate import MultiStateValueObject -from bacpypes3.primitivedata import Real +from bacpypes3.primitivedata import Real, Unsigned +from bacpypes3.service.object import ReadWritePropertyMultipleServices + + +# bacpypes3 0.0.102 ships a stub WritePropertyMultiple handler that raises +# UnrecognizedService(). Replace it with a working dispatch so the plc4go +# integration tests can verify WPM wire-format end-to-end. Mirrors the +# per-property cast_out + obj.write_property flow that do_WritePropertyRequest +# uses, applied to each (object, property) pair in the request. +async def _do_write_property_multiple(self, apdu): + for spec in apdu.listOfWriteAccessSpecs: + obj = self.get_object_id(spec.objectIdentifier) + if not obj: + raise ExecutionError(errorClass="object", errorCode="unknownObject") + for prop in spec.listOfProperties: + property_type = obj.get_property_type(prop.propertyIdentifier) + array_index = prop.propertyArrayIndex + priority = prop.priority + if issubclass(property_type, Array): + if array_index is None: + pass + elif array_index == 0: + property_type = Unsigned + else: + property_type = property_type._subtype + value = prop.value.cast_out(property_type, null=(priority is not None)) + await obj.write_property(prop.propertyIdentifier, value, array_index, priority) + await self.response(SimpleAckPDU(context=apdu)) + + +ReadWritePropertyMultipleServices.do_WritePropertyMultipleRequest = _do_write_property_multiple + + +# In docker-compose the test runs on bridge 172.30.0.10/24; the host can +# override this for a non-docker invocation. +DEFAULT_ADDRESS = os.environ.get("BACNET_LOCAL_ADDRESS", "172.30.0.10/24:47808") +DEFAULT_INSTANCE = os.environ.get("BACNET_INSTANCE", "1234") +DEFAULT_NAME = os.environ.get("BACNET_NAME", "plc4x-it") +# bacpypes3 ships a bundled vendor-id → vendor-info map and refuses to start +# with an unknown vendor id. The library pre-registers two ids: 0 (ASHRAE, +# spec maintainer) and 999 (intended for tests / unregistered vendors). +DEFAULT_VENDOR = os.environ.get("BACNET_VENDOR", "999") async def sawtooth(obj: AnalogInputObject) -> None: - """Background coroutine that bumps AnalogInput.0 by 1.0 every 2 seconds so + """Background coroutine bumping AnalogInput.0 by 1.0 every 2 seconds so SubscribeCOV consumers see a stream of notifications.""" value = 0.0 while True: @@ -51,68 +100,78 @@ async def sawtooth(obj: AnalogInputObject) -> None: obj.presentValue = Real(value) -def build_app() -> Application: - device = DeviceObject( - objectIdentifier=("device", 1234), - objectName="plc4x-it", - vendorIdentifier=0x4D4D, - modelName="plc4x-it-simulator", - maxApduLengthAccepted=1476, - segmentationSupported="segmentedBoth", - maxSegmentsAccepted=16, - protocolVersion=1, - protocolRevision=14, - ) - - app = Application(device, ("0.0.0.0/24", 47808)) - +async def main() -> None: + # Build the argv that SimpleArgumentParser expects. + parser = SimpleArgumentParser() + args = parser.parse_args([ + "--address", DEFAULT_ADDRESS, + "--instance", DEFAULT_INSTANCE, + "--name", DEFAULT_NAME, + "--vendoridentifier", DEFAULT_VENDOR, + "--debug", + "bacpypes3.ipv4.IPv4DatagramServer", + "bacpypes3.ipv4.bvll", + "bacpypes3.ipv4.link", + "bacpypes3.ipv4.service", + "bacpypes3.npdu", + "bacpypes3.apdu", + "bacpypes3.app.Application", + ]) + print(f"starting bacpypes3 device {args.instance} @ {args.address}", flush=True) + + app = Application.from_args(args) + + # Writable scratchpad objects for Read/Write integration tests. for i in range(5): app.add_object( AnalogValueObject( - objectIdentifier=("analogValue", i), + objectIdentifier=("analog-value", i), objectName=f"AV-{i}", presentValue=Real(0.0), outOfService=False, ) ) - for i in range(2): app.add_object( BinaryValueObject( - objectIdentifier=("binaryValue", i), + objectIdentifier=("binary-value", i), objectName=f"BV-{i}", presentValue="inactive", ) ) + # Read-only analog input with a slow generator — exercises SubscribeCOV. + # covIncrement must be non-None or bacpypes3's present_value_filter + # crashes inside property_change with `Real - NoneType`. 0.5 means every + # sawtooth tick (1.0) triggers a COV notification. ai0 = AnalogInputObject( - objectIdentifier=("analogInput", 0), + objectIdentifier=("analog-input", 0), objectName="AI-0", presentValue=Real(0.0), outOfService=False, + covIncrement=Real(0.5), ) app.add_object(ai0) app.add_object( MultiStateValueObject( - objectIdentifier=("multiStateValue", 0), + objectIdentifier=("multi-state-value", 0), objectName="MSV-0", presentValue=1, numberOfStates=4, ) ) + print("device ready", flush=True) asyncio.create_task(sawtooth(ai0)) - return app - -async def main() -> None: - build_app() - # Application.run() blocks on the BACnet event loop forever; this - # container only exists to host that loop, so just await the - # never-completing future. + # Idle forever; bacpypes3's event loop keeps the network stack alive. await asyncio.Future() if __name__ == "__main__": - asyncio.run(main()) + # -u (PYTHONUNBUFFERED) on the CMD makes stdout flush per line for compose logs. + try: + asyncio.run(main()) + except KeyboardInterrupt: + sys.exit(0) diff --git a/plc4go/tests/integration/bacnetip/docker-compose.yml b/plc4go/tests/integration/bacnetip/docker-compose.yml index 1198d05987..eafbea38f5 100644 --- a/plc4go/tests/integration/bacnetip/docker-compose.yml +++ b/plc4go/tests/integration/bacnetip/docker-compose.yml @@ -17,20 +17,56 @@ # under the License. # -# Spins a single bacpypes3 BACnet/IP virtual device for plc4go integration -# tests. Usage: +# Two-service setup so both ends of the BACnet conversation get their own +# network namespace. This matters because BACnet/IP uses port 47808 on both +# sides — a single host network can only honour that for one process at a +# time. With a bridge network, the simulator and the Go test-runner each +# bind 47808 against their own interface, and UDP broadcasts traverse the +# bridge so discovery works. # -# docker compose -f tests/integration/bacnetip/docker-compose.yml up -d -# BACNET_IT=1 go test -tags integration ./tests/integration/bacnetip/... -# docker compose -f tests/integration/bacnetip/docker-compose.yml down +# Usage: +# make integration-bacnetip +# +# or directly: +# docker compose -f tests/integration/bacnetip/docker-compose.yml \ +# --project-directory ../.. \ +# up --build --abort-on-container-exit --exit-code-from test-runner # services: bacnet-device: build: + # `context` is resolved relative to this compose file's directory. + # `.` here is plc4go/tests/integration/bacnetip — the dir holding the + # simulator's Dockerfile and device.py. context: . dockerfile: Dockerfile - # host networking is the easiest way to make BACnet broadcast traffic - # work without per-OS bridge tweaks. The downside is the container - # claims UDP 47808 on the host. - network_mode: host - restart: unless-stopped + networks: + bacnet: + ipv4_address: 172.30.0.10 + # 47808 is reachable inside the bacnet bridge via the service name + # `bacnet-device` or the fixed IP above. No host port mapping — the host + # network namespace stays untouched. + expose: + - "47808/udp" + restart: "no" + + test-runner: + build: + # Walk back up to the plc4go module root so Dockerfile.test can COPY + # go.mod, go.sum, and the whole source tree. + context: ../../.. + dockerfile: tests/integration/bacnetip/Dockerfile.test + depends_on: + - bacnet-device + networks: + bacnet: + ipv4_address: 172.30.0.11 + restart: "no" + +networks: + bacnet: + driver: bridge + ipam: + driver: default + config: + - subnet: 172.30.0.0/24 diff --git a/plc4go/tests/integration/bacnetip/integration_test.go b/plc4go/tests/integration/bacnetip/integration_test.go index ce68d93662..5684b86db3 100644 --- a/plc4go/tests/integration/bacnetip/integration_test.go +++ b/plc4go/tests/integration/bacnetip/integration_test.go @@ -29,7 +29,10 @@ package bacnetip_test import ( "context" + "fmt" + "math" "os" + "strings" "testing" "time" @@ -45,6 +48,10 @@ import ( const ( envSimulatorHost = "BACNET_IT_HOST" // optional override; defaults to 127.0.0.1 envEnabled = "BACNET_IT" // master switch (must be set, e.g. BACNET_IT=1) + // deviceInstance must mirror BACNET_INSTANCE in docker-compose / device.py. + // Assertions on the discovered name use this so a stray BACnet packet + // from another device on the bridge would not pass the test. + deviceInstance = "1234" ) func skipIfDisabled(t *testing.T) { @@ -69,6 +76,15 @@ func newDriverManager(t *testing.T) plc4go.PlcDriverManager { return dm } +func connect(t *testing.T, ctx context.Context) plc4go.PlcConnection { + t.Helper() + dm := newDriverManager(t) + conn, err := dm.GetConnection(ctx, "bacnet-ip:udp://"+simulatorHost()+":47808") + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + // readPlcReadResult drains a one-shot result channel with a timeout. The // per-request channel is closed by the driver after exactly one send, so a // successful receive plus a timeout case is sufficient. @@ -111,39 +127,57 @@ func readPlcSubscriptionResult(t *testing.T, ch <-chan apiModel.PlcSubscriptionR return nil } +func readPlcUnsubscriptionResult(t *testing.T, ch <-chan apiModel.PlcUnsubscriptionRequestResult) apiModel.PlcUnsubscriptionRequestResult { + t.Helper() + select { + case r := <-ch: + require.NotNil(t, r) + require.NoError(t, r.GetErr()) + return r + case <-time.After(5 * time.Second): + t.Fatal("unsubscribe request timed out") + } + return nil +} + func TestIT_Discover_FindsSimulator(t *testing.T) { skipIfDisabled(t) dm := newDriverManager(t) - ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), 12*time.Second) defer cancel() - found := make(chan struct{}, 1) + // Discoverer formats the event name as "device DEVICE:<instance>" — assert + // the instance matches so an unrelated IAm broadcast can't pass the test. + found := make(chan apiModel.PlcDiscoveryItem, 8) err := dm.Discover(ctx, func(event apiModel.PlcDiscoveryItem) { select { - case found <- struct{}{}: + case found <- event: default: } }, plc4go.WithDiscoveryOptionProtocol("bacnet-ip")) require.NoError(t, err) - select { - case <-found: - case <-ctx.Done(): - t.Fatal("discovery timed out without finding the simulator") + for { + select { + case ev := <-found: + if strings.Contains(ev.GetName(), deviceInstance) { + assert.Equal(t, "bacnet-ip", ev.GetProtocolCode()) + assert.Equal(t, "udp", ev.GetTransportCode()) + return + } + t.Logf("discovered non-matching device %q, waiting for instance %s", ev.GetName(), deviceInstance) + case <-ctx.Done(): + t.Fatalf("discovery timed out without finding device instance %s", deviceInstance) + } } } func TestIT_Read_AnalogValueRoundTrip(t *testing.T) { skipIfDisabled(t) - dm := newDriverManager(t) - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) defer cancel() - - conn, err := dm.GetConnection(ctx, "bacnet-ip:udp://"+simulatorHost()+":47808") - require.NoError(t, err) - defer func() { _ = conn.Close() }() + conn := connect(t, ctx) rrb := conn.ReadRequestBuilder() rrb.AddTagAddress("av0", "ANALOG_VALUE,0/PRESENT_VALUE") @@ -158,16 +192,71 @@ func TestIT_Read_AnalogValueRoundTrip(t *testing.T) { assert.InDelta(t, 0.0, val.GetFloat32(), 0.01) } -func TestIT_Write_AnalogValueRoundTrip(t *testing.T) { +func TestIT_Read_BinaryValue(t *testing.T) { skipIfDisabled(t) - dm := newDriverManager(t) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + conn := connect(t, ctx) - ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress("bv0", "BINARY_VALUE,0/PRESENT_VALUE") + req, err := rrb.Build() + require.NoError(t, err) + + r := readPlcReadResult(t, req.Execute(ctx)) + resp := r.GetResponse() + require.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("bv0")) + val := resp.GetValue("bv0") + require.NotNil(t, val) + // BV PresentValue is Enumerated (0=inactive, 1=active). Decoder returns + // it as PlcUDINT. device.py initializes BV-0 with presentValue="inactive". + assert.Equal(t, uint32(0), val.GetUint32(), "BV-0 should be inactive (0)") +} + +func TestIT_Read_MultiStateValue(t *testing.T) { + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) defer cancel() + conn := connect(t, ctx) - conn, err := dm.GetConnection(ctx, "bacnet-ip:udp://"+simulatorHost()+":47808") + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress("msv0", "MULTI_STATE_VALUE,0/PRESENT_VALUE") + req, err := rrb.Build() require.NoError(t, err) - defer func() { _ = conn.Close() }() + + r := readPlcReadResult(t, req.Execute(ctx)) + resp := r.GetResponse() + require.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("msv0")) + val := resp.GetValue("msv0") + require.NotNil(t, val) + // MSV PresentValue is Unsigned, initialized to state 1 in device.py. + assert.Equal(t, uint64(1), val.GetUint64(), "MSV-0 should be in state 1") +} + +func TestIT_Read_UnknownObject_ReturnsNotFound(t *testing.T) { + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + conn := connect(t, ctx) + + rrb := conn.ReadRequestBuilder() + // AV.99 isn't defined in device.py — bacpypes3 returns + // APDUError(unknown-object) which the driver maps to NOT_FOUND. + rrb.AddTagAddress("ghost", "ANALOG_VALUE,99/PRESENT_VALUE") + req, err := rrb.Build() + require.NoError(t, err) + + r := readPlcReadResult(t, req.Execute(ctx)) + resp := r.GetResponse() + assert.Equal(t, apiModel.PlcResponseCode_NOT_FOUND, resp.GetResponseCode("ghost"), + "reading a non-existent AV should map to NOT_FOUND") +} + +func TestIT_Write_AnalogValueRoundTrip(t *testing.T) { + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + conn := connect(t, ctx) wrb := conn.WriteRequestBuilder() wrb.AddTagAddress("av1", "ANALOG_VALUE,1/PRESENT_VALUE", float32(42.5)) @@ -185,16 +274,46 @@ func TestIT_Write_AnalogValueRoundTrip(t *testing.T) { assert.InDelta(t, 42.5, r.GetResponse().GetValue("av1").GetFloat32(), 0.001) } -func TestIT_Subscribe_AnalogInput(t *testing.T) { +func TestIT_WritePropertyMultiple_ThreeAnalogValues(t *testing.T) { skipIfDisabled(t) - dm := newDriverManager(t) - - ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) defer cancel() + conn := connect(t, ctx) - conn, err := dm.GetConnection(ctx, "bacnet-ip:udp://"+simulatorHost()+":47808") + // >1 tag in a single WriteRequest triggers WritePropertyMultiple in Writer.go. + wrb := conn.WriteRequestBuilder() + wrb.AddTagAddress("av2", "ANALOG_VALUE,2/PRESENT_VALUE", float32(11.25)) + wrb.AddTagAddress("av3", "ANALOG_VALUE,3/PRESENT_VALUE", float32(22.5)) + wrb.AddTagAddress("av4", "ANALOG_VALUE,4/PRESENT_VALUE", float32(33.75)) + wreq, err := wrb.Build() + require.NoError(t, err) + w := readPlcWriteResult(t, wreq.Execute(ctx)) + resp := w.GetResponse() + assert.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("av2")) + assert.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("av3")) + assert.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("av4")) + + // Re-read all three via ReadPropertyMultiple to confirm they actually + // committed on the server — a SimpleAck alone doesn't prove the writes + // landed in the right slots. + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress("av2", "ANALOG_VALUE,2/PRESENT_VALUE") + rrb.AddTagAddress("av3", "ANALOG_VALUE,3/PRESENT_VALUE") + rrb.AddTagAddress("av4", "ANALOG_VALUE,4/PRESENT_VALUE") + rreq, err := rrb.Build() require.NoError(t, err) - defer func() { _ = conn.Close() }() + r := readPlcReadResult(t, rreq.Execute(ctx)) + rresp := r.GetResponse() + assert.InDelta(t, 11.25, rresp.GetValue("av2").GetFloat32(), 0.001) + assert.InDelta(t, 22.5, rresp.GetValue("av3").GetFloat32(), 0.001) + assert.InDelta(t, 33.75, rresp.GetValue("av4").GetFloat32(), 0.001) +} + +func TestIT_Subscribe_AnalogInput_InitialNotification(t *testing.T) { + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 8*time.Second) + defer cancel() + conn := connect(t, ctx) srb := conn.SubscriptionRequestBuilder() srb.AddChangeOfStateTagAddress("ai0", "ANALOG_INPUT,0/PRESENT_VALUE") @@ -205,6 +324,11 @@ func TestIT_Subscribe_AnalogInput(t *testing.T) { sresp := sr.GetResponse() require.Equal(t, apiModel.PlcResponseCode_OK, sresp.GetResponseCode("ai0")) + // bacpypes3 sends an initial UnconfirmedCOVNotification right after the + // subscription is accepted — that's what this test waits for. Verifies + // the unsolicited-message path (codec.HandleMessages fallthrough → + // defaultIncomingMessageChannel → Connection.routeIncomingMessage → + // Subscriber.HandleUnconfirmedCOVNotification → consumer). notif := make(chan apiModel.PlcSubscriptionEvent, 4) for _, h := range sresp.GetSubscriptionHandles() { h.Register(func(event apiModel.PlcSubscriptionEvent) { @@ -215,10 +339,306 @@ func TestIT_Subscribe_AnalogInput(t *testing.T) { }) } - // device.py bumps AI-0 every 2 seconds — allow up to 10s. select { - case <-notif: + case ev := <-notif: + require.Equal(t, apiModel.PlcResponseCode_OK, ev.GetResponseCode("ai0")) + val := ev.GetValue("ai0") + require.NotNil(t, val, "subscription event should carry a value") + // The initial notification carries whatever AI-0 currently reads. + // The simulator's sawtooth has been ticking since container start — + // docker compose builds run for tens of seconds — so the value is + // usually large (we've seen ~35.0). Assert that it decoded as a + // finite, non-negative Real, which is the actual property type + // guarantee from device.py. The follow-on Sawtooth test verifies + // that subsequent ticks generate fresh notifications. + f := val.GetFloat32() + assert.False(t, math.IsNaN(float64(f)), "AI-0 should decode as a real number") + assert.GreaterOrEqual(t, f, float32(0.0), "sawtooth values are non-negative") + assert.Less(t, f, float32(100.0), "sawtooth wraps at 100, so values stay in [0,100)") + t.Logf("initial COV notification: AI-0 = %v", f) + case <-ctx.Done(): + t.Fatal("did not receive the initial COV notification within timeout") + } +} + +func TestIT_Subscribe_ReceivesSawtoothChange(t *testing.T) { + skipIfDisabled(t) + // Sawtooth ticks every 2s; covIncrement=0.5 means every tick fires a COV. + // We need ≥one full sawtooth period (2s) after the initial notification, + // plus headroom for startup variance. + ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) + defer cancel() + conn := connect(t, ctx) + + srb := conn.SubscriptionRequestBuilder() + srb.AddChangeOfStateTagAddress("ai0", "ANALOG_INPUT,0/PRESENT_VALUE") + sreq, err := srb.Build() + require.NoError(t, err) + + sr := readPlcSubscriptionResult(t, sreq.Execute(ctx)) + sresp := sr.GetResponse() + require.Equal(t, apiModel.PlcResponseCode_OK, sresp.GetResponseCode("ai0")) + + notif := make(chan apiModel.PlcSubscriptionEvent, 8) + for _, h := range sresp.GetSubscriptionHandles() { + h.Register(func(event apiModel.PlcSubscriptionEvent) { + select { + case notif <- event: + default: + } + }) + } + + // Capture the initial notification. bacpypes3 sends one immediately on + // subscribe with the *current* AI-0 value — which has been ticking since + // the simulator started, so it's already non-zero. Just accepting any + // non-zero value (which my earlier attempt did) would pass on this + // initial echo alone and prove nothing about sawtooth-driven COV. + var initialVal float32 + select { + case ev := <-notif: + require.Equal(t, apiModel.PlcResponseCode_OK, ev.GetResponseCode("ai0")) + initialVal = ev.GetValue("ai0").GetFloat32() + t.Logf("initial COV notification (will discard): AI-0 = %v", initialVal) case <-ctx.Done(): - t.Fatal("did not receive a COV notification within timeout") + t.Fatal("never received the initial COV notification") + } + + // Now wait for a CHANGE: a subsequent COV with a value different from + // initialVal. Sawtooth bumps by 1.0 every 2s and covIncrement=0.5, so + // the next tick after initial should trigger one. + deadline := time.After(6 * time.Second) + for { + select { + case ev := <-notif: + require.Equal(t, apiModel.PlcResponseCode_OK, ev.GetResponseCode("ai0")) + val := ev.GetValue("ai0").GetFloat32() + t.Logf("subsequent COV: AI-0 = %v (initial was %v)", val, initialVal) + if val != initialVal { + return + } + case <-deadline: + t.Fatalf("AI-0 never changed after subscribe; stuck at %v — sawtooth or covIncrement filter not firing", initialVal) + case <-ctx.Done(): + t.Fatalf("context cancelled while waiting for sawtooth-driven COV after initialVal=%v", initialVal) + } + } +} + +// ── Tier 2: extended integration coverage ────────────────────────────────── + +func TestIT_Read_DeviceObjectName(t *testing.T) { + // Exercises the CharacterString decode path. device.py sets + // DEVICE:1234/OBJECT_NAME = "plc4x-it". + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + conn := connect(t, ctx) + + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress("name", "DEVICE,1234/OBJECT_NAME") + req, err := rrb.Build() + require.NoError(t, err) + + r := readPlcReadResult(t, req.Execute(ctx)) + resp := r.GetResponse() + require.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("name")) + val := resp.GetValue("name") + require.NotNil(t, val) + assert.Equal(t, "plc4x-it", val.GetString()) +} + +func TestIT_Read_MultipleTags_ReadPropertyMultiple(t *testing.T) { + // Hits Reader.go's >1-tag branch which builds a + // BACnetConfirmedServiceRequestReadPropertyMultiple. The earlier + // WPM-readback only proved this works when all tags are the same + // object-type; here we mix AV / BV / MSV to force the multi-typed + // dispatch in the response decoder. + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + conn := connect(t, ctx) + + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress("av0", "ANALOG_VALUE,0/PRESENT_VALUE") + rrb.AddTagAddress("bv0", "BINARY_VALUE,0/PRESENT_VALUE") + rrb.AddTagAddress("msv0", "MULTI_STATE_VALUE,0/PRESENT_VALUE") + req, err := rrb.Build() + require.NoError(t, err) + + r := readPlcReadResult(t, req.Execute(ctx)) + resp := r.GetResponse() + assert.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("av0")) + assert.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("bv0")) + assert.Equal(t, apiModel.PlcResponseCode_OK, resp.GetResponseCode("msv0")) + assert.InDelta(t, 0.0, resp.GetValue("av0").GetFloat32(), 0.01) + assert.Equal(t, uint32(0), resp.GetValue("bv0").GetUint32()) + assert.Equal(t, uint64(1), resp.GetValue("msv0").GetUint64()) +} + +func TestIT_Write_BinaryValue(t *testing.T) { + // Exercises plcValueToApplicationTag's Boolean → Enumerated hint path, + // the BV setter on bacpypes3, and the readback's BACnetBinaryPVTagged + // decode through taggedEnumToPlcValue. + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + conn := connect(t, ctx) + + wrb := conn.WriteRequestBuilder() + wrb.AddTagAddress("bv1", "BINARY_VALUE,1/PRESENT_VALUE", true) + wreq, err := wrb.Build() + require.NoError(t, err) + w := readPlcWriteResult(t, wreq.Execute(ctx)) + require.Equal(t, apiModel.PlcResponseCode_OK, w.GetResponse().GetResponseCode("bv1")) + + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress("bv1", "BINARY_VALUE,1/PRESENT_VALUE") + rreq, err := rrb.Build() + require.NoError(t, err) + r := readPlcReadResult(t, rreq.Execute(ctx)) + assert.Equal(t, uint32(1), r.GetResponse().GetValue("bv1").GetUint32(), + "BV-1 should read back as 1 (active) after Write(true)") +} + +func TestIT_Write_MultiStateValue(t *testing.T) { + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + conn := connect(t, ctx) + + wrb := conn.WriteRequestBuilder() + // MSV.0 has numberOfStates=4; state 3 is valid. + wrb.AddTagAddress("msv0", "MULTI_STATE_VALUE,0/PRESENT_VALUE", uint32(3)) + wreq, err := wrb.Build() + require.NoError(t, err) + w := readPlcWriteResult(t, wreq.Execute(ctx)) + require.Equal(t, apiModel.PlcResponseCode_OK, w.GetResponse().GetResponseCode("msv0")) + + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress("msv0", "MULTI_STATE_VALUE,0/PRESENT_VALUE") + rreq, err := rrb.Build() + require.NoError(t, err) + r := readPlcReadResult(t, rreq.Execute(ctx)) + assert.Equal(t, uint64(3), r.GetResponse().GetValue("msv0").GetUint64(), + "MSV-0 should read back as 3 after the write") +} + +func TestIT_Write_ReadOnlyProperty_Fails(t *testing.T) { + // OBJECT_TYPE is universally read-only — bacpypes3 returns + // APDUError(property, write-access-denied), which the driver should + // map to ACCESS_DENIED. (AI.PresentValue would be a more obvious + // candidate, but bacpypes3 doesn't enforce outOfService=False for + // writes on AI in 0.0.102 — so we'd silently get an OK.) Confirms + // the error mapping on the write path (counterpart to read NOT_FOUND). + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + conn := connect(t, ctx) + + wrb := conn.WriteRequestBuilder() + wrb.AddTagAddress("av0type", "ANALOG_VALUE,0/OBJECT_TYPE", uint32(2)) + wreq, err := wrb.Build() + require.NoError(t, err) + w := readPlcWriteResult(t, wreq.Execute(ctx)) + code := w.GetResponse().GetResponseCode("av0type") + // We accept either ACCESS_DENIED (the canonical mapping for + // write-access-denied) or INVALID_DATA (some BACnet stacks REJECT + // the request outright before producing an APDUError). The key is + // that it MUST NOT be OK. + assert.NotEqual(t, apiModel.PlcResponseCode_OK, code, + "writing to read-only OBJECT_TYPE should not succeed; got %v", code) + t.Logf("write to read-only OBJECT_TYPE returned response code %v", code) +} + +func TestIT_ConcurrentReads_AllSucceed(t *testing.T) { + // Fire 5 reads in parallel on the same connection. The transaction + // manager must serialize them onto the wire and dispatch responses + // to the right caller. A race in invoke-id allocation or expectation + // matching surfaces as a deadlock or cross-talk. + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + conn := connect(t, ctx) + + const N = 5 + results := make([]<-chan apiModel.PlcReadRequestResult, N) + for i := range N { + rrb := conn.ReadRequestBuilder() + rrb.AddTagAddress(fmt.Sprintf("av%d", i), fmt.Sprintf("ANALOG_VALUE,%d/PRESENT_VALUE", i)) + req, err := rrb.Build() + require.NoError(t, err) + results[i] = req.Execute(ctx) + } + for i, ch := range results { + r := readPlcReadResult(t, ch) + key := fmt.Sprintf("av%d", i) + require.Equal(t, apiModel.PlcResponseCode_OK, r.GetResponse().GetResponseCode(key), + "AV.%d should read OK; transaction manager may have crossed wires", i) + } +} + +func TestIT_Unsubscribe_StopsNotifications(t *testing.T) { + // Subscribe, capture one notification, unsubscribe, then verify no more + // notifications arrive in the next 3s (longer than sawtooth period). + // Exercises Subscriber.Unsubscribe (sends SubscribeCOV with lifetime=0). + skipIfDisabled(t) + ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) + defer cancel() + conn := connect(t, ctx) + + srb := conn.SubscriptionRequestBuilder() + srb.AddChangeOfStateTagAddress("ai0", "ANALOG_INPUT,0/PRESENT_VALUE") + sreq, err := srb.Build() + require.NoError(t, err) + sr := readPlcSubscriptionResult(t, sreq.Execute(ctx)) + sresp := sr.GetResponse() + require.Equal(t, apiModel.PlcResponseCode_OK, sresp.GetResponseCode("ai0")) + + notif := make(chan apiModel.PlcSubscriptionEvent, 16) + handles := sresp.GetSubscriptionHandles() + require.NotEmpty(t, handles) + for _, h := range handles { + h.Register(func(event apiModel.PlcSubscriptionEvent) { + select { + case notif <- event: + default: + } + }) + } + + // Drain the initial notification. + select { + case <-notif: + case <-time.After(5 * time.Second): + t.Fatal("never got the initial notification — subscribe likely failed silently") + } + + // Unsubscribe. + ureq, err := conn.UnsubscriptionRequestBuilder().AddHandles(handles...).Build() + require.NoError(t, err) + uresp := readPlcUnsubscriptionResult(t, ureq.Execute(ctx)) + require.NotNil(t, uresp.GetResponse()) + + // Drain any in-flight notifications that snuck in between unsubscribe + // send and bacpypes3 processing it. + drainDeadline := time.After(500 * time.Millisecond) +drain: + for { + select { + case <-notif: + case <-drainDeadline: + break drain + } + } + + // Now wait 3s — sawtooth ticks every 2s, so a working subscription + // would deliver at least one new notification in this window. Zero + // means unsubscribe took effect. + select { + case ev := <-notif: + t.Fatalf("got a notification after unsubscribe: AI-0=%v", ev.GetValue("ai0").GetFloat32()) + case <-time.After(3 * time.Second): + // Expected — no more notifications. } }
