Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[cherry-pick]Add GC accessory API test case #19463

Merged
merged 1 commit into from
Oct 20, 2023
Merged

[cherry-pick]Add GC accessory API test case #19463

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 167 additions & 15 deletions tests/apitests/python/test_garbage_collection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,16 +3,18 @@
import unittest
import time

from testutils import ADMIN_CLIENT, suppress_urllib3_warning
from testutils import ADMIN_CLIENT, suppress_urllib3_warning, files_directory
from testutils import TEARDOWN
from testutils import harbor_server
from library.user import User
from library.project import Project
from library.repository import Repository
from library.base import _assert_status_code
from library.repository import push_special_image_to_project
from library.repository import push_special_image_to_project, push_self_build_image_to_project
from library.artifact import Artifact
from library.gc import GC
from library import docker_api, cosign


class TestProjects(unittest.TestCase):
@suppress_urllib3_warning
Expand All @@ -25,14 +27,8 @@ def setUp(self):
self.repo_name = "test_repo"
self.repo_name_untag = "test_untag"
self.tag = "v1.0"
self.gc_success_status = "Success"

@unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
def tearDown(self):
#2. Delete project(PA);
self.project.delete_project(TestProjects.project_gc_id, **TestProjects.USER_GC_CLIENT)

#3. Delete user(UA);
self.user.delete_user(TestProjects.user_gc_id, **ADMIN_CLIENT)

def testGarbageCollection(self):
"""
Expand All @@ -51,9 +47,6 @@ def testGarbageCollection(self):
10. Check garbage collection job was finished;
11. Repository with untag image should be still there;
12. But no any artifact in repository anymore.
Tear down:
1. Delete project(PA);
2. Delete user(UA).
"""
url = ADMIN_CLIENT["endpoint"]
admin_name = ADMIN_CLIENT["username"]
Expand Down Expand Up @@ -85,7 +78,7 @@ def testGarbageCollection(self):
gc_id = self.gc.gc_now(**ADMIN_CLIENT)

#6. Check garbage collection job was finished;
self.gc.validate_gc_job_status(gc_id, "Success", **ADMIN_CLIENT)
self.gc.validate_gc_job_status(gc_id, self.gc_success_status, **ADMIN_CLIENT)

#7. Get garbage collection log, check there is a number of files was deleted;
self.gc.validate_deletion_success(gc_id, **ADMIN_CLIENT)
Expand All @@ -99,7 +92,7 @@ def testGarbageCollection(self):
gc_id = self.gc.gc_now(is_delete_untagged=True, **ADMIN_CLIENT)

#10. Check garbage collection job was finished;
self.gc.validate_gc_job_status(gc_id, "Success", **ADMIN_CLIENT)
self.gc.validate_gc_job_status(gc_id, self.gc_success_status, **ADMIN_CLIENT)

#7. Get garbage collection log, check there is a number of files was deleted;
self.gc.validate_deletion_success(gc_id, **ADMIN_CLIENT)
Expand All @@ -114,6 +107,165 @@ def testGarbageCollection(self):
self.assertEqual(artifacts,[])


def testGarbageCollectionAccessory(self):
"""
Test case:
Garbage Collection Accessory
Test step and expected result:
1. Create a new user(UA);
2. Create a new project(PA) by user(UA);
3. Push a new image(IA) in project(PA) by user(UA);
4. Push image(IA) SBOM to project(PA) by user(UA);
5. Sign image(IA) with cosign;
6. Sign image(IA) SBOM with cosign;
7. Sign image(IA) Signature with cosign;
8. Delete image(IA) Signature of signature by user(UA);
9. Trigger GC and wait for GC to succeed;
10. Get the GC log and check that the image(IA) Signature of signature is deleted;
11. Delete image(IA) Signature by user(UA);
12. Trigger GC and wait for GC to succeed;
13. Get the GC log and check that the image(IA) Signature is deleted;
14. Delete image(IA) SBOM by user(UA);
15. Trigger GC and wait for GC to succeed;
16. Get the GC log and check that the image(IA) SBOM and Signature of SBOM is deleted;
17. Push image(IA) SBOM to project(PA) by user(UA);
18. Sign image(IA) with cosign;
19. Sign image(IA) SBOM with cosign;
20. Sign image(IA) Signature with cosign;
21. Trigger GC and wait for GC to succeed;
22. Get the GC log and check that it is not deleted;
23. Delete tag of image(IA) by user(UA);
24. Trigger GC and wait for GC to succeed;
25. Get the GC log and check that the image(IA) and all aeecssory is deleted;
"""
url = ADMIN_CLIENT["endpoint"]
user_password = "Aa123456"
# 1. Create user(UA)
_, user_name = self.user.create_user(user_password = user_password, **ADMIN_CLIENT)
user_client = dict(endpoint = url, username = user_name, password = user_password, with_accessory = True)

# 2. Create private project(PA) by user(UA)
self.image = "test_image"
_, project_name = self.project.create_project(metadata = {"public": "false"}, **user_client)

# 3. Push a new image(IA) in project(PA) by user(UA)
push_self_build_image_to_project(project_name, harbor_server, user_name, user_password, self.image, self.tag)

# 4. Push image(IA) SBOM to project(PA) by user(UA)
self.sbom_path = files_directory + "sbom_test.json"
docker_api.docker_login_cmd(harbor_server, user_name, user_password, enable_manifest = False)
cosign.push_artifact_sbom("{}/{}/{}:{}".format(harbor_server, project_name, self.image, self.tag), self.sbom_path)
artifact_info = self.artifact.get_reference_info(project_name, self.image, self.tag, **user_client)
sbom_digest = artifact_info.accessories[0].digest

# 5. Sign image(IA) with cosign
cosign.generate_key_pair()
cosign.sign_artifact("{}/{}/{}:{}".format(harbor_server, project_name, self.image, self.tag))
artifact_info = self.artifact.get_reference_info(project_name, self.image, self.tag, **user_client)
self.assertEqual(len(artifact_info.accessories), 2)
image_signature_digest = None
for accessory in artifact_info.accessories:
if accessory.digest != sbom_digest:
image_signature_digest = accessory.digest
break

# 6. Sign image(IA) SBOM cosign
cosign.sign_artifact("{}/{}/{}@{}".format(harbor_server, project_name, self.image, sbom_digest))
sbom_info = self.artifact.get_reference_info(project_name, self.image, sbom_digest, **user_client)
sbom_signature_digest = sbom_info.accessories[0].digest

# 7. Sign image(IA) Signature with cosign
cosign.sign_artifact("{}/{}/{}@{}".format(harbor_server, project_name, self.image, image_signature_digest))
signature_info = self.artifact.get_reference_info(project_name, self.image, image_signature_digest, **user_client)
signature_signature_digest = signature_info.accessories[0].digest

# 8. Delete image(IA) Signature of signature by user(UA)
self.artifact.delete_artifact(project_name, self.image, signature_signature_digest, **user_client)

# 9. Trigger GC and wait for GC to succeed
gc_id = self.gc.gc_now(**ADMIN_CLIENT)
self.gc.validate_gc_job_status(gc_id, self.gc_success_status, **ADMIN_CLIENT)

# 10. Get the GC log and check that the image(IA) Signature of signature is deleted
gc_log = self.gc.get_gc_log_by_id(gc_id, **ADMIN_CLIENT)
self.assertIn(signature_signature_digest, gc_log)

# 11. Delete image(IA) Signature by user(UA)
self.artifact.delete_artifact(project_name, self.image, image_signature_digest, **user_client)

# 12. Trigger GC and wait for GC to succeed
gc_id = self.gc.gc_now(**ADMIN_CLIENT)
self.gc.validate_gc_job_status(gc_id, self.gc_success_status, **ADMIN_CLIENT)

# 13. Get the GC log and check that the image(IA) Signature is deleted
gc_log = self.gc.get_gc_log_by_id(gc_id, **ADMIN_CLIENT)
self.assertIn(image_signature_digest, gc_log)

# 14. Delete image(IA) SBOM by user(UA)
self.artifact.delete_artifact(project_name, self.image, sbom_digest, **user_client)

# 15. Trigger GC and wait for GC to succeed
gc_id = self.gc.gc_now(**ADMIN_CLIENT)
self.gc.validate_gc_job_status(gc_id, self.gc_success_status, **ADMIN_CLIENT)

# 16. Get the GC log and check that the image(IA) SBOM and Signature of SBOM is deleted
gc_log = self.gc.get_gc_log_by_id(gc_id, **ADMIN_CLIENT)
self.assertIn(sbom_digest, gc_log)
self.assertIn(sbom_signature_digest, gc_log)

# 17. Push image(IA) SBOM to project(PA) by user(UA)
self.sbom_path = files_directory + "sbom_test.json"
docker_api.docker_login_cmd(harbor_server, user_name, user_password, enable_manifest = False)
cosign.push_artifact_sbom("{}/{}/{}:{}".format(harbor_server, project_name, self.image, self.tag), self.sbom_path)
artifact_info = self.artifact.get_reference_info(project_name, self.image, self.tag, **user_client)
sbom_digest = artifact_info.accessories[0].digest

# 18. Sign image(IA) with cosign
cosign.sign_artifact("{}/{}/{}:{}".format(harbor_server, project_name, self.image, self.tag))
artifact_info = self.artifact.get_reference_info(project_name, self.image, self.tag, **user_client)
self.assertEqual(len(artifact_info.accessories), 2)
image_signature_digest = None
for accessory in artifact_info.accessories:
if accessory.digest != sbom_digest:
image_signature_digest = accessory.digest
break

# 19. Sign image(IA) SBOM cosign
cosign.sign_artifact("{}/{}/{}@{}".format(harbor_server, project_name, self.image, sbom_digest))
sbom_info = self.artifact.get_reference_info(project_name, self.image, sbom_digest, **user_client)
sbom_signature_digest = sbom_info.accessories[0].digest

# 20. Sign image(IA) Signature with cosign
cosign.sign_artifact("{}/{}/{}@{}".format(harbor_server, project_name, self.image, image_signature_digest))
signature_info = self.artifact.get_reference_info(project_name, self.image, image_signature_digest, **user_client)
signature_signature_digest = signature_info.accessories[0].digest

# 21. Trigger GC and wait for GC to succeed
gc_id = self.gc.gc_now(**ADMIN_CLIENT)
self.gc.validate_gc_job_status(gc_id, self.gc_success_status, **ADMIN_CLIENT)

# 22. Get the GC log and check that it is not deleted
gc_log = self.gc.get_gc_log_by_id(gc_id, **ADMIN_CLIENT)
self.assertNotIn(sbom_digest, gc_log)
self.assertNotIn(sbom_signature_digest, gc_log)
self.assertNotIn(image_signature_digest, gc_log)
self.assertNotIn(signature_signature_digest, gc_log)

# 23. Delete tag of image(IA) by user(UA)
self.artifact.delete_tag(project_name, self.image, self.tag, self.tag, **user_client)

# 24. Trigger GC and wait for GC to succeed
gc_id = self.gc.gc_now(is_delete_untagged=True, **ADMIN_CLIENT)
self.gc.validate_gc_job_status(gc_id, self.gc_success_status, **ADMIN_CLIENT)

# 25. Get the GC log and check that the image(IA) and all aeecssory is deleted
gc_log = self.gc.get_gc_log_by_id(gc_id, **ADMIN_CLIENT)
self.assertIn(self.image, gc_log)
self.assertIn(sbom_digest, gc_log)
self.assertIn(sbom_signature_digest, gc_log)
self.assertIn(image_signature_digest, gc_log)
self.assertIn(signature_signature_digest, gc_log)


if __name__ == '__main__':
unittest.main()
unittest.main()
Loading