Enable and fix tests

Signed-off-by: Leo Siepel <leosiepel@gmail.com>

Author: lsiepel

itests/org.openhab.binding.homie.tests/src/main/java/org/openhab/binding/mqtt/homie/HomieImplementationTest.java

@@ -18,6 +18,7 @@
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -30,7 +31,6 @@
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
@@ -55,6 +55,7 @@
 import org.openhab.core.io.transport.mqtt.MqttBrokerConnection;
 import org.openhab.core.io.transport.mqtt.MqttConnectionObserver;
 import org.openhab.core.io.transport.mqtt.MqttConnectionState;
+import org.openhab.core.io.transport.mqtt.MqttMessageSubscriber;
 import org.openhab.core.library.types.OnOffType;
 import org.openhab.core.library.types.QuantityType;
 import org.openhab.core.library.unit.SIUnits;
@@ -72,6 +73,7 @@ public class HomieImplementationTest extends MqttOSGiTest {
     private static final String BASE_TOPIC = "homie";
     private static final String DEVICE_ID = ThingChannelConstants.TEST_HOMIE_THING.getId();
     private static final String DEVICE_TOPIC = BASE_TOPIC + "/" + DEVICE_ID;
+    private static final int MQTT_TIMEOUT = 10000;
 
     private @NonNullByDefault({}) MqttBrokerConnection homieConnection;
     private int registeredTopics = 100;
@@ -165,15 +167,13 @@ public void retrieveAllTopics() throws Exception {
                 "Connection " + homieConnection.getClientId() + " not retrieving all topics ");
     }
 
-    @Disabled("https://github.com/openhab/openhab-addons/issues/12667")
     @Test
     public void retrieveOneAttribute() throws Exception {
         WaitForTopicValue watcher = new WaitForTopicValue(homieConnection, DEVICE_TOPIC + "/$homie");
-        assertThat(watcher.waitForTopicValue(1000), is("3.0"));
+        assertThat(watcher.waitForTopicValue(MQTT_TIMEOUT), is("3.0"));
     }
 
     @SuppressWarnings("null")
-    @Disabled("Temporarily disabled: unstable")
     @Test
     public void retrieveAttributes() throws Exception {
         assertThat(homieConnection.hasSubscribers(), is(false));
@@ -183,9 +183,6 @@ public void retrieveAttributes() throws Exception {
         Property property = spy(
                 new Property(DEVICE_TOPIC + "/testnode", node, "temperature", callback, new PropertyAttributes()));
 
-        // Create a scheduler
-        ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(4);
-
         property.subscribe(homieConnection, scheduler, 500).get();
 
         assertThat(property.attributes.settable, is(true));
@@ -202,8 +199,8 @@ public void retrieveAttributes() throws Exception {
 
         property.startChannel(homieConnection, scheduler, 500).get();
         verify(channelState).start(any(), any(), anyInt());
-        verify(channelState, timeout(500)).processMessage(any(), any());
-        verify(callback).updateChannelState(any(), any());
+        verify(callback, timeout(MQTT_TIMEOUT)).updateChannelState(any(), any());
+        verify(channelState).processMessage(any(), any());
 
         assertThat(property.getChannelState().getCache().getChannelState(),
                 is(new QuantityType<>(10, SIUnits.CELSIUS)));
@@ -231,7 +228,6 @@ public Node createSpyNode(InvocationOnMock invocation) {
     }
 
     @SuppressWarnings("null")
-    @Disabled("Temporarily disabled: unstable")
     @Test
     public void parseHomieTree() throws Exception {
         // Create a Homie Device object. Because spied Nodes are required for call verification,
@@ -297,14 +293,24 @@ public void parseHomieTree() throws Exception {
         device.startChannels(homieConnection, scheduler, 50, handler).get();
         assertThat(propertyBell.getChannelState().isStateful(), is(false));
         assertThat(propertyBell.getChannelState().getCache().getChannelState(), is(UnDefType.UNDEF));
-        assertThat(property.getChannelState().getCache().getChannelState(),
-                is(new QuantityType<>(10, SIUnits.CELSIUS)));
+        Property temperatureProperty = property;
+        waitForAssert(() -> assertThat(temperatureProperty.getChannelState().getCache().getChannelState(),
+                is(new QuantityType<>(10, SIUnits.CELSIUS))));
 
         property = node.properties.get("testRetain");
-        WaitForTopicValue watcher = new WaitForTopicValue(brokerConnection, propertyTestTopic + "/set");
+        String commandTopic = propertyTestTopic + "/set";
+        CompletableFuture<String> receivedValue = new CompletableFuture<>();
+        MqttMessageSubscriber subscriber = (topic, payload) -> receivedValue
+                .complete(new String(payload, StandardCharsets.UTF_8));
+        assertTrue(brokerConnection.subscribe(commandTopic, subscriber).get(5, TimeUnit.SECONDS));
+
         // Watch the topic. Publish a retain=false value to MQTT
-        property.getChannelState().publishValue(OnOffType.OFF).get();
-        assertThat(watcher.waitForTopicValue(10000), is("false"));
+        try {
+            property.getChannelState().publishValue(OnOffType.OFF).get();
+            assertThat(receivedValue.get(MQTT_TIMEOUT, TimeUnit.MILLISECONDS), is("false"));
+        } finally {
+            brokerConnection.unsubscribe(commandTopic, subscriber).get(5, TimeUnit.SECONDS);
+        }
 
         // Publish a retain=false value to MQTT.
         property.getChannelState().publishValue(OnOffType.ON).get();