spacecruft
/
farmbot_os
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Update `mox` to be used correctly. 

I previously misunderstood how the black magic of mox actually works.
This updates `farmbot_ext` to not require setting excess data in every
config.exs entry. Also removes calls to `Application.get_env/2`
pull/974/head
Connor Rigby 2019-06-05 09:21:56 -07:00
parent 2d66f09485
commit 26578a7ae2
No known key found for this signature in database
GPG Key ID: 29A88B24B70456E0
11 changed files with 179 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
farmbot_ext/config/config.exs
View File

@ -9,4 +9,3 @@ config :farmbot_celery_script, FarmbotCeleryScript.SysCalls,
import_config "ecto.exs"
import_config "farmbot_core.exs"
import_config "lagger.exs"
import_config "#{Mix.env()}.exs"

1
farmbot_ext/config/dev.exs
View File

@ -1 +0,0 @@
use Mix.Config

1
farmbot_ext/config/prod.exs
View File

@ -1 +0,0 @@
use Mix.Config

6
farmbot_ext/config/test.exs
View File

@ -1,6 +0,0 @@
use Mix.Config
config :farmbot_ext, FarmbotExt.API.Preloader, preloader_impl: MockPreloader
config :farmbot_ext, FarmbotExt.AMQP.ConnectionWorker, network_impl: MockConnectionWorker
config :farmbot_ext, FarmbotExt.AMQP.AutoSyncChannel, query_impl: MockQuery
config :farmbot_ext, FarmbotExt.AMQP.AutoSyncChannel, command_impl: MockCommand

18
farmbot_ext/lib/farmbot_ext/amqp/auto_sync_channel.ex
View File

@ -103,9 +103,9 @@ defmodule FarmbotExt.AMQP.AutoSyncChannel do
end
def handle_asset(asset_kind, id, params) do
if query().auto_sync?() do
if Asset.Query.auto_sync?() do
:ok = BotState.set_sync_status("syncing")
command().update(asset_kind, params, id)
Asset.Command.update(asset_kind, params, id)
:ok = BotState.set_sync_status("synced")
else
cache_sync(asset_kind, params, id)
@ -114,13 +114,13 @@ defmodule FarmbotExt.AMQP.AutoSyncChannel do
def cache_sync(kind, params, id) when kind in @cache_kinds do
:ok = BotState.set_sync_status("syncing")
:ok = command().update(kind, params, id)
:ok = Asset.Command.update(kind, params, id)
:ok = BotState.set_sync_status("synced")
end
def cache_sync(asset_kind, params, id) do
Logger.info("Autocaching sync #{asset_kind} #{id} #{inspect(params)}")
changeset = command().new_changeset(asset_kind, id, params)
changeset = Asset.Command.new_changeset(asset_kind, id, params)
:ok = EagerLoader.cache(changeset)
:ok = BotState.set_sync_status("sync_now")
end
@ -136,14 +136,4 @@ defmodule FarmbotExt.AMQP.AutoSyncChannel do
{:noreply, %{state | conn: nil, chan: nil}}
end
defp query() do
mod = Application.get_env(:farmbot_ext, __MODULE__) || []
Keyword.get(mod, :query_impl, Asset.Query)
end
defp command() do
mod = Application.get_env(:farmbot_ext, __MODULE__) || []
Keyword.get(mod, :command_impl, Asset.Command)
end
end

84
farmbot_ext/lib/farmbot_ext/amqp/connection_worker.ex
View File

