neiam /lambent_ex

defmodule LambentExWeb.MQTTTest do
  use LambentExWeb.ConnCase

  import Phoenix.LiveViewTest

  alias LambentEx.Meta
  alias LambentEx.MQTTSupervisor

  setup do
    on_exit(fn ->
      Meta.get_saved_cfg_mqtt()
      |> Enum.filter(fn m -> String.starts_with?(m[:opts][:name], "mqtt-test") end)
      |> Enum.each(fn m -> MQTTSupervisor.remove(m[:opts][:name]) end)
    end)

    :ok
  end

  defp saved(name) do
    Meta.get_saved_cfg_mqtt() |> Enum.find(fn m -> m[:opts][:name] == name end)
  end

  defp running?(name), do: Registry.lookup(:lambent_mqtt, "mqtt-#{name}") != []

  # answers one CONNECT with the given CONNACK return code, and hands the
  # decoded CONNECT back to the test
  defp fake_broker(return_code) do
    {:ok, listen} = :gen_tcp.listen(0, [:binary, packet: :raw, active: false, reuseaddr: true])
    {:ok, port} = :inet.port(listen)
    test = self()

    spawn_link(fn ->
      {:ok, sock} = :gen_tcp.accept(listen, 5000)
      {:ok, connect} = :gen_tcp.recv(sock, 0, 5000)
      send(test, {:connect, Tortoise.Package.Connect.decode(connect)})
      :ok = :gen_tcp.send(sock, <<0x20, 2, 0, return_code>>)
      :gen_tcp.recv(sock, 0, 1000)
      :gen_tcp.close(sock)
      :gen_tcp.close(listen)
    end)

    port
  end

  # enough of a broker to keep tortoise happy: accept the CONNECT, then ack
  # every SUBSCRIBE and PINGREQ that comes in
  defp ackging_broker do
    {:ok, listen} = :gen_tcp.listen(0, [:binary, packet: :raw, active: false, reuseaddr: true])
    {:ok, port} = :inet.port(listen)

    pid =
      spawn(fn ->
        {:ok, sock} = :gen_tcp.accept(listen, 5000)
        {:ok, _connect} = :gen_tcp.recv(sock, 0, 5000)
        :ok = :gen_tcp.send(sock, <<0x20, 2, 0, 0>>)
        serve(sock, <<>>)
      end)

    on_exit(fn ->
      Process.exit(pid, :kill)
      :gen_tcp.close(listen)
    end)

    port
  end

  defp serve(sock, buf) do
    case buf do
      # every packet here is short, so the length fits in one byte
      <<type::4, _flags::4, len, body::binary-size(len), rest::binary>> ->
        case type do
          8 ->
            <<id::16, _topics::binary>> = body
            :gen_tcp.send(sock, <<0x90, 3, id::16, 0>>)

          12 ->
            :gen_tcp.send(sock, <<0xD0, 0>>)

          _other ->
            :ok
        end

        serve(sock, rest)

      _partial ->
        case :gen_tcp.recv(sock, 0) do
          {:ok, more} -> serve(sock, buf <> more)
          {:error, _closed} -> :ok
        end
    end
  end

  test "a broker that connects and subscribes stays up", %{conn: _conn} do
    port = ackging_broker()
    Phoenix.PubSub.subscribe(LambentEx.PubSub, "svc_idx")

    {:ok, pid} = MQTTSupervisor.add("mqtt-test-up", "127.0.0.1", port)
    ref = Process.monitor(pid)

    assert_receive {:mqtt_pub, %{name: "mqtt-test-up", connected: :up}}, 3000
    # the subscribe acks land just after the connection comes up
    refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 1000
  end

  describe "probe" do
    alias LambentEx.MQTT.Probe

    test "a broker that accepts is ok, and is sent the login" do
      port = fake_broker(0)

      assert :ok = Probe.check("127.0.0.1", port, user: "lex", password: "hunter2")
      assert_receive {:connect, connect}
      assert connect.user_name == "lex"
      assert connect.password == "hunter2"
    end

    test "no login is sent when none is given" do
      port = fake_broker(0)

      assert :ok = Probe.check("127.0.0.1", port, user: "", password: nil)
      assert_receive {:connect, connect}
      assert connect.user_name == nil
      assert connect.password == nil
    end

    test "a refused login says so" do
      port = fake_broker(4)
      assert {:error, :bad_user_name_or_password} = Probe.check("127.0.0.1", port)
    end

    test "nothing listening says so" do
      assert {:error, :econnrefused} = Probe.check("127.0.0.1", 1)
    end
  end

  test "testing from the form reports what the broker said", %{conn: conn} do
    port = fake_broker(5)
    {:ok, live, _html} = live(conn, "/cfg/services/mqtt")

    live
    |> form("#mqtt-form", mqtt: %{host: "127.0.0.1", port: "#{port}", user: "lex", password: "x"})
    |> render_change()

    live |> element("#mqtt-test") |> render_click()
    assert_receive {:connect, %{user_name: "lex"}}

    assert render_async(live) =~ "Not authorized"
  end

  test "testing from the form says when it connects", %{conn: conn} do
    port = fake_broker(0)
    {:ok, live, _html} = live(conn, "/cfg/services/mqtt")

    live |> form("#mqtt-form", mqtt: %{host: "127.0.0.1", port: "#{port}"}) |> render_change()
    live |> element("#mqtt-test") |> render_click()

    assert render_async(live) =~ "Connected"
  end

  test "a password without a user is turned away", %{conn: conn} do
    {:ok, live, _html} = live(conn, "/cfg/services/mqtt")

    html =
      live
      |> form("#mqtt-form",
        mqtt: %{name: "mqtt-test-nouser", host: "127.0.0.1", port: "1", password: "x"}
      )
      |> render_submit()

    assert html =~ "needed with a password"
    refute saved("mqtt-test-nouser")
  end

  test "a login is saved with the broker", %{conn: conn} do
    {:ok, live, _html} = live(conn, "/cfg/services/mqtt")

    live
    |> form("#mqtt-form",
      mqtt: %{name: "mqtt-test-auth", host: "127.0.0.1", port: "1", user: "lex", password: "pw"}
    )
    |> render_submit()

    assert %{opts: opts} = saved("mqtt-test-auth")
    assert opts[:user] == "lex"
    assert opts[:password] == "pw"
    assert running?("mqtt-test-auth")
  end

  test "the broker form renders", %{conn: conn} do
    {:ok, _live, html} = live(conn, "/cfg/services/mqtt")

    assert html =~ "Configure MQTT Broker"
    assert html =~ "mqtt-form"
    assert html =~ "1883"
  end

  test "a broker is started and saved in the shape the supervisor replays", %{conn: conn} do
    {:ok, live, _html} = live(conn, "/cfg/services/mqtt")

    # nothing listens on port 1, tortoise just keeps backing off
    live
    |> form("#mqtt-form", mqtt: %{name: "mqtt-test-a", host: "127.0.0.1", port: "1"})
    |> render_submit()

    assert %{opts: opts} = saved("mqtt-test-a")
    assert opts[:host] == "127.0.0.1"
    assert opts[:port] == 1
    assert opts[:persist] == true
    # left blank, so nothing is stored for them
    refute Keyword.has_key?(opts, :user)
    refute Keyword.has_key?(opts, :password)
    assert running?("mqtt-test-a")
  end

  test "a name already in use is turned away", %{conn: conn} do
    {:ok, _pid} = MQTTSupervisor.add("mqtt-test-dup", "127.0.0.1", 1)
    {:ok, live, _html} = live(conn, "/cfg/services/mqtt")

    html =
      live
      |> form("#mqtt-form", mqtt: %{name: "mqtt-test-dup", host: "127.0.0.2", port: "1"})
      |> render_submit()

    assert html =~ "is already a broker"
    assert saved("mqtt-test-dup")[:opts][:host] == "127.0.0.1"
  end

  test "a bad port is not saved", %{conn: conn} do
    {:ok, live, _html} = live(conn, "/cfg/services/mqtt")

    html =
      live
      |> form("#mqtt-form", mqtt: %{name: "mqtt-test-port", host: "127.0.0.1", port: "70000"})
      |> render_submit()

    assert html =~ "must be less than 65536"
    refute saved("mqtt-test-port")
    refute running?("mqtt-test-port")
  end

  test "deleting a broker stops it and forgets it", %{conn: conn} do
    {:ok, live, _html} = live(conn, "/cfg/services")
    Phoenix.PubSub.subscribe(LambentEx.PubSub, "svc_idx")
    {:ok, _pid} = MQTTSupervisor.add("mqtt-test-del", "127.0.0.1", 1)
    # the index only learns of a broker from its publishes
    assert_receive {:mqtt_pub, %{name: "mqtt-test-del"}}, 1000

    assert render(live) =~ "mqtt-test-del"

    live |> element("a[phx-value-tgt=mqtt-test-del]") |> render_click()

    refute running?("mqtt-test-del")
    refute saved("mqtt-test-del")
    refute render(live) =~ "mqtt-test-del"
  end
end