Test device deletion support

config/test.exs

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 Ispirata Srl
+# Copyright 2017 - 2023 SECO Mind Srl
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -26,9 +26,6 @@ config :lager,
 
 config :astarte_vmq_plugin, :amqp_options, host: System.get_env("RABBITMQ_HOST") || "rabbitmq"
 
-# make amqp supervisors logs less verbose
-config :logger, handle_otp_reports: false
-
 config :logger, :console,
   format: {PrettyLog.UserFriendlyFormatter, :format},
   metadata: [:function]

test/astarte_vmq_plugin/config_test.exs

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 Ispirata Srl
+# Copyright 2017 - 2023 SECO Mind Srl
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -53,4 +53,19 @@ defmodule Astarte.VMQ.Plugin.ConfigTest do
       Application.put_env(:astarte_vmq_plugin, :data_queue_prefix, old_data_queue_prefix)
     end)
   end
+
+  test "config init correctly converts cassandra_nodes to elixir string" do
+    erlang_cassandra_nodes = ['something']
+
+    elixir_cassandra_nodes = ["something"]
+
+    Application.put_env(:astarte_vmq_plugin, :cassandra_nodes, erlang_cassandra_nodes)
+    Config.init()
+
+    assert elixir_cassandra_nodes == Config.xandra_options!()[:nodes]
+
+    on_exit(fn ->
+      Application.put_env(:astarte_vmq_plugin, :data_queue_prefix, elixir_cassandra_nodes)
+    end)
+  end
 end

test/astarte_vmq_plugin_test.exs

@@ -1,7 +1,7 @@
 #
 # This file is part of Astarte.
 #
-# Copyright 2017 Ispirata Srl
+# Copyright 2017 - 2023 SECO Mind Srl
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -23,11 +23,13 @@ defmodule Astarte.VMQ.PluginTest do
   alias AMQP.{Channel, Connection, Queue}
   alias Astarte.VMQ.Plugin
   alias Astarte.VMQ.Plugin.Config
+  alias Astarte.VMQ.Plugin.DatabaseTestHelper
 
   @device_id :crypto.strong_rand_bytes(16) |> Base.url_encode64(padding: false)
   @other_device_id :crypto.strong_rand_bytes(16) |> Base.url_encode64(padding: false)
   @realm "test"
   @device_base_path "#{@realm}/#{@device_id}"
+  @other_device_base_path "#{@realm}/#{@other_device_id}"
   @other_mqtt_user "other"
   @another_mqtt_user "another"
   @queue_name "#{Config.data_queue_prefix()}0"
@@ -37,6 +39,9 @@ defmodule Astarte.VMQ.PluginTest do
     {:ok, conn} = Connection.open(amqp_opts)
     {:ok, chan} = Channel.open(conn)
     Queue.declare(chan, @queue_name)
+    :ok = DatabaseTestHelper.await_xandra_cluster_connected!()
+    DatabaseTestHelper.setup_db!()
+    on_exit(&DatabaseTestHelper.teardown_db!/0)
     {:ok, chan: chan}
   end
 
@@ -55,28 +60,60 @@ defmodule Astarte.VMQ.PluginTest do
     :ok
   end
 
-  test "auth_on_register for a device" do
-    assert {:ok, modifiers} =
-             Plugin.auth_on_register(
-               :dontcare,
-               {"/", :dontcare},
-               @device_base_path,
-               :dontcare,
-               :dontcare
-             )
+  describe "auth_on_register" do
+    test "for an existing device succeeds" do
+      DatabaseTestHelper.insert_device_into_devices!(@device_id)
+
+      assert {:ok, modifiers} =
+               Plugin.auth_on_register(
+                 :dontcare,
+                 {"/", :dontcare},
+                 @device_base_path,
+                 :dontcare,
+                 :dontcare
+               )
+
+      assert Keyword.get(modifiers, :subscriber_id) == {"/", @device_base_path}
+      DatabaseTestHelper.cleanup_db!()
+    end
 
-    assert Keyword.get(modifiers, :subscriber_id) == {"/", @device_base_path}
-  end
+    test "for a non-existing device fails" do
+      assert {:error, :device_does_not_exist} =
+               Plugin.auth_on_register(
+                 :dontcare,
+                 {"/", :dontcare},
+                 @other_device_base_path,
+                 :dontcare,
+                 :dontcare
+               )
+    end
 