@ -5,30 +5,75 @@ defmodule FarmbotExt.AMQP.ConnectionWorker do
use GenServer
require Logger
alias FarmbotExt.JWT
alias FarmbotCore.Project
alias FarmbotExt.AMQP.ConnectionWorker
alias AMQP.{Basic, Channel, Queue}
alias FarmbotExt.{JWT, AMQP.ConnectionWorker}
alias FarmbotCore.{Project, JSON}
@exchange "amq.topic"
@type connection :: map()
@type channel :: map()
defstruct [:opts, :conn]
@doc "Get the current active connection"
@callback connection(GenServer.server()) :: connection()
def connection(connection_worker \\ __MODULE__) do
GenServer.call(connection_worker, :connection)
end
@doc "Cleanly close an AMQP channel"
@callback close_channel(channel) :: nil
def close_channel(chan) do
Channel.close(chan)
end
@doc "Takes the 'bot' claim seen in the JWT and connects to the AMQP broker."
@callback maybe_connect(String.t()) :: connection() | {:error, term()}
def maybe_connect(jwt_dot_bot) do
bot = jwt_dot_bot
auto_sync = bot <> "_auto_sync"
route = "bot.#{bot}.sync.#"
with %{} = conn <- FarmbotExt.AMQP.ConnectionWorker.connection(),
{:ok, chan} <- Channel.open(conn),
:ok <- Basic.qos(chan, global: true),
{:ok, _} <- Queue.declare(chan, auto_sync, auto_delete: false),
:ok <- Queue.bind(chan, auto_sync, @exchange, routing_key: route),
{:ok, _} <- Basic.consume(chan, auto_sync, self(), no_ack: true) do
%{conn: conn, chan: chan}
else
nil -> %{conn: nil, chan: nil}
error -> error
end
end
@doc "Respond with an OK message to a CeleryScript(TM) RPC message."
@callback rpc_reply(map(), String.t(), String.t()) :: :ok
def rpc_reply(chan, jwt_dot_bot, label) do
json = JSON.encode!(%{args: %{label: label}, kind: "rpc_ok"})
Basic.publish(chan, @exchange, "bot.#{jwt_dot_bot}.from_device", json)
end
@doc false
def start_link(args, opts \\ [name: __MODULE__]) do
GenServer.start_link(__MODULE__, args, opts)
end
def connection do
GenServer.call(__MODULE__, :connection)
end
@impl GenServer
def init(opts) do
Process.flag(:sensitive, true)
Process.flag(:trap_exit, true)
{:ok, %ConnectionWorker{conn: nil, opts: opts}, 0}
end
@impl GenServer
def terminate(reason, %{conn: nil}) do
Logger.info("AMQP connection not open: #{inspect(reason)}")
end
@impl GenServer
def terminate(reason, %{conn: conn}) do
if Process.alive?(conn.pid) do
try do
@ -42,6 +87,7 @@ defmodule FarmbotExt.AMQP.ConnectionWorker do
end
end
@impl GenServer
def handle_info(:timeout, state) do
token = Keyword.fetch!(state.opts, :token)
email = Keyword.fetch!(state.opts, :email)
@ -64,9 +110,12 @@ defmodule FarmbotExt.AMQP.ConnectionWorker do
{:stop, reason, conn}
end
def handle_call(:connection, _, %{conn: conn} = state), do: {:reply, conn, state}
@impl GenServer
def handle_call(:connection, _, %{conn: conn} = state) do
{:reply, conn, state}
end
def open_connection(token, email, bot, mqtt_server, vhost) do
defp open_connection(token, email, bot, mqtt_server, vhost) do
Logger.info("Opening new AMQP connection.")
# Make sure the types of these fields are correct. If they are not
@ -92,21 +141,4 @@ defmodule FarmbotExt.AMQP.ConnectionWorker do
AMQP.Connection.open(opts)
end
def network_impl() do
mod = Application.get_env(:farmbot_ext, __MODULE__) || []
Keyword.get(mod, :network_impl, FarmbotExt.AMQP.ConnectionWorker.Network)
end
def maybe_connect(jwt_bot_claim) do
network_impl().maybe_connect(jwt_bot_claim)
end
def close_channel(channel) do
network_impl().close_channel(channel)
end
def rpc_reply(chan, jwt_dot_bot, label) do
network_impl().rpc_reply(chan, jwt_dot_bot, label)
end
end

42
farmbot_ext/lib/farmbot_ext/amqp/connection_worker/network.ex
View File

@ -1,42 +0,0 @@
defmodule FarmbotExt.AMQP.ConnectionWorker.Network do
@moduledoc """
Real-world implementation of AMQP socket IO handlers.
"""
alias AMQP.{Basic, Channel, Queue}
alias FarmbotCore.JSON
@exchange "amq.topic"
@doc "Cleanly close an AMQP channel"
@callback close_channel(map()) :: nil
def close_channel(chan) do
Channel.close(chan)
end
@doc "Takes the 'bot' claim seen in the JWT and connects to the AMQP broker."
@callback maybe_connect(String.t()) :: map()
def maybe_connect(jwt_dot_bot) do
bot = jwt_dot_bot
auto_sync = bot <> "_auto_sync"
route = "bot.#{bot}.sync.#"
with %{} = conn <- FarmbotExt.AMQP.ConnectionWorker.connection(),
{:ok, chan} <- Channel.open(conn),
:ok <- Basic.qos(chan, global: true),
{:ok, _} <- Queue.declare(chan, auto_sync, auto_delete: false),
:ok <- Queue.bind(chan, auto_sync, @exchange, routing_key: route),
{:ok, _} <- Basic.consume(chan, auto_sync, self(), no_ack: true) do
%{conn: conn, chan: chan}
else
nil -> %{conn: nil, chan: nil}
error -> error
end
end
@doc "Respond with an OK message to a CeleryScript(TM) RPC message."
@callback rpc_reply(map(), String.t(), String.t()) :: :ok
def rpc_reply(chan, jwt_dot_bot, label) do
json = JSON.encode!(%{args: %{label: label}, kind: "rpc_ok"})
Basic.publish(chan, @exchange, "bot.#{jwt_dot_bot}.from_device", json)
end
end

67
farmbot_ext/lib/farmbot_ext/api/preloader.ex
View File

@ -6,13 +6,70 @@ defmodule FarmbotExt.API.Preloader do
* FarmbotCore.Asset.FirmwareConfig
"""
alias Ecto.{Changeset, Multi}
require FarmbotCore.Logger
alias FarmbotExt.API
alias API.{Reconciler, SyncGroup, EagerLoader}
alias FarmbotCore.{
Asset,
Asset.Repo,
Asset.Sync
}
@doc """
Syncronous call to sync or preload assets.
Starts with `group_0` to check if `auto_sync` is enabled. If it is,
actually sync all resources. If it is not, preload all resources.
"""
@callback preload_all :: :ok | :error
def preload_all do
preloader_impl().preload_all()
def preload_all() do
sync_changeset = API.get_changeset(Sync)
sync = Changeset.apply_changes(sync_changeset)
multi = Multi.new()
with {:ok, multi} <- Reconciler.sync_group(multi, sync, SyncGroup.group_0()),
{:ok, _} <- Repo.transaction(multi) do
auto_sync_change =
Enum.find_value(multi.operations, fn {{key, _id}, {:changeset, change, []}} ->
key == :fbos_configs && Changeset.get_change(change, :auto_sync)
end)
FarmbotCore.Logger.success(3, "Successfully synced bootup resources.")
:ok = maybe_auto_sync(sync_changeset, auto_sync_change || Asset.fbos_config().auto_sync)
end
end
defp preloader_impl() do
mod = Application.get_env(:farmbot_ext, __MODULE__) || []
Keyword.get(mod, :preloader_impl, FarmbotExt.API.Preloader.HTTP)
# When auto_sync is enabled, do the full sync.
defp maybe_auto_sync(sync_changeset, true) do
FarmbotCore.Logger.busy(3, "bootup auto sync")
sync = Changeset.apply_changes(sync_changeset)
multi = Multi.new()
with {:ok, multi} <- Reconciler.sync_group(multi, sync, SyncGroup.group_1()),
{:ok, multi} <- Reconciler.sync_group(multi, sync, SyncGroup.group_2()),
{:ok, multi} <- Reconciler.sync_group(multi, sync, SyncGroup.group_3()),
{:ok, multi} <- Reconciler.sync_group(multi, sync, SyncGroup.group_4()) do
Multi.insert(multi, :syncs, sync_changeset)
|> Repo.transaction()
FarmbotCore.Logger.success(3, "bootup auto sync complete")
else
error -> FarmbotCore.Logger.error(3, "bootup auto sync failed #{inspect(error)}")
end
:ok
end
# When auto_sync is disabled preload the sync.
defp maybe_auto_sync(sync_changeset, false) do
FarmbotCore.Logger.busy(3, "preloading sync")
sync = Changeset.apply_changes(sync_changeset)
EagerLoader.preload(sync)
FarmbotCore.Logger.success(3, "preloaded sync ok")
:ok
end
end

89
farmbot_ext/test/farmbot_ext/amqp/auto_sync_channel_test.exs
View File

@ -1,7 +1,12 @@
defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
defmodule AutoSyncChannelTest do
alias FarmbotExt.AMQP.AutoSyncChannel
use ExUnit.Case
import Mox
alias FarmbotExt.JWT
alias FarmbotCore.JSON
alias FarmbotCore.Asset.{Query, Command}
alias FarmbotExt.{JWT, API.Preloader, AMQP.ConnectionWorker}
@fake_jwt "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZ" <>
"G1pbkBhZG1pbi5jb20iLCJpYXQiOjE1MDIxMjcxMTcsImp0a" <>
@ -30,31 +35,31 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
test_pid = self()
expect(MockPreloader, :preload_all, fn ->
expect(Preloader, :preload_all, fn ->
send(test_pid, :preload_all_called)
:ok
end)
expect(MockConnectionWorker, :maybe_connect, fn jwt ->
expect(ConnectionWorker, :maybe_connect, fn jwt ->
send(test_pid, {:maybe_connect_called, jwt})
fake_value
end)
stub(MockConnectionWorker, :close_channel, fn _ ->
stub(ConnectionWorker, :close_channel, fn _ ->
send(test_pid, :close_channel_called)
:ok
end)
stub(MockConnectionWorker, :rpc_reply, fn chan, jwt_dot_bot, label ->
stub(ConnectionWorker, :rpc_reply, fn chan, jwt_dot_bot, label ->
send(test_pid, {:rpc_reply_called, chan, jwt_dot_bot, label})
:ok
end)
{:ok, pid} = FarmbotExt.AMQP.AutoSyncChannel.start_link([jwt: jwt], [])
{:ok, pid} = AutoSyncChannel.start_link([jwt: jwt], [])
assert_receive :preload_all_called
assert_receive {:maybe_connect_called, "device_15"}
Map.merge(%{pid: pid}, FarmbotExt.AMQP.AutoSyncChannel.network_status(pid))
Map.merge(%{pid: pid}, AutoSyncChannel.network_status(pid))
end
def under_normal_conditions() do
@ -105,7 +110,7 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
%{pid: pid} = pretend_network_returned(fake_response)
payload =
FarmbotCore.JSON.encode!(%{
JSON.encode!(%{
args: %{label: "xyz"}
})
@ -119,7 +124,7 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
payload = '{"args":{"label":"foo"}}'
key = "bot.device_15.sync.Device.999"
stub(MockQuery, :auto_sync?, fn ->
stub(Query, :auto_sync?, fn ->
send(test_pid, :called_auto_sync?)
false
end)
@ -133,9 +138,9 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
test_pid = self()
payload = '{"args":{"label":"foo"},"body":{}}'
key = "bot.device_15.sync.Device.999"
stub(MockQuery, :auto_sync?, fn -> true end)
stub(Query, :auto_sync?, fn -> true end)
stub(MockCommand, :update, fn x, y, z ->
stub(Command, :update, fn x, y, z ->
send(test_pid, {:update_called, x, y, z})
:ok
end)
@ -150,14 +155,14 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
payload = '{"args":{"label":"foo"},"body":{"foo": "bar"}}'
key = "bot.device_15.sync.#{module_name}.999"
stub(MockQuery, :auto_sync?, fn -> true end)
stub(Query, :auto_sync?, fn -> true end)
stub(MockCommand, :update, fn x, y, z ->
stub(Command, :update, fn x, y, z ->
send(test_pid, {:update_called, x, y, z})
:ok
end)
stub(MockCommand, :update, fn x, y, z ->
stub(Command, :update, fn x, y, z ->
send(test_pid, {:update_called, x, y, z})
:ok
end)
@ -167,29 +172,6 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
assert_receive {:update_called, module_name, %{"foo" => "bar"}, 999}, 10
end
def simple_asset_test_plural(module_name) do
%{pid: pid} = under_normal_conditions()
test_pid = self()
payload = '{"args":{"label":"foo"},"body":{"foo": "bar"}}'
key = "bot.device_15.sync.#{module_name}.999"
stub(MockQuery, :auto_sync?, fn -> true end)
stub(MockCommand, :update, fn x, y, z ->
send(test_pid, {:update_called, x, y, z})
:ok
end)
send(pid, {:basic_deliver, payload, %{routing_key: key}})
assert_receive {:update_called, module_name, %{"foo" => "bar"}, 999}, 10
end
test "handles FbosConfig", do: simple_asset_test_singleton("FbosConfig")
test "handles FirmwareConfig", do: simple_asset_test_singleton("FirmwareConfig")
test "handles FarmwareEnv", do: simple_asset_test_plural("FarmwareEnv")
test "handles FarmwareInstallation", do: simple_asset_test_plural("FarmwareInstallation")
test "handles auto_sync of 'cache_assets' when auto_sync is false" do
test_pid = self()
%{pid: pid} = under_normal_conditions()
@ -197,12 +179,12 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
key = "bot.device_15.sync.FbosConfig.999"
payload = '{"args":{"label":"foo"},"body":{"foo": "bar"}}'
stub(MockQuery, :auto_sync?, fn ->
stub(Query, :auto_sync?, fn ->
send(test_pid, :called_auto_sync?)
false
end)
stub(MockCommand, :update, fn x, y, z ->
stub(Command, :update, fn x, y, z ->
send(test_pid, {:update_called, x, y, z})
:ok
end)
@ -219,12 +201,12 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
key = "bot.device_15.sync.Point.999"
payload = '{"args":{"label":"foo"},"body":{"foo": "bar"}}'
stub(MockQuery, :auto_sync?, fn ->
stub(Query, :auto_sync?, fn ->
send(test_pid, :called_auto_sync?)
false
end)
stub(MockCommand, :new_changeset, fn kind, id, params ->
stub(Command, :new_changeset, fn kind, id, params ->
send(test_pid, {:new_changeset_called, kind, id, params})
:ok
end)
@ -232,4 +214,27 @@ defmodule FarmbotExt.AMQP.AutoSyncChannelTest do
send(pid, {:basic_deliver, payload, %{routing_key: key}})
assert_receive {:new_changeset_called, "Point", 999, %{"foo" => "bar"}}, 10
end
test "handles FbosConfig", do: simple_asset_test_singleton("FbosConfig")
test "handles FirmwareConfig", do: simple_asset_test_singleton("FirmwareConfig")
test "handles FarmwareEnv", do: simple_asset_test_plural("FarmwareEnv")
test "handles FarmwareInstallation", do: simple_asset_test_plural("FarmwareInstallation")
defp simple_asset_test_plural(module_name) do
%{pid: pid} = under_normal_conditions()
test_pid = self()
payload = '{"args":{"label":"foo"},"body":{"foo": "bar"}}'
key = "bot.device_15.sync.#{module_name}.999"
stub(Query, :auto_sync?, fn -> true end)
stub(Command, :update, fn x, y, z ->
send(test_pid, {:update_called, x, y, z})
:ok
end)
send(pid, {:basic_deliver, payload, %{routing_key: key}})
assert_receive {:update_called, module_name, %{"foo" => "bar"}, 999}, 10
end
end

3
farmbot_ext/test/preloader_test.exs
View File

@ -6,8 +6,7 @@ defmodule FarmbotExt.API.PreloaderTest do
setup :verify_on_exit!
test "preload" do
expect(MockPreloader, :preload_all, fn -> :ok end)
expect(FarmbotExt.API.Preloader, :preload_all, fn -> :ok end)
assert FarmbotExt.API.Preloader.preload_all() == :ok
end
end

11
farmbot_ext/test/test_helper.exs
View File

@ -1,6 +1,9 @@
Mox.defmock(MockPreloader, for: FarmbotExt.API.Preloader)
Mox.defmock(MockConnectionWorker, for: FarmbotExt.AMQP.ConnectionWorker.Network)
Mox.defmock(MockQuery, for: FarmbotCore.Asset.Query)
Mox.defmock(MockCommand, for: FarmbotCore.Asset.Command)
# Mocking for FarmbotCore
Mox.defmock(FarmbotCore.Asset.Query, for: FarmbotCore.Asset.Query)
Mox.defmock(FarmbotCore.Asset.Command, for: FarmbotCore.Asset.Command)
# Mocking for FarmbotExt
Mox.defmock(FarmbotExt.API.Preloader, for: FarmbotExt.API.Preloader)
Mox.defmock(FarmbotExt.AMQP.ConnectionWorker, for: FarmbotExt.AMQP.ConnectionWorker)
ExUnit.start()