git.squeep.com Git - akkoma/blob - test/support/data_case.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.DataCase do
   6   @moduledoc """
   7   This module defines the setup for tests requiring
   8   access to the application's data layer.
   9
  10   You may define functions here to be used as helpers in
  11   your tests.
  12
  13   Finally, if the test case interacts with the database,
  14   it cannot be async. For this reason, every test runs
  15   inside a transaction which is reset at the beginning
  16   of the test unless the test case is marked as async.
  17   """
  18
  19   use ExUnit.CaseTemplate
  20
  21   using do
  22     quote do
  23       alias Pleroma.Repo
  24
  25       import Ecto
  26       import Ecto.Changeset
  27       import Ecto.Query
  28       import Pleroma.DataCase
  29       use Pleroma.Tests.Helpers
  30
  31       # Sets up OAuth access with specified scopes
  32       defp oauth_access(scopes, opts \\ []) do
  33         user =
  34           Keyword.get_lazy(opts, :user, fn ->
  35             Pleroma.Factory.insert(:user)
  36           end)
  37
  38         token =
  39           Keyword.get_lazy(opts, :oauth_token, fn ->
  40             Pleroma.Factory.insert(:oauth_token, user: user, scopes: scopes)
  41           end)
  42
  43         %{user: user, token: token}
  44       end
  45     end
  46   end
  47
  48   def clear_cachex do
  49     Pleroma.Supervisor
  50     |> Supervisor.which_children()
  51     |> Enum.each(fn
  52       {name, _, _, [Cachex]} ->
  53         name
  54         |> to_string
  55         |> String.trim_leading("cachex_")
  56         |> Kernel.<>("_cache")
  57         |> String.to_existing_atom()
  58         |> Cachex.clear()
  59
  60       _ ->
  61         nil
  62     end)
  63   end
  64
  65   setup tags do
  66     :ok = Ecto.Adapters.SQL.Sandbox.checkout(Pleroma.Repo)
  67
  68     if tags[:async] do
  69       Mox.stub_with(Pleroma.CachexMock, Pleroma.NullCache)
  70       Mox.set_mox_private()
  71     else
  72       Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
  73       Mox.stub_with(Pleroma.CachexMock, Pleroma.CachexProxy)
  74       Mox.set_mox_global()
  75       clear_cachex()
  76     end
  77
  78     if tags[:needs_streamer] do
  79       start_supervised(%{
  80         id: Pleroma.Web.Streamer.registry(),
  81         start:
  82           {Registry, :start_link, [[keys: :duplicate, name: Pleroma.Web.Streamer.registry()]]}
  83       })
  84     end
  85
  86     stub_pipeline()
  87
  88     Mox.verify_on_exit!()
  89
  90     :ok
  91   end
  92
  93   def stub_pipeline do
  94     Mox.stub_with(Pleroma.Web.ActivityPub.SideEffectsMock, Pleroma.Web.ActivityPub.SideEffects)
  95
  96     Mox.stub_with(
  97       Pleroma.Web.ActivityPub.ObjectValidatorMock,
  98       Pleroma.Web.ActivityPub.ObjectValidator
  99     )
 100
 101     Mox.stub_with(Pleroma.Web.ActivityPub.MRFMock, Pleroma.Web.ActivityPub.MRF)
 102     Mox.stub_with(Pleroma.Web.ActivityPub.ActivityPubMock, Pleroma.Web.ActivityPub.ActivityPub)
 103     Mox.stub_with(Pleroma.Web.FederatorMock, Pleroma.Web.Federator)
 104     Mox.stub_with(Pleroma.ConfigMock, Pleroma.Config)
 105   end
 106
 107   def ensure_local_uploader(context) do
 108     test_uploader = Map.get(context, :uploader, Pleroma.Uploaders.Local)
 109     uploader = Pleroma.Config.get([Pleroma.Upload, :uploader])
 110     filters = Pleroma.Config.get([Pleroma.Upload, :filters])
 111
 112     Pleroma.Config.put([Pleroma.Upload, :uploader], test_uploader)
 113     Pleroma.Config.put([Pleroma.Upload, :filters], [])
 114
 115     on_exit(fn ->
 116       Pleroma.Config.put([Pleroma.Upload, :uploader], uploader)
 117       Pleroma.Config.put([Pleroma.Upload, :filters], filters)
 118     end)
 119
 120     :ok
 121   end
 122
 123   @doc """
 124   A helper that transform changeset errors to a map of messages.
 125
 126       changeset = Accounts.create_user(%{password: "short"})
 127       assert "password is too short" in errors_on(changeset).password
 128       assert %{password: ["password is too short"]} = errors_on(changeset)
 129
 130   """
 131   def errors_on(changeset) do
 132     Ecto.Changeset.traverse_errors(changeset, fn {message, opts} ->
 133       Enum.reduce(opts, message, fn {key, value}, acc ->
 134         String.replace(acc, "%{#{key}}", to_string(value))
 135       end)
 136     end)
 137   end
 138 end