@@ -1,0 +1,268 @@
# Copyright 2026 Cole Christensen
#
# 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.
defmodule ExGitObjectstore.Protocol.ReceivePackHooksTest do
use ExUnit.Case, async: true
alias ExGitObjectstore.{Object, Ref}
alias ExGitObjectstore.Object.{Blob, Commit, Tree}
alias ExGitObjectstore.Protocol.{PktLine, ReceivePack}
alias ExGitObjectstore.Test.RepoHelper
@zero_sha String.duplicate("0", 40)
# -- Helpers --
defp build_repo_with_commit do
repo = RepoHelper.memory_repo()
ExGitObjectstore.init(repo)
blob = Blob.from_content("hello\n")
{:ok, blob_sha} = Object.write(repo, blob)
tree = Tree.new([%{mode: "100644", name: "file.txt", sha: blob_sha}])
{:ok, tree_sha} = Object.write(repo, tree)
commit = %Commit{
tree: tree_sha,
parents: [],
author: "Test <test@test.com> 1000000000 +0000",
committer: "Test <test@test.com> 1000000000 +0000",
message: "initial\n"
}
{:ok, commit_sha} = Object.write(repo, commit)
%{repo: repo, commit_sha: commit_sha, tree_sha: tree_sha, blob_sha: blob_sha}
end
defp push_command(old_sha, new_sha, ref) do
cmd_line = "#{old_sha} #{new_sha} #{ref}\0report-status"
PktLine.encode(cmd_line) <> PktLine.flush()
end
defp push_commands(cmds) do
{first, rest} =
case cmds do
[{old, new, ref} | t] -> {PktLine.encode("#{old} #{new} #{ref}\0report-status"), t}
end
rest_lines =
Enum.map(rest, fn {old, new, ref} ->
PktLine.encode("#{old} #{new} #{ref}")
end)
IO.iodata_to_binary([first | rest_lines] ++ [PktLine.flush()])
end
# Minimal valid pack with 0 objects (header + SHA checksum)
defp empty_pack do
header = <<"PACK", 2::unsigned-big-32, 0::unsigned-big-32>>
checksum = :crypto.hash(:sha, header)
header <> checksum
end
defp assert_ref_ok(response, ref) do
assert response =~ "ok #{ref}"
end
defp assert_ref_ng(response, ref) do
assert response =~ "ng #{ref}"
end
# -- update hook tests --
describe "update hook" do
test "called per-ref, can accept individual refs" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
test_pid = self()
update_hook = fn ref, _old_sha, _new_sha ->
send(test_pid, {:update_called, ref})
:ok
end
{_advert, state} = ReceivePack.init(repo, update_hook: update_hook)
data = push_command(@zero_sha, sha, "refs/heads/main")
{response, _state} = ReceivePack.feed(state, data <> empty_pack())
assert_received {:update_called, "refs/heads/main"}
assert_ref_ok(response, "refs/heads/main")
end
test "rejection blocks only the affected ref" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
update_hook = fn ref, _old_sha, _new_sha ->
if ref == "refs/heads/blocked" do
{:error, "branch is protected"}
else
:ok
end
end
{_advert, state} = ReceivePack.init(repo, update_hook: update_hook)
data =
push_commands([
{@zero_sha, sha, "refs/heads/allowed"},
{@zero_sha, sha, "refs/heads/blocked"}
])
{response, _state} = ReceivePack.feed(state, data <> empty_pack())
assert_ref_ok(response, "refs/heads/allowed")
assert_ref_ng(response, "refs/heads/blocked")
# allowed ref was actually created
assert {:ok, ^sha} = Ref.get(repo, "refs/heads/allowed")
# blocked ref was not created
assert {:error, :not_found} = Ref.get(repo, "refs/heads/blocked")
end
test "called with correct arguments" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
test_pid = self()
update_hook = fn ref, old_sha, new_sha ->
send(test_pid, {:update_args, ref, old_sha, new_sha})
:ok
end
{_advert, state} = ReceivePack.init(repo, update_hook: update_hook)
data = push_command(@zero_sha, sha, "refs/heads/main")
{_response, _state} = ReceivePack.feed(state, data <> empty_pack())
assert_received {:update_args, "refs/heads/main", @zero_sha, ^sha}
end
test "no update hook means all refs accepted" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
{_advert, state} = ReceivePack.init(repo)
data = push_command(@zero_sha, sha, "refs/heads/main")
{response, _state} = ReceivePack.feed(state, data <> empty_pack())
assert_ref_ok(response, "refs/heads/main")
end
end
# -- post_receive hook tests --
describe "post_receive hook" do
test "called after all refs updated with list of changes" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
test_pid = self()
post_receive_hook = fn changes ->
send(test_pid, {:post_receive, changes})
:ok
end
{_advert, state} = ReceivePack.init(repo, post_receive_hook: post_receive_hook)
data = push_command(@zero_sha, sha, "refs/heads/main")
{_response, _state} = ReceivePack.feed(state, data <> empty_pack())
assert_received {:post_receive, changes}
assert [%{ref: "refs/heads/main", old_sha: @zero_sha, new_sha: ^sha, status: :ok}] = changes
end
test "includes rejected refs in changes list" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
test_pid = self()
update_hook = fn ref, _old, _new ->
if ref == "refs/heads/blocked", do: {:error, "nope"}, else: :ok
end
post_receive_hook = fn changes ->
send(test_pid, {:post_receive, changes})
:ok
end
{_advert, state} =
ReceivePack.init(repo,
update_hook: update_hook,
post_receive_hook: post_receive_hook
)
data =
push_commands([
{@zero_sha, sha, "refs/heads/ok"},
{@zero_sha, sha, "refs/heads/blocked"}
])
{_response, _state} = ReceivePack.feed(state, data <> empty_pack())
assert_received {:post_receive, changes}
assert length(changes) == 2
ok_change = Enum.find(changes, &(&1.ref == "refs/heads/ok"))
blocked_change = Enum.find(changes, &(&1.ref == "refs/heads/blocked"))
assert ok_change.status == :ok
assert match?({:error, _}, blocked_change.status)
end
test "failure is logged but does not affect push result" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
post_receive_hook = fn _changes ->
{:error, "webhook failed"}
end
{_advert, state} = ReceivePack.init(repo, post_receive_hook: post_receive_hook)
data = push_command(@zero_sha, sha, "refs/heads/main")
{response, state} = ReceivePack.feed(state, data <> empty_pack())
# Push should still succeed even though post_receive failed
assert_ref_ok(response, "refs/heads/main")
assert {:ok, ^sha} = Ref.get(repo, "refs/heads/main")
# The state result should reflect success (post_receive doesn't affect it)
assert state.result == :ok
end
test "not called when pre_receive_hook rejects" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
test_pid = self()
pre_receive_hook = fn _commands -> {:error, "rejected"} end
post_receive_hook = fn _changes ->
send(test_pid, :post_receive_called)
:ok
end
{_advert, state} =
ReceivePack.init(repo,
pre_receive_hook: pre_receive_hook,
post_receive_hook: post_receive_hook
)
data = push_command(@zero_sha, sha, "refs/heads/main")
{_response, _state} = ReceivePack.feed(state, data <> empty_pack())
refute_received :post_receive_called
end
test "no post_receive hook is fine" do
%{repo: repo, commit_sha: sha} = build_repo_with_commit()
{_advert, state} = ReceivePack.init(repo)
data = push_command(@zero_sha, sha, "refs/heads/main")
{response, _state} = ReceivePack.feed(state, data <> empty_pack())
assert_ref_ok(response, "refs/heads/main")
end
end
end