-  test "auth_on_register for non-devices" do
-    assert :next =
-             Plugin.auth_on_register(
-               :dontcare,
-               {"/", :dontcare},
-               @other_mqtt_user,
-               :dontcare,
-               :dontcare
-             )
+    test "for an existing device that's being deleted fails" do
+      DatabaseTestHelper.insert_device_into_devices!(@other_device_id)
+      DatabaseTestHelper.insert_device_into_deletion_in_progress!(@other_device_id)
+
+      assert {:error, :device_deletion_in_progress} =
+               Plugin.auth_on_register(
+                 :dontcare,
+                 {"/", :dontcare},
+                 @other_device_base_path,
+                 :dontcare,
+                 :dontcare
+               )
+
+      DatabaseTestHelper.cleanup_db!()
+    end
+
+    test "ignores non-devices" do
+      assert :next =
+               Plugin.auth_on_register(
+                 :dontcare,
+                 {"/", :dontcare},
+                 @other_mqtt_user,
+                 :dontcare,
+                 :dontcare
+               )
+    end
   end
 
   test "partially authorized auth_on_subscribe for devices" do

test/support/database_test_helper.ex

@@ -0,0 +1,130 @@
+#
+# This file is part of Astarte.
+#
+# Copyright 2023 SECO Mind Srl
+#
+# Licensed 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
+#
+#    http://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.
+#
+
+defmodule Astarte.VMQ.Plugin.DatabaseTestHelper do
+  require Logger
+  alias Astarte.Core.Device
+  import ExUnit.Assertions
+
+  @create_test_keyspace """
+    CREATE KEYSPACE test
+      WITH
+        replication = {'class': 'SimpleStrategy', 'replication_factor': '1'} AND
+        durable_writes = true;
+  """
+
+  @create_devices_table """
+  CREATE TABLE test.devices (
+    device_id uuid,
+    PRIMARY KEY (device_id)
+  );
+  """
+
+  @create_deletion_in_progress_table """
+  CREATE TABLE test.deletion_in_progress (
+    device_id uuid,
+    vmq_ack boolean,
+    PRIMARY KEY (device_id)
+  );
+  """
+
+  @insert_device_into_devices """
+    INSERT INTO test.devices (device_id)
+      VALUES (:device_id);
+  """
+
+  @insert_device_into_deletion_in_progress """
+    INSERT INTO test.deletion_in_progress (device_id, vmq_ack)
+      VALUES (:device_id, :vmq_ack);
+  """
+
+  @truncate_devices_table """
+  TRUNCATE test.devices;
+  """
+
+  @truncate_deletion_in_progress_table """
+  TRUNCATE test.deletion_in_progress;
+  """
+
+  @drop_test_keyspace """
+  DROP KEYSPACE test;
+  """
+
+  def setup_db!() do
+    Xandra.Cluster.run(:xandra, fn conn ->
+      Xandra.execute!(conn, @create_test_keyspace, %{}, consistency: :local_quorum)
+      Xandra.execute!(conn, @create_devices_table, %{}, consistency: :local_quorum)
+      Xandra.execute!(conn, @create_deletion_in_progress_table, %{}, consistency: :local_quorum)
+    end)
+  end
+
+  def insert_device_into_devices!(encoded_device_id) do
+    Xandra.Cluster.run(:xandra, fn conn ->
+      {:ok, device_id} = Device.decode_device_id(encoded_device_id)
+
+      params = %{
+        "device_id" => device_id
+      }
+
+      prepared = Xandra.prepare!(conn, @insert_device_into_devices)
+      Xandra.execute!(conn, prepared, params, consistency: :quorum)
+    end)
+  end
+
+  def insert_device_into_deletion_in_progress!(encoded_device_id) do
+    Xandra.Cluster.run(:xandra, fn conn ->
+      {:ok, device_id} = Device.decode_device_id(encoded_device_id)
+
+      params = %{
+        "device_id" => device_id,
+        "vmq_ack" => false
+      }
+
+      prepared = Xandra.prepare!(conn, @insert_device_into_deletion_in_progress)
+      Xandra.execute!(conn, prepared, params, consistency: :quorum)
+    end)
+  end
+
+  def cleanup_db!() do
+    Xandra.Cluster.run(:xandra, fn conn ->
+      Xandra.execute!(conn, @truncate_devices_table, %{}, consistency: :local_quorum)
+      Xandra.execute!(conn, @truncate_deletion_in_progress_table, %{}, consistency: :local_quorum)
+    end)
+  end
+
+  def teardown_db!() do
+    Xandra.Cluster.run(:xandra, fn conn ->
+      {:ok, %Xandra.SchemaChange{}} = Xandra.execute(conn, @drop_test_keyspace)
+    end)
+  end
+
+  def await_xandra_cluster_connected!(tries \\ 10) do
+    case Xandra.Cluster.run(:xandra, &Xandra.execute(&1, "SELECT * FROM system.local")) do
+      {:error, %Xandra.ConnectionError{}} ->
+        if tries > 0 do
+          Process.sleep(100)
+          await_xandra_cluster_connected!(tries - 1)
+        else
+          flunk("exceeded maximum number of attempts")
+        end
+
+      _other ->
+        :ok
+    end
+  end
+end