import asyncio
import base64
import os
import httpx
import gitlab
LOGIN = "root"
PASSWORD = "5iveL!fe"
SSH_KEY = (
"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDZAjAX8vTiHD7Yi3/EzuVaDChtih"
"79HyJZ6H9dEqxFfmGA1YnncE0xujQ64TCebhkYJKzmTJCImSVkOu9C4hZgsw6eE76n"
"+Cg3VwEeDUFy+GXlEJWlHaEyc3HWioxgOALbUp3rOezNh+d8BDwwqvENGoePEBsz5l"
"a6WP5lTi/HJIjAl6Hu+zHgdj1XVExeH+S52EwpZf/ylTJub0Bl5gHwf/siVE48mLMI"
"sqrukXTZ6Zg+8EHAIvIQwJ1dKcXe8P5IoLT7VKrbkgAnolS0I8J+uH7KtErZJb5oZh"
"S4OEwsNpaXMAr+6/wWSpircV2/e7sFLlhlKBC4Iq1MpqlZ7G3p foo@bar"
)
DEPLOY_KEY = (
"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDFdRyjJQh+1niBpXqE2I8dzjG"
"MXFHlRjX9yk/UfOn075IdaockdU58sw2Ai1XIWFpZpfJkW7z+P47ZNSqm1gzeXI"
"rtKa9ZUp8A7SZe8vH4XVn7kh7bwWCUirqtn8El9XdqfkzOs/+FuViriUWoJVpA6"
"WZsDNaqINFKIA5fj/q8XQw+BcS92L09QJg9oVUuH0VVwNYbU2M2IRmSpybgC/gu"
"uWTrnCDMmLItksATifLvRZwgdI8dr+q6tbxbZknNcgEPrI2jT0hYN9ZcjNeWuyv"
"rke9IepE7SPBT41C+YtUX4dfDZDmczM1cE0YL/krdUCfuZHMa4ZS2YyNd6slufc"
"vn bar@foo"
)
GPG_KEY = """-----BEGIN PGP PUBLIC KEY BLOCK-----
mQENBFn5mzYBCADH6SDVPAp1zh/hxmTi0QplkOfExBACpuY6OhzNdIg+8/528b3g
Y5YFR6T/HLv/PmeHskUj21end1C0PNG2T9dTx+2Vlh9ISsSG1kyF9T5fvMR3bE0x
Dl6S489CXZrjPTS9SHk1kF+7dwjUxLJyxF9hPiSihFefDFu3NeOtG/u8vbC1mewQ
ZyAYue+mqtqcCIFFoBz7wHKMWjIVSJSyTkXExu4OzpVvy3l2EikbvavI3qNz84b+
Mgkv/kiBlNoCy3CVuPk99RYKZ3lX1vVtqQ0OgNGQvb4DjcpyjmbKyibuZwhDjIOh
au6d1OyEbayTntd+dQ4j9EMSnEvm/0MJ4eXPABEBAAG0G0dpdGxhYlRlc3QxIDxm
YWtlQGZha2UudGxkPokBNwQTAQgAIQUCWfmbNgIbAwULCQgHAgYVCAkKCwIEFgID
AQIeAQIXgAAKCRBgxELHf8f3hF3yB/wNJlWPKY65UsB4Lo0hs1OxdxCDqXogSi0u
6crDEIiyOte62pNZKzWy8TJcGZvznRTZ7t8hXgKFLz3PRMcl+vAiRC6quIDUj+2V
eYfwaItd1lUfzvdCaC7Venf4TQ74f5vvNg/zoGwE6eRoSbjlLv9nqsxeA0rUBUQL
LYikWhVMP3TrlfgfduYvh6mfgh57BDLJ9kJVpyfxxx9YLKZbaas9sPa6LgBtR555
JziUxHmbEv8XCsUU8uoFeP1pImbNBplqE3wzJwzOMSmmch7iZzrAwfN7N2j3Wj0H
B5kQddJ9dmB4BbU0IXGhWczvdpxboI2wdY8a1JypxOdePoph/43iuQENBFn5mzYB
CADnTPY0Zf3d9zLjBNgIb3yDl94uOcKCq0twNmyjMhHzGqw+UMe9BScy34GL94Al
xFRQoaL+7P8hGsnsNku29A/VDZivcI+uxTx4WQ7OLcn7V0bnHV4d76iky2ufbUt/
GofthjDs1SonePO2N09sS4V4uK0d5N4BfCzzXgvg8etCLxNmC9BGt7AaKUUzKBO4
2QvNNaC2C/8XEnOgNWYvR36ylAXAmo0sGFXUsBCTiq1fugS9pwtaS2JmaVpZZ3YT
pMZlS0+SjC5BZYFqSmKCsA58oBRzCxQz57nR4h5VEflgD+Hy0HdW0UHETwz83E6/
U0LL6YyvhwFr6KPq5GxinSvfABEBAAGJAR8EGAEIAAkFAln5mzYCGwwACgkQYMRC
x3/H94SJgwgAlKQb10/xcL/epdDkR7vbiei7huGLBpRDb/L5fM8B5W77Qi8Xmuqj
cCu1j99ZCA5hs/vwVn8j8iLSBGMC5gxcuaar/wtmiaEvT9fO/h6q4opG7NcuiJ8H
wRj8ccJmRssNqDD913PLz7T40Ts62blhrEAlJozGVG/q7T3RAZcskOUHKeHfc2RI
YzGsC/I9d7k6uxAv1L9Nm5F2HaAQDzhkdd16nKkGaPGR35cT1JLInkfl5cdm7ldN
nxs4TLO3kZjUTgWKdhpgRNF5hwaz51ZjpebaRf/ZqRuNyX4lIRolDxzOn/+O1o8L
qG2ZdhHHmSK2LaQLFiSprUkikStNU9BqSQ==
=5OGa
-----END PGP PUBLIC KEY BLOCK-----"""
AVATAR_PATH = os.path.join(os.path.dirname(__file__), "avatar.png")
async def main():
# token authentication from config file
gl = gitlab.AsyncGitlab.from_config(config_files=["/tmp/python-gitlab.cfg"])
await gl.auth()
assert isinstance(gl.user, gitlab.v4.objects.CurrentUser)
# markdown
html = await gl.markdown("foo")
assert "foo" in html
success, errors = await gl.lint("Invalid")
assert success is False
assert errors
# sidekiq
out = await gl.sidekiq.queue_metrics()
assert isinstance(out, dict)
assert "pages" in out["queues"]
out = await gl.sidekiq.process_metrics()
assert isinstance(out, dict)
assert "hostname" in out["processes"][0]
out = await gl.sidekiq.job_stats()
assert isinstance(out, dict)
assert "processed" in out["jobs"]
out = await gl.sidekiq.compound_metrics()
assert isinstance(out, dict)
assert "jobs" in out
assert "processes" in out
assert "queues" in out
# settings
settings = await gl.settings.get()
settings.default_projects_limit = 42
await settings.save()
settings = await gl.settings.get()
assert settings.default_projects_limit == 42
# users
new_user = await gl.users.create(
{
"email": "foo@bar.com",
"username": "foo",
"name": "foo",
"password": "foo_password",
"avatar": open(AVATAR_PATH, "rb"),
}
)
avatar_url = new_user.avatar_url.replace("gitlab.test", "localhost:8080")
uploaded_avatar = httpx.get(avatar_url).content
assert uploaded_avatar == open(AVATAR_PATH, "rb").read()
users_list = await gl.users.list()
for user in users_list:
if user.username == "foo":
break
assert new_user.username == user.username
assert new_user.email == user.email
await new_user.block()
await new_user.unblock()
# user projects list
assert len(await new_user.projects.list()) == 0
# events list
await new_user.events.list()
foobar_user = await gl.users.create(
{
"email": "foobar@example.com",
"username": "foobar",
"name": "Foo Bar",
"password": "foobar_password",
}
)
assert (await gl.users.list(search="foobar"))[0].id == foobar_user.id
expected = [new_user, foobar_user]
actual = list(await gl.users.list(search="foo"))
assert len(expected) == len(actual)
assert len(await gl.users.list(search="asdf")) == 0
foobar_user.bio = "This is the user bio"
await foobar_user.save()
# GPG keys
gkey = await new_user.gpgkeys.create({"key": GPG_KEY})
assert len(await new_user.gpgkeys.list()) == 1
# Seems broken on the gitlab side
# gkey = new_user.gpgkeys.get(gkey.id)
await gkey.delete()
assert len(await new_user.gpgkeys.list()) == 0
# SSH keys
key = await new_user.keys.create({"title": "testkey", "key": SSH_KEY})
assert len(await new_user.keys.list()) == 1
await key.delete()
assert len(await new_user.keys.list()) == 0
# emails
email = await new_user.emails.create({"email": "foo2@bar.com"})
assert len(await new_user.emails.list()) == 1
await email.delete()
assert len(await new_user.emails.list()) == 0
# custom attributes
attrs = await new_user.customattributes.list()
assert len(attrs) == 0
attr = await new_user.customattributes.set("key", "value1")
assert len(await gl.users.list(custom_attributes={"key": "value1"})) == 1
assert attr.key == "key"
assert attr.value == "value1"
assert len(await new_user.customattributes.list()) == 1
attr = await new_user.customattributes.set("key", "value2")
attr = await new_user.customattributes.get("key")
assert attr.value == "value2"
assert len(await new_user.customattributes.list()) == 1
await attr.delete()
assert len(await new_user.customattributes.list()) == 0
# impersonation tokens
user_token = await new_user.impersonationtokens.create(
{"name": "token1", "scopes": ["api", "read_user"]}
)
l = await new_user.impersonationtokens.list(state="active")
assert len(l) == 1
await user_token.delete()
l = await new_user.impersonationtokens.list(state="active")
assert len(l) == 0
l = await new_user.impersonationtokens.list(state="inactive")
assert len(l) == 1
await new_user.delete()
await foobar_user.delete()
assert len(await gl.users.list()) == 3 + len(
[u for u in await gl.users.list() if u.username == "ghost"]
)
# current user mail
mail = await gl.user.emails.create({"email": "current@user.com"})
assert len(await gl.user.emails.list()) == 1
await mail.delete()
assert len(await gl.user.emails.list()) == 0
# current user GPG keys
gkey = await gl.user.gpgkeys.create({"key": GPG_KEY})
assert len(await gl.user.gpgkeys.list()) == 1
# Seems broken on the gitlab side
gkey = await gl.user.gpgkeys.get(gkey.id)
await gkey.delete()
assert len(await gl.user.gpgkeys.list()) == 0
# current user key
key = await gl.user.keys.create({"title": "testkey", "key": SSH_KEY})
assert len(await gl.user.keys.list()) == 1
await key.delete()
assert len(await gl.user.keys.list()) == 0
# templates
assert await gl.dockerfiles.list()
dockerfile = await gl.dockerfiles.get("Node")
assert dockerfile.content is not None
assert await gl.gitignores.list()
gitignore = await gl.gitignores.get("Node")
assert gitignore.content is not None
assert await gl.gitlabciymls.list()
gitlabciyml = await gl.gitlabciymls.get("Nodejs")
assert gitlabciyml.content is not None
assert await gl.licenses.list()
license = await gl.licenses.get(
"bsd-2-clause", project="mytestproject", fullname="mytestfullname"
)
assert "mytestfullname" in license.content
# groups
user1 = await gl.users.create(
{
"email": "user1@test.com",
"username": "user1",
"name": "user1",
"password": "user1_pass",
}
)
user2 = await gl.users.create(
{
"email": "user2@test.com",
"username": "user2",
"name": "user2",
"password": "user2_pass",
}
)
group1 = await gl.groups.create({"name": "group1", "path": "group1"})
group2 = await gl.groups.create({"name": "group2", "path": "group2"})
p_id = (await gl.groups.list(search="group2"))[0].id
group3 = await gl.groups.create(
{"name": "group3", "path": "group3", "parent_id": p_id}
)
assert len(await gl.groups.list()) == 3
assert len(await gl.groups.list(search="oup1")) == 1
assert group3.parent_id == p_id
assert (await group2.subgroups.list())[0].id == group3.id
await group1.members.create(
{"access_level": gitlab.const.OWNER_ACCESS, "user_id": user1.id}
)
await group1.members.create(
{"access_level": gitlab.const.GUEST_ACCESS, "user_id": user2.id}
)
await group2.members.create(
{"access_level": gitlab.const.OWNER_ACCESS, "user_id": user2.id}
)
# Administrator belongs to the groups
assert len(await group1.members.list()) == 3
assert len(await group2.members.list()) == 2
await group1.members.delete(user1.id)
assert len(await group1.members.list()) == 2
assert len(await group1.members.all())
member = await group1.members.get(user2.id)
member.access_level = gitlab.const.OWNER_ACCESS
await member.save()
member = await group1.members.get(user2.id)
assert member.access_level == gitlab.const.OWNER_ACCESS
await group2.members.delete(gl.user.id)
# group custom attributes
attrs = await group2.customattributes.list()
assert len(attrs) == 0
attr = await group2.customattributes.set("key", "value1")
assert len(await gl.groups.list(custom_attributes={"key": "value1"})) == 1
assert attr.key == "key"
assert attr.value == "value1"
assert len(await group2.customattributes.list()) == 1
attr = await group2.customattributes.set("key", "value2")
attr = await group2.customattributes.get("key")
assert attr.value == "value2"
assert len(await group2.customattributes.list()) == 1
await attr.delete()
assert len(await group2.customattributes.list()) == 0
# group notification settings
settings = await group2.notificationsettings.get()
settings.level = "disabled"
await settings.save()
settings = await group2.notificationsettings.get()
assert settings.level == "disabled"
# group badges
badge_image = "http://example.com"
badge_link = "http://example/img.svg"
badge = await group2.badges.create(
{"link_url": badge_link, "image_url": badge_image}
)
assert len(await group2.badges.list()) == 1
badge.image_url = "http://another.example.com"
await badge.save()
badge = await group2.badges.get(badge.id)
assert badge.image_url == "http://another.example.com"
await badge.delete()
assert len(await group2.badges.list()) == 0
# group milestones
gm1 = await group1.milestones.create({"title": "groupmilestone1"})
assert len(await group1.milestones.list()) == 1
gm1.due_date = "2020-01-01T00:00:00Z"
await gm1.save()
gm1.state_event = "close"
await gm1.save()
gm1 = await group1.milestones.get(gm1.id)
assert gm1.state == "closed"
assert len(await gm1.issues()) == 0
assert len(await gm1.merge_requests()) == 0
# group variables
await group1.variables.create({"key": "foo", "value": "bar"})
g_v = await group1.variables.get("foo")
assert g_v.value == "bar"
g_v.value = "baz"
await g_v.save()
g_v = await group1.variables.get("foo")
assert g_v.value == "baz"
assert len(await group1.variables.list()) == 1
await g_v.delete()
assert len(await group1.variables.list()) == 0
# group labels
# group1.labels.create({"name": "foo", "description": "bar", "color": "#112233"})
# g_l = group1.labels.get("foo")
# assert g_l.description == "bar"
# g_l.description = "baz"
# g_l.save()
# g_l = group1.labels.get("foo")
# assert g_l.description == "baz"
# assert len(group1.labels.list()) == 1
# g_l.delete()
# assert len(group1.labels.list()) == 0
# hooks
hook = await gl.hooks.create({"url": "http://whatever.com"})
assert len(await gl.hooks.list()) == 1
await hook.delete()
assert len(await gl.hooks.list()) == 0
# projects
admin_project = await gl.projects.create({"name": "admin_project"})
gr1_project = await gl.projects.create(
{"name": "gr1_project", "namespace_id": group1.id}
)
gr2_project = await gl.projects.create(
{"name": "gr2_project", "namespace_id": group2.id}
)
sudo_project = await gl.projects.create({"name": "sudo_project"}, sudo=user1.name)
assert len(await gl.projects.list(owned=True)) == 2
assert len(await gl.projects.list(search="admin")) == 1
# test pagination
l1 = await gl.projects.list(per_page=1, page=1)
l2 = await gl.projects.list(per_page=1, page=2)
assert len(l1) == 1
assert len(l2) == 1
assert l1[0].id != l2[0].id
# group custom attributes
attrs = await admin_project.customattributes.list()
assert len(attrs) == 0
attr = await admin_project.customattributes.set("key", "value1")
assert len(await gl.projects.list(custom_attributes={"key": "value1"})) == 1
assert attr.key == "key"
assert attr.value == "value1"
assert len(await admin_project.customattributes.list()) == 1
attr = await admin_project.customattributes.set("key", "value2")
attr = await admin_project.customattributes.get("key")
assert attr.value == "value2"
assert len(await admin_project.customattributes.list()) == 1
await attr.delete()
assert len(await admin_project.customattributes.list()) == 0
# project pages domains
domain = await admin_project.pagesdomains.create({"domain": "foo.domain.com"})
assert len(await admin_project.pagesdomains.list()) == 1
assert len(await gl.pagesdomains.list()) == 1
domain = await admin_project.pagesdomains.get("foo.domain.com")
assert domain.domain == "foo.domain.com"
await domain.delete()
assert len(await admin_project.pagesdomains.list()) == 0
# project content (files)
await admin_project.files.create(
{
"file_path": "README",
"branch": "master",
"content": "Initial content",
"commit_message": "Initial commit",
}
)
readme = await admin_project.files.get(file_path="README", ref="master")
readme.content = base64.b64encode(b"Improved README").decode()
await asyncio.sleep(2)
await readme.save(branch="master", commit_message="new commit")
await readme.delete(commit_message="Removing README", branch="master")
await admin_project.files.create(
{
"file_path": "README.rst",
"branch": "master",
"content": "Initial content",
"commit_message": "New commit",
}
)
readme = await admin_project.files.get(file_path="README.rst", ref="master")
# The first decode() is the ProjectFile method, the second one is the bytes
# object method
assert readme.decode().decode() == "Initial content"
blame = await admin_project.files.blame(file_path="README.rst", ref="master")
data = {
"branch": "master",
"commit_message": "blah blah blah",
"actions": [{"action": "create", "file_path": "blah", "content": "blah"}],
}
await admin_project.commits.create(data)
assert "@@" in (await (await admin_project.commits.list())[0].diff())[0]["diff"]
# commit status
commit = (await admin_project.commits.list())[0]
# size = len(commit.statuses.list())
# status = commit.statuses.create({"state": "success", "sha": commit.id})
# assert len(commit.statuses.list()) == size + 1
# assert commit.refs()
# assert commit.merge_requests()
# commit comment
await commit.comments.create({"note": "This is a commit comment"})
# assert len(commit.comments.list()) == 1
# commit discussion
count = len(await commit.discussions.list())
discussion = await commit.discussions.create({"body": "Discussion body"})
# assert len(commit.discussions.list()) == (count + 1)
d_note = await discussion.notes.create({"body": "first note"})
d_note_from_get = await discussion.notes.get(d_note.id)
d_note_from_get.body = "updated body"
await d_note_from_get.save()
discussion = await commit.discussions.get(discussion.id)
# assert discussion.attributes["notes"][-1]["body"] == "updated body"
await d_note_from_get.delete()
discussion = await commit.discussions.get(discussion.id)
# assert len(discussion.attributes["notes"]) == 1
# housekeeping
await admin_project.housekeeping()
# repository
tree = await admin_project.repository_tree()
assert len(tree) != 0
assert tree[0]["name"] == "README.rst"
blob_id = tree[0]["id"]
blob = await admin_project.repository_raw_blob(blob_id)
assert blob.decode() == "Initial content"
archive1 = await admin_project.repository_archive()
archive2 = await admin_project.repository_archive("master")
assert archive1 == archive2
snapshot = await admin_project.snapshot()
# project file uploads
filename = "test.txt"
file_contents = "testing contents"
uploaded_file = await admin_project.upload(filename, file_contents)
assert uploaded_file["alt"] == filename
assert uploaded_file["url"].startswith("/uploads/")
assert uploaded_file["url"].endswith("/" + filename)
assert uploaded_file["markdown"] == "[{}]({})".format(
uploaded_file["alt"], uploaded_file["url"]
)
# environments
await admin_project.environments.create(
{"name": "env1", "external_url": "http://fake.env/whatever"}
)
envs = await admin_project.environments.list()
assert len(envs) == 1
env = envs[0]
env.external_url = "http://new.env/whatever"
await env.save()
env = (await admin_project.environments.list())[0]
assert env.external_url == "http://new.env/whatever"
await env.stop()
await env.delete()
assert len(await admin_project.environments.list()) == 0
# Project clusters
await admin_project.clusters.create(
{
"name": "cluster1",
"platform_kubernetes_attributes": {
"api_url": "http://url",
"token": "tokenval",
},
}
)
clusters = await admin_project.clusters.list()
assert len(clusters) == 1
cluster = clusters[0]
cluster.platform_kubernetes_attributes = {"api_url": "http://newurl"}
await cluster.save()
cluster = (await admin_project.clusters.list())[0]
assert cluster.platform_kubernetes["api_url"] == "http://newurl"
await cluster.delete()
assert len(await admin_project.clusters.list()) == 0
# Group clusters
await group1.clusters.create(
{
"name": "cluster1",
"platform_kubernetes_attributes": {
"api_url": "http://url",
"token": "tokenval",
},
}
)
clusters = await group1.clusters.list()
assert len(clusters) == 1
cluster = clusters[0]
cluster.platform_kubernetes_attributes = {"api_url": "http://newurl"}
await cluster.save()
cluster = (await group1.clusters.list())[0]
assert cluster.platform_kubernetes["api_url"] == "http://newurl"
await cluster.delete()
assert len(await group1.clusters.list()) == 0
# project events
await admin_project.events.list()
# forks
fork = await admin_project.forks.create({"namespace": user1.username})
p = await gl.projects.get(fork.id)
assert p.forked_from_project["id"] == admin_project.id
forks = await admin_project.forks.list()
assert fork.id in map(lambda p: p.id, forks)
# project hooks
hook = await admin_project.hooks.create({"url": "http://hook.url"})
assert len(await admin_project.hooks.list()) == 1
hook.note_events = True
await hook.save()
hook = await admin_project.hooks.get(hook.id)
assert hook.note_events is True
await hook.delete()
# deploy keys
deploy_key = await admin_project.keys.create(
{"title": "foo@bar", "key": DEPLOY_KEY}
)
project_keys = list(await admin_project.keys.list())
assert len(project_keys) == 1
await sudo_project.keys.enable(deploy_key.id)
assert len(await sudo_project.keys.list()) == 1
await sudo_project.keys.delete(deploy_key.id)
assert len(await sudo_project.keys.list()) == 0
# labels
# label1 = admin_project.labels.create({"name": "label1", "color": "#778899"})
# label1 = admin_project.labels.list()[0]
# assert len(admin_project.labels.list()) == 1
# label1.new_name = "label1updated"
# label1.save()
# assert label1.name == "label1updated"
# label1.subscribe()
# assert label1.subscribed == True
# label1.unsubscribe()
# assert label1.subscribed == False
# label1.delete()
# milestones
m1 = await admin_project.milestones.create({"title": "milestone1"})
assert len(await admin_project.milestones.list()) == 1
m1.due_date = "2020-01-01T00:00:00Z"
await m1.save()
m1.state_event = "close"
await m1.save()
m1 = await admin_project.milestones.get(m1.id)
assert m1.state == "closed"
assert len(await m1.issues()) == 0
assert len(await m1.merge_requests()) == 0
# issues
issue1 = await admin_project.issues.create(
{"title": "my issue 1", "milestone_id": m1.id}
)
issue2 = await admin_project.issues.create({"title": "my issue 2"})
issue3 = await admin_project.issues.create({"title": "my issue 3"})
assert len(await admin_project.issues.list()) == 3
issue3.state_event = "close"
await issue3.save()
assert len(await admin_project.issues.list(state="closed")) == 1
assert len(await admin_project.issues.list(state="opened")) == 2
assert len(await admin_project.issues.list(milestone="milestone1")) == 1
assert (await (await m1.issues()).anext()).title == "my issue 1"
size = len(await issue1.notes.list())
note = await issue1.notes.create({"body": "This is an issue note"})
assert len(await issue1.notes.list()) == size + 1
emoji = await note.awardemojis.create({"name": "tractor"})
assert len(await note.awardemojis.list()) == 1
await emoji.delete()
assert len(await note.awardemojis.list()) == 0
await note.delete()
assert len(await issue1.notes.list()) == size
assert isinstance(await issue1.user_agent_detail(), dict)
assert (await issue1.user_agent_detail())["user_agent"]
assert await issue1.participants()
assert type(await issue1.closed_by()) == list
assert type(await issue1.related_merge_requests()) == list
# issues labels and events
label2 = await admin_project.labels.create({"name": "label2", "color": "#aabbcc"})
issue1.labels = ["label2"]
await issue1.save()
events = await issue1.resourcelabelevents.list()
assert events
event = await issue1.resourcelabelevents.get(events[0].id)
assert event
size = len(await issue1.discussions.list())
discussion = await issue1.discussions.create({"body": "Discussion body"})
assert len(await issue1.discussions.list()) == size + 1
d_note = await discussion.notes.create({"body": "first note"})
d_note_from_get = await discussion.notes.get(d_note.id)
d_note_from_get.body = "updated body"
await d_note_from_get.save()
discussion = await issue1.discussions.get(discussion.id)
assert discussion.attributes["notes"][-1]["body"] == "updated body"
await d_note_from_get.delete()
discussion = await issue1.discussions.get(discussion.id)
assert len(discussion.attributes["notes"]) == 1
# tags
tag1 = await admin_project.tags.create({"tag_name": "v1.0", "ref": "master"})
assert len(await admin_project.tags.list()) == 1
await tag1.set_release_description("Description 1")
await tag1.set_release_description("Description 2")
assert tag1.release["description"] == "Description 2"
await tag1.delete()
# project snippet
admin_project.snippets_enabled = True
await admin_project.save()
snippet = await admin_project.snippets.create(
{
"title": "snip1",
"file_name": "foo.py",
"content": "initial content",
"visibility": gitlab.v4.objects.VISIBILITY_PRIVATE,
}
)
assert (await snippet.user_agent_detail())["user_agent"]
size = len(await snippet.discussions.list())
discussion = await snippet.discussions.create({"body": "Discussion body"})
assert len(await snippet.discussions.list()) == size + 1
d_note = await discussion.notes.create({"body": "first note"})
d_note_from_get = await discussion.notes.get(d_note.id)
d_note_from_get.body = "updated body"
await d_note_from_get.save()
discussion = await snippet.discussions.get(discussion.id)
assert discussion.attributes["notes"][-1]["body"] == "updated body"
await d_note_from_get.delete()
discussion = await snippet.discussions.get(discussion.id)
assert len(discussion.attributes["notes"]) == 1
snippet.file_name = "bar.py"
await snippet.save()
snippet = await admin_project.snippets.get(snippet.id)
assert (await snippet.content()).decode() == "initial content"
assert snippet.file_name == "bar.py"
size = len(await admin_project.snippets.list())
await snippet.delete()
assert len(await admin_project.snippets.list()) == (size - 1)
# triggers
tr1 = await admin_project.triggers.create({"description": "trigger1"})
assert len(await admin_project.triggers.list()) == 1
await tr1.delete()
# variables
v1 = await admin_project.variables.create({"key": "key1", "value": "value1"})
assert len(await admin_project.variables.list()) == 1
v1.value = "new_value1"
await v1.save()
v1 = await admin_project.variables.get(v1.key)
assert v1.value == "new_value1"
await v1.delete()
# branches and merges
to_merge = await admin_project.branches.create(
{"branch": "branch1", "ref": "master"}
)
await admin_project.files.create(
{
"file_path": "README2.rst",
"branch": "branch1",
"content": "Initial content",
"commit_message": "New commit in new branch",
}
)
mr = await admin_project.mergerequests.create(
{"source_branch": "branch1", "target_branch": "master", "title": "MR readme2"}
)
# discussion
size = len(await mr.discussions.list())
discussion = await mr.discussions.create({"body": "Discussion body"})
assert len(await mr.discussions.list()) == size + 1
d_note = await discussion.notes.create({"body": "first note"})
d_note_from_get = await discussion.notes.get(d_note.id)
d_note_from_get.body = "updated body"
await d_note_from_get.save()
discussion = await mr.discussions.get(discussion.id)
assert discussion.attributes["notes"][-1]["body"] == "updated body"
await d_note_from_get.delete()
discussion = await mr.discussions.get(discussion.id)
assert len(discussion.attributes["notes"]) == 1
# mr labels and events
mr.labels = ["label2"]
await mr.save()
events = await mr.resourcelabelevents.list()
assert events
event = await mr.resourcelabelevents.get(events[0].id)
assert event
# rebasing
assert await mr.rebase()
# basic testing: only make sure that the methods exist
await mr.commits()
await mr.changes()
assert await mr.participants()
await mr.merge()
await admin_project.branches.delete("branch1")
try:
await mr.merge()
except gitlab.GitlabMRClosedError:
pass
# protected branches
p_b = await admin_project.protectedbranches.create({"name": "*-stable"})
assert p_b.name == "*-stable"
p_b = await admin_project.protectedbranches.get("*-stable")
# master is protected by default when a branch has been created
assert len(await admin_project.protectedbranches.list()) == 2
await admin_project.protectedbranches.delete("master")
await p_b.delete()
assert len(await admin_project.protectedbranches.list()) == 0
# stars
await admin_project.star()
assert admin_project.star_count == 1
await admin_project.unstar()
assert admin_project.star_count == 0
# project boards
# boards = admin_project.boards.list()
# assert(len(boards))
# board = boards[0]
# lists = board.lists.list()
# begin_size = len(lists)
# last_list = lists[-1]
# last_list.position = 0
# last_list.save()
# last_list.delete()
# lists = board.lists.list()
# assert(len(lists) == begin_size - 1)
# project badges
badge_image = "http://example.com"
badge_link = "http://example/img.svg"
badge = await admin_project.badges.create(
{"link_url": badge_link, "image_url": badge_image}
)
assert len(await admin_project.badges.list()) == 1
badge.image_url = "http://another.example.com"
await badge.save()
badge = await admin_project.badges.get(badge.id)
assert badge.image_url == "http://another.example.com"
await badge.delete()
assert len(await admin_project.badges.list()) == 0
# project wiki
wiki_content = "Wiki page content"
wp = await admin_project.wikis.create(
{"title": "wikipage", "content": wiki_content}
)
assert len(await admin_project.wikis.list()) == 1
wp = await admin_project.wikis.get(wp.slug)
assert wp.content == wiki_content
# update and delete seem broken
# wp.content = 'new content'
# wp.save()
# wp.delete()
# assert(len(admin_project.wikis.list()) == 0)
# namespaces
ns = await gl.namespaces.list(all=True)
assert len(ns) != 0
ns = (await gl.namespaces.list(search="root", all=True))[0]
assert ns.kind == "user"
# features
# Disabled as this fails with GitLab 11.11
# feat = gl.features.set("foo", 30)
# assert feat.name == "foo"
# assert len(gl.features.list()) == 1
# feat.delete()
# assert len(gl.features.list()) == 0
# broadcast messages
msg = await gl.broadcastmessages.create({"message": "this is the message"})
msg.color = "#444444"
await msg.save()
msg_id = msg.id
msg = (await gl.broadcastmessages.list(all=True))[0]
assert msg.color == "#444444"
msg = await gl.broadcastmessages.get(msg_id)
assert msg.color == "#444444"
await msg.delete()
assert len(await gl.broadcastmessages.list()) == 0
# notification settings
settings = await gl.notificationsettings.get()
settings.level = gitlab.NOTIFICATION_LEVEL_WATCH
await settings.save()
settings = await gl.notificationsettings.get()
assert settings.level == gitlab.NOTIFICATION_LEVEL_WATCH
# services
service = await admin_project.services.get("asana")
service.api_key = "whatever"
await service.save()
service = await admin_project.services.get("asana")
assert service.active == True
await service.delete()
service = await admin_project.services.get("asana")
assert service.active == False
# snippets
snippets = await gl.snippets.list(all=True)
assert len(snippets) == 0
snippet = await gl.snippets.create(
{"title": "snippet1", "file_name": "snippet1.py", "content": "import gitlab"}
)
snippet = await gl.snippets.get(snippet.id)
snippet.title = "updated_title"
await snippet.save()
snippet = await gl.snippets.get(snippet.id)
assert snippet.title == "updated_title"
content = await snippet.content()
assert content.decode() == "import gitlab"
assert (await snippet.user_agent_detail())["user_agent"]
await snippet.delete()
snippets = await gl.snippets.list(all=True)
assert len(snippets) == 0
# user activities
await gl.user_activities.list(query_parameters={"from": "2019-01-01"})
# events
await gl.events.list()
# rate limit
settings = await gl.settings.get()
settings.throttle_authenticated_api_enabled = True
settings.throttle_authenticated_api_requests_per_period = 1
settings.throttle_authenticated_api_period_in_seconds = 3
await settings.save()
projects = list()
for i in range(0, 20):
projects.append(await gl.projects.create({"name": str(i) + "ok"}))
error_message = None
for i in range(20, 40):
try:
projects.append(
await gl.projects.create(
{"name": str(i) + "shouldfail"}, obey_rate_limit=False
)
)
except gitlab.GitlabCreateError as e:
error_message = e.error_message
break
assert "Retry later" in error_message
settings.throttle_authenticated_api_enabled = False
await settings.save()
[await current_project.delete() for current_project in projects]
# project import/export
ex = await admin_project.exports.create({})
await ex.refresh()
count = 0
while ex.export_status != "finished":
await asyncio.sleep(1)
await ex.refresh()
count += 1
if count == 10:
raise Exception("Project export taking too much time")
with open("/tmp/gitlab-export.tgz", "wb") as f:
await ex.download(streamed=True, action=f.write)
output = await gl.projects.import_project(
open("/tmp/gitlab-export.tgz", "rb"), "imported_project"
)
project_import = await gl.projects.get(output["id"], lazy=True).imports.get()
count = 0
while project_import.import_status != "finished":
await asyncio.sleep(1)
await project_import.refresh()
count += 1
if count == 10:
raise Exception("Project import taking too much time")
# project releases
release_test_project = await gl.projects.create(
{"name": "release-test-project", "initialize_with_readme": True}
)
release_name = "Demo Release"
release_tag_name = "v1.2.3"
release_description = "release notes go here"
await release_test_project.releases.create(
{
"name": release_name,
"tag_name": release_tag_name,
"description": release_description,
"ref": "master",
}
)
assert len(await release_test_project.releases.list()) == 1
# get single release
retrieved_project = await release_test_project.releases.get(release_tag_name)
assert retrieved_project.name == release_name
assert retrieved_project.tag_name == release_tag_name
assert retrieved_project.description == release_description
# delete release
await release_test_project.releases.delete(release_tag_name)
assert len(await release_test_project.releases.list()) == 0
await release_test_project.delete()
# status
message = "Test"
emoji = "thumbsup"
status = await gl.user.status.get()
status.message = message
status.emoji = emoji
await status.save()
new_status = await gl.user.status.get()
assert new_status.message == message
assert new_status.emoji == emoji
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())