Adding list tags and get catalog API endpoints

endpoints/permissions.py

@@ -166,6 +166,7 @@ def get_users_of_organization_repos(quay_url, org, repos, users):
         target_name = "'GET %s'" % path
         Attacker().run_vegeta('get_users_of_organizations_repos', reqs, target_name=target_name)
 
+    @staticmethod
     def list_users_of_organization_repos(quay_url, org, repos):
         """
         Lists all users permission info in all specified repos.

endpoints/tags.py

@@ -1,7 +1,9 @@
 import json
+import requests
 from subprocess import run, Popen, PIPE, STDOUT
 from utils.attacker import Attacker
 from utils.util import print_header
+from urllib3.exceptions import InsecureRequestWarning
 
 
 class Tags:
@@ -12,6 +14,85 @@ class Tags:
     def __init__(self):
         pass
 
+    @staticmethod
+    def fetch_repo_tokens(quay_host, user, repo):
+        """
+        Fetches v2 token for a given repo of the user.
+
+        :param quay_host: quay host name
+        :param user: username
+        :param repo: repository of a specified user
+        :return: returns the token
+        """
+        url = f"https://{quay_host}/v2/auth?service={quay_host}&scope=repository:{user}/{repo}:pull,push"
+        auth = (user, 'password')
+        try:
+            requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
+            response = requests.get(url, auth=auth, verify=False)
+            response.raise_for_status()
+            command_output = response.json()
+            return command_output.get('token', '')
+        except requests.exceptions.RequestException as e:
+            print("An error occurred during the request:", e)
+            return ''
+
+    @staticmethod
+    def list_tags(quay_url, quay_host, users):
+        """
+        List all tags for all given user repos.
+        We query it on top of the tags created in the load phase.
+
+        :param quay_url: quay host base url
+        :param quay_host: quay host name
+        :param users: list of usernames
+        :return: None
+        """
+        print_header('Listing tags for given users repos')
+        test_name = 'list_tags_for_user_repos'
+        list_of_repos = ['repo_with_100_tags']
+
+        reqs = []
+        for user in users:
+            for repo in list_of_repos:
+                path = '/v2/%s/%s/tags/list' % (user, repo)
+                url = quay_url + path
+                token = Tags.fetch_repo_tokens(quay_host, user, repo)
+                headers = {
+                    "Content-Type": ["application/json"],
+                    "Authorization": [f"Bearer {token}"]
+                }
+                request = {
+                    "url": url,
+                    "method": "GET",
+                    "header": headers,
+                }
+                reqs.append(request)
+        target_name = "'GET %s'" % path
+        Attacker().run_vegeta(test_name, reqs, target_name=target_name)
+
+    @staticmethod
+    def get_catalog(quay_url, target_hit_size):
+        """
+        Gets catalog specified number of times.
+
+        :param quay_url: quay host base url
+        :param target_hit_size: number of times to hit catalog endpoint
+        :return: None
+        """
+        print_header("Running: Get Catalog")
+        path = '/v2/_catalog'
+        url = quay_url + path
+        reqs = []
+        for each in range(target_hit_size):
+            request = {
+                'header': None,
+                'url': url,
+                'method': 'GET',
+            }
+            reqs.append(request)
+        target_name = "'GET %s'" % path
+        Attacker().run_vegeta('get_catalog', reqs, target_name=target_name)
+
     @staticmethod
     def delete_repository_tags(quay_url, org, repo, tags, target_hit_size):
         """

main.py

@@ -707,6 +707,8 @@ def batch_process(users_chunk, batch_args):
     Permissions.list_teams_of_organization_repos(env_config['base_url'], organization, repos)
     Permissions.get_users_of_organization_repos(env_config['base_url'], organization, repos, users)
     Permissions.list_users_of_organization_repos(env_config['base_url'], organization, repos)
+    Tags.get_catalog(env_config['base_url'], env_config["target_hit_size"])
+    Tags.list_tags(env_config['base_url'], env_config['quay_host'], users)
 
     # Cleanup Phase
     # These tests are ran at the end to cleanup stuff