This is an automated email from the ASF dual-hosted git repository. nickva pushed a commit to branch gun-for-tests in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 6bf4a91ad5ae7f2890ba22e8b88c51bec6234df6 Author: Nick Vatamaniuc <[email protected]> AuthorDate: Fri Aug 21 17:30:33 2026 -0400 Use gun in tests Since we have gun and it's proven its worth in nouveau let's use for tests as well. The bigger idea is to use for replication but let's make a smaller step first and see how it fares in unit and elixir tests. To help with unit tests created a simpler compatibility couch_gun module. It accepted some of the existing options and shapes of headers and auth parameters we already use everywhere. This helps avoid rewritting a whole lot of tests in one go. Dependency-wise we just have to ensure gun is started but otherwise it's already present there for nouveau so we don't need to bring anything new in, which is nice. Elixir tests used httpotion based on ibrowse. That client is deprecated we have been getting "httpotion is unmaintained" warnings for a while now (this was another reason to attempt this PR). Since httpotion was based on ibrowse we had to create a similar "helper" as for unit tests, just to handle some expected APIs from the callers. In many ways gun provides a nicer abstraction to streamingm, so we could also simplify some of the worker start/stop and direct or pid:once hacks. Dependency-wise we don't have to do anything gun and cowlib are already present we can just clean up httpotion and ibrowse from elixir deps. --- .credo.exs | 1 - .gitignore | 1 - mix.exs | 9 +- .../test/eunit/chttpd_db_attachment_size_tests.erl | 5 +- src/chttpd/test/eunit/chttpd_dbs_info_test.erl | 13 +- src/couch/src/couch.app.src | 1 + src/couch/src/couch_gun.erl | 448 ++++++++++++++++++ src/couch/src/test_request.erl | 114 ++++- src/couch/src/test_util.erl | 5 +- .../test/eunit/couch_prometheus_e2e_tests.erl | 2 +- test/elixir/lib/couch.ex | 131 ++---- test/elixir/lib/couch/dbtest.ex | 12 +- test/elixir/lib/couch/http.ex | 503 +++++++++++++++++++++ test/elixir/lib/couch_raw.ex | 123 +---- test/elixir/lib/step/start.ex | 2 +- test/elixir/test/attachments_multipart_test.exs | 7 +- test/elixir/test/attachments_test.exs | 4 +- test/elixir/test/changes_async_test.exs | 162 +++---- test/elixir/test/design_paths_test.exs | 16 +- test/elixir/test/replication_test.exs | 44 +- 20 files changed, 1228 insertions(+), 375 deletions(-) diff --git a/.credo.exs b/.credo.exs index a07bb48a3..384b19155 100644 --- a/.credo.exs +++ b/.credo.exs @@ -30,7 +30,6 @@ ~r"/src/fast_pbkdf2/", ~r"/src/jason", ~r"/src/hackney", - ~r"/src/httpotion", ~r"/src/file_system", ~r"/src/credo", ~r"/src/idna", diff --git a/.gitignore b/.gitignore index 28dcadf05..cb42cdb75 100644 --- a/.gitignore +++ b/.gitignore @@ -130,7 +130,6 @@ test/javascript/junit.xml /_build/ /src/bunt /src/credo/ -/src/httpotion/ /src/jason/ /src/junit_formatter/ diff --git a/mix.exs b/mix.exs index 701bef5f6..cb6962c27 100644 --- a/mix.exs +++ b/mix.exs @@ -73,7 +73,7 @@ defmodule CouchDBTest.Mixfile do end # Run "mix help compile.app" to learn about applications. - def application, do: [applications: [:logger, :httpotion]] + def application, do: [applications: [:logger]] # Specifies which paths to compile per environment. defp elixirc_paths(:test), do: ["test/elixir/lib", "test/elixir/test/support"] @@ -84,9 +84,7 @@ defmodule CouchDBTest.Mixfile do defp deps() do deps1 = [ {:junit_formatter, "~> 3.4", only: [:dev, :test, :integration]}, - {:httpotion, ">= 3.2.0", only: [:dev, :test, :integration], runtime: false}, {:excoveralls, "~> 0.18.5", only: :test}, - {:ibrowse, path: path("ibrowse"), override: true}, {:credo, "== 1.7.19", only: [:dev, :test, :integration], runtime: false} ] @@ -95,14 +93,14 @@ defmodule CouchDBTest.Mixfile do deps_list = deps1 ++ deps2 - [:config, :couch, :fabric] + [:config, :couch, :fabric, :gun, :cowlib] |> Enum.map(&path("#{&1}/ebin")) |> Enum.map(&String.to_charlist/1) |> Enum.each(&:code.add_patha/1) # Some deps may be missing during source check # Besides we don't want to spend time checking them anyway - List.foldl([:ibrowse | extra_deps], deps_list, fn dep, acc -> + List.foldl(extra_deps, deps_list, fn dep, acc -> if File.dir?(acc[dep][:path]) do acc else @@ -133,7 +131,6 @@ defmodule CouchDBTest.Mixfile do "credo", "excoveralls", "hackney", - "httpotion", "ibrowse", "idna", "jason", diff --git a/src/chttpd/test/eunit/chttpd_db_attachment_size_tests.erl b/src/chttpd/test/eunit/chttpd_db_attachment_size_tests.erl index 420291616..1b49beb92 100644 --- a/src/chttpd/test/eunit/chttpd_db_attachment_size_tests.erl +++ b/src/chttpd/test/eunit/chttpd_db_attachment_size_tests.erl @@ -394,8 +394,9 @@ req(Method, Url, Headers, Body) -> {ok, Code, _, Res} = test_request:request(Method, Url, Headers1, Body), {Code, json_decode(Res)}. -% Data streaming generator for ibrowse client. ibrowse will repeatedly call the -% function with State and it should return {ok, Data, NewState} or eof at end. +% Data streaming generator for the test http client, which repeatedly calls +% the function with State; it should return {ok, Data, NewState} or eof at +% the end. data_stream_fun(Size) -> Fun = fun (0) -> eof; diff --git a/src/chttpd/test/eunit/chttpd_dbs_info_test.erl b/src/chttpd/test/eunit/chttpd_dbs_info_test.erl index a53442f98..0749c2f70 100644 --- a/src/chttpd/test/eunit/chttpd_dbs_info_test.erl +++ b/src/chttpd/test/eunit/chttpd_dbs_info_test.erl @@ -21,7 +21,7 @@ -define(CONTENT_JSON, {"Content-Type", "application/json"}). start() -> - Ctx = test_util:start_couch([inets, chttpd]), + Ctx = test_util:start_couch([chttpd]), DbDir = config:get("couchdb", "database_dir"), Suffix = ?b2l(couch_uuids:random()), test_util:with_couch_server_restart(fun() -> @@ -183,17 +183,12 @@ should_return_nothing_when_db_not_exist_for_get_dbs_info(_) -> should_return_500_time_out_when_time_is_not_enough_for_get_dbs_info(_) -> mock_timeout(), - Auth = base64:encode_to_string(?USER ++ ":" ++ ?PASS), - Headers = [{"Authorization", "Basic " ++ Auth}], - Request = {dbs_info_url("buffer_response=true"), Headers}, + Url = dbs_info_url("buffer_response=true"), {Props} = test_util:wait( fun() -> - % Use httpc to avoid ibrowse returning {error, - % retry_later} in some cases, causing test_request to - % sleep and retry, resulting in timeout failures. - case httpc:request(get, Request, [], []) of - {ok, {{_, Code, _}, _, Body}} -> + case test_request:get(Url, [?CONTENT_JSON, ?AUTH]) of + {ok, Code, _, Body} -> ?assertEqual(500, Code), jiffy:decode(Body); _ -> diff --git a/src/couch/src/couch.app.src b/src/couch/src/couch.app.src index 5f1fb9800..140140f82 100644 --- a/src/couch/src/couch.app.src +++ b/src/couch/src/couch.app.src @@ -36,6 +36,7 @@ os_mon, % Upstream deps + gun, ibrowse, mochiweb, diff --git a/src/couch/src/couch_gun.erl b/src/couch/src/couch_gun.erl new file mode 100644 index 000000000..ede2b0ade --- /dev/null +++ b/src/couch/src/couch_gun.erl @@ -0,0 +1,448 @@ +% Licensed under the Apache License, Version 2.0 (the "License"); you may not +% use this file except in compliance with the License. You may obtain a copy of +% the License at +% +% http://www.apache.org/licenses/LICENSE-2.0 +% +% Unless required by applicable law or agreed to in writing, software +% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +% License for the specific language governing permissions and limitations under +% the License. + +% Helper module to use gun instead of other http clients we had. +% +% Functions: +% +% * req/3,4,5: For basic synchronous requests +% +% * parse_url/1: Helper parser to turn urls with possible auth bits +% embedded into a gun uri map +% +% * open/3, close/1: Open/close connections +% +% * send/5,6: Send requests on opened connections. Can take {Fun, State} for +% a streaming body. +% +% * await/3: Wait for response a send. +% +% * headers/1, method/1, basic_auth/2: convert existing calling conventions +% to gun's format. These are helpers to avoid modifying all the call sites +% right off the bat. For example, the codebase expects headers to be +% strings and methods to be atom so we transform them here accordingly. +% +% * norm_error/1: shorten/normalize gun's error reason + +-module(couch_gun). + +-export([ + req/3, + req/4, + req/5, + parse_url/1, + open/3, + close/1, + send/5, + send/6, + await/3, + headers/1, + method/1, + basic_auth/2, + norm_error/1 +]). + +-define(DEFAULT_TIMEOUT, 30000). + +req(Method, Url, Headers) -> + req(Method, Url, Headers, <<>>, #{}). + +req(Method, Url, Headers, Body) -> + req(Method, Url, Headers, Body, #{}). + +req(Method, Url, Headers, Body, #{} = Opts) when is_atom(Method), is_list(Url) -> + case parse_url(Url) of + {ok, #{transport := Transport, host := Host, port := Port} = Parsed} -> + #{path := Path, userinfo := UserInfo} = Parsed, + Timeout = maps:get(timeout, Opts, ?DEFAULT_TIMEOUT), + Headers1 = auth_headers(headers(Headers), UserInfo, Opts), + OpenOpts = maps:with([tls_opts, tcp_opts, gun_opts], Opts), + case open(Host, Port, OpenOpts#{transport => Transport}) of + {ok, Pid} -> + try + case gun:await_up(Pid, Timeout) of + {ok, _} -> + Ref = send(Pid, Method, Path, Headers1, Body), + await(Pid, Ref, Timeout); + {error, Reason} -> + {error, norm_error(Reason)} + end + after + close(Pid) + end; + {error, Reason} -> + {error, norm_error(Reason)} + end; + {error, _} = Error -> + Error + end. + +parse_url("http://"; ++ Rest) -> + parse_auth(tcp, 80, Rest); +parse_url("https://"; ++ Rest) -> + parse_auth(tls, 443, Rest); +parse_url(_) -> + {error, invalid_uri}. + +parse_auth(Transport, DefaultPort, Rest) -> + {Auth, Path} = + case lists:splitwith(fun(C) -> C /= $/ andalso C /= $? end, Rest) of + {A, ""} -> {A, "/"}; + {A, "?" ++ _ = Query} -> {A, "/" ++ Query}; + {A, P} -> {A, P} + end, + {UserInfo, HostPort} = + case string:split(Auth, "@", trailing) of + [Creds, HP] -> {Creds, HP}; + [HP] -> {undefined, HP} + end, + case parse_host_port(HostPort, DefaultPort) of + {error, _} = Error -> + Error; + {Host, Port} -> + {ok, #{ + transport => Transport, + host => Host, + port => Port, + path => Path, + userinfo => UserInfo + }} + end. + +parse_host_port("", _DefaultPort) -> + {error, invalid_uri}; +parse_host_port("[" ++ Rest, DefaultPort) -> + % ipv6 with brackets [...] + case string:split(Rest, "]") of + ["", _] -> + {error, invalid_uri}; + [Host, ""] -> + {Host, DefaultPort}; + [Host, ":" ++ PortStr] -> + case string:to_integer(PortStr) of + {Port, ""} when is_integer(Port) -> {Host, Port}; + _ -> {error, invalid_uri} + end; + _ -> + {error, invalid_uri} + end; +parse_host_port(HostPort, DefaultPort) -> + case string:split(HostPort, ":", trailing) of + ["", _] -> + {error, invalid_uri}; + [Host, PortStr] -> + case string:to_integer(PortStr) of + {Port, ""} when is_integer(Port) -> {Host, Port}; + _ -> {error, invalid_uri} + end; + [Host] -> + {Host, DefaultPort} + end. + +% Open an http 1.1 connection. Opts #{} may have these fields: +% transport - tcp (default) | tls +% tls_opts - TSL client opts. Default is [] +% tcp_opts - Options for gen_tcp +% gun_opts - Other options for gun:open/3 +% +% Host maybe a string or address tuple. Connection pid returned but we don't +% wait for it to be up. Caller should do that (call gun:wait_up/2) or if they +% don't care just do a send right away. +% +open(Host, Port, #{} = Opts) -> + Transport = maps:get(transport, Opts, tcp), + OpenOpts0 = #{transport => Transport, protocols => [http], retry => 0}, + OpenOpts1 = + case Opts of + #{tcp_opts := TcpOpts} -> OpenOpts0#{tcp_opts => TcpOpts}; + #{} -> OpenOpts0 + end, + OpenOpts2 = + case Transport of + tls -> OpenOpts1#{tls_opts => maps:get(tls_opts, Opts, [])}; + tcp -> OpenOpts1 + end, + OpenOpts = maps:merge(OpenOpts2, maps:get(gun_opts, Opts, #{})), + gun:open(host(Host), Port, OpenOpts). + +close(Pid) when is_pid(Pid) -> + try + gun:close(Pid) + catch + _:_ -> ok + end, + ok. + +% Send a request and get back a stream reference. The body may be given as +% {Fun, State} tuple. So Fun(State) should return {ok, Data, NewState} and then +% return eof at the end. That will be sent chunked unless a content-length +% header is set. +send(Pid, Method, Path, Headers, Body) -> + send(Pid, Method, Path, Headers, Body, #{}). + +send(Pid, Method, Path, Headers, {Fun, State}, ReqOpts) when is_function(Fun, 1) -> + SRef = gun:headers(Pid, method(Method), Path, headers(Headers), ReqOpts), + ok = send_body(Pid, SRef, Fun, State), + SRef; +send(Pid, Method, Path, Headers, Body, ReqOpts) when is_pid(Pid) -> + gun:request(Pid, method(Method), Path, headers(Headers), body(Body), ReqOpts). + +body([]) -> + <<>>; +body(Body) -> + Body. + +% Send streaming body helper. +send_body(Pid, SRef, Fun, State) -> + case Fun(State) of + {ok, Data, State1} -> + send_body(Pid, SRef, Fun, State1, Data); + eof -> + ok = gun:data(Pid, SRef, fin, <<>>) + end. + +send_body(Pid, SRef, Fun, State, Data0) -> + case Fun(State) of + {ok, Data, State1} -> + % Send pending data before sending the next + % we're doing one chunk at a time here + ok = gun:data(Pid, SRef, nofin, Data0), + send_body(Pid, SRef, Fun, State1, Data); + eof -> + ok = gun:data(Pid, SRef, fin, Data0) + end. + +% Wait for a response. First wait for status + headers then body. 1xx info +% responses are skipped and we don't care about trailer either. If we got a bad +% connection and didn't find out until calling send and await we'll get the +% error here. +await(Pid, SRef, Timeout) when is_pid(Pid) -> + MRef = monitor(process, Pid), + try await_headers(Pid, SRef, Timeout, MRef) of + {response, fin, Code, RespHeaders} -> + {ok, Code, headers_from_gun(RespHeaders), <<>>}; + {response, nofin, Code, RespHeaders} -> + case gun:await_body(Pid, SRef, Timeout, MRef) of + {ok, RespBody} -> + {ok, Code, headers_from_gun(RespHeaders), RespBody}; + {ok, RespBody, _Trailers} -> + {ok, Code, headers_from_gun(RespHeaders), RespBody}; + {error, Reason} -> + {error, norm_error(Reason)} + end; + {error, Reason} -> + {error, norm_error(Reason)} + after + demonitor(MRef, [flush]) + end. + +await_headers(Pid, SRef, Timeout, MRef) -> + case gun:await(Pid, SRef, Timeout, MRef) of + {inform, _Status, _Headers} -> await_headers(Pid, SRef, Timeout, MRef); + Other -> Other + end. + +headers_from_gun(Headers) -> + [{binary_to_list(K), binary_to_list(V)} || {K, V} <- Headers]. + +% Transform our request headers into gun's lowercase binary shape. Previous +% http client accepted special atom headers like {basic_auth, {User, Pass}} and +% {cookie, Value}, {content_type, Type}, {content_length, Len}. We handle those +% here to avoid modifying all the call sites. + +headers(Headers) -> + [header(H) || H <- Headers]. + +header({basic_auth, {User, Pass}}) -> + basic_auth(User, Pass); +header({cookie, Cookie}) -> + {~"cookie", to_bin(Cookie)}; +header({content_type, Value}) -> + {~"content-type", to_bin(Value)}; +header({content_length, Value}) -> + {~"content-length", to_bin(Value)}; +header({Name, Value}) -> + {string:lowercase(to_bin(Name)), to_bin(Value)}. + +method(Method) when is_atom(Method) -> + string:uppercase(atom_to_binary(Method, utf8)). + +basic_auth(User, Pass) -> + UserPass = base64:encode(iolist_to_binary([User, $:, Pass])), + {~"authorization", <<"Basic ", UserPass/binary>>}. + +% If headers already have authorization set use that, otherwise take from the +% userinfo field from the url +auth_headers(Headers, UserInfo, Opts) -> + case lists:keymember(~"authorization", 1, Headers) of + true -> + Headers; + false -> + case {Opts, UserInfo} of + {#{basic_auth := {User, Pass}}, _} -> + [basic_auth(User, Pass) | Headers]; + {#{}, undefined} -> + Headers; + {#{}, UserInfo} -> + case string:split(UserInfo, ":") of + [User, Pass] -> [basic_auth(User, Pass) | Headers]; + [User] -> [basic_auth(User, "") | Headers] + end + end + end. + +to_bin(V) when is_atom(V) -> + atom_to_binary(V, utf8); +to_bin(V) -> + iolist_to_binary(V). + +% gun expects IPs as address tuples +host(Host) when is_list(Host) -> + case inet:parse_strict_address(Host) of + {ok, Ip} -> Ip; + {error, _} -> Host + end; +host(Host) -> + Host. + +% To implify error handling return just the {error, Reason} to make it easier +% for callers to handle it instead of the multi-level nested error shapes from +% gun. +norm_error({stream_error, Reason}) -> + norm_error(Reason); +norm_error({connection_error, Reason}) -> + norm_error(Reason); +norm_error({down, {shutdown, Reason}}) -> + norm_error(Reason); +norm_error({down, Reason}) -> + norm_error(Reason); +norm_error({shutdown, Reason}) -> + norm_error(Reason); +norm_error(Reason) -> + Reason. + +-ifdef(TEST). + +-include_lib("couch/include/couch_eunit.hrl"). + +parse_url_test() -> + ?assertEqual( + {ok, #{transport => tcp, host => "h", port => 80, path => "/", userinfo => undefined}}, + parse_url("http://h";) + ), + ?assertEqual( + {ok, #{ + transport => tcp, host => "h", port => 5984, path => "/db?a=b", userinfo => undefined + }}, + parse_url("http://h:5984/db?a=b";) + ), + ?assertEqual( + {ok, #{transport => tls, host => "h", port => 443, path => "/", userinfo => undefined}}, + parse_url("https://h";) + ), + ?assertEqual( + {ok, #{ + transport => tcp, host => "127.0.0.1", port => 80, path => "/", userinfo => undefined + }}, + parse_url("http://127.0.0.1";) + ), + ?assertEqual( + {ok, #{transport => tcp, host => "::1", port => 5984, path => "/db", userinfo => undefined}}, + parse_url("http://[::1]:5984/db";) + ), + ?assertEqual( + {ok, #{transport => tcp, host => "::1", port => 80, path => "/", userinfo => undefined}}, + parse_url("http://[::1]";) + ), + ?assertEqual( + {ok, #{transport => tcp, host => "h", port => 80, path => "/", userinfo => "u:p"}}, + parse_url("http://u:p@h";) + ), + ?assertEqual( + {ok, #{ + transport => tcp, + host => "h", + port => 15984, + path => "/_dbs_info?startkey=\"db1\"&endkey=\"db2\"", + userinfo => undefined + }}, + parse_url("http://h:15984/_dbs_info?startkey=\"db1\"&endkey=\"db2\"";) + ), + ?assertEqual( + {ok, #{transport => tcp, host => "h", port => 80, path => "/?q=1", userinfo => undefined}}, + parse_url("http://h?q=1";) + ), + ?assertEqual({error, invalid_uri}, parse_url("a potato")), + ?assertEqual({error, invalid_uri}, parse_url("ftp://h/";)), + ?assertEqual({error, invalid_uri}, parse_url("http://";)), + ?assertEqual({error, invalid_uri}, parse_url("http://:80/";)), + ?assertEqual({error, invalid_uri}, parse_url("http://h:x/";)), + ?assertEqual({error, invalid_uri}, parse_url("http://[::1";)), + ?assertEqual({error, invalid_uri}, parse_url("http://[]:80/";)). + +host_test() -> + ?assertEqual("cdb.example.com", host("cdb.example.com")), + ?assertEqual({127, 0, 0, 1}, host("127.0.0.1")), + ?assertEqual({0, 0, 0, 0, 0, 0, 0, 1}, host("::1")), + ?assertEqual({1, 2, 3, 4}, host({1, 2, 3, 4})). + +headers_test() -> + ?assertEqual([], headers([])), + ?assertEqual( + [{~"content-type", ~"application/json"}], + headers([{"Content-Type", "application/json"}]) + ), + ?assertEqual( + [{~"x-foo", ~"1"}, {~"accept", ~"*/*"}], + headers([{'X-Foo', "1"}, {~"Accept", ~"*/*"}]) + ), + ?assertEqual( + [basic_auth("u", "p"), {~"cookie", ~"k=v"}], + headers([{basic_auth, {"u", "p"}}, {cookie, "k=v"}]) + ), + ?assertEqual( + [{~"content-type", ~"text/plain"}, {~"content-length", ~"3"}], + headers([{content_type, "text/plain"}, {content_length, "3"}]) + ). + +method_test() -> + ?assertEqual(~"GET", method(get)), + ?assertEqual(~"COPY", method(copy)), + ?assertEqual(~"DELETE", method('Delete')). + +basic_auth_test() -> + ?assertEqual( + {~"authorization", <<"Basic ", (base64:encode(~"u:p"))/binary>>}, + basic_auth("u", "p") + ), + ?assertEqual(basic_auth("u", "p"), basic_auth(~"u", ~"p")). + +auth_headers_test() -> + Auth = basic_auth("u", "p"), + Override = [{~"authorization", ~"Bearer dabears"}], + ?assertEqual([], auth_headers([], undefined, #{})), + ?assertEqual([Auth], auth_headers([], undefined, #{basic_auth => {"u", "p"}})), + ?assertEqual([Auth], auth_headers([], "u:p", #{})), + ?assertEqual([basic_auth("u", "")], auth_headers([], "u", #{})), + % The option take effect when header not set + ?assertEqual([Auth], auth_headers([], "x:y", #{basic_auth => {"u", "p"}})), + % If header is set go with the header + ?assertEqual(Override, auth_headers(Override, "x:y", #{basic_auth => {"u", "p"}})). + +norm_error_test() -> + ?assertEqual(econnrefused, norm_error({down, {shutdown, econnrefused}})), + ?assertEqual(closed, norm_error({stream_error, closed})), + ?assertEqual(closed, norm_error({connection_error, closed})), + ?assertEqual(timeout, norm_error(timeout)), + ?assertEqual(normal, norm_error({down, normal})). + +-endif. diff --git a/src/couch/src/test_request.erl b/src/couch/src/test_request.erl index d7364012f..743f2ee55 100644 --- a/src/couch/src/test_request.erl +++ b/src/couch/src/test_request.erl @@ -20,6 +20,8 @@ -export([options/1, options/2, options/3]). -export([request/3, request/4, request/5]). +-define(TIMEOUT, 30000). + copy(Url) -> copy(Url, []). @@ -86,25 +88,101 @@ request(Method, Url, Headers, Body, Opts) -> request(_Method, _Url, _Headers, _Body, _Opts, 0) -> {error, request_failed}; request(Method, Url, Headers, Body, Opts, N) -> - case code:is_loaded(ibrowse) of - false -> - {ok, _} = ibrowse:start(); - _ -> - ok - end, - case ibrowse:send_req(Url, Headers, Method, Body, Opts) of - {ok, Code0, RespHeaders, RespBody0} -> - Code = list_to_integer(Code0), - RespBody = iolist_to_binary(RespBody0), - {ok, Code, RespHeaders, RespBody}; - {error, {'EXIT', {normal, _}}} -> - % Connection closed right after a successful request that - % used the same connection. - request(Method, Url, Headers, Body, Opts, N - 1); - {error, retry_later} -> - % CouchDB is busy, let’s wait a bit - timer:sleep(3000 div N), + {ok, _} = application:ensure_all_started(gun), + Headers1 = headers(Headers, Opts), + ReqOpts = #{timeout => ?TIMEOUT, tls_opts => [{verify, verify_none}]}, + case couch_gun:req(Method, Url, Headers1, Body, ReqOpts) of + {ok, Code, RespHeaders, RespBody} -> + {ok, Code, canonical_headers(RespHeaders), RespBody}; + {error, closed} -> + % Retry. Possible race with the server starting. request(Method, Url, Headers, Body, Opts, N - 1); Error -> Error end. + +headers(Headers, Opts) -> + lists:foldl(fun apply_opt/2, couch_gun:headers(Headers), Opts). + +apply_opt({host_header, Value}, Headers) -> + [Host] = couch_gun:headers([{host, Value}]), + lists:keystore(~"host", 1, Headers, Host); +apply_opt({basic_auth, {User, Pass}}, Headers) -> + Auth = couch_gun:basic_auth(User, Pass), + lists:keystore(~"authorization", 1, Headers, Auth); +apply_opt(_Other, Headers) -> + Headers. + +% Gun returns headers as lower case we we update them to camel case to avoid +% updating all the test call sites at this moment. Some are non-standard so we +% handle them as special cases +% +canonical_headers(Headers) -> + [{canonical_name(K), V} || {K, V} <- Headers]. + +canonical_name("etag") -> + "ETag"; +canonical_name("www-authenticate") -> + "WWW-Authenticate"; +canonical_name("content-md5") -> + "Content-MD5"; +canonical_name("x-couchdb-body-time") -> + "X-CouchDB-Body-Time"; +canonical_name("x-couch-request-id") -> + "X-Couch-Request-ID"; +canonical_name("x-couch-update-newrev") -> + "X-Couch-Update-NewRev"; +canonical_name("x-couchdb-vhost-path") -> + "x-couchdb-vhost-path"; +canonical_name(Name) -> + Parts = string:split(Name, "-", all), + lists:flatten(lists:join("-", [cap(S) || S <- Parts])). + +cap("") -> + ""; +cap([C | Rest]) -> + string:uppercase([C]) ++ string:lowercase(Rest). + +-ifdef(TEST). + +-include_lib("couch/include/couch_eunit.hrl"). + +canonical_name_test() -> + ?assertEqual("Content-Type", canonical_name("content-type")), + ?assertEqual("ETag", canonical_name("etag")), + ?assertEqual("X-Couch-Request-ID", canonical_name("x-couch-request-id")), + ?assertEqual("X-Foo-", canonical_name("x-foo-")). + +headers_test() -> + ?assertEqual([], headers([], [])), + ?assertEqual( + [{~"content-type", ~"application/json"}], + headers([{"Content-Type", "application/json"}], []) + ), + Auth = couch_gun:basic_auth("u", "p"), + ?assertEqual( + [{~"authorization", ~"Basic dTpw"}], + headers([{basic_auth, {"u", "p"}}], []) + ), + ?assertEqual( + [{~"cookie", ~"k=v"}], + headers([{cookie, "k=v"}], []) + ), + ?assertEqual( + [{~"accept", ~"*/*"}, {~"host", ~"potato.local"}], + headers([{"Accept", "*/*"}], [{host_header, "potato.local"}]) + ), + ?assertEqual( + [{~"host", ~"b"}], + headers([{"Host", "a"}], [{host_header, "b"}]) + ), + ?assertEqual( + [Auth], + headers([], [{basic_auth, {"u", "p"}}]) + ), + ?assertEqual( + [Auth], + headers([{basic_auth, {"x", "y"}}], [{basic_auth, {"u", "p"}}]) + ). + +-endif. diff --git a/src/couch/src/test_util.erl b/src/couch/src/test_util.erl index fd5364fb7..dabbb3382 100644 --- a/src/couch/src/test_util.erl +++ b/src/couch/src/test_util.erl @@ -46,7 +46,7 @@ -record(test_context, {mocked = [], started = [], module}). --define(DEFAULT_APPS, [inets, ibrowse, ssl, config, couch_epi, couch_event, couch]). +-define(DEFAULT_APPS, [inets, gun, ssl, config, couch_epi, couch_event, couch]). srcdir() -> code:priv_dir(couch) ++ "/../../". @@ -58,7 +58,8 @@ init_code_path() -> Paths = [ "couchdb", "jiffy", - "ibrowse", + "gun", + "cowlib", "mochiweb", "snappy" ], diff --git a/src/couch_prometheus/test/eunit/couch_prometheus_e2e_tests.erl b/src/couch_prometheus/test/eunit/couch_prometheus_e2e_tests.erl index 913b80834..1d4e44447 100644 --- a/src/couch_prometheus/test/eunit/couch_prometheus_e2e_tests.erl +++ b/src/couch_prometheus/test/eunit/couch_prometheus_e2e_tests.erl @@ -105,7 +105,7 @@ t_prometheus_port(_) -> t_reject_prometheus_port(Port) -> Response = test_request:get(node_local_url(Port), [?CONTENT_JSON, ?AUTH]), - ?assertEqual({error, {conn_failed, {error, econnrefused}}}, Response). + ?assertEqual({error, econnrefused}, Response). t_no_duplicate_metrics(Port) -> Url = node_local_url(Port), diff --git a/test/elixir/lib/couch.ex b/test/elixir/lib/couch.ex index a119095a9..dde356bed 100644 --- a/test/elixir/lib/couch.ex +++ b/test/elixir/lib/couch.ex @@ -42,9 +42,12 @@ defmodule Couch.Session do # if the need arises. def go(%Couch.Session{} = sess, method, url, opts) do parse_response = Keyword.get(opts, :parse_response, true) - opts = opts - |> Keyword.merge(cookie: sess.cookie) - |> Keyword.delete(:parse_response) + + opts = + opts + |> Keyword.merge(cookie: sess.cookie) + |> Keyword.delete(:parse_response) + if parse_response do Couch.request(method, url, opts) else @@ -54,9 +57,12 @@ defmodule Couch.Session do def go!(%Couch.Session{} = sess, method, url, opts) do parse_response = Keyword.get(opts, :parse_response, true) - opts = opts - |> Keyword.merge(cookie: sess.cookie) - |> Keyword.delete(:parse_response) + + opts = + opts + |> Keyword.merge(cookie: sess.cookie) + |> Keyword.delete(:parse_response) + if parse_response do Couch.request!(method, url, opts) else @@ -66,112 +72,37 @@ defmodule Couch.Session do end defmodule Couch do - use HTTPotion.Base - @moduledoc """ CouchDB library to power test suite. """ - # These constants are supplied to the underlying HTTP client and control - # how long we will wait before timing out a test. The inactivity timeout - # specifically fires during an active HTTP response and defaults to 10_000 - # if not specified. We're defining it to a different value than the - # request_timeout largely just so we know which timeout fired. - @request_timeout 60_000 - @inactivity_timeout 55_000 - - def process_url("http://"; <> _ = url) do - url - end - - def process_url(url) do - base_url = System.get_env("EX_COUCH_URL") || "http://127.0.0.1:15984"; - base_url <> url - end - - def process_request_headers(headers, _body, options) do - headers = Keyword.put(headers, :"User-Agent", "couch-potion") - - headers = - if headers[:"Content-Type"] do - headers - else - Keyword.put(headers, :"Content-Type", "application/json") - end - - case Keyword.get(options, :cookie) do - nil -> - headers - - cookie -> - Keyword.put(headers, :Cookie, cookie) - end - end - - def process_options(options) do - options - |> set_auth_options() - |> set_inactivity_timeout() - |> set_request_timeout() - end + defdelegate process_url(url), to: Couch.Http - def process_request_body(body) do - if is_map(body) do - :jiffy.encode(body, [:use_nil]) - else - body - end - end + def get(url, opts \\ []), do: request(:get, url, opts) + def get!(url, opts \\ []), do: request!(:get, url, opts) + def put(url, opts \\ []), do: request(:put, url, opts) + def put!(url, opts \\ []), do: request!(:put, url, opts) + def post(url, opts \\ []), do: request(:post, url, opts) + def post!(url, opts \\ []), do: request!(:post, url, opts) + def delete(url, opts \\ []), do: request(:delete, url, opts) + def delete!(url, opts \\ []), do: request!(:delete, url, opts) + def head(url, opts \\ []), do: request(:head, url, opts) + def head!(url, opts \\ []), do: request!(:head, url, opts) - def process_response_body(_headers, body) when body == [] do - "" + def request(method, url, opts \\ []) do + Couch.Http.request(method, url, opts, :json) end - def process_response_body(headers, body) do - content_type = headers[:"Content-Type"] - - if !!content_type and String.match?(content_type, ~r/application\/json/) do - body |> IO.iodata_to_binary() |> :jiffy.decode([:return_maps, :use_nil]) - else - process_response_body(body) - end - end + def request!(method, url, opts \\ []) do + case request(method, url, opts) do + %Couch.ErrorResponse{message: message} -> + raise "HTTP request failed: #{method} #{url}: #{message}" - def set_auth_options(options) do - cond do - Keyword.get(options, :no_auth, false) -> - options - Keyword.get(options, :cookie) == nil -> - headers = Keyword.get(options, :headers, []) - if headers[:basic_auth] != nil or headers[:authorization] != nil - or List.keymember?(headers, :"X-Auth-CouchDB-UserName", 0) do - options - else - username = System.get_env("EX_USERNAME") || "adm" - password = System.get_env("EX_PASSWORD") || "pass" - Keyword.put(options, :basic_auth, {username, password}) - end - true -> - options + resp -> + resp end end - def set_inactivity_timeout(options) do - Keyword.update( - options, - :ibrowse, - [{:inactivity_timeout, @inactivity_timeout}], - fn ibrowse -> - Keyword.put_new(ibrowse, :inactivity_timeout, @inactivity_timeout) - end - ) - end - - def set_request_timeout(options) do - timeout = Application.get_env(:httpotion, :default_timeout, @request_timeout) - Keyword.put_new(options, :timeout, timeout) - end - def login(userinfo) do [user, pass] = String.split(userinfo, ":", parts: 2) login(user, pass) diff --git a/test/elixir/lib/couch/dbtest.ex b/test/elixir/lib/couch/dbtest.ex index 693e6f0f3..d221151a3 100644 --- a/test/elixir/lib/couch/dbtest.ex +++ b/test/elixir/lib/couch/dbtest.ex @@ -63,7 +63,7 @@ defmodule Couch.DBTest do on_exit(fn -> query = %{:rev => user["_rev"]} resp = Couch.delete("/_users/#{user["_id"]}", query: query) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) end) context = Map.put(context, :user, user) @@ -159,7 +159,7 @@ defmodule Couch.DBTest do end resp = Couch.post("/_users", body: user_doc) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["ok"] Map.put(user_doc, "_rev", resp.body["rev"]) end @@ -337,7 +337,7 @@ defmodule Couch.DBTest do options = Map.put(options, :body, body) resp = Couch.post("/_replicate", Enum.to_list(options)) - assert HTTPotion.Response.success?(resp), "#{inspect(resp)}" + assert Couch.Response.success?(resp), "#{inspect(resp)}" resp.body end @@ -562,7 +562,7 @@ defmodule Couch.DBTest do defp restart_node(node, port) do url = "http://127.0.0.1:#{port}/_node/#{node}/_restart"; resp = Couch.post(url) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["ok"] # make sure node went down. we assuming the node can't bounce quick # enough to inroduce a race here @@ -575,7 +575,7 @@ defmodule Couch.DBTest do url = "http://127.0.0.1:#{port}/_up"; resp = Couch.get(url) - case HTTPotion.Response.success?(resp) do + case Couch.Response.success?(resp) do true -> resp.status_code in 200..399 false -> false end @@ -584,7 +584,7 @@ defmodule Couch.DBTest do defp node_to_port(node) do url = "/_node/#{node}/_config/chttpd/port" resp = Couch.get(url) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) resp.body end end diff --git a/test/elixir/lib/couch/http.ex b/test/elixir/lib/couch/http.ex new file mode 100644 index 000000000..35b37b2d6 --- /dev/null +++ b/test/elixir/lib/couch/http.ex @@ -0,0 +1,503 @@ +defmodule Couch.Headers do + @moduledoc """ + Response headers with case-insensitive access. + + Headers are stored in hdrs map in lower-case format (as gun returns them). + Then we normalize at access time. Same key headers are appended in order + of arrival. + """ + + @behaviour Access + + defstruct hdrs: %{} + + def new(headers) when is_list(headers) do + hdrs = + Enum.reduce(headers, %{}, fn {name, value}, acc -> + value = to_string(value) + + Map.update(acc, norm(name), value, fn + values when is_list(values) -> values ++ [value] + value0 -> [value0, value] + end) + end) + + %__MODULE__{hdrs: hdrs} + end + + def get(headers, key, default \\ nil) + + def get(%__MODULE__{hdrs: hdrs}, key, default) do + Map.get(hdrs, norm(key), default) + end + + @impl Access + def fetch(%__MODULE__{hdrs: hdrs}, key), do: Map.fetch(hdrs, norm(key)) + + @impl Access + def get_and_update(%__MODULE__{hdrs: hdrs} = headers, key, fun) do + {value, hdrs} = Map.get_and_update(hdrs, norm(key), fun) + {value, %{headers | hdrs: hdrs}} + end + + @impl Access + def pop(%__MODULE__{hdrs: hdrs} = headers, key) do + {value, hdrs} = Map.pop(hdrs, norm(key)) + {value, %{headers | hdrs: hdrs}} + end + + defp norm(key), do: key |> to_string() |> String.downcase() +end + +defmodule Couch.Response do + @moduledoc """ + Response to the request + """ + defstruct status_code: nil, headers: %Couch.Headers{}, body: "" + def success?(%__MODULE__{status_code: code}), do: code in 200..299 + def success?(_), do: false +end + +defmodule Couch.ErrorResponse do + @moduledoc """ + Error response to the request + """ + defstruct message: "" +end + +defmodule Couch.AsyncResponse do + @moduledoc """ + Response to a streaming request iniated by the stream_to: pid option + """ + defstruct [:id] +end + +defmodule Couch.AsyncHeaders do + @moduledoc """ + Streaming header response to a streaming request initiated by the stream_to: pid option + """ + defstruct [:id, :status_code, :headers] +end + +defmodule Couch.AsyncChunk do + @moduledoc """ + Streaming body chunk to a streaming request initiated by the stream_to: pid option + """ + defstruct [:id, :chunk] +end + +defmodule Couch.AsyncEnd do + @moduledoc """ + Stream end for a streaming request initiated by the stream_to: pid option + """ + defstruct [:id] +end + +defmodule Couch.Http do + @moduledoc """ + + Small http client built on gun. It looks a bit odd because it's trying to + mimick the now removed httpotion client shape a bit. Some of the patterns + here are also copied from couch_gun.erl. The one differense if we keep a + connected process cached in the process dict to speed up test runs here. + + Normal requests return Couch.Response | Couch.ErrorResponse results. + + Streaming requests should pass `stream_to: pid` as the option. Their response + will be Couch.AsyncResponse then followed by Couch.AsyncChunk messages and + finally Couch.AsyncEnd. + + Some requests options are: + :body + :headers + :query + :timeout + :cookie + :no_auth + """ + + @request_timeout 60_000 + @inactivity_timeout 55_000 + @attempts 3 + + def base_url do + System.get_env("EX_COUCH_URL") || "http://127.0.0.1:15984"; + end + + def process_url("http://"; <> _ = url), do: url + def process_url("https://"; <> _ = url), do: url + def process_url(url), do: base_url() <> url + + def request(method, url, options, body_mode) when body_mode in [:json, :raw] do + ensure_gun_started() + url = url |> to_string() |> process_url() + url = append_query(url, Keyword.get(options, :query)) + method = method |> to_string() |> String.upcase() + headers = build_headers(options) + body = encode_body(Keyword.get(options, :body, "")) + + case Keyword.get(options, :stream_to) do + nil -> + sync_request(method, url, headers, body, options, body_mode) + + target when is_pid(target) -> + async_request(method, url, headers, body, target) + end + end + + defp sync_request(method, url, headers, body, options, body_mode) do + with {:ok, origin, path} <- parse_url(url), + {:ok, status, resp_headers, resp_body} <- + do_sync_request(origin, method, path, headers, body, options, 1) do + resp_headers = Couch.Headers.new(resp_headers) + + %Couch.Response{ + status_code: status, + headers: resp_headers, + body: process_body(body_mode, resp_headers, resp_body) + } + else + {:error, reason} -> %Couch.ErrorResponse{message: error_message(reason)} + end + end + + defp do_sync_request(origin, method, path, headers, body, options, attempt) do + timeout = Keyword.get(options, :timeout, @request_timeout) + deadline = now_ms() + timeout + conn = cached_conn(origin) + stream = :gun.request(conn, method, path, headers, body, %{}) + mref = Process.monitor(conn) + result = await_response(conn, stream, mref, deadline) + Process.demonitor(mref, [:flush]) + + case result do + {:ok, _status, resp_headers, _body} = ok -> + # If server closed the connection we drop it as well + if close_after?(resp_headers), do: invalidate(origin, conn) + ok + + {:error, reason} -> + reason = norm_error(reason) + # Teardown the cached connection on error and start fresh + invalidate(origin, conn) + + if conn_lost?(reason) and attempt < @attempts do + # Retry a few times on times or server start/stop race + do_sync_request(origin, method, path, headers, body, options, attempt + 1) + else + {:error, reason} + end + end + end + + defp await_response(conn, stream, mref, deadline) do + case gun_await(conn, stream, mref, deadline) do + {:inform, _status, _headers} -> + # These are 1xx and such and we don't care about them + await_response(conn, stream, mref, deadline) + + {:response, :fin, status, headers} -> + {:ok, status, headers, ""} + + {:response, :nofin, status, headers} -> + collect_body(conn, stream, mref, deadline, status, headers, []) + + {:error, _} = error -> + error + end + end + + defp collect_body(conn, stream, mref, deadline, status, headers, acc) do + case gun_await(conn, stream, mref, deadline) do + {:data, :nofin, data} -> + collect_body(conn, stream, mref, deadline, status, headers, [acc | data]) + + {:data, :fin, data} -> + {:ok, status, headers, IO.iodata_to_binary([acc | data])} + + {:trailers, _} -> + # We don't care about trailers + {:ok, status, headers, IO.iodata_to_binary(acc)} + + {:error, _} = error -> + error + end + end + + defp gun_await(conn, stream, mref, deadline) do + remaining = deadline - now_ms() + + if remaining <= 0 do + {:error, :req_timedout} + else + :gun.await(conn, stream, min(remaining, @inactivity_timeout), mref) + end + end + + # Streaming stuff + # + # A helper relay process opens the connection, makes the request forward + # response to the target as Couch.Async* messages. + + defp async_request(method, url, headers, body, target) do + case parse_url(url) do + {:ok, origin, path} -> + relay = spawn(fn -> relay_init(target, origin, method, path, headers, body) end) + %Couch.AsyncResponse{id: relay} + + {:error, reason} -> + %Couch.ErrorResponse{message: error_message(reason)} + end + end + + defp relay_init(target, origin, method, path, headers, body) do + tref = Process.monitor(target) + conn = open_conn(origin) + cref = Process.monitor(conn) + stream = :gun.request(conn, method, path, headers, body, %{}) + relay_loop(%{target: target, conn: conn, stream: stream, tref: tref, cref: cref}) + end + + defp relay_loop(state) do + %{target: target, conn: conn, stream: stream} = state + + receive do + {:gun_inform, ^conn, ^stream, _status, _headers} -> + # skip 1xx stuff + relay_loop(state) + + {:gun_response, ^conn, ^stream, fin, status, headers} -> + async_headers = %Couch.AsyncHeaders{ + id: self(), + status_code: status, + headers: Couch.Headers.new(headers) + } + + send(target, async_headers) + if fin == :fin, do: relay_done(state), else: relay_loop(state) + + {:gun_data, ^conn, ^stream, fin, data} -> + send(target, %Couch.AsyncChunk{id: self(), chunk: data}) + if fin == :fin, do: relay_done(state), else: relay_loop(state) + + {:gun_trailers, ^conn, ^stream, _trailers} -> + # don't care about trailers + relay_done(state) + + {:gun_error, ^conn, ^stream, _reason} -> + relay_done(state) + + {:gun_error, ^conn, _reason} -> + relay_done(state) + + {:DOWN, mref, :process, _pid, _reason} -> + cond do + mref == state.cref -> + # connection died + send(target, %Couch.AsyncEnd{id: self()}) + :ok + + mref == state.tref -> + # target (test) process died, clean up + close_conn(conn) + :ok + + true -> + relay_loop(state) + end + end + end + + defp relay_done(state) do + send(state.target, %Couch.AsyncEnd{id: self()}) + close_conn(state.conn) + :ok + end + + # Connection handling. This works for the test with one + # test client and one server decently enough. + defp cached_conn(origin) do + key = {:couch_http_conn, origin} + + case Process.get(key) do + pid when is_pid(pid) -> + if Process.alive?(pid) do + pid + else + Process.delete(key) + cached_conn(origin) + end + + nil -> + conn = open_conn(origin) + Process.put(key, conn) + conn + end + end + + defp invalidate(origin, conn) do + key = {:couch_http_conn, origin} + if Process.get(key) == conn, do: Process.delete(key) + close_conn(conn) + end + + defp open_conn({transport, host, port}) do + host_chars = String.to_charlist(host) + # gun expects IP address tuples + host_addr = + case :inet.parse_strict_address(host_chars) do + {:ok, addr} -> addr + {:error, _} -> host_chars + end + + opts = %{transport: transport, protocols: [:http], retry: 0} + + opts = + case transport do + :tls -> Map.put(opts, :tls_opts, [{:verify, :verify_none}]) + :tcp -> opts + end + + {:ok, conn} = :gun.open(host_addr, port, opts) + conn + end + + defp close_conn(conn) do + try do + :gun.close(conn) + catch + _, _ -> :ok + end + end + + defp ensure_gun_started() do + case Process.get(:couch_http_gun_started) do + true -> + :ok + + _ -> + {:ok, _} = Application.ensure_all_started(:gun) + Process.put(:couch_http_gun_started, true) + :ok + end + end + + defp parse_url(url) do + case URI.parse(url) do + %URI{scheme: scheme, host: host} = uri + when scheme in ["http", "https"] and is_binary(host) and host != "" -> + transport = if scheme == "https", do: :tls, else: :tcp + path = uri.path || "/" + path = if uri.query, do: path <> "?" <> uri.query, else: path + {:ok, {transport, host, uri.port}, path} + + _ -> + {:error, :invalid_uri} + end + end + + defp append_query(url, query) when query == nil or query == [] or query == %{} do + url + end + + defp append_query(url, query) do + sep = if String.contains?(url, "?"), do: "&", else: "?" + url <> sep <> URI.encode_query(query) + end + + defp encode_body(nil), do: "" + defp encode_body(body) when is_map(body), do: :jiffy.encode(body, [:use_nil]) + defp encode_body(body), do: body + + defp build_headers(options) do + headers = + for {k, v} <- Keyword.get(options, :headers, []) do + {k |> to_string() |> String.downcase(), to_string(v)} + end + + headers = + headers + |> put_new_header("user-agent", "couch-potion") + |> put_new_header("content-type", "application/json") + + case Keyword.get(options, :cookie) do + nil -> set_auth(headers, options) + cookie -> put_new_header(headers, "cookie", cookie) + end + end + + # Auth may come from the environtment test setup + defp set_auth(headers, options) do + conf_auth? = + List.keymember?(headers, "authorization", 0) or + List.keymember?(headers, "x-auth-couchdb-username", 0) + + if Keyword.get(options, :no_auth, false) or conf_auth? do + headers + else + username = System.get_env("EX_USERNAME") || "adm" + password = System.get_env("EX_PASSWORD") || "pass" + credentials = Base.encode64("#{username}:#{password}") + [{"authorization", "Basic #{credentials}"} | headers] + end + end + + defp put_new_header(headers, key, value) do + if List.keymember?(headers, key, 0) do + headers + else + [{key, value} | headers] + end + end + + # Response stuff + + defp process_body(:raw, _headers, body), do: body + + defp process_body(:json, headers, body) do + content_type = headers["content-type"] + + json? = + is_binary(content_type) and + String.match?(content_type, ~r/application\/json/) + + if json? and body != "" do + :jiffy.decode(body, [:return_maps, :use_nil]) + else + body + end + end + + # Errors + + defp norm_error({:stream_error, reason}), do: norm_error(reason) + defp norm_error({:connection_error, reason}), do: norm_error(reason) + defp norm_error({:down, {:shutdown, reason}}), do: norm_error(reason) + defp norm_error({:down, reason}), do: norm_error(reason) + defp norm_error({:shutdown, reason}), do: norm_error(reason) + defp norm_error(reason), do: reason + + defp conn_lost?(:closed), do: true + defp conn_lost?({:closed, _}), do: true + defp conn_lost?(:normal), do: true + defp conn_lost?(:shutdown), do: true + defp conn_lost?(:noproc), do: true + defp conn_lost?(:einval), do: true + defp conn_lost?(:socket_closed_remotely), do: true + defp conn_lost?(_), do: false + + defp close_after?(headers) do + case List.keyfind(headers, "connection", 0) do + {_, value} -> String.downcase(value) == "close" + nil -> false + end + end + + defp error_message(:timeout), do: "req_timedout" + defp error_message(:req_timedout), do: "req_timedout" + defp error_message(reason) when is_atom(reason), do: Atom.to_string(reason) + defp error_message(reason), do: inspect(reason) + + # Helpers + defp now_ms(), do: System.monotonic_time(:millisecond) +end diff --git a/test/elixir/lib/couch_raw.ex b/test/elixir/lib/couch_raw.ex index 641612c9c..afe5f85b9 100644 --- a/test/elixir/lib/couch_raw.ex +++ b/test/elixir/lib/couch_raw.ex @@ -1,105 +1,32 @@ defmodule Rawresp do - use HTTPotion.Base - @moduledoc """ - HTTP client that provides raw response as result + HTTP client that provides raw response as result. Same as `Couch` but + response bodies are returned as-is, without JSON decoding. """ - @request_timeout 60_000 - @inactivity_timeout 55_000 - - def process_url("http://"; <> _ = url) do - url - end - - def process_url(url) do - base_url = System.get_env("EX_COUCH_URL") || "http://127.0.0.1:15984"; - base_url <> url - end - - def process_request_headers(headers, _body, options) do - headers = - headers - |> Keyword.put(:"User-Agent", "couch-potion") - - headers = - if headers[:"Content-Type"] do - headers - else - Keyword.put(headers, :"Content-Type", "application/json") - end - - case Keyword.get(options, :cookie) do - nil -> - headers - - cookie -> - Keyword.put(headers, :Cookie, cookie) - end - end - - def process_options(options) do - options - |> set_auth_options() - |> set_inactivity_timeout() - |> set_request_timeout() - end - - def process_request_body(body) do - if is_map(body) do - :jiffy.encode(body, [:use_nil]) - else - body - end - end - - def set_auth_options(options) do - if Keyword.get(options, :cookie) == nil do - headers = Keyword.get(options, :headers, []) - - if headers[:basic_auth] != nil or headers[:authorization] != nil do - options - else - username = System.get_env("EX_USERNAME") || "adm" - password = System.get_env("EX_PASSWORD") || "pass" - Keyword.put(options, :basic_auth, {username, password}) - end - else - options - end - end - - def set_inactivity_timeout(options) do - Keyword.update( - options, - :ibrowse, - [{:inactivity_timeout, @inactivity_timeout}], - fn ibrowse -> - Keyword.put_new(ibrowse, :inactivity_timeout, @inactivity_timeout) - end - ) - end - - def set_request_timeout(options) do - timeout = Application.get_env(:httpotion, :default_timeout, @request_timeout) - Keyword.put_new(options, :timeout, timeout) - end - - def login(userinfo) do - [user, pass] = String.split(userinfo, ":", parts: 2) - login(user, pass) - end - - def login(user, pass, expect \\ :success) do - resp = Couch.post("/_session", body: %{:username => user, :password => pass}) - if expect == :success do - true = resp.body["ok"] - cookie = resp.headers[:"set-cookie"] - [token | _] = String.split(cookie, ";") - %Couch.Session{cookie: token} - else - true = Map.has_key?(resp.body, "error") - %Couch.Session{error: resp.body["error"]} + def get(url, opts \\ []), do: request(:get, url, opts) + def get!(url, opts \\ []), do: request!(:get, url, opts) + def put(url, opts \\ []), do: request(:put, url, opts) + def put!(url, opts \\ []), do: request!(:put, url, opts) + def post(url, opts \\ []), do: request(:post, url, opts) + def post!(url, opts \\ []), do: request!(:post, url, opts) + def delete(url, opts \\ []), do: request(:delete, url, opts) + def delete!(url, opts \\ []), do: request!(:delete, url, opts) + def head(url, opts \\ []), do: request(:head, url, opts) + def head!(url, opts \\ []), do: request!(:head, url, opts) + def options(url, opts \\ []), do: request(:options, url, opts) + + def request(method, url, opts \\ []) do + Couch.Http.request(method, url, opts, :raw) + end + + def request!(method, url, opts \\ []) do + case request(method, url, opts) do + %Couch.ErrorResponse{message: message} -> + raise "HTTP request failed: #{method} #{url}: #{message}" + + resp -> + resp end end end diff --git a/test/elixir/lib/step/start.ex b/test/elixir/lib/step/start.ex index b86b14a4f..1f1ca2ac9 100644 --- a/test/elixir/lib/step/start.ex +++ b/test/elixir/lib/step/start.ex @@ -4,7 +4,7 @@ defmodule Couch.Test.Setup.Step.Start do list of applications from DEFAULT_APPS macro defined in `test_util.erl`. At the time of writing this list included: - inets - - ibrowse + - gun - ssl - config - couch_epi diff --git a/test/elixir/test/attachments_multipart_test.exs b/test/elixir/test/attachments_multipart_test.exs index f940bae10..0abee4682 100644 --- a/test/elixir/test/attachments_multipart_test.exs +++ b/test/elixir/test/attachments_multipart_test.exs @@ -312,14 +312,13 @@ defmodule AttachmentMultipartTest do end defp put_multipart_new_edits_false(db_name, rev, multipart_data) do - # Help ensure we're re-using client connections - ibrowse_opts = [{:max_sessions, 1}, {:max_pipeline_size, 1}] + # The client reuses one connection per process, which is what this + # test needs: repeating the request must not wedge the connection resp = Couch.put( "/#{db_name}/multipart_replicated_changes?new_edits=false&rev=#{rev}", body: multipart_data, - headers: ["Content-Type": "multipart/related;boundary=\"abc123\""], - ibrowse: ibrowse_opts + headers: ["Content-Type": "multipart/related;boundary=\"abc123\""] ) assert resp.status_code in [201, 202] diff --git a/test/elixir/test/attachments_test.exs b/test/elixir/test/attachments_test.exs index 2d3251b19..5e1376eef 100644 --- a/test/elixir/test/attachments_test.exs +++ b/test/elixir/test/attachments_test.exs @@ -131,10 +131,10 @@ defmodule AttachmentsTest do assert resp.status_code in [201, 202] rev = resp.body["rev"] - resp = Couch.delete("/#{db_name}/bin_doc/foo.txt", body: "some payload", query: %{w: 3, rev: rev}, ibrowse: [{:max_sessions, 1}, {:max_pipeline_size, 1}]) + resp = Couch.delete("/#{db_name}/bin_doc/foo.txt", body: "some payload", query: %{w: 3, rev: rev}) assert resp.status_code == 200 - resp = Couch.get("/", timeout: 1000, ibrowse: [{:max_sessions, 1}, {:max_pipeline_size, 1}]) + resp = Couch.get("/", timeout: 1000) assert resp.status_code == 200 end diff --git a/test/elixir/test/changes_async_test.exs b/test/elixir/test/changes_async_test.exs index 75362d8a9..97a4b2ebb 100644 --- a/test/elixir/test/changes_async_test.exs +++ b/test/elixir/test/changes_async_test.exs @@ -38,12 +38,10 @@ defmodule ChangesAsyncTest do assert last_seq_prefix == "1-", "seq must start with 1-" last_seq = changes["last_seq"] - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) req_id = Couch.get("/#{db_name}/_changes?feed=longpoll&since=#{last_seq}", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) @@ -60,8 +58,7 @@ defmodule ChangesAsyncTest do req_id = Couch.get("/#{db_name}/_changes?feed=longpoll&since=now", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) @@ -82,12 +79,10 @@ defmodule ChangesAsyncTest do check_empty_db(db_name) create_doc(db_name, sample_doc_foo()) - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) req_id = Rawresp.get("/#{db_name}/_changes?feed=eventsource&timeout=500", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) @@ -99,8 +94,6 @@ defmodule ChangesAsyncTest do assert length(changes) == 2 assert Enum.at(changes, 0)["id"] == "foo" assert Enum.at(changes, 1)["id"] == "bar" - - HTTPotion.stop_worker_process(worker_pid) end @tag :with_db @@ -110,12 +103,10 @@ defmodule ChangesAsyncTest do check_empty_db(db_name) create_doc(db_name, sample_doc_foo()) - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) req_id = Rawresp.get("/#{db_name}/_changes?feed=eventsource&limit=1", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) @@ -125,8 +116,6 @@ defmodule ChangesAsyncTest do changes = process_response(req_id.id, &parse_event/1) assert length(changes) == 1 assert Enum.at(changes, 0)["id"] == "foo" - - HTTPotion.stop_worker_process(worker_pid) end @tag :with_db @@ -136,12 +125,10 @@ defmodule ChangesAsyncTest do check_empty_db(db_name) create_doc(db_name, sample_doc_foo()) - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) req_id = Rawresp.get("/#{db_name}/_changes?feed=eventsource&limit=2", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) @@ -152,8 +139,6 @@ defmodule ChangesAsyncTest do assert length(changes) == 2 assert Enum.at(changes, 0)["id"] == "foo" assert Enum.at(changes, 1)["id"] == "bar" - - HTTPotion.stop_worker_process(worker_pid) end @tag :with_db @@ -166,12 +151,9 @@ defmodule ChangesAsyncTest do t0 = :erlang.monotonic_time(:millisecond) - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) - req_id = Rawresp.get("/#{db_name}/_changes?feed=eventsource&timeout=1100&limit=2", - stream_to: self(), - direct: worker_pid + stream_to: self() ) changes = process_response(req_id.id, &parse_event/1, 5000) @@ -181,8 +163,6 @@ defmodule ChangesAsyncTest do assert length(changes) == 1 assert Enum.at(changes, 0)["id"] == "foo" assert dt_msec > 1000 - - HTTPotion.stop_worker_process(worker_pid) end @tag :with_db @@ -197,34 +177,31 @@ defmodule ChangesAsyncTest do lines = String.split(resp.body, "\n") - all_lines = lines - |> Enum.map(fn p -> Enum.at(String.split(p, ":"), 0) end) + all_lines = + lines + |> Enum.map(fn p -> Enum.at(String.split(p, ":"), 0) end) allowed = ["", "data", "id", "event"] - allowed_lines = all_lines - |> Enum.filter(fn p -> Enum.member?(allowed, p) end) + allowed_lines = + all_lines + |> Enum.filter(fn p -> Enum.member?(allowed, p) end) assert length(all_lines) == length(allowed_lines) - end @tag :with_db test "eventsource heartbeat", context do db_name = context[:db_name] - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) - req_id = Rawresp.get("/#{db_name}/_changes?feed=eventsource&heartbeat=10", - stream_to: {self(), :once}, - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) beats = wait_for_heartbeats(req_id.id, 0, 3) assert beats == 3 - HTTPotion.stop_worker_process(worker_pid) end @tag :with_db @@ -247,13 +224,11 @@ defmodule ChangesAsyncTest do last_seq = changes["last_seq"] # longpoll waits until a matching change before returning - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) req_id = Couch.get( "/#{db_name}/_changes?feed=longpoll&filter=changes_filter/bop&since=#{last_seq}", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) @@ -279,20 +254,17 @@ defmodule ChangesAsyncTest do create_doc(db_name, %{bop: false}) create_doc(db_name, %{_id: "bingo", bop: "bingo"}) - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) - req_id = Rawresp.get( "/#{db_name}/_changes?feed=continuous&filter=changes_filter/bop&timeout=500", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) create_doc(db_name, %{_id: "rusty", bop: "plankton"}) retry_until(fn -> - changes = process_response(req_id.id, &parse_changes_line_chunk/1) + changes = process_response(req_id.id, &parse_changes_line/1) changes_ids = changes @@ -300,8 +272,8 @@ defmodule ChangesAsyncTest do |> Enum.map(fn p -> p["id"] end) Enum.member?(changes_ids, "bingo") and - Enum.member?(changes_ids, "rusty") and - length(changes_ids) == 2 + Enum.member?(changes_ids, "rusty") and + length(changes_ids) == 2 end) end @@ -313,21 +285,18 @@ defmodule ChangesAsyncTest do create_doc(db_name, %{_id: "doc1", value: 1}) create_doc(db_name, %{_id: "doc2", value: 2}) - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) - req_id = Rawresp.post( "/#{db_name}/_changes?feed=continuous&timeout=500&filter=_doc_ids", body: doc_ids, headers: ["Content-Type": "application/json"], - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id.id, 200) create_doc(db_name, %{_id: "doc3", value: 3}) - changes = process_response(req_id.id, &parse_changes_line_chunk/1) + changes = process_response(req_id.id, &parse_changes_line/1) changes_ids = changes @@ -352,15 +321,12 @@ defmodule ChangesAsyncTest do assert length(resp.body["results"]) == 4 seq = Enum.at(resp.body["results"], 1)["seq"] - {:ok, worker_pid} = HTTPotion.spawn_link_worker_process(Couch.process_url("")) - # simulate an EventSource request with a Last-Event-ID header req_id = Rawresp.get( "/#{db_name}/_changes?feed=eventsource&timeout=100&since=0", headers: [Accept: "text/event-stream", "Last-Event-ID": seq], - stream_to: self(), - direct: worker_pid + stream_to: self() ) changes = process_response(req_id.id, &parse_event/1) @@ -369,13 +335,16 @@ defmodule ChangesAsyncTest do defp wait_for_heartbeats(id, beats, expexted_beats) do if beats < expexted_beats do - :ibrowse.stream_next(id) - is_heartbeat = process_response(id, &parse_heartbeat/1) - - case is_heartbeat do - :heartbeat -> wait_for_heartbeats(id, beats + 1, expexted_beats) - :timeout -> beats - _ -> wait_for_heartbeats(id, beats, expexted_beats) + case next_chunk(id) do + :timeout -> + beats + + chunk -> + if Regex.match?(~r/event: heartbeat/, chunk) do + wait_for_heartbeats(id, beats + 1, expexted_beats) + else + wait_for_heartbeats(id, beats, expexted_beats) + end end else beats @@ -384,7 +353,7 @@ defmodule ChangesAsyncTest do defp wait_for_headers(id, status, timeout \\ 1000) do receive do - %HTTPotion.AsyncHeaders{id: ^id, status_code: ^status} -> + %Couch.AsyncHeaders{id: ^id, status_code: ^status} -> :ok _ -> @@ -394,24 +363,51 @@ defmodule ChangesAsyncTest do end end - defp process_response(id, chunk_parser, timeout \\ 3000) do + # Gather response until stream ends and also handles timeouts as we'd expect + # them in the _changes feeds responses normally and we have tests for those + defp process_response(id, parser, timeout \\ 3000) do + acc = Process.delete({:chunk_acc, id}) || [] + case gather_response(id, acc, timeout) do + {:done, body} -> + parser.(body) + {:timeout, []} -> + :timeout + {:timeout, acc} -> + Process.put({:chunk_acc, id}, acc) + :timeout + end + end + + defp gather_response(id, acc, timeout) do receive do - %HTTPotion.AsyncChunk{id: ^id} = msg -> - chunk_parser.(msg) + %Couch.AsyncChunk{id: ^id, chunk: chunk} -> + gather_response(id, [acc | chunk], timeout) + %Couch.AsyncEnd{id: ^id} -> + if acc == [], do: {:timeout, []}, else: {:done, IO.iodata_to_binary(acc)} + _ -> + gather_response(id, acc, timeout) + after + timeout -> {:timeout, acc} + end + end + defp next_chunk(id, timeout \\ 3000) do + receive do + %Couch.AsyncChunk{id: ^id, chunk: chunk} -> + chunk _ -> - process_response(id, chunk_parser, timeout) + next_chunk(id, timeout) after timeout -> :timeout end end - defp parse_chunk(msg) do - msg.chunk |> IO.iodata_to_binary() |> :jiffy.decode([:return_maps, :use_nil]) + defp parse_chunk(body) do + :jiffy.decode(body, [:return_maps, :use_nil]) end - defp parse_event(msg) do - captures = Regex.scan(~r/data: (.*)/, msg.chunk) + defp parse_event(body) do + captures = Regex.scan(~r/data: (.*)/, body) captures |> Enum.map(fn p -> Enum.at(p, 1) end) @@ -423,16 +419,6 @@ defmodule ChangesAsyncTest do end) end - defp parse_heartbeat(msg) do - is_heartbeat = Regex.match?(~r/event: heartbeat/, msg.chunk) - - if is_heartbeat do - :heartbeat - else - :other - end - end - defp parse_changes_response(changes) do {length(changes["results"]), String.slice(changes["last_seq"], 0..1)} end @@ -466,31 +452,23 @@ defmodule ChangesAsyncTest do assert String.at(change["last_seq"], 0) == "1" # create_doc_bar(db_name,"bar") - {:ok, worker_pid} = HTTPotion.spawn_worker_process(Couch.process_url("")) - %HTTPotion.AsyncResponse{id: req_id} = + %Couch.AsyncResponse{id: req_id} = Rawresp.get("/#{db_name}/_changes?feed=#{feed}&timeout=500", - stream_to: self(), - direct: worker_pid + stream_to: self() ) :ok = wait_for_headers(req_id, 200) create_doc_bar(db_name, "bar") - changes = process_response(req_id, &parse_changes_line_chunk/1) + changes = process_response(req_id, &parse_changes_line/1) assert length(changes) == 3 - - HTTPotion.stop_worker_process(worker_pid) end def create_doc_bar(db_name, id) do create_doc(db_name, %{:_id => id, :bar => 1}) end - defp parse_changes_line_chunk(msg) do - parse_changes_line(msg.chunk) - end - defp parse_changes_line(body) do body_lines = String.split(body, "\n") diff --git a/test/elixir/test/design_paths_test.exs b/test/elixir/test/design_paths_test.exs index b3e10c165..1485eb878 100644 --- a/test/elixir/test/design_paths_test.exs +++ b/test/elixir/test/design_paths_test.exs @@ -25,11 +25,9 @@ defmodule DesignPathTest do resp = Couch.get("/#{db_name}/_design/test") assert resp.body["_id"] == "_design/test" - resp = - Couch.get(Couch.process_url("/#{db_name}/_design%2Ftest"), - follow_redirects: true - ) - + resp = Couch.get("/#{db_name}/_design%2Ftest") + assert resp.status_code == 301 + resp = Couch.get(resp.headers["location"]) assert resp.body["_id"] == "_design/test" resp = Couch.get("/#{db_name}/_design/test/_view/testing") @@ -50,11 +48,9 @@ defmodule DesignPathTest do resp = Couch.get("/#{db_name}/_design/test2") assert resp.body["_id"] == "_design/test2" - resp = - Couch.get(Couch.process_url("/#{db_name}/_design%2Ftest2"), - follow_redirects: true - ) - + resp = Couch.get("/#{db_name}/_design%2Ftest2") + assert resp.status_code == 301 + resp = Couch.get(resp.headers["location"]) assert resp.body["_id"] == "_design/test2" resp = Couch.get("/#{db_name}/_design/test2/_view/testing") diff --git a/test/elixir/test/replication_test.exs b/test/elixir/test/replication_test.exs index 75e69ed2d..b86d28574 100644 --- a/test/elixir/test/replication_test.exs +++ b/test/elixir/test/replication_test.exs @@ -87,7 +87,7 @@ defmodule ReplicationTest do opts = [headers: [Accept: "application/json"], query: query] resp = Couch.get("/#{tgt_db_name}/#{doc["_id"]}", opts) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert is_map(resp.body) refute Map.has_key?(resp.body, "_conflicts") refute Map.has_key?(resp.body, "_deleted_conflicts") @@ -904,11 +904,11 @@ defmodule ReplicationTest do assert history["doc_write_failures"] == 0 resp = Couch.get!("/#{tgt_db_name}/foo1") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["value"] == 1 resp = Couch.get!("/#{tgt_db_name}/foo2") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["value"] == 2 resp = Couch.get!("/#{tgt_db_name}/foo3") @@ -932,23 +932,23 @@ defmodule ReplicationTest do assert history["doc_write_failures"] == 0 resp = Couch.get!("/#{tgt_db_name}/foo1") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["value"] == 1 resp = Couch.get!("/#{tgt_db_name}/foo2") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["value"] == 2 resp = Couch.get!("/#{tgt_db_name}/foo3") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["value"] == 3 resp = Couch.get!("/#{tgt_db_name}/foo4") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["value"] == 4 resp = Couch.get!("/#{tgt_db_name}/_design/mydesign") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) end def run_by_id_repl(src_prefix, tgt_prefix) do @@ -1046,8 +1046,8 @@ defmodule ReplicationTest do assert orig.status_code == 404 assert copy.status_code == 404 else - assert HTTPotion.Response.success?(orig) - assert HTTPotion.Response.success?(copy) + assert Couch.Response.success?(orig) + assert Couch.Response.success?(copy) assert cmp_json(orig.body, copy.body) end end) @@ -1059,7 +1059,7 @@ defmodule ReplicationTest do is_doc_id = &Enum.member?(doc_ids, &1) if is_doc_id.(doc["_id"]) or is_doc_id.(encoded_id) do - assert HTTPotion.Response.success?(copy) + assert Couch.Response.success?(copy) else assert copy.status_code == 404 end @@ -1101,8 +1101,8 @@ defmodule ReplicationTest do assert orig.status_code == 404 assert copy.status_code == 404 else - assert HTTPotion.Response.success?(orig) - assert HTTPotion.Response.success?(copy) + assert Couch.Response.success?(orig) + assert Couch.Response.success?(copy) assert cmp_json(orig.body, copy.body) end end) @@ -1116,7 +1116,7 @@ defmodule ReplicationTest do is_doc_id = &Enum.member?(all_doc_ids, &1) if is_doc_id.(doc["_id"]) or is_doc_id.(encoded_id) do - assert HTTPotion.Response.success?(copy) + assert Couch.Response.success?(copy) else assert copy.status_code == 404 end @@ -1161,7 +1161,7 @@ defmodule ReplicationTest do query = %{"conflicts" => "true"} copy = Couch.get!("/#{tgt_db_name}/#{conflict_id}", query: query) - assert HTTPotion.Response.success?(copy) + assert Couch.Response.success?(copy) assert copy.body["integer"] == 666 assert String.starts_with?(copy.body["_rev"], "4-") assert not Map.has_key?(doc, "_conflicts") @@ -1537,7 +1537,7 @@ defmodule ReplicationTest do if String.starts_with?(doc["_id"], "_design/") do assert resp.status_code == 404 else - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert cmp_json(doc, resp.body) end end) @@ -1593,7 +1593,7 @@ defmodule ReplicationTest do def get_db_info(db_name) do resp = Couch.get("/#{db_name}") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) resp.body end @@ -1610,7 +1610,7 @@ defmodule ReplicationTest do def get_db_changes(db_name, query \\ %{}) do resp = Couch.get("/#{db_name}/_changes", query: query) - assert HTTPotion.Response.success?(resp), "#{inspect(resp)} #{inspect(query)}" + assert Couch.Response.success?(resp), "#{inspect(resp)} #{inspect(query)}" resp.body end @@ -1618,7 +1618,7 @@ defmodule ReplicationTest do query = %{w: 3} body = %{docs: docs} resp = Couch.post("/#{db_name}/_bulk_docs", query: query, body: body) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) for {doc, resp} <- Enum.zip(docs, resp.body) do assert resp["ok"], "Error saving doc: #{doc["_id"]}" @@ -1628,7 +1628,7 @@ defmodule ReplicationTest do def set_security(db_name, sec_props) do resp = Couch.put("/#{db_name}/_security", body: :jiffy.encode(sec_props, [:use_nil])) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert resp.body["ok"] end @@ -1652,7 +1652,7 @@ defmodule ReplicationTest do retry_until(fn -> resp = Couch.put(uri, headers: headers, query: params, body: att[:body]) - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) Map.put(doc, "_rev", resp.body["rev"]) end) end @@ -1718,7 +1718,7 @@ defmodule ReplicationTest do def try_get_task(repl_id) do resp = Couch.get("/_active_tasks") - assert HTTPotion.Response.success?(resp) + assert Couch.Response.success?(resp) assert is_list(resp.body) Enum.find(resp.body, nil, fn task ->
