Pour tout problème contactez-nous par mail : support@froggit.fr | La FAQ :grey_question: | Rejoignez-nous sur le Chat :speech_balloon:

Skip to content
Snippets Groups Projects
Commit f993bc7c authored by Dorian Turba's avatar Dorian Turba
Browse files

OOP

parent fc040bf7
No related branches found
No related tags found
No related merge requests found
...@@ -10,15 +10,12 @@ authors = [ ...@@ -10,15 +10,12 @@ authors = [
] ]
description = "Release a new version of a software based on CHANGELOG.md file." description = "Release a new version of a software based on CHANGELOG.md file."
readme = "README.md" readme = "README.md"
requires-python = ">=3.8" requires-python = ">=3.11"
classifiers = [ classifiers = [
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License", "License :: OSI Approved :: MIT License",
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Development Status :: 3 - Alpha", "Development Status :: 3 - Alpha",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.12",
"Typing :: Typed", "Typing :: Typed",
...@@ -210,7 +207,7 @@ legacy_tox_ini = """ ...@@ -210,7 +207,7 @@ legacy_tox_ini = """
[tox] [tox]
min_version = 4.0 min_version = 4.0
env_list = env_list =
py{38,39,310,311,312} py{311,312}
isolated_build = true isolated_build = true
[testenv] [testenv]
......
from __future__ import annotations
import typer
from release_by_changelog import logging
def get_token(ci_job_token: str | None, token: str | None) -> str:
"""
Get the token from the environment variables or the CLI.
:raises typer.Exit: If no token is provided.
"""
match (token, ci_job_token):
case (str(value), _) | (_, str(value)):
return value
case _:
logging.error(
logging.err_panel(
"You need to provide a PRIVATE_TOKEN or a CI_JOB_TOKEN",
title="Error: Missing token",
)
)
raise typer.Exit(code=1)
...@@ -5,14 +5,9 @@ import typing ...@@ -5,14 +5,9 @@ import typing
import typer import typer
import release_by_changelog.services.release
from release_by_changelog.app import app from release_by_changelog.app import app
from release_by_changelog.services.release import ( from release_by_changelog.cmds._release import get_token
find_changelog_file,
get_project,
get_token,
last_changelog,
publish,
)
@app.command() @app.command()
...@@ -33,11 +28,6 @@ def release( ...@@ -33,11 +28,6 @@ def release(
envvar="CI_SERVER_HOST", envvar="CI_SERVER_HOST",
), ),
] = "gitlab.com", ] = "gitlab.com",
interact: bool = typer.Option(
True,
help="CLI ask for confirmation before creating the release. "
"No interaction means automatic confirmation.",
),
token: str = typer.Option( token: str = typer.Option(
None, None,
help="[red]Required[/red] for [yellow]user-based[/yellow] authentication.", help="[red]Required[/red] for [yellow]user-based[/yellow] authentication.",
...@@ -48,6 +38,11 @@ def release( ...@@ -48,6 +38,11 @@ def release(
help="[red]Required[/red] for [yellow]CI-based[/yellow] authentication.", help="[red]Required[/red] for [yellow]CI-based[/yellow] authentication.",
envvar="CI_JOB_TOKEN", envvar="CI_JOB_TOKEN",
), ),
interact: bool = typer.Option(
True,
help="CLI ask for confirmation before creating the release. "
"No interaction means automatic confirmation.",
),
tag_only: bool = typer.Option( tag_only: bool = typer.Option(
False, False,
help="Only tag the commit with the changelog version.", help="Only tag the commit with the changelog version.",
...@@ -55,22 +50,32 @@ def release( ...@@ -55,22 +50,32 @@ def release(
), ),
) -> None: ) -> None:
token = get_token(ci_job_token, token) token = get_token(ci_job_token, token)
changelog_path = find_changelog_file(changelog_path, token, host, project, ref) release_ = release_by_changelog.services.release.Release(
changelog_entry = last_changelog(changelog_path) project_id=project,
project_, project_name = get_project(token, host, project) ref=ref,
changelog_path=changelog_path,
if interact: host=host,
typer.confirm("Do you confirm release?", default=True, abort=True) interact=interact,
token=token,
publish(changelog_entry, project_, project_name, ref, tag_only) )
release_.publish(tag_only)
# changelog_path = find_changelog_file(changelog_path, token, host, project, ref)
# changelog_entry = last_changelog(changelog_path)
# project_, project_name = get_project(token, host, project)
#
# if interact:
# typer.confirm("Do you confirm release?", default=True, abort=True)
#
# publish(changelog_entry, project_, project_name, ref, tag_only)
if __name__ == "__main__": if __name__ == "__main__":
release( r = release_by_changelog.services.release.Release(
project="1484", project_id="1484",
ref="main", ref="main",
changelog_path=pathlib.Path("test_dir/TEST_CHANGELOG.md"), changelog_path=pathlib.Path("test_dir/TEST_CHANGELOG.md"),
host="lab.frogg.it", host="lab.frogg.it",
interact=True, interact=True,
token="glpat-nxkztk41T6-WDxfaifxs", token="glpat-nxkztk41T6-WDxfaifxs",
) )
r.publish(False)
...@@ -13,248 +13,223 @@ import gitlab.v4.objects ...@@ -13,248 +13,223 @@ import gitlab.v4.objects
import typer import typer
from release_by_changelog import logging from release_by_changelog import logging
from release_by_changelog.typings_ import ChangelogEntry from release_by_changelog.typings_ import Version
def publish(
changelog_entry: ChangelogEntry,
project: gitlab.v4.objects.Project,
project_name: str,
ref: str,
tag_only: bool,
) -> None:
logging.info(
f"Creating release [bold cyan]{changelog_entry.version}[/bold cyan] for "
f"project [bold cyan]{project_name}[/bold cyan]"
)
data = {
"tag_name": changelog_entry.version,
"ref": ref,
}
try:
getattr(project, TARGET_RELEASE[tag_only]).create(data)
except gitlab.GitlabAuthenticationError as e:
if e.response_code == http.HTTPStatus.UNAUTHORIZED:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Check if the provided token is correct.\n"
"\t- Check if the provided token has the correct permissions and "
"scope, is not expired or revoked.",
title="Error: Unauthorized",
)
)
raise typer.Exit(code=1)
raise e
except gitlab.GitlabCreateError as e:
if e.response_code == http.HTTPStatus.CONFLICT:
logging.error(
logging.err_panel(
f"It looks like the {TARGET_RELEASE[tag_only]} [bold cyan]"
f"{changelog_entry.version}[/bold cyan] already exists.\n"
"Possible remediation: Bump the version in the changelog file.",
title=f"Error: {TARGET_RELEASE[tag_only].capitalize()} already "
f"exists",
)
)
raise typer.Exit(code=1)
raise e
logging.success(f"Release created: {changelog_entry.version}")
TARGET_RELEASE: typing.Final[dict[bool, typing.Literal["tags", "releases"]]] = { TARGET_RELEASE: typing.Final[dict[bool, typing.Literal["tags", "releases"]]] = {
True: "tags", True: "tags",
False: "releases", False: "releases",
} }
REGEX: typing.Final = re.compile(
r"^## \[(?P<version>(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]"
r"\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*"
r"|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA"
r"-Z-]+)*))?)\]"
)
def get_token(ci_job_token: str | None, token: str | None) -> str:
"""
Get the token from the environment variables or the CLI.
:raises typer.Exit: If no token is provided.
"""
if token is not None:
return token
if ci_job_token is not None:
return ci_job_token
logging.error( class Release:
logging.err_panel( project_id: str
"You need to provide a PRIVATE_TOKEN or a CI_JOB_TOKEN", ref: str
title="Error: Missing token", changelog_path: pathlib.Path
host: str
interact: bool
token: str
def __init__(
self: typing.Self,
project_id: str,
ref: str,
changelog_path: pathlib.Path,
host: str,
interact: bool,
token: str,
) -> None:
self.project_id = project_id
self.ref = ref
self.changelog_path = changelog_path
self.host = host
self.interact = interact
self.token = token
@functools.cached_property
def tmp_changelog(self: typing.Self) -> pathlib.Path:
"""
Save the remote changelog file to a temporary file.
:raises NotImplementedError: If the OS is not supported.
"""
if os.name == "posix":
tmp_file = pathlib.Path(f"/tmp/{self.changelog_path.name}")
elif os.name == "nt":
temp_path = pathlib.Path(os.environ["Temp"])
tmp_file = temp_path / self.changelog_path.name
else:
raise NotImplementedError(f"OS {os.name} not supported")
tmp_file.write_bytes(self.remote_changelog.decode())
return tmp_file
@functools.cached_property
def remote_changelog(
self: typing.Self,
) -> gitlab.v4.objects.ProjectFile:
try:
changelog_file: gitlab.v4.objects.ProjectFile = self.project.files.get(
file_path=str(self.changelog_path), ref=self.ref
)
except gitlab.GitlabGetError as e:
if e.response_code == http.HTTPStatus.NOT_FOUND:
logging.error(
f"[bold red]Changelog file {self.changelog_path} not found in the "
f"remote project files[/bold red]"
)
raise typer.Exit(code=1)
raise e
logging.success(
f"[bold cyan]{self.changelog_path}[/bold cyan] found in the remote project "
"files"
) )
) return changelog_file
raise typer.Exit(code=1)
@functools.cached_property
def local_changelog(self: typing.Self) -> pathlib.Path:
def _extract_body(f: typing.TextIO) -> str: logging.info(f"Look for local [bold cyan]{self.changelog_path}[/bold cyan]")
body = [] if self.changelog_path.exists():
for lines in f: logging.success(f"Found local [bold cyan]{self.changelog_path}[/bold cyan]")
matches = regex.finditer(lines) return self.changelog_path
with contextlib.suppress(StopIteration):
next(matches) logging.warn(
break f"Local [bold cyan]{self.changelog_path}[/bold cyan] file not found, "
"looking for file in the remote project files"
body.append(lines)
return "".join(body)
def _get_remote_changelog_file(
changelog_path: pathlib.Path,
project_: gitlab.v4.objects.Project,
ref: str,
) -> gitlab.v4.objects.ProjectFile:
try:
changelog_file: gitlab.v4.objects.ProjectFile = project_.files.get(
file_path=str(changelog_path), ref=ref
) )
except gitlab.GitlabGetError as e:
if e.response_code == http.HTTPStatus.NOT_FOUND:
logging.error(
f"[bold red]Changelog file {changelog_path} not found in the remote "
f"project files[/bold red]"
)
raise typer.Exit(code=1)
raise e
logging.success(
f"[bold cyan]{changelog_path}[/bold cyan] found in the remote project files"
)
return changelog_file
def _save_remote_changelog_file( return self.tmp_changelog
changelog_file: gitlab.v4.objects.ProjectFile, changelog_path: pathlib.Path
) -> pathlib.Path: @staticmethod
""" def extract_last_version(f: typing.TextIO) -> str:
Save the remote changelog file to a temporary file. for lines in f:
matches = REGEX.finditer(lines)
:raises NotImplementedError: If the OS is not supported. try:
""" match = next(matches)
if os.name == "posix": except StopIteration: # test
tmp_file = pathlib.Path(f"/tmp/{changelog_path.name}") continue # test
elif os.name == "nt": return match.group("version")
temp_path = pathlib.Path(os.environ["Temp"]) raise ValueError("No changelog entry found")
tmp_file = temp_path / changelog_path.name
else: @staticmethod
raise NotImplementedError(f"OS {os.name} not supported") def extract_description(f: typing.TextIO) -> str:
tmp_file.write_bytes(changelog_file.decode()) body = []
return tmp_file for lines in f:
matches = REGEX.finditer(lines)
with contextlib.suppress(StopIteration):
def find_changelog_file( next(matches)
changelog_path: pathlib.Path, break
token: str,
host: str, body.append(lines)
project: str, return "".join(body)
ref: str,
) -> pathlib.Path: @functools.cached_property
""" def version(self: typing.Self) -> Version:
Find usable changelog file path. """Extract the last changelog entry from the changelog file."""
logging.info(f"Processing [bold cyan]{self.local_changelog}[/bold cyan]")
If the changelog file is not found locally, it will look for it in the remote with self.local_changelog.open() as f:
project files. version = self.extract_last_version(f)
description = self.extract_description(f)
:raises NotImplementedError: If the OS is not supported. logging.success(f"Found changelog entry: [bold cyan]{version}[/bold cyan]")
""" return Version(name=version, description=description)
logging.info(f"Look for local [bold cyan]{changelog_path}[/bold cyan]")
if changelog_path.exists(): @functools.cached_property
logging.success(f"Found local [bold cyan]{changelog_path}[/bold cyan]") def project(self: typing.Self) -> gitlab.v4.objects.Project:
return changelog_path logging.info(
f"Retrieving project [bold cyan]{self.project_id}[/bold cyan] from "
logging.warn( f"[bold cyan]{self.host}[/bold cyan]"
f"Local [bold cyan]{changelog_path}[/bold cyan] file not found, looking for " )
"file in the remote project files" url = f"https://{self.host}"
) gl = gitlab.Gitlab(url=url, oauth_token=self.token)
project_, _ = get_project(
token,
host,
project,
)
changelog_file = _get_remote_changelog_file(changelog_path, project_, ref)
tmp_file = _save_remote_changelog_file(changelog_file, changelog_path)
return tmp_file
def last_changelog(changelog_path: pathlib.Path) -> ChangelogEntry:
"""Extract the last changelog entry from the changelog file."""
logging.info(f"Processing [bold cyan]{changelog_path}[/bold cyan]")
with changelog_path.open() as f:
version = _extract_last_version(f)
body = _extract_body(f)
logging.success(f"Found changelog entry: [bold cyan]{version}[/bold cyan]")
return ChangelogEntry(version=version, body=body)
def _extract_last_version(f: typing.TextIO) -> str:
for lines in f:
matches = regex.finditer(lines)
try: try:
match = next(matches) project = gl.projects.get(self.project_id)
except StopIteration: except gitlab.GitlabAuthenticationError as e:
continue if e.response_code == http.HTTPStatus.UNAUTHORIZED:
return match.group("version") logging.error(
raise ValueError("No changelog entry found") logging.err_panel(
"Possible remediation:\n"
"\t- Check if the provided token is correct.\n"
regex: typing.Final = re.compile( "\t- Check if the provided token has the correct permissions "
r"^## \[(?P<version>(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]" "and scope, is not expired or revoked.",
r"\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*" title="Error: Unauthorized",
r"|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA" )
r"-Z-]+)*))?)\]" )
) raise typer.Exit(code=1)
raise e
except gitlab.GitlabGetError as e:
if e.response_code == http.HTTPStatus.NOT_FOUND:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Provide the project ID. To find the project ID, go to the "
"project page, click on the three vertical dot button on the "
"top right corner and press the 'Copy project ID: XXXX' "
"button.\n"
"\t- Provide the full path to the project. To find it, go to "
"the project page, check at the url and copy the path after the"
" host. If the following link doesn't point to your project, "
"you could have the namespace wrong: "
f"'https://{self.host}/{self.project_id}'\n"
"\t- Check if the host is correct. If you are using a "
"self-hosted GitLab, you need to provide the correct host. "
f"Current host: 'https://{self.host}', is that where your "
f"project is hosted?\n",
title="Error: Project not found.",
)
)
raise typer.Exit(code=1)
raise e
project_path = f"{project.namespace.get('full_path')}/{project.name}"
logging.success(f"Project found: [bold cyan]{project_path}[bold cyan]")
return project
@functools.cached_property
def project_path(self: typing.Self) -> str:
return f"{self.project.namespace.get('full_path')}/{self.project.name}"
def publish(self: typing.Self, tag_only: bool) -> None:
logging.info(
f"Creating release [bold cyan]{self.version.name}[/bold cyan] for "
f"project [bold cyan]{self.project_path}[/bold cyan]"
)
data = {
"tag_name": self.version.name,
"ref": self.ref,
}
target_attr = TARGET_RELEASE[tag_only]
target = getattr(self.project, target_attr)
@functools.cache try:
def get_project( target.create(data)
token: str, except gitlab.GitlabAuthenticationError as e:
host: str, if e.response_code == http.HTTPStatus.UNAUTHORIZED:
project: str, logging.error(
) -> tuple[gitlab.v4.objects.Project, str]: logging.err_panel(
logging.info( "Possible remediation:\n"
f"Retrieving project [bold cyan]{project}[/bold cyan] from " "\t- Check if the provided token is correct.\n"
f"[bold cyan]{host}[/bold cyan]" "\t- Check if the provided token has the correct permissions "
) "and scope, is not expired or revoked.",
url = f"https://{host}" title="Error: Unauthorized",
gl = gitlab.Gitlab(url=url, oauth_token=token) )
try:
project_ = gl.projects.get(project)
except gitlab.GitlabAuthenticationError as e:
if e.response_code == http.HTTPStatus.UNAUTHORIZED:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Check if the provided token is correct.\n"
"\t- Check if the provided token has the correct permissions and "
"scope, is not expired or revoked.",
title="Error: Unauthorized",
) )
) raise typer.Exit(code=1)
raise typer.Exit(code=1) raise e
raise e except gitlab.GitlabCreateError as e:
except gitlab.GitlabGetError as e: if e.response_code == http.HTTPStatus.CONFLICT:
if e.response_code == http.HTTPStatus.NOT_FOUND: logging.error(
logging.error( logging.err_panel(
logging.err_panel( f"It looks like the {target_attr} [bold cyan]"
"Possible remediation:\n" f"{self.version.name}[/bold cyan] already exists.\n"
"\t- Provide the project ID. To find the project ID, go to the " "Possible remediation: Bump the version in the changelog file.",
"project page, click on the three vertical dot button on the top " title=f"Error: {target_attr.capitalize()} already " f"exists",
"right corner and press the 'Copy project ID: XXXX' button.\n" )
"\t- Provide the full path to the project. To find it, go to the "
"project page, check at the url and copy the path after the host. "
"If the following link doesn't point to your project, you could "
f"have the namespace wrong: 'https://{host}/{project}'\n"
"\t- Check if the host is correct. If you are using a self-hosted "
"GitLab, you need to provide the correct host. Current host: "
f"'https://{host}', is that where your project is hosted?\n",
title="Error: Project not found.",
) )
) raise typer.Exit(code=1)
raise typer.Exit(code=1) raise e
raise e logging.success(f"Release created: {self.version.name}")
project_name = f"{project_.namespace.get('full_path')}/{project_.name}"
logging.success(f"Project found: [bold cyan]{project_name}[bold cyan]")
return project_, project_name
...@@ -3,6 +3,6 @@ from __future__ import annotations ...@@ -3,6 +3,6 @@ from __future__ import annotations
import typing import typing
class ChangelogEntry(typing.NamedTuple): class Version(typing.NamedTuple):
version: str name: str
body: str description: str
...@@ -7,7 +7,6 @@ import pathlib ...@@ -7,7 +7,6 @@ import pathlib
import click.testing import click.testing
import pytest import pytest
import pytest_mock import pytest_mock
import release_by_changelog.services.release
import typer.testing import typer.testing
from release_by_changelog.app import app as rbc_app from release_by_changelog.app import app as rbc_app
...@@ -326,8 +325,3 @@ def fake_post() -> collections.abc.Callable: ...@@ -326,8 +325,3 @@ def fake_post() -> collections.abc.Callable:
def successful_gitlab_interaction(mocker: pytest_mock.MockFixture) -> None: def successful_gitlab_interaction(mocker: pytest_mock.MockFixture) -> None:
mocker.patch("gitlab.Gitlab.http_get", side_effect=fake_get()) mocker.patch("gitlab.Gitlab.http_get", side_effect=fake_get())
mocker.patch("gitlab.Gitlab.http_post", side_effect=fake_post()) mocker.patch("gitlab.Gitlab.http_post", side_effect=fake_post())
@pytest.fixture(autouse=True)
def clear_cached_functions() -> None:
release_by_changelog.services.release.get_project.cache_clear()
...@@ -18,7 +18,7 @@ def test_release( ...@@ -18,7 +18,7 @@ def test_release(
in result.output in result.output
) )
assert "Retrieving project 3 from HOST" in result.output assert "Retrieving project 3 from HOST" in result.output
assert "Project found: diaspora/Diaspora Project Site" in result.output assert "Project found: diaspora/Diaspora" in result.output
assert "FILE found in the remote project files" in result.output assert "FILE found in the remote project files" in result.output
assert "Found changelog entry: 1.0.0" in result.output assert "Found changelog entry: 1.0.0" in result.output
assert ( assert (
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment