First
[anni] / test / support / conn_case.ex
1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2022 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   alias Pleroma.DataCase
23
24   using do
25     quote do
26       # Import conveniences for testing with connections
27       import Plug.Conn
28       import Phoenix.ConnTest
29       use Pleroma.Tests.Helpers
30       import Pleroma.Web.Router.Helpers
31
32       alias Pleroma.Config
33
34       # The default endpoint for testing
35       @endpoint Pleroma.Web.Endpoint
36
37       # Sets up OAuth access with specified scopes
38       defp oauth_access(scopes, opts \\ []) do
39         user =
40           Keyword.get_lazy(opts, :user, fn ->
41             Pleroma.Factory.insert(:user)
42           end)
43
44         token =
45           Keyword.get_lazy(opts, :oauth_token, fn ->
46             Pleroma.Factory.insert(:oauth_token, user: user, scopes: scopes)
47           end)
48
49         conn =
50           build_conn()
51           |> assign(:user, user)
52           |> assign(:token, token)
53
54         %{user: user, token: token, conn: conn}
55       end
56
57       defp request_content_type(%{conn: conn}) do
58         conn = put_req_header(conn, "content-type", "multipart/form-data")
59         [conn: conn]
60       end
61
62       defp empty_json_response(conn) do
63         body = response(conn, 204)
64         response_content_type(conn, :json)
65
66         body
67       end
68
69       defp json_response_and_validate_schema(
70              %{private: %{operation_id: op_id}} = conn,
71              status
72            ) do
73         {spec, lookup} = OpenApiSpex.Plug.PutApiSpec.get_spec_and_operation_lookup(conn)
74
75         content_type =
76           conn
77           |> Plug.Conn.get_resp_header("content-type")
78           |> List.first()
79           |> String.split(";")
80           |> List.first()
81
82         status = Plug.Conn.Status.code(status)
83
84         unless lookup[op_id].responses[status] do
85           err = "Response schema not found for #{status} #{conn.method} #{conn.request_path}"
86           flunk(err)
87         end
88
89         schema = lookup[op_id].responses[status].content[content_type].schema
90         json = if status == 204, do: empty_json_response(conn), else: json_response(conn, status)
91
92         case OpenApiSpex.cast_value(json, schema, spec) do
93           {:ok, _data} ->
94             json
95
96           {:error, errors} ->
97             errors =
98               Enum.map(errors, fn error ->
99                 message = OpenApiSpex.Cast.Error.message(error)
100                 path = OpenApiSpex.Cast.Error.path_to_string(error)
101                 "#{message} at #{path}"
102               end)
103
104             flunk(
105               "Response does not conform to schema of #{op_id} operation: #{Enum.join(errors, "\n")}\n#{inspect(json)}"
106             )
107         end
108       end
109
110       defp json_response_and_validate_schema(conn, _status) do
111         flunk("Response schema not found for #{conn.method} #{conn.request_path} #{conn.status}")
112       end
113     end
114   end
115
116   setup tags do
117     DataCase.setup_multi_process_mode(tags)
118     DataCase.setup_streamer(tags)
119     DataCase.stub_pipeline()
120
121     Mox.verify_on_exit!()
122
123     {:ok, conn: Phoenix.ConnTest.build_conn()}
124   end
125 end