mirror of
https://github.com/bitechdev/ResolveSpec.git
synced 2026-01-05 03:14:25 +00:00
feature: mqtt support
This commit is contained in:
Hein
409
pkg/mqttspec/broker_test.go
Normal file
409
pkg/mqttspec/broker_test.go
Normal file
@@ -0,0 +1,409 @@
|
||||
package mqttspec
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestNewEmbeddedBroker(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 1883,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
|
||||
assert.NotNil(t, broker)
|
||||
assert.Equal(t, config, broker.config)
|
||||
assert.Equal(t, cm, broker.clientManager)
|
||||
assert.NotNil(t, broker.subscriptions)
|
||||
assert.False(t, broker.started)
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_StartStop(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11883, // Use non-standard port for testing
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
ctx := context.Background()
|
||||
|
||||
// Start broker
|
||||
err := broker.Start(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify started
|
||||
assert.True(t, broker.IsConnected())
|
||||
|
||||
// Stop broker
|
||||
err = broker.Stop(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify stopped
|
||||
assert.False(t, broker.IsConnected())
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_StartTwice(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11884,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
ctx := context.Background()
|
||||
|
||||
// Start broker
|
||||
err := broker.Start(ctx)
|
||||
require.NoError(t, err)
|
||||
defer broker.Stop(ctx)
|
||||
|
||||
// Try to start again - should fail
|
||||
err = broker.Start(ctx)
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "already started")
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_StopWithoutStart(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11885,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
ctx := context.Background()
|
||||
|
||||
// Stop without starting - should not error
|
||||
err := broker.Stop(ctx)
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_PublishWithoutStart(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11886,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
|
||||
// Try to publish without starting - should fail
|
||||
err := broker.Publish("test/topic", 1, []byte("test"))
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "broker not started")
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_SubscribeWithoutStart(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11887,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
|
||||
// Try to subscribe without starting - should fail
|
||||
err := broker.Subscribe("test/topic", 1, func(topic string, payload []byte) {})
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "broker not started")
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_PublishSubscribe(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11888,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
ctx := context.Background()
|
||||
|
||||
// Start broker
|
||||
err := broker.Start(ctx)
|
||||
require.NoError(t, err)
|
||||
defer broker.Stop(ctx)
|
||||
|
||||
// Subscribe to topic
|
||||
callback := func(topic string, payload []byte) {
|
||||
// Callback for subscription - actual message delivery would require
|
||||
// integration with Mochi MQTT's hook system
|
||||
}
|
||||
|
||||
err = broker.Subscribe("test/topic", 1, callback)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Note: Embedded broker's Subscribe is simplified and doesn't fully integrate
|
||||
// with Mochi MQTT's internal pub/sub. This test verifies the subscription
|
||||
// is registered but actual message delivery would require more complex
|
||||
// integration with Mochi MQTT's hook system.
|
||||
|
||||
// Verify subscription was registered
|
||||
broker.subMu.RLock()
|
||||
_, exists := broker.subscriptions["test/topic"]
|
||||
broker.subMu.RUnlock()
|
||||
assert.True(t, exists)
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_Unsubscribe(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11889,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
ctx := context.Background()
|
||||
|
||||
// Start broker
|
||||
err := broker.Start(ctx)
|
||||
require.NoError(t, err)
|
||||
defer broker.Stop(ctx)
|
||||
|
||||
// Subscribe
|
||||
callback := func(topic string, payload []byte) {}
|
||||
err = broker.Subscribe("test/topic", 1, callback)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify subscription exists
|
||||
broker.subMu.RLock()
|
||||
_, exists := broker.subscriptions["test/topic"]
|
||||
broker.subMu.RUnlock()
|
||||
assert.True(t, exists)
|
||||
|
||||
// Unsubscribe
|
||||
err = broker.Unsubscribe("test/topic")
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify subscription removed
|
||||
broker.subMu.RLock()
|
||||
_, exists = broker.subscriptions["test/topic"]
|
||||
broker.subMu.RUnlock()
|
||||
assert.False(t, exists)
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_SetHandler(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11890,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
|
||||
// Create a mock handler (nil is fine for this test)
|
||||
var handler *Handler = nil
|
||||
|
||||
// Set handler
|
||||
broker.SetHandler(handler)
|
||||
|
||||
// Verify handler was set
|
||||
broker.mu.RLock()
|
||||
assert.Equal(t, handler, broker.handler)
|
||||
broker.mu.RUnlock()
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_GetClientManager(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11891,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
|
||||
// Get client manager
|
||||
retrievedCM := broker.GetClientManager()
|
||||
|
||||
// Verify it's the same instance
|
||||
assert.Equal(t, cm, retrievedCM)
|
||||
}
|
||||
|
||||
func TestEmbeddedBroker_ConcurrentPublish(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := BrokerConfig{
|
||||
Host: "localhost",
|
||||
Port: 11892,
|
||||
MaxConnections: 100,
|
||||
KeepAlive: 60 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewEmbeddedBroker(config, cm)
|
||||
ctx := context.Background()
|
||||
|
||||
// Start broker
|
||||
err := broker.Start(ctx)
|
||||
require.NoError(t, err)
|
||||
defer broker.Stop(ctx)
|
||||
|
||||
// Test concurrent publishing
|
||||
var wg sync.WaitGroup
|
||||
numPublishers := 10
|
||||
|
||||
for i := 0; i < numPublishers; i++ {
|
||||
wg.Add(1)
|
||||
go func(id int) {
|
||||
defer wg.Done()
|
||||
for j := 0; j < 10; j++ {
|
||||
err := broker.Publish("test/topic", 1, []byte("test"))
|
||||
// Errors are acceptable in concurrent scenario
|
||||
_ = err
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func TestNewExternalBrokerClient(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := ExternalBrokerConfig{
|
||||
BrokerURL: "tcp://localhost:1883",
|
||||
ClientID: "test-client",
|
||||
Username: "user",
|
||||
Password: "pass",
|
||||
CleanSession: true,
|
||||
KeepAlive: 60 * time.Second,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReconnectDelay: 1 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewExternalBrokerClient(config, cm)
|
||||
|
||||
assert.NotNil(t, broker)
|
||||
assert.Equal(t, config, broker.config)
|
||||
assert.Equal(t, cm, broker.clientManager)
|
||||
assert.NotNil(t, broker.subscriptions)
|
||||
assert.False(t, broker.connected)
|
||||
}
|
||||
|
||||
func TestExternalBrokerClient_SetHandler(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := ExternalBrokerConfig{
|
||||
BrokerURL: "tcp://localhost:1883",
|
||||
ClientID: "test-client",
|
||||
Username: "user",
|
||||
Password: "pass",
|
||||
CleanSession: true,
|
||||
KeepAlive: 60 * time.Second,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReconnectDelay: 1 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewExternalBrokerClient(config, cm)
|
||||
|
||||
// Create a mock handler (nil is fine for this test)
|
||||
var handler *Handler = nil
|
||||
|
||||
// Set handler
|
||||
broker.SetHandler(handler)
|
||||
|
||||
// Verify handler was set
|
||||
broker.mu.RLock()
|
||||
assert.Equal(t, handler, broker.handler)
|
||||
broker.mu.RUnlock()
|
||||
}
|
||||
|
||||
func TestExternalBrokerClient_GetClientManager(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := ExternalBrokerConfig{
|
||||
BrokerURL: "tcp://localhost:1883",
|
||||
ClientID: "test-client",
|
||||
Username: "user",
|
||||
Password: "pass",
|
||||
CleanSession: true,
|
||||
KeepAlive: 60 * time.Second,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReconnectDelay: 1 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewExternalBrokerClient(config, cm)
|
||||
|
||||
// Get client manager
|
||||
retrievedCM := broker.GetClientManager()
|
||||
|
||||
// Verify it's the same instance
|
||||
assert.Equal(t, cm, retrievedCM)
|
||||
}
|
||||
|
||||
func TestExternalBrokerClient_IsConnected(t *testing.T) {
|
||||
cm := NewClientManager(context.Background())
|
||||
defer cm.Shutdown()
|
||||
|
||||
config := ExternalBrokerConfig{
|
||||
BrokerURL: "tcp://localhost:1883",
|
||||
ClientID: "test-client",
|
||||
Username: "user",
|
||||
Password: "pass",
|
||||
CleanSession: true,
|
||||
KeepAlive: 60 * time.Second,
|
||||
ConnectTimeout: 5 * time.Second,
|
||||
ReconnectDelay: 1 * time.Second,
|
||||
}
|
||||
|
||||
broker := NewExternalBrokerClient(config, cm)
|
||||
|
||||
// Should not be connected initially
|
||||
assert.False(t, broker.IsConnected())
|
||||
}
|
||||
|
||||
// Note: Tests for ExternalBrokerClient Start/Stop/Publish/Subscribe require
|
||||
// a running MQTT broker and are better suited for integration tests.
|
||||
// These tests would be included in integration_test.go with proper test
|
||||
// broker setup (e.g., using Docker Compose).
|
||||
Reference in New Issue
Block a user