Local test MQTT broker

.github/workflows/elixir.yml

@@ -14,8 +14,19 @@ jobs:
     name: Build and test
     runs-on: ubuntu-22.04
 
+    services:
+      mosquitto:
+        image: eclipse-mosquitto:2.0
+        ports:
+          - 1883:1883
+        volumes:
+          - ./mosquitto:/mosquitto/config/
+        options: --name mqtt
+
     steps:
       - uses: actions/checkout@v2
+      - name: Restart MQTT to load mosquitto/mosquitto.conf from checkout
+        run: docker restart mqtt
       # cache the ASDF directory, using the values from .tool-versions
       - name: ASDF cache
         uses: actions/cache@v2

README.md

@@ -28,10 +28,11 @@ ln -s ../../hooks/pre-commit .git/hooks/pre-commit
 # make sure everything passes! (slowest to fastest)
 mix format --check-formatted
 mix credo
-mix test
 mix dialyzer
 ```
 
+See [the section on tests below](#tests) for information on running unit tests, which requires having a local MQTT broker running.
+
 If you run into issues compiling `snabbkaffe`:
 
 ``` shell
@@ -40,6 +41,22 @@ mix deps.get
 mix deps.compile
 ```
 
+### Tests
+[tests]: #tests
+
+To run the tests, first install and setup Colima, Docker, and docker-compose:
+
+```shell
+brew install docker docker-compose colima
+colima start
+mkdir -p ${DOCKER_CONFIG:-"~/.docker"}/cli-plugins
+ln -sfn /opt/homebrew/opt/docker-compose/bin/docker-compose ${DOCKER_CONFIG:-"~/.docker"}/cli-plugins/docker-compose
+```
+
+Then, start the Compose configuration in a separate window or tab and run the tests: 
+1. `docker compose up` 
+1. `mix test`
+
 ## Docker
 
 Concentrate comes with a Dockerfile, allowing you to build an image that can

docker-compose.yml

@@ -0,0 +1,8 @@
+services:
+  mosquitto:
+    image: "eclipse-mosquitto:2.0"
+    container_name: "concentrate-dev-mosquitto"
+    ports:
+      - "1883:1883"
+    volumes:
+      - ./mosquitto:/mosquitto/config/

mosquitto/mosquitto.conf

@@ -0,0 +1,5 @@
+listener 1883
+allow_anonymous true
+# simple password file created with the mosquitto_passwd utility
+# defines a single test_user with password test_password
+password_file /mosquitto/config/passwd

mosquitto/passwd

@@ -0,0 +1 @@
+test_user:$7$101$qoz9oyG9w86ZBioJ$tx75ylhhYLX3FTf8fdUrwf9+ISmWpA4EO2SNnCqqJYXzmpnG4tM3Lm7Qh8NwpdzSi03zl1CDA/2xrbHmkRtS7g==

test/concentrate/producer/mqtt_test.exs

@@ -9,62 +9,92 @@ defmodule Concentrate.Producer.MqttTest do
     on_exit(fn -> Logger.configure(level: old_level) end)
     Logger.configure(level: :warning)
 
-    :ok
+    {:ok, emqtt_writer} =
+      :emqtt.start_link(%{
+        host: "localhost",
+        port: 1883
+      })
+
+    {:ok, _} = :emqtt.connect(emqtt_writer)
+
+    test_topic = "concentrate/test_producer/#{System.unique_integer()}"
+
+    {:ok, %{emqtt_writer: emqtt_writer, test_topic: test_topic}}
   end
 
-  test "can dispatch events from an MQTT stream" do
+  test "can dispatch events from an MQTT stream", %{
+    emqtt_writer: emqtt_writer,
+    test_topic: test_topic
+  } do
     config = {
-      url = "mqtt+ssl://test.mosquitto.org:8886",
-      topics: ["home/#"], parser: __MODULE__.PassThroughParser
+      url = "mqtt://localhost:1883",
+      topics: [test_topic], parser: __MODULE__.PassThroughParser
     }
 
     {:ok, pid} = Mqtt.start_link(config)
 
+    :emqtt.publish(emqtt_writer, test_topic, %{}, "payload", qos: 1, retain: true)
+
     [[{:parsed, body, opts}]] = Enum.take(GenStage.stream([pid]), 1)
     assert <<_::binary>> = body
     assert String.starts_with?(Keyword.fetch!(opts, :feed_url), url <> "/")
   end
 
-  test "can accept a function as a parser" do
+  test "can accept a function as a parser", %{
+    emqtt_writer: emqtt_writer,
+    test_topic: test_topic
+  } do
     config = {
-      "mqtt+ssl://test.mosquitto.org:8886",
-      topics: ["home/#"], parser: &__MODULE__.PassThroughParser.parse/2
+      "mqtt://localhost:1883",
+      topics: [test_topic], parser: &__MODULE__.PassThroughParser.parse/2
     }
 
     {:ok, pid} = Mqtt.start_link(config)
 
+    :emqtt.publish(emqtt_writer, test_topic, %{}, "payload", qos: 1, retain: true)
+
     [[{:parsed, body, _opts}]] = Enum.take(GenStage.stream([pid]), 1)
     assert <<_::binary>> = body
   end
 
-  test "can authenticate with a password" do
+  test "can authenticate with a password", %{
+    emqtt_writer: emqtt_writer,
+    test_topic: test_topic
+  } do
     config = {
-      "mqtt://test.mosquitto.org:1884",
-      username: "ro",
-      password: "readonly",
-      topics: ["home/#"],
+      "mqtt://localhost:1883",
+      username: "test_user",
+      password: "test_password",
+      topics: [test_topic],
       parser: __MODULE__.PassThroughParser
     }
 
     {:ok, pid} = Mqtt.start_link(config)
 
+    :emqtt.publish(emqtt_writer, test_topic, %{}, "payload", qos: 1, retain: true)
+
     assert [[{:parsed, _, _}]] = Enum.take(GenStage.stream([pid]), 1)
   end
 
   # we expect a warning here
   @tag :capture_log
-  test "can authenticate with one of multiple passwords" do
+  test "can authenticate with one of multiple passwords", %{
+    emqtt_writer: emqtt_writer,
+    test_topic: test_topic
+  } do
     config = {
-      "mqtt://test.mosquitto.org:1884",
-      username: "ro",
-      password: "notvalid readonly",
-      topics: ["home/#"],
+      "mqtt://localhost:1883",
+      username: "test_user",
+      password: "notvalid test_password",
+      topics: [test_topic],
       backoff: 0,
       parser: __MODULE__.PassThroughParser
     }
 
     {:ok, pid} = Mqtt.start_link(config)
 
+    :emqtt.publish(emqtt_writer, test_topic, %{}, "payload", qos: 1, retain: true)
+
     assert [[{:parsed, _, _}]] = Enum.take(GenStage.stream([pid]), 1)
   end
 
@@ -73,18 +103,14 @@ defmodule Concentrate.Producer.MqttTest do
     payload = "payload"
 
     config = {
-      "mqtt+ssl://test.mosquitto.org:8886",
+      "mqtt://localhost:1883",
       topics: [test_topic], parser: __MODULE__.PassThroughParser
     }
 
     {:ok, writer} =
       :emqtt.start_link(%{
-        host: "test.mosquitto.org",
-        port: 8885,
-        username: "wo",
-        password: "writeonly",
-        ssl: true,
-        ssl_opts: [verify: :verify_none]
+        host: "localhost",
+        port: 1883
       })
 
     {:ok, _} = :emqtt.connect(writer)

test/concentrate/sink/mqtt_test.exs

@@ -9,7 +9,7 @@ defmodule Concentrate.Sink.MqttTest do
     prefix = "concentrate/test_sink/#{System.unique_integer()}/"
 
     config = [
-      url: "mqtt://test.mosquitto.org",
+      url: "mqtt://localhost:1883",
       prefix: prefix
     ]