versionTest.py

import re, os
from kebechet.utils import cloned_repo
from kebechet.managers import VersionManager
import typing 
from git import Repo
import semver
import logging
from datetime import datetime


_LOGGER = logging.getLogger(__name__)
_VERSION_PULL_REQUEST_NAME = 'Release of version {}'
_NO_VERSION_FOUND_ISSUE_NAME = f"No version identifier found in sources to perform a release"
_MULTIPLE_VERSIONS_FOUND_ISSUE_NAME = f"Multiple version identifiers found in sources to perform a new release"
_NO_MAINTAINERS_ERROR = "No release maintainers stated for this repository"
_DIRECT_VERSION_TITLE = ' release'
_RELEASE_TITLES = {
    "new calendar release": lambda _: datetime.utcnow().strftime("%Y.%m.%d"),
    "new major release": lambda current_version : semver.VersionInfo.bump_major,
    "new minor release": lambda current_version : str(semver.VersionInfo.parse(current_version).bump_minor()),
    "new patch release": lambda current_version : str(semver.VersionInfo.parse(current_version).bump_patch()),
    "new pre-release": lambda current_version : str(semver.VersionInfo.parse(current_version).bump_prerelease()),
    "new build release": lambda current_version : str(semver.VersionInfo.parse(current_version).bump_build()),
    "finalize version": lambda current_version : str(semver.VersionInfo.parse(current_version).finalize_version()),
}


def _get_new_version(issue_title: str, current_version: str) -> typing.Optional[str]:
        """Get next version based on user request."""
        issue_title = issue_title.lower()

        handler = _RELEASE_TITLES.get(issue_title)
        if handler:
            try:
                return handler(current_version)
            except ValueError as exc:  # Semver raises ValueError when version cannot be parsed.
                raise 

        if issue_title.endswith(_DIRECT_VERSION_TITLE):  # a specific release
            parts = issue_title.split(' ')
            if len(parts) == 2:
                return parts[0]

        return None

def _adjust_version_file(file_path, issue) -> typing.Optional[tuple]:
        """Adjust version in the given file, return signalizes whether the return value indicates change in file."""
        with open(file_path, 'r') as input_file:
            content = input_file.read().splitlines()

        changed = False
        new_version = None
        old_version = None
        for idx, line in enumerate(content):
            if line.startswith('__version__ = '):
                parts = line.split(' = ', maxsplit=1)
                if len(parts) != 2:
                    print(
                        "Found '__version__' identifier but unable to parse old version, skipping: %r", line
                    )
                    continue

                old_version = parts[1][1:-1]  # Remove ' and " in string representation.
                print("Old version found in sources: %s", old_version)

                new_version = _get_new_version("new patch release", old_version)
                print("Computed new version: %s", new_version)

                content[idx] = f'__version__ = "{new_version}"'
                changed = True

        if not changed:
            return None

        # # Apply changes.
        # with open(file_path, 'w') as output_file:
        #     output_file.write("\n".join(content))
        #     # Add new line at the of file explicitly.
        #     output_file.write("\n")

        return new_version, old_version


def _adjust_version_in_sources(repo, labels, issue) -> typing.Optional[tuple]:
        """Walk through the directory structure and try to adjust version identifier in sources."""
        adjusted = []
        for root, _, files in os.walk('./'):
            for file_name in files:
                if file_name in ('setup.py', '__init__.py', '__about__.py', 'version.py', 'app.py', 'wsgi.py'):
                    file_path = os.path.join(root, file_name)
                    adjusted_version = _adjust_version_file(file_path, issue)
                    if adjusted_version:
                        repo.git.add(file_path)
                        adjusted.append((file_path, adjusted_version[0], adjusted_version[1]))

        if len(adjusted) == 0:
            error_msg = print("Automated version release cannot be performed. No change")

        if len(adjusted) > 1:
            error_msg = print("Automated version release cannot be performed.")

        print(f"Adjusted - {adjusted}")
        # Return old and new version identifier.
        return adjusted[0][1], adjusted[0][2]

def _compute_changelog(repo: Repo, old_version: str, new_version: str,
                           version_file: bool = False) -> typing.List[str]:
        """Compute changelog for the given repo.

        If version file is used, add changelog to the version file and add changes to git.
        """
        _LOGGER.debug("Computing changelog for new release from version %r to version %r", old_version, new_version)

        tags = repo.git.tag().splitlines()

        is_tagged_version = False
        for tag in tags:
            if old_version == tag or re.match(f"v?{old_version}", tag):
                old_version = tag
                is_tagged_version = True
                break

        if not is_tagged_version:
            _LOGGER.debug(
                "Old version was not found in the git tag history, assuming initial release"
            )
            # Use the initial commit if this the previous tag was not found - this
            # can be in case of the very first release.
            old_version = repo.git.rev_list("HEAD", max_parents=0)

        changelog = repo.git.log(f'{old_version}..HEAD', no_merges=True, format='* %s').splitlines()
        if version_file:
            # TODO: We should prepend changes instead of appending them.
            _LOGGER.info("Adding changelog to the CHANGELOG.md file")
            with open('CHANGELOG.md', 'a') as changelog_file:
                changelog_file.write(
                    f"\n## Release {new_version} ({datetime.now().replace(microsecond=0).isoformat()})\n"
                )
                changelog_file.write('\n'.join(changelog))
                changelog_file.write('\n')
            repo.git.add('CHANGELOG.md')

        print("Computed changelog has %d entries", len(changelog))
        return changelog

with cloned_repo("https://github.com", "thoth-station/storages") as repo:
    # changelog = repo.git.log('v0.23.0..HEAD', no_merges=True, format='* %s').splitlines()
    # print(changelog)
    # old_version = repo.git.rev_list("HEAD", max_parents=0)
    # print(old_version)
    # changelog = repo.git.log(f'{old_version}..HEAD', no_merges=True, format='* %s').splitlines()
    # print(changelog)
    
    tags = repo.git.tag().splitlines()
    print(tags)

    version_identifier, old_version = _adjust_version_in_sources(repo, None, None)
    print(version_identifier, old_version)
    
    changelog = _compute_changelog(
                    repo, old_version, version_identifier, version_file=True
                )

    # changelog = repo.git.log(f'{old_version}..HEAD', no_merges=True, format='* %s').splitlines()
    print(changelog)