Michał Szczepanik
b89f2afb0d
Forgejo-aneksajo 16.0.1-git-annex-2 now bundles a more recent git-annex, which has the fix for the double HEAD request (likely 10.20260717; still based on alpine 3.23 but now installing git-annex from edge). See: https://codeberg.org/forgejo-aneksajo/forgejo-aneksajo/pulls/123
296 lines
11 KiB
Python
296 lines
11 KiB
Python
from pathlib import Path
|
|
import random
|
|
import subprocess
|
|
import tempfile
|
|
|
|
from datalad_core.create import create_annexrepo
|
|
from datalad_core.git_utils import get_current_branch
|
|
from datalad_core.runners import call_annex_json_lines, call_git, call_git_oneline
|
|
from furl import furl
|
|
import pytest
|
|
import requests
|
|
|
|
|
|
def test_push_to_create(pytestconfig, forgejo, words):
|
|
"""Test the push to create functionality
|
|
|
|
This tests:
|
|
|
|
- whether push to create creates the forgejo repo (this depends on
|
|
a feature which is off by default, but became a de facto
|
|
standard for us)
|
|
|
|
- whether the default branch in forgejo matches the local branch
|
|
(ie. not git-annex); this most likely depends on the order in
|
|
which branches are pushed, changed in git-annex 10.20250605
|
|
("sync: Push the current branch first, rather than a synced
|
|
branch, to better support git forges (gitlab, gitea, forgejo,
|
|
etc.) which use push-to-create with the first pushed branch
|
|
becoming the default branch.")
|
|
|
|
- whether the UUID becomes known after a single push; this
|
|
probably changed in git-annex 10.20260213 ("When used with git
|
|
forges that allow Push to Create, the remote's annex-uuid is
|
|
re-probed after the initial push.")
|
|
|
|
This uses a single call to `git annex push`.
|
|
|
|
See:
|
|
|
|
https://forgejo.org/docs/latest/user/push-to-create/
|
|
https://git-annex.branchable.com/todo/Pushing_current_branch_before_synced__42___or_git-annex/
|
|
https://git-annex.branchable.com/todo/support_push_to_create/
|
|
|
|
"""
|
|
|
|
if pytestconfig.getoption("org") is not None:
|
|
owner = pytestconfig.getoption("org")
|
|
else:
|
|
# we have a token, we need to check which user it belongs to
|
|
r = forgejo.get_user()
|
|
owner = r.json().get("login")
|
|
assert owner is not None and len(owner) > 0
|
|
|
|
remote_repo_name = words.random_name()
|
|
push_url = furl(
|
|
pytestconfig.getoption("instance_url"), path=f"{owner}/{remote_repo_name}"
|
|
).tostr()
|
|
|
|
# the name is random, but best make sure repo doesn't exist
|
|
r = forgejo.get_repo(owner, remote_repo_name, do_raise=False)
|
|
assert r.status_code == 404
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = create_annexrepo(repo_path)
|
|
(wt.path / "foo.dat").write_bytes(random.randbytes(1024))
|
|
|
|
_ = list(call_annex_json_lines(annex_args=["add", "foo.dat"], cwd=wt.path))
|
|
_ = call_git(args=["commit", "-m", "Add foo"], cwd=wt.path)
|
|
_ = call_git(args=["remote", "add", "origin", push_url], cwd=wt.path)
|
|
_ = subprocess.run(
|
|
args=["git", "annex", "push", "origin"],
|
|
cwd=wt.path,
|
|
check=True,
|
|
)
|
|
|
|
# check if the remote exists
|
|
r = forgejo.get_repo(
|
|
owner, remote_repo_name
|
|
) # 200 OK or raises for 404 Not Found
|
|
|
|
# check if the default branch is the branch that got pushed (not git-annex)
|
|
current_branch = get_current_branch(repo_path)
|
|
remote_branch = r.json().get("default_branch")
|
|
assert current_branch == remote_branch
|
|
|
|
# check if the remote got initialized with a UUID, and local knows it
|
|
res = list(call_annex_json_lines(annex_args=["info", "origin"], cwd=wt.path))
|
|
assert len(res[0]["uuid"]) > 0
|
|
|
|
|
|
@pytest.mark.xfail(reason="Known issue, forgejo-aneksajo/issues/103")
|
|
def test_pull_does_not_create(pytestconfig, forgejo, words):
|
|
"""Test git annex pull from a non-existing repo
|
|
|
|
Pulling from a remote URL which points to a non-existing repo
|
|
should not auto-create that repo. Pull creating a repo has been
|
|
observed as an unwanted side-effect of push-to-create workarounds
|
|
(what happens when git-annex requests /config).
|
|
|
|
A test for
|
|
https://codeberg.org/forgejo-aneksajo/forgejo-aneksajo/issues/103
|
|
|
|
"""
|
|
remote_repo_name = words.random_name()
|
|
|
|
if pytestconfig.getoption("org") is not None:
|
|
owner = pytestconfig.getoption("org")
|
|
else:
|
|
# we have a token, we need to check which user it belongs to
|
|
r = forgejo.get_user()
|
|
owner = r.json().get("login")
|
|
assert owner is not None and len(owner) > 0
|
|
|
|
# the name is random, but best make sure repo doesn't exist
|
|
clone_url = furl(
|
|
pytestconfig.getoption("instance_url"), path=f"{owner}/{remote_repo_name}"
|
|
).tostr()
|
|
r = forgejo.get_repo(owner, remote_repo_name, do_raise=False)
|
|
assert r.status_code == 404
|
|
|
|
with tempfile.TemporaryDirectory(delete=True) as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = create_annexrepo(repo_path)
|
|
|
|
_ = call_git(args=["remote", "add", "foo", clone_url], cwd=wt.path)
|
|
|
|
_ = subprocess.run(
|
|
args=["git", "annex", "pull", "foo"],
|
|
cwd=wt.path,
|
|
check=False, # ok to return non-zero, hide error
|
|
)
|
|
|
|
# the repo should not get created
|
|
r = forgejo.get_repo(owner, remote_repo_name, do_raise=False)
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_create_same_name(pytestconfig, forgejo, words):
|
|
"""Test deleting and recreating a repo
|
|
|
|
Deleting an annex-initialized forgejo repo and creating a new one
|
|
under the same name should not cause any issues.
|
|
|
|
A test for
|
|
https://codeberg.org/forgejo-aneksajo/forgejo-aneksajo/issues/83
|
|
|
|
"""
|
|
|
|
# create a Forgejo repo (push-to-create to be tested elsewhere)
|
|
remote_repo_name = words.random_name()
|
|
r = forgejo.create_repo(remote_repo_name, org=pytestconfig.getoption("org"))
|
|
create_response = r.json()
|
|
|
|
# create a local repo with an annexed file & push
|
|
with tempfile.TemporaryDirectory(delete=True) as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = create_annexrepo(repo_path)
|
|
(wt.path / "bigstuff").write_bytes(random.randbytes(1024))
|
|
|
|
# git annex add, see https://hub.datalad.org/datalad/datalad-core/commit/a1feef64
|
|
_ = list(call_annex_json_lines(annex_args=["add", "bigstuff"], cwd=wt.path))
|
|
_ = call_git(args=["commit", "-m", "dummy"], cwd=wt.path)
|
|
_ = call_git(
|
|
args=["remote", "add", "origin", create_response["clone_url"]], cwd=wt.path
|
|
)
|
|
|
|
# git annex sync
|
|
# note: does not support --json, thus basic subprocess.run
|
|
_ = subprocess.run(
|
|
args=["git", "annex", "sync", "--content"],
|
|
cwd=wt.path,
|
|
check=True, # raises CalledProcessError on non-zero exit
|
|
)
|
|
|
|
# check if origin got a UUID
|
|
res = list(call_annex_json_lines(annex_args=["info", "origin"], cwd=wt.path))
|
|
uuid_there_first_try = res[0]["uuid"]
|
|
assert len(uuid_there_first_try) > 0
|
|
|
|
# now, delete the repo
|
|
create_owner = create_response["owner"]["login"] # login works for user and org
|
|
r = forgejo.delete_repo(create_owner, remote_repo_name)
|
|
|
|
# create a repo again under the same name
|
|
r = forgejo.create_repo(remote_repo_name, org=pytestconfig.getoption("org"))
|
|
recreate_response = r.json()
|
|
|
|
# repeate create & sync
|
|
with tempfile.TemporaryDirectory(delete=True) as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = create_annexrepo(repo_path)
|
|
(wt.path / "bigstuff").write_bytes(random.randbytes(1024))
|
|
|
|
_ = list(call_annex_json_lines(annex_args=["add", "bigstuff"], cwd=wt.path))
|
|
_ = call_git(args=["commit", "-m", "dummy"], cwd=wt.path)
|
|
_ = call_git(
|
|
args=["remote", "add", "origin", recreate_response["clone_url"]],
|
|
cwd=wt.path,
|
|
)
|
|
_ = subprocess.run(
|
|
args=["git", "annex", "sync", "--content"],
|
|
cwd=wt.path,
|
|
check=True, # raises CalledProcessError on non-zero exit
|
|
)
|
|
|
|
# check if origin got a (new) UUID
|
|
for res in call_annex_json_lines(annex_args=["info", "origin"], cwd=wt.path):
|
|
assert len(res["uuid"]) > 0
|
|
assert res["uuid"] != uuid_there_first_try
|
|
|
|
|
|
def test_annex_init_priv_pub(pytestconfig, forgejo, words):
|
|
"""Test whether remote annex gets initialized (priv/publ)
|
|
|
|
We expect that git-annex sync for an empty forgejo repo would
|
|
initialize it with an annex UUID. This used not to be the case for
|
|
public repositories, so we test two cases (private and
|
|
public). This is related to forgejo-aneksajo's method of
|
|
initializing an annex.
|
|
|
|
A test for
|
|
https://codeberg.org/forgejo-aneksajo/forgejo-aneksajo/issues/93
|
|
|
|
"""
|
|
|
|
# create a Forgejo repo (push-to-create to be tested elsewhere)
|
|
public_repo_name = words.random_name()
|
|
org: str | None = pytestconfig.getoption("org")
|
|
r = forgejo.create_repo(name=public_repo_name, private=False, org=org)
|
|
public_clone_url = r.json()["clone_url"]
|
|
|
|
private_repo_name = words.random_name()
|
|
r = forgejo.create_repo(name=private_repo_name, private=True, org=org)
|
|
private_clone_url = r.json()["clone_url"]
|
|
|
|
with tempfile.TemporaryDirectory(delete=True) as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = create_annexrepo(repo_path)
|
|
(wt.path / "foo.dat").write_bytes(random.randbytes(1024))
|
|
|
|
_ = list(call_annex_json_lines(annex_args=["add", "foo.dat"], cwd=wt.path))
|
|
_ = call_git(args=["commit", "-m", "Add foo"], cwd=wt.path)
|
|
_ = call_git(args=["remote", "add", "publ", public_clone_url], cwd=wt.path)
|
|
_ = call_git(args=["remote", "add", "priv", private_clone_url], cwd=wt.path)
|
|
_ = subprocess.run(
|
|
args=["git", "annex", "sync", "--content"],
|
|
cwd=wt.path,
|
|
check=True, # raises CalledProcessError on non-zero exit
|
|
)
|
|
|
|
# check if both remotes have an UUID (according to local info)
|
|
for remote in ("priv", "publ"):
|
|
for res in call_annex_json_lines(annex_args=["info", remote], cwd=wt.path):
|
|
assert len(res["uuid"]) > 0
|
|
|
|
|
|
def test_double_head(pytestconfig, forgejo, words):
|
|
"""Make sure p2phttps does not hang on HEAD requests
|
|
|
|
A test for
|
|
https://git-annex.branchable.com/bugs/p2phttp_key_endpoints_stuck_after_2_head_requests/
|
|
|
|
The behavior should have been fixed in git-annex 10.20260525
|
|
|
|
"""
|
|
repo_name = words.random_name()
|
|
org: str | None = pytestconfig.getoption("org")
|
|
r = forgejo.create_repo(name=repo_name, private=False, org=org)
|
|
clone_url = r.json()["clone_url"]
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = create_annexrepo(repo_path)
|
|
(wt.path / "foo.dat").write_bytes(random.randbytes(1024))
|
|
|
|
_ = list(call_annex_json_lines(annex_args=["add", "foo.dat"], cwd=wt.path))
|
|
_ = call_git(args=["commit", "-m", "Add foo"], cwd=wt.path)
|
|
_ = call_git(args=["remote", "add", "origin", clone_url], cwd=wt.path)
|
|
_ = call_git(args=["annex", "push", "origin"], cwd=wt.path)
|
|
|
|
key = call_git_oneline(args=["annex", "lookupkey", "foo.dat"], cwd=wt.path)
|
|
|
|
info = next(call_annex_json_lines(annex_args=["info", "origin"], cwd=wt.path))
|
|
|
|
# build p2phttp file download URL
|
|
f = furl(clone_url)
|
|
f.path.segments = ["git-annex-p2phttp", "git-annex", info["uuid"], "key", key]
|
|
|
|
f = furl(clone_url)
|
|
f.path.segments = ["git-annex-p2phttp", "git-annex", info["uuid"], "key", key]
|
|
|
|
# make three requests in succession; lockup raises requests.exceptions.ReadTimeout
|
|
_ = requests.head(f.tostr(), timeout=15)
|
|
_ = requests.head(f.tostr(), timeout=15)
|
|
_ = requests.head(f.tostr(), timeout=15)
|