okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.peterl: replace deprecated catch expressions with try for OTP 2931ff8342026-10-06

%% A fake usage CLI, a local HTTP endpoint and a command whose whole job is to
%% dump the environment it ran with: the three things usage_feed's loop has to
%% get right (envelope on stdin, responses replayed, allowlist on commands)
%% without a real provider, a real credential or the network.
%%
%% ALBEDO_USAGE_CORE points the driver's binary resolution at the fake, so the
%% daemon is never involved: no route exercises this loop end to end yet.
-module(albedo_usage_test_support).

-export([start/0, stop/1, envelope/1, env_dump/1, request_line/1,
         break_binary/1, mono_ms/0, trickle_pid_gone/1]).

%% eunit runs a module's tests in parallel ({inparallel, _} reaches into the
%% module), and ALBEDO_USAGE_CORE is one global environment variable, so the
%% fakes must not overlap. A registered name is the lock: it is held from
%% start to stop, and the death of a crashed test process releases it.
-define(LOCK, albedo_usage_fake).

start() ->
    await_lock(),
    Dir = filename:join(tmp(), "albedo-usage-" ++ os:getpid() ++ "-"
                              ++ integer_to_list(erlang:unique_integer([positive]))),
    ok = file:make_dir(Dir),
    {ok, Listen} = gen_tcp:listen(0, [binary, {active, false}, {reuseaddr, true}, {packet, raw}]),
    {ok, {_Address, Port}} = inet:sockname(Listen),
    Server = spawn(fun() -> serve(Listen, []) end),
    Url = io_lib:format("http://127.0.0.1:~B/usage", [Port]),
    Usage = filename:join(Dir, "usage"),
    Dump = filename:join(Dir, "env-dump.sh"),
    Trickle = filename:join(Dir, "trickle.sh"),
    write(Usage, fake_usage(Url, Dump, Trickle, filename:join(Dir, "envelope.txt"))),
    write(Dump, ["#!/bin/sh\nenv >", filename:join(Dir, "env.txt"), "\necho dumped\n"]),
    %% Prints its own pid, then a line forever: a command that always has
    %% output but never exits, for the deadline-kill test.
    write(Trickle, ["#!/bin/sh\necho $$ >", filename:join(Dir, "trickle.pid"),
                    "\nwhile true; do\n  echo tick\n  sleep 0.2\ndone\n"]),
    ok = file:change_mode(Usage, 8#755),
    ok = file:change_mode(Dump, 8#755),
    ok = file:change_mode(Trickle, 8#755),
    os:putenv("ALBEDO_USAGE_CORE", Usage),
    os:putenv("ALBEDO_USAGE_TEST_CANARY", "leak"),
    {ok, #{dir => Dir, listen => Listen, server => Server,
           envelope => filename:join(Dir, "envelope.txt"),
           env => filename:join(Dir, "env.txt"),
           trickle_pid => filename:join(Dir, "trickle.pid")}}.

stop(Ref) ->
    exit(maps:get(server, Ref), kill),
    gen_tcp:close(maps:get(listen, Ref)),
    os:unsetenv("ALBEDO_USAGE_CORE"),
    os:unsetenv("ALBEDO_USAGE_TEST_CANARY"),
    _ = file:del_dir_r(maps:get(dir, Ref)),
    _ = try unregister(?LOCK) catch error:badarg -> false end,
    nil.

await_lock() ->
    case whereis(?LOCK) of
        Pid when is_pid(Pid) ->
            Ref = erlang:monitor(process, Pid),
            receive
                {'DOWN', Ref, process, Pid, _} -> ok
            after 15000 ->
                exit(usage_fake_lock_timeout)
            end,
            await_lock();
        undefined ->
            try register(?LOCK, self()) of
                true -> ok
            catch
                _:_ -> await_lock()
            end
    end.

envelope(Ref) ->
    read(maps:get(envelope, Ref)).

env_dump(Ref) ->
    read(maps:get(env, Ref)).

request_line(Ref) ->
    Server = maps:get(server, Ref),
    Server ! {requests, self()},
    receive
        {requests, Server, Lines} -> Lines
    after 2000 -> []
    end.

%% Points resolution at a path that does not exist: the override must be
%% authoritative, never silently falling back to PATH.
break_binary(Ref) ->
    os:putenv("ALBEDO_USAGE_CORE", filename:join(maps:get(dir, Ref), "not-there")),
    nil.

mono_ms() ->
    erlang:monotonic_time(millisecond).

%% The trickle command is killed on timeout, not orphaned: its pid stops
%% answering signal 0. The beam reaps its port children asynchronously, so
%% the check polls briefly rather than asserting the very first probe.
trickle_pid_gone(Ref) ->
    Path = maps:get(trickle_pid, Ref),
    case file:read_file(Path) of
        {ok, Bin} ->
            Pid = binary_to_integer(string:trim(Bin)),
            gone(Pid, 20);
        _ ->
            false
    end.

gone(Pid, 0) ->
    signal(Pid) =/= 0;
gone(Pid, N) ->
    case signal(Pid) of
        0 -> timer:sleep(50), gone(Pid, N - 1);
        _ -> true
    end.

%% Exit status of `kill -0 Pid`: 0 while the process exists.
signal(Pid) ->
    Kill = case os:find_executable("kill") of
        false -> erlang:error({missing_test_dependency, "kill executable not found in PATH"});
        Executable -> Executable
    end,
    Port = open_port({spawn_executable, Kill},
                     [{args, ["-0", integer_to_list(Pid)]}, exit_status, binary, hide]),
    receive
        {Port, {exit_status, Status}} -> Status
    after 1000 ->
        _ = try port_close(Port) catch _:_ -> ok end,
        0
    end.

read(Path) ->
    case file:read_file(Path) of
        {ok, Bin} -> Bin;
        _ -> <<>>
    end.

write(Path, Content) ->
    ok = file:write_file(Path, Content).

tmp() ->
    case os:getenv("TMPDIR") of
        false -> "/tmp";
        T -> T
    end.

%% The fake core speaks the real protocol: one envelope line in (the whole
%% history, on stdin), one step line out, and the provider name decides what
%% kind of feed it is. The default feed answers round one with an http request
%% and a command, and round two with a report built from the http body - so
%% the loop, the envelope and the command environment are all load-bearing.
fake_usage(Url, Dump, Trickle, EnvelopeLog) ->
    ["#!/usr/bin/env python3\n"
     "import json, sys\n"
     "line = sys.stdin.readline()\n"
     "open(", quote(EnvelopeLog), ", 'a').write(line)\n"
     "envelope = json.loads(line)\n"
     "provider = envelope.get('provider')\n"
     "responses = envelope.get('responses', [])\n"
     "if provider == 'endless':\n"
     "    step = {'requests': [{'kind': 'command', 'command': 'true', 'args': []}]}\n"
     "elif provider == 'broken':\n"
     "    step = {'report': {'limits': [], 'error': 'the provider is unreachable'}}\n"
     "elif provider == 'trickle':\n"
     "    if responses:\n"
     "        status = responses[-1][0]['status']\n"
     "        step = {'report': {'limits': [], 'error': 'command answered %d' % status}}\n"
     "    else:\n"
     "        step = {'requests': [\n"
     "            {'kind': 'command', 'command': ", quote(Trickle), ", 'args': [],\n"
     "             'timeoutMs': 1000}]}\n"
     "elif provider == 'missing-command' and responses:\n"
     "    status = responses[-1][0]['status']\n"
     "    step = {'report': {'limits': [], 'error': 'command answered %d' % status}}\n"
     "elif responses:\n"
     "    body = json.loads(responses[-1][0]['body'])\n"
     "    step = {'report': {\n"
     "        'accountId': 'acct-1', 'email': 'dawn@example.com', 'planType': 'token',\n"
     "        'limits': [{'id': 'five_hour', 'label': 'Session',\n"
     "                    'usedPercent': body['pct'], 'windowLabel': '5h window',\n"
     "                    'windowSeconds': 18000, 'resetsAt': 1893456000000,\n"
     "                    'status': 'ok'}],\n"
     "        'error': None}}\n"
     "elif provider == 'missing-command':\n"
     "    step = {'requests': [\n"
     "        {'kind': 'command', 'command': 'no-such-binary-xyz', 'args': []}]}\n"
     "else:\n"
     "    step = {'requests': [\n"
     "        {'kind': 'http', 'method': 'GET', 'url': ", quote(Url), ",\n"
     "         'headers': [['accept', 'application/json']]},\n"
     "        {'kind': 'command', 'command': ", quote(Dump), ", 'args': [],\n"
     "         'timeoutMs': 5000}]}\n"
     "print(json.dumps(step))\n"].

quote(Text) ->
    [$', Text, $'].

serve(Listen, Acc) ->
    receive
        {requests, From} ->
            From ! {requests, self(), lists:reverse(Acc)}
    after 0 ->
        ok
    end,
    case gen_tcp:accept(Listen, 200) of
        {ok, Socket} ->
            case read_head(Socket, <<>>) of
                {ok, Head} ->
                    [RequestLine | _] = binary:split(Head, <<"\r\n">>),
                    Body = <<"{\"pct\":37}">>,
                    ok = gen_tcp:send(Socket, [
                        <<"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n">>,
                        <<"content-length: ">>, integer_to_binary(byte_size(Body)),
                        <<"\r\n\r\n">>, Body]),
                    gen_tcp:close(Socket),
                    serve(Listen, [RequestLine | Acc]);
                _ ->
                    gen_tcp:close(Socket),
                    serve(Listen, Acc)
            end;
        {error, timeout} ->
            serve(Listen, Acc);
        _ ->
            ok
    end.

read_head(Socket, Acc) ->
    case binary:match(Acc, <<"\r\n\r\n">>) of
        nomatch ->
            case gen_tcp:recv(Socket, 0, 5000) of
                {ok, Bytes} -> read_head(Socket, <<Acc/binary, Bytes/binary>>);
                Error -> Error
            end;
        _ ->
            {ok, Acc}
    end.