git.squeep.com Git - akkoma/blob - test/support/conn_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.Web.ConnCase do
   6   @moduledoc """
   7   This module defines the test case to be used by
   8   tests that require setting up a connection.
   9
  10   Such tests rely on `Phoenix.ConnTest` and also
  11   import other functionality to make it easier
  12   to build common datastructures and query the data layer.
  13
  14   Finally, if the test case interacts with the database,
  15   it cannot be async. For this reason, every test runs
  16   inside a transaction which is reset at the beginning
  17   of the test unless the test case is marked as async.
  18   """
  19
  20   use ExUnit.CaseTemplate
  21
  22   using do
  23     quote do
  24       # Import conveniences for testing with connections
  25       use Phoenix.ConnTest
  26       use Pleroma.Tests.Helpers
  27       import Pleroma.Web.Router.Helpers
  28
  29       alias Pleroma.Config
  30
  31       # The default endpoint for testing
  32       @endpoint Pleroma.Web.Endpoint
  33
  34       # Sets up OAuth access with specified scopes
  35       defp oauth_access(scopes, opts \\ []) do
  36         user =
  37           Keyword.get_lazy(opts, :user, fn ->
  38             Pleroma.Factory.insert(:user)
  39           end)
  40
  41         token =
  42           Keyword.get_lazy(opts, :oauth_token, fn ->
  43             Pleroma.Factory.insert(:oauth_token, user: user, scopes: scopes)
  44           end)
  45
  46         conn =
  47           build_conn()
  48           |> assign(:user, user)
  49           |> assign(:token, token)
  50
  51         %{user: user, token: token, conn: conn}
  52       end
  53
  54       defp request_content_type(%{conn: conn}) do
  55         conn = put_req_header(conn, "content-type", "multipart/form-data")
  56         [conn: conn]
  57       end
  58
  59       defp empty_json_response(conn) do
  60         body = response(conn, 204)
  61         response_content_type(conn, :json)
  62
  63         body
  64       end
  65
  66       defp json_response_and_validate_schema(
  67              %{
  68                private: %{
  69                  open_api_spex: %{operation_id: op_id, operation_lookup: lookup, spec: spec}
  70                }
  71              } = conn,
  72              status
  73            ) do
  74         content_type =
  75           conn
  76           |> Plug.Conn.get_resp_header("content-type")
  77           |> List.first()
  78           |> String.split(";")
  79           |> List.first()
  80
  81         status = Plug.Conn.Status.code(status)
  82
  83         unless lookup[op_id].responses[status] do
  84           err = "Response schema not found for #{status} #{conn.method} #{conn.request_path}"
  85           flunk(err)
  86         end
  87
  88         schema = lookup[op_id].responses[status].content[content_type].schema
  89         json = if status == 204, do: empty_json_response(conn), else: json_response(conn, status)
  90
  91         case OpenApiSpex.cast_value(json, schema, spec) do
  92           {:ok, _data} ->
  93             json
  94
  95           {:error, errors} ->
  96             errors =
  97               Enum.map(errors, fn error ->
  98                 message = OpenApiSpex.Cast.Error.message(error)
  99                 path = OpenApiSpex.Cast.Error.path_to_string(error)
 100                 "#{message} at #{path}"
 101               end)
 102
 103             flunk(
 104               "Response does not conform to schema of #{op_id} operation: #{
 105                 Enum.join(errors, "\n")
 106               }\n#{inspect(json)}"
 107             )
 108         end
 109       end
 110
 111       defp json_response_and_validate_schema(conn, _status) do
 112         flunk("Response schema not found for #{conn.method} #{conn.request_path} #{conn.status}")
 113       end
 114
 115       defp ensure_federating_or_authenticated(conn, url, user) do
 116         initial_setting = Config.get([:instance, :federating])
 117         on_exit(fn -> Config.put([:instance, :federating], initial_setting) end)
 118
 119         Config.put([:instance, :federating], false)
 120
 121         conn
 122         |> get(url)
 123         |> response(403)
 124
 125         conn
 126         |> assign(:user, user)
 127         |> get(url)
 128         |> response(200)
 129
 130         Config.put([:instance, :federating], true)
 131
 132         conn
 133         |> get(url)
 134         |> response(200)
 135       end
 136     end
 137   end
 138
 139   setup tags do
 140     Cachex.clear(:user_cache)
 141     Cachex.clear(:object_cache)
 142     :ok = Ecto.Adapters.SQL.Sandbox.checkout(Pleroma.Repo)
 143
 144     unless tags[:async] do
 145       Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
 146     end
 147
 148     if tags[:needs_streamer] do
 149       start_supervised(%{
 150         id: Pleroma.Web.Streamer.registry(),
 151         start:
 152           {Registry, :start_link, [[keys: :duplicate, name: Pleroma.Web.Streamer.registry()]]}
 153       })
 154     end
 155
 156     {:ok, conn: Phoenix.ConnTest.build_conn()}
 157   end
 158 end