1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.ConnCase do
7 This module defines the test case to be used by
8 tests that require setting up a connection.
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.
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.
20 use ExUnit.CaseTemplate
24 # Import conveniences for testing with connections
26 import Phoenix.ConnTest
27 use Pleroma.Tests.Helpers
28 import Pleroma.Web.Router.Helpers
32 # The default endpoint for testing
33 @endpoint Pleroma.Web.Endpoint
35 # Sets up OAuth access with specified scopes
36 defp oauth_access(scopes, opts \\ []) do
38 Keyword.get_lazy(opts, :user, fn ->
39 Pleroma.Factory.insert(:user)
43 Keyword.get_lazy(opts, :oauth_token, fn ->
44 Pleroma.Factory.insert(:oauth_token, user: user, scopes: scopes)
49 |> assign(:user, user)
50 |> assign(:token, token)
52 %{user: user, token: token, conn: conn}
55 defp request_content_type(%{conn: conn}) do
56 conn = put_req_header(conn, "content-type", "multipart/form-data")
60 defp empty_json_response(conn) do
61 body = response(conn, 204)
62 response_content_type(conn, :json)
67 defp json_response_and_validate_schema(
70 open_api_spex: %{operation_id: op_id, operation_lookup: lookup, spec: spec}
77 |> Plug.Conn.get_resp_header("content-type")
82 status = Plug.Conn.Status.code(status)
84 unless lookup[op_id].responses[status] do
85 err = "Response schema not found for #{status} #{conn.method} #{conn.request_path}"
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)
92 case OpenApiSpex.cast_value(json, schema, spec) do
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}"
105 "Response does not conform to schema of #{op_id} operation: #{
106 Enum.join(errors, "\n")
112 defp json_response_and_validate_schema(conn, _status) do
113 flunk("Response schema not found for #{conn.method} #{conn.request_path} #{conn.status}")
119 Cachex.clear(:user_cache)
120 Cachex.clear(:object_cache)
121 :ok = Ecto.Adapters.SQL.Sandbox.checkout(Pleroma.Repo)
123 unless tags[:async] do
124 Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
127 if tags[:needs_streamer] do
129 id: Pleroma.Web.Streamer.registry(),
131 {Registry, :start_link, [[keys: :duplicate, name: Pleroma.Web.Streamer.registry()]]}
135 {:ok, conn: Phoenix.ConnTest.build_conn()}