StackSpot Action: action responsible for creating a Gitlab repository written in Python

README.md

Create Repository Gitlab

Creates a new repository (aka project) in a specific Gitlab group.

How to set up this project

This project was developed with Python v3.9, and it also uses Python's Virtualenv tool. So before working on this project we need to set up it.

Inside this project's folder, execute those commands to configure your environment:

python3 -m venv .venv
source env/bin/activate
pip install -r requirements.txt

Now, before starting to code, run all unit tests:

pytest -v

If everything is ok, you can code now :-)

⚠️ You can also run the unit tests with the standard unittest library, as below:

python3 -m unittest script_test.py

For more detail on how to configure Virtualenv, read this article.

Running tests with coverage

Unfortunately, the feature "run tests with coverage" doesn't work with Pycharm Community Edition, so that you need to run it via command line:

pytest -v --cov --cov-report=html:.coverage_reports

A directory called .coverage_reports will be created with all the coverage reports in HTML format inside of it. Then, just open the index.html on the browser to see the report details.

Adding new dependencies

If you add new dependencies in the project, don't forget to run this command below so that other developers can import the environment in their machines correctly:

pip freeze > requirements.txt

action.yaml

schema-version: v2
kind: action
metadata:
  name: create-repo-gitlab
  display-name: create-repo-gitlab
  description: Action used to create a Git repository in Gitlab
  version: 1.0.0
spec:
  type: python
  about: docs/about.md
  usage: docs/usage.md
  release-notes: docs/release-notes.md
  requirements: docs/requirements.md
  implementation: docs/implementation.md
  inputs:
    - label: Gitlab project name
      name: project_name
      type: text
      required: true
      help: Input the Gitlab devOps project/repository name, if it does not exist it will be created
    - label: Gitlab group name
      name: group_name
      type: text
      required: true
      help: Input the Gitlab group name. It must exist in your Gitlab account as a top-level group
    - label: Repository visibility
      name: visibility
      type: text
      required: true
      help: Input the Gitlab repository visibility
      items:
        - public
        - private
      default: private
  python:
    workdir: .
    script: script.py

docs.usage.md

Usage

Here are some commands to execute and test this action locally.

Notice that all those commands are executed inside the action's folder and they use the variable stk_access_token containing the user's access token. So that, before executing these commands, you must define the variable as below:

export stk_access_token="my-access-token-pat"

Also notice that you must set the JSON attributes org (similar to group), project_name (similar to repository name) and ``visibility` (public or private) properly.

1. How to execute this action locally

Inside the action's folder, execute the commando below:

stk run action -i "{\"token\": \"$stk_access_token\", \"project_name\": \"demo-$(date +%s)\", \"group_name\": \"stackspot\", \"visibility\": \"private\"}" --non-interactive .

If the command executes correctly, you will see an output like this:

> Getting the Group ID by the informed group name 'StackSpot'.
> Checking if the project (repository) 'demo-1693864337' exists in the group 'StackSpot' (group_id=71401813).
> Project (repository) 'demo-1693864337' not found in the group 'StackSpot' (group_id=71401813).
> Creating project (repository) 'demo-1693864337' in the group 'StackSpot' (group_id=71401813).
> Project (repository) 'demo-1693864337' created in the group 'StackSpot' with visibility 'private'.

2. Executing with disabled HTTP SSL Verify (GITLAB_HTTP_SSL_VERIFY_ENABLED=false)

By default, the SSL verify is enabled. But if you have any SSL issue, like expired root certificate or something like that, you may disable the SSL verify in Python's Requests libary. For that, just set the dev__enable_ssl_verify property to false:

export GITLAB_HTTP_SSL_VERIFY_ENABLED=false
stk run action -i "{\"token\": \"$stk_access_token\", \"project_name\": \"demo-$(date +%s)\", \"group_name\": \"stackspot\", \"visibility\": \"private\"}" --non-interactive .

You should disable this check for testing purposes only.

3. Executing with enabled HTTP logging (GITLAB_HTTP_LOGGING_ENABLED=true)

By default, the HTTP logging is disabled due to its verbosity. However, enabling it can be helpful when doing troubleshooting. For that, just set the dev__enable_logging property to true:

export GITLAB_HTTP_LOGGING_ENABLED=true
stk run action -i "{\"token\": \"$stk_access_token\", \"project_name\": \"demo-$(date +%s)\", \"group_name\": \"stackspot\", \"visibility\": \"private\"}" --non-interactive .

You should enable the logging for testing purposes only.

4. Executing with specific HTTP timeout (GITLAB_HTTP_REQUEST_TIMEOUT=<number>)

By default, all HTTP requests have a timeout configured to 10 seconds. If needed, you can change the default timeout for any number. For that, just set the dev__default_timeout property to any integer or float value. For example, in the command below, we're setting the timeout to 3 seconds:

export GITLAB_HTTP_REQUEST_TIMEOUT=3
stk run action -i "{\"token\": \"$stk_access_token\", \"project_name\": \"demo-$(date +%s)\", \"group_name\": \"stackspot\", \"visibility\": \"private\"}" --non-interactive .

For more information about Python's Request Library timeout, just read this article: Timeouts in Python requests

script.py

import json
import os
from typing import Any

import requests


class GitlabCreateRepoException(Exception):
    def __init__(self, message):
        self.message = message
        super().__init__(self.message)


class InvalidInputException(GitlabCreateRepoException):
    pass


class GroupNotFound(GitlabCreateRepoException):
    def __init__(self, group_name):
        super().__init__(f"The '{group_name}' group was not found.")


class ProjectAlreadyExistsInTheGroup(GitlabCreateRepoException):
    def __init__(self, group_name, group_id, project_name):
        super().__init__(
            f"The project (repository) '{project_name}' already exists in the group '{group_name}' (group_id={group_id})."
        )


class StkLocalContext:
    """
    An instance of this class represents the STK Local Context
    """
    OUTPUT_FILE_NAME = "stk-local-context.json"

    def __init__(self, json_indent=4):
        self.json_indent = json_indent

    def export_repository_url(self, repository_url):
        """
        Exports Gitlab repository URL to STK local context in the disk
        """
        data = {
            "outputs": {
                "created_repository": repository_url
            }
        }
        with open(self.OUTPUT_FILE_NAME, "w") as file:
            json.dump(data, file, indent=self.json_indent)

    def get_content(self) -> dict:
        """
        Loads STK local context from the disk
        """
        with open(self.OUTPUT_FILE_NAME) as output:
            return json.load(output)

    def clear(self):
        """
        Clears STK local context by deleting the file from the disk
        """
        if os.path.exists(self.OUTPUT_FILE_NAME):
            os.remove(self.OUTPUT_FILE_NAME)


class Runner:
    def __init__(self, metadata, base_url="https://gitlab.com") -> None:
        self.base_url = base_url
        self.group_name = metadata.inputs.get("group_name")
        self.project_name = metadata.inputs.get("project_name")
        self.visibility = metadata.inputs.get("visibility", "private")  # default=private
        self.token = metadata.inputs.get("token")  # injected by the StackSpot Portal
        self._enable_ssl_verify_env = os.getenv("GITLAB_HTTP_SSL_VERIFY_ENABLED", "true")  # default=enabled
        self._enable_logging_env = os.getenv("GITLAB_HTTP_LOGGING_ENABLED", "false")  # default=disabled
        self._request_timeout_env = os.getenv("GITLAB_HTTP_REQUEST_TIMEOUT", "10")  # default=10s
        self.headers = {
            "Private-Token": self.token,
            "Content-Type": "application/json"
        }
        if self.enable_logging:
            self.__enable_logging()

    @property
    def enable_ssl_verify(self) -> bool:
        return (self._enable_ssl_verify_env
                and self._enable_ssl_verify_env.lower() == "true")

    @property
    def enable_logging(self) -> bool:
        return (self._enable_logging_env
                and self._enable_logging_env.lower() == "true")

    @property
    def request_timeout(self) -> int:
        return int(self._request_timeout_env)

    def __call__(self) -> Any:

        self.__validate_inputs()

        group_id = self.__get_group_id(self.group_name)
        if group_id is None:
            raise GroupNotFound(self.group_name)

        if self.__project_exists(group_id, self.project_name):
            raise ProjectAlreadyExistsInTheGroup(self.group_name, group_id, self.project_name)

        created_repo = self.__create_repository_in_group(group_id, self.project_name, self.visibility)
        StkLocalContext().export_repository_url(
            repository_url=created_repo.get("http_url_to_repo")
        )

    def __validate_inputs(self):
        """
        Validates all inputs
        """
        if not self.token or not self.token.strip():
            raise InvalidInputException("The private token ('token') must not be blank.")

        if not self.group_name or not self.group_name.strip():
            raise InvalidInputException("The group name ('group_name') must not be blank.")

        if not self.project_name or not self.project_name.strip():
            raise InvalidInputException("The project name ('project_name') must not be blank.")

        if len(self.project_name.strip()) < 3:
            raise InvalidInputException(
                f"The project name '{self.project_name}' is too short. It must contain at least 3 characters."
            )

        if not self.visibility or not self.visibility.strip():
            raise InvalidInputException("The visibility type ('visibility') must not be blank.")

        if self.visibility not in ["public", "private"]:
            raise InvalidInputException(
                f"The visibility type informed is invalid: '{self.visibility}'. It must be 'public' or 'private'."
            )

    def __get_group_id(self, group_name):
        """
        Gets the Group ID by the informed group name
        """
        print(
            f"> Getting the Group ID by the informed group name '{group_name}'."
        )

        url = f"{self.base_url}/api/v4/groups"
        params = {"search": group_name}

        try:
            response = requests.get(
                url, headers=self.headers, params=params,
                timeout=self.request_timeout, verify=self.enable_ssl_verify
            )
            response.raise_for_status()
            groups = response.json()
            for group in groups:
                if group['name'] == group_name:
                    print(f"> Group ID has been found: '{group['id']}'")
                    return group['id']

            print(f"> Group '{group_name}' not found among all existing groups.")
            return None
        except requests.exceptions.HTTPError as e:
            self.__log_and_raise_error(
                f"Error while searching for group_id by the group name '{group_name}': {e.response.text}"
            )

    def __project_exists(self, group_id, project_name):
        """
        Checks if the project (repository) exists
        """
        print(
            f"> Checking if the project (repository) '{project_name}' exists in the group '{self.group_name}' (group_id={group_id})."
        )

        url = f"{self.base_url}/api/v4/groups/{group_id}/projects"

        try:
            response = requests.get(url, headers=self.headers, timeout=self.request_timeout,
                                    verify=self.enable_ssl_verify)
            response.raise_for_status()
            projects = response.json()
            for project in projects:
                if project['name'] == project_name:
                    return True
                if project['path'] == project_name:
                    return True

            print(
                f"> Project (repository) '{project_name}' not found in the group '{self.group_name}' (group_id={group_id})."
            )
            return False
        except requests.exceptions.HTTPError as e:
            self.__log_and_raise_error(
                f"Error while checking if project (repository) '{project_name}' already exists in the group '{self.group_name} (group_id={group_id})': {e.response.text}"
            )

    def __create_repository_in_group(self, group_id, project_name, visibility):
        """
        Creates a new Gitlab repository (project) in the informed group
        """
        print(
            f"> Creating project (repository) '{project_name}' in the group '{self.group_name}' (group_id={group_id})."
        )

        url = f"{self.base_url}/api/v4/projects"
        data = {
            "name": project_name,
            "namespace_id": group_id,
            "visibility": visibility
        }

        try:
            response = requests.post(
                url, json=data, headers=self.headers,
                timeout=self.request_timeout, verify=self.enable_ssl_verify
            )
            response.raise_for_status()
            print(
                f"> Project (repository) '{project_name}' created in the group '{self.group_name}' with visibility '{visibility}'."
            )
            return response.json()
        except requests.exceptions.HTTPError as e:
            error_message = e.response.text
            if e.response.status_code == 400 and "has already been taken" in e.response.text:
                error_message = "Project already exists in the group. (possible race condition)"

            self.__log_and_raise_error(
                f"Error while creating project (repository) '{project_name}' in the group '{self.group_name}': {error_message}"
            )

    @staticmethod
    def __log_and_raise_error(message):
        """
        Logs error message and raise the main exception
        """
        print(f"> {message}")
        raise GitlabCreateRepoException(message)

    @staticmethod
    def __enable_logging():
        """
        Enables logging for debugging purposes only
        """
        import logging
        import http.client

        http.client.HTTPConnection.debuglevel = 1
        logging.basicConfig()
        logging.getLogger().setLevel(logging.DEBUG)
        requests_log = logging.getLogger("requests.packages.urllib3")
        requests_log.setLevel(logging.DEBUG)
        requests_log.propagate = True


def run(metadata):
    runner = Runner(metadata)
    runner()

script_test.py

import unittest
from unittest.mock import patch, Mock
from script import *
from requests import HTTPError


class TestRunner(unittest.TestCase):

    def setUp(self) -> None:
        # resets stk local context
        StkLocalContext().clear()
        # resets env vars
        if "GITLAB_HTTP_SSL_VERIFY_ENABLED" in os.environ:
            os.environ.pop("GITLAB_HTTP_SSL_VERIFY_ENABLED")
        if "GITLAB_HTTP_LOGGING_ENABLED" in os.environ:
            os.environ.pop("GITLAB_HTTP_LOGGING_ENABLED")
        if "GITLAB_HTTP_REQUEST_TIMEOUT" in os.environ:
            os.environ.pop("GITLAB_HTTP_REQUEST_TIMEOUT")

    ##
    # Happy-path
    ##
    @patch('requests.post')
    @patch('requests.get')
    def test_create_repository_in_group_success(self, mock_get, mock_post):
        # scenario
        metadata = Metadata({
            "group_name": "StackSpot",
            "project_name": "My Java App",
            "visibility": "private",
            "token": "valid-token"
        })
        mock_get.side_effect = [
            Mock(status_code=200, json=lambda: [{"name": "StackSpot", "id": 123, "path": "stackspot1"}]),
            Mock(status_code=200, json=lambda: [
                {"id": 1, "name": "project 1", "path": "project1"},
                {"id": 2, "name": "project 2", "path": "project2"},
                {"id": 3, "name": "project 3", "path": "project3"}
            ])
        ]
        mock_post.return_value.status_code = 201
        mock_post.return_value.json.return_value = {
            "id": 24,
            "name": "My Java App",
            "path": "my-java-app",
            "ssh_url_to_repo": "[email protected]:stackspot1/meu-projeto-b3.git",
            "http_url_to_repo": "https://gitlab.com/stackspot1/meu-projeto-b3.git"
        }

        # action
        runner = Runner(metadata, "http://localhost:9090")
        runner()

        # validation
        mock_post.assert_called_once_with(
            "http://localhost:9090/api/v4/projects",
            json={
                "name": metadata.get("project_name"),
                "namespace_id": 123,
                "visibility": metadata.get("visibility"),
            },
            headers={
                "Private-Token": "valid-token",
                "Content-Type": "application/json"
            },
            timeout=10,
            verify=True
        )

        # validates STK local context
        ctx_content = StkLocalContext().get_content()
        expected_content = dict(outputs=dict(created_repository="https://gitlab.com/stackspot1/meu-projeto-b3.git"))
        self.assertDictEqual(expected_content, ctx_content, "STK local context")

    def test_inputs_has_default_values(self):
        # scenario
        empty_metadata = Metadata({})

        # action
        runner = Runner(empty_metadata)

        # validation
        self.assertEqual("https://gitlab.com", runner.base_url, "Gitlab base URL")
        self.assertEqual("private", runner.visibility, "visibility type")
        self.assertEqual(None, runner.project_name, "project name")
        self.assertEqual(None, runner.group_name, "group name")

    def test_env_variables_GITLAB_HTTP_SSL_VERIFY_ENABLED(self):
        # scenario
        empty_metadata = Metadata({})

        # validation: default values
        runner = Runner(empty_metadata)
        self.assertEqual(True, runner.enable_ssl_verify, "HTTP SSL verify: enabled by default")

        # validation: toggle on
        os.environ["GITLAB_HTTP_SSL_VERIFY_ENABLED"] = "true"
        runner = Runner(empty_metadata)
        self.assertEqual(True, runner.enable_ssl_verify, "HTTP SSL verify: enabled")

        # validation: toggle off
        os.environ["GITLAB_HTTP_SSL_VERIFY_ENABLED"] = "false"
        runner = Runner(empty_metadata)
        self.assertEqual(False, runner.enable_ssl_verify, "HTTP SSL verify: disabled")

        # validation: toggle on (insensitive-case)
        os.environ["GITLAB_HTTP_SSL_VERIFY_ENABLED"] = "TRUE"
        runner = Runner(empty_metadata)
        self.assertEqual(True, runner.enable_ssl_verify, "HTTP SSL verify: enabled")

        # validation: invalid value
        os.environ["GITLAB_HTTP_SSL_VERIFY_ENABLED"] = "anything different than 'true' is 'false'"
        runner = Runner(empty_metadata)
        self.assertEqual(False, runner.enable_ssl_verify, "HTTP SSL verify: disabled")

    def test_env_variables_GITLAB_HTTP_LOGGING_ENABLED(self):
        # scenario
        empty_metadata = Metadata({})

        # validation: default values
        runner = Runner(empty_metadata)
        self.assertEqual(False, runner.enable_logging, "HTTP logging: disabled by default")

        # validation: toggle on
        os.environ["GITLAB_HTTP_LOGGING_ENABLED"] = "true"
        runner = Runner(empty_metadata)
        self.assertEqual(True, runner.enable_logging, "HTTP logging: enabled")

        # validation: toggle off
        os.environ["GITLAB_HTTP_LOGGING_ENABLED"] = "false"
        runner = Runner(empty_metadata)
        self.assertEqual(False, runner.enable_logging, "HTTP logging: disabled")

        # validation: toggle on (insensitive-case)
        os.environ["GITLAB_HTTP_LOGGING_ENABLED"] = "TRUE"
        runner = Runner(empty_metadata)
        self.assertEqual(True, runner.enable_logging, "HTTP logging: enabled")

        # validation: invalid value
        os.environ["GITLAB_HTTP_LOGGING_ENABLED"] = "anything different than 'true' is 'false'"
        runner = Runner(empty_metadata)
        self.assertEqual(False, runner.enable_logging, "HTTP logging: disabled")

    def test_env_variables_GITLAB_HTTP_REQUEST_TIMEOUT(self):
        # scenario
        empty_metadata = Metadata({})

        # validation: default values
        runner = Runner(empty_metadata)
        self.assertEqual(10, runner.request_timeout, "HTTP request timeout: '10s' by default")

        # validation: informed
        os.environ["GITLAB_HTTP_REQUEST_TIMEOUT"] = "2"
        runner = Runner(empty_metadata)
        self.assertEqual(2, runner.request_timeout, "HTTP request timeout: 2s")

    def test_group_name_is_not_informed_error(self):
        # scenario
        metadata = Metadata({
            "token": "valid-token-injected-by-portal"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual("The group name ('group_name') must not be blank.", str(em.exception))

    def test_group_name_is_blank_error(self):
        # scenario
        metadata = Metadata({
            "group_name": (" " * 10),
            "token": "valid-token-injected-by-portal"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual("The group name ('group_name') must not be blank.", str(em.exception))

    def test_token_is_not_informed_error(self):
        # scenario
        metadata = Metadata({
            "group_name": "valid-group"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual("The private token ('token') must not be blank.", str(em.exception))

    def test_token_is_blank_error(self):
        # scenario
        metadata = Metadata({
            "group_name": "valid-group",
            "token": (" " * 10)
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual("The private token ('token') must not be blank.", str(em.exception))

    def test_project_name_is_not_informed_error(self):
        # scenario
        metadata = Metadata({
            "group_name": "my_group",
            "token": "valid-token"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual("The project name ('project_name') must not be blank.", str(em.exception))

    def test_project_name_is_blank_error(self):
        # scenario
        metadata = Metadata({
            "group_name": "my_group",
            "project_name": (" " * 10),
            "token": "valid-token"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual("The project name ('project_name') must not be blank.", str(em.exception))

    def test_project_name_is_too_short_error(self):
        # scenario
        metadata = Metadata({
            "group_name": "my_group",
            "project_name": "p1",
            "token": "valid-token"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(
            "The project name 'p1' is too short. It must contain at least 3 characters.",
            str(em.exception)
        )

    def test_project_name_is_too_short_with_whitespaces_error(self):
        # scenario
        project_name_with_whitespaces = "   p2   "
        metadata = Metadata({
            "group_name": "my_group",
            "project_name": project_name_with_whitespaces,
            "token": "valid-token"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(
            f"The project name '{project_name_with_whitespaces}' is too short. It must contain at least 3 characters.",
            str(em.exception)
        )

    def test_visibility_type_is_blank_error(self):
        # scenario
        blank_visibility = (" " * 10)
        metadata = Metadata({
            "group_name": "my_group",
            "project_name": "my-project",
            "visibility": blank_visibility,
            "token": "valid-token"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual("The visibility type ('visibility') must not be blank.", str(em.exception))

    def test_visibility_type_is_invalid_error(self):
        # scenario
        invalid_visibility = "internal"
        metadata = Metadata({
            "group_name": "my_group",
            "project_name": "my-project",
            "visibility": invalid_visibility,
            "token": "valid-token"
        })

        # action and validation
        with self.assertRaises(InvalidInputException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(
            f"The visibility type informed is invalid: '{invalid_visibility}'. It must be 'public' or 'private'.",
            str(em.exception)
        )

    @patch('requests.get')
    def test_group_not_found_error(self, mock_get):
        # scenario
        metadata = Metadata({
            "group_name": "invalid-group",
            "project_name": "blank-project",
            "token": "valid-token"
        })
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = [{"name": "group1", "id": 1}, {"name": "group2", "id": 2}]

        # action and validation
        with self.assertRaises(GroupNotFound) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(f"The 'invalid-group' group was not found.", str(em.exception))

    @patch('requests.get')
    def test_project_already_exists_in_name_error(self, mock_get):
        # scenario
        metadata = Metadata({
            "group_name": "stackspot_group",
            "project_name": "Existing project",
            "token": "valid-token"
        })
        mock_get.side_effect = [
            Mock(status_code=200, json=lambda: [{"name": "stackspot_group", "id": 123}]),
            Mock(status_code=200, json=lambda: [
                {"id": 1, "name": "Simple project", "path": "simple-project"},
                {"id": 2, "name": metadata.get("project_name"), "path": "existing-project-app"},
                {"id": 3, "name": "Another java project", "path": "another-java-project"}
            ])
        ]

        # action and validation
        with self.assertRaises(ProjectAlreadyExistsInTheGroup) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(
            "The project (repository) 'Existing project' already exists in the group 'stackspot_group' (group_id=123).",
            str(em.exception)
        )

    @patch('requests.get')
    def test_project_already_exists_in_path_error(self, mock_get):
        # scenario
        metadata = Metadata({
            "group_name": "stackspot_group",
            "project_name": "existing-project",
            "token": "valid-token"
        })
        mock_get.side_effect = [
            Mock(status_code=200, json=lambda: [{"name": "stackspot_group", "id": 456}]),
            Mock(status_code=200, json=lambda: [
                {"id": 1, "name": "Simple project", "path": "simple-project"},
                {"id": 2, "name": "Existing project App", "path": metadata.get("project_name")},
                {"id": 3, "name": "Another java project", "path": "another-java-project"}
            ])
        ]

        # action and validation
        with self.assertRaises(ProjectAlreadyExistsInTheGroup) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(
            "The project (repository) 'existing-project' already exists in the group 'stackspot_group' (group_id=456).",
            str(em.exception)
        )

    ##
    # Validates a possible race condition scenario when creating a repository
    ##
    @patch('requests.post')
    @patch('requests.get')
    def test_create_repository_in_group_when_project_already_exists_error(self, mock_get, mock_post):
        # scenario
        metadata = Metadata({
            "group_name": "group1",
            "project_name": "my repo",
            "visibility": "public",
            "token": "valid-token"
        })
        mock_get.side_effect = [
            Mock(status_code=200, json=lambda: [{"name": "group1", "id": 123}]),
            Mock(status_code=200, json=lambda: [])
        ]

        mock_post.return_value = self.__mock_http_error_400_with_payload(json={
            "message": {
                "project_namespace.name": [
                    "has already been taken"
                ],
                "name": [
                    "has already been taken"
                ],
                "path": [
                    "has already been taken"
                ]
            }
        })

        # action and validation
        with self.assertRaises(GitlabCreateRepoException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(
            f"Error while creating project (repository) 'my repo' in the group 'group1': Project already exists in the group. (possible race condition)",
            str(em.exception)
        )

    @patch('requests.get')
    def test_group_unexpected_error(self, mock_get):
        # scenario
        metadata = Metadata({
            "group_name": "group1",
            "project_name": "blank-project",
            "token": "valid-token"
        })
        mock_get.return_value = self.__mock_http_error_401()

        # action and validation
        with self.assertRaises(GitlabCreateRepoException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(f"Error while searching for group_id by the group name 'group1': 401-Unauthorized",
                         str(em.exception))

    @patch('requests.get')
    def test_project_unexpected_error(self, mock_get):
        # scenario
        metadata = Metadata({
            "group_name": "group1",
            "project_name": "my-new-project",
            "token": "valid-token"
        })
        mock_get.side_effect = [
            Mock(status_code=200, json=lambda: [{"name": "group1", "id": 123}]),
            self.__mock_http_error_401()
        ]

        # action and validation
        with self.assertRaises(GitlabCreateRepoException) as em:
            runner = Runner(metadata, "http://localhost:9090")
            runner()
        self.assertEqual(
            "Error while checking if project (repository) 'my-new-project' already exists in the group 'group1 (group_id=123)': 401-Unauthorized",
            str(em.exception)
        )

    @staticmethod
    def __mock_http_error_400_with_payload(json):
        mock_response = Mock()
        mock_response.status_code = 400
        mock_response.raise_for_status.side_effect = HTTPError("Bad Request", response=mock_response)
        mock_response.text = str(json)
        mock_response.json.return_value = json
        return mock_response

    @staticmethod
    def __mock_http_error_401():
        mock_response = Mock()
        mock_response.status_code = 401
        mock_response.raise_for_status.side_effect = HTTPError(
            "401-Unauthorized: Private Token invalid or not informed", response=mock_response)
        mock_response.text = "401-Unauthorized"
        return mock_response


class Metadata:
    def __init__(self, inputs):
        self.inputs = inputs

    def get(self, key):
        return self.inputs.get(key)


if __name__ == "__main__":
    unittest.main()

https://github.com/stack-spot/stackspot-workflows-action/blob/7ec55d4a645959100b6c990c6367b11937a95e96/stackspot-actions/gitlab/README.md