Skip to content

Instantly share code, notes, and snippets.

@rponte
Last active January 5, 2024 19:19
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save rponte/efe3a428cb5c4e1fcc9b1a948f5831de to your computer and use it in GitHub Desktop.
Save rponte/efe3a428cb5c4e1fcc9b1a948f5831de to your computer and use it in GitHub Desktop.
StackSpot Action: action responsible for creating a Gitlab repository written in Python

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
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

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

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()
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": "git@gitlab.com: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()