Source code for freedom.history

# -*- coding: utf-8 -*-
"""
:summary: All historian related activity is done from here.

:author: francis.horsman@gmail.com
"""

from datetime import datetime as dt

from .helper import Helper


class AllHistory(object):
    def __init__(self, core, lines, meta, delimiter, executor, logger=None):
        self.comments = []
        self.lines = self._strip_meta(lines, delimiter)
        self.meta = meta
        self.executor = executor
        self.logger = logger or core.monitor.get_logger('history-archive')

    def add(self, new_version, comment, format_, date=None):
        if date is None:
            date = dt.utcnow().strftime('%Y-%m-%d')
        self.comments.insert(0, self._new_comment(comment or '', format_, date,
                                                  new_version))

    @staticmethod
    def _new_comment(comment, format_, date, new_version):
        return format_ % (new_version, date, comment or '')

    @staticmethod
    def _strip_meta(lines, delimiter):
        for index, line in enumerate(lines):
            if delimiter in line:
                return lines[index + 1:]
        return lines

    def write(self, filename):
        payload = ''.join([self.meta] + self.comments + self.lines)
        self.executor.file_write(filename, payload, 'update history file ?')


class History(Helper):
    """
    All functionality relating to historian and versions.
    """
    NAME = 'history'

    def __init__(self, core, logger=None, **kwargs):
        assert core.vcs
        assert core.executor
        assert core.monitor
        Helper.__init__(self, core, logger=logger, **kwargs)
        self._current_version = None

    @property
    def current_version(self):
        if self._current_version is None:
            self._current_version = self._get_current_version_from_vcs()
        return self._current_version

    @staticmethod
    def generate_new_version(history_version_overwrite, current_version):
        """
        Increment the current version.

        :param history_version_overwrite: new version to use irrespective of
            current version.
        :param current_version: Existing 'major.minor.revision', eg: '0.1.2'
        :return: Incremented 'major.minor.revision', eg: '0.1.3'
        """
        tokens = current_version.split('.')
        if history_version_overwrite:
            tokens_overwrite = history_version_overwrite.split('.')
            if tokens_overwrite < tokens:
                raise ValueError(
                    'attempt to create new version which is <= '
                    'than the current version!')
            tokens = tokens_overwrite
        else:
            tokens[2] = str(int(tokens[2]) + 1)
        return '.'.join(tokens)

    def _get_current_version_from_vcs(self):
        current_version = self.vcs.current_tag or '0.0.1'
        self.logger.debug('Current tagged version: %s' % current_version)
        return current_version

    def commit_history(self, history_file):
        self.logger.info('Adding history file: %s to git' % history_file)
        self.vcs.add(history_file)
        self.vcs.commit(commit_all=True,
                        comment='Updating tag and history file.')
        self.vcs.push(push_all=True)

    def read_history(self, history_file):
        self.logger.debug('Reading history file: %s' % history_file)
        return open(history_file).readlines()

    def rewrite_history_from_profile(self, new_version):
        h = self.profile.history
        file_ = h.history_file
        comment = h.history_comment
        format_ = h.history_format
        date = h.history_date
        meta = h.history_meta
        delimiter = h.history_delimiter
        return self.rewrite_history(new_version, file_, format_, comment, date,
                                    meta, delimiter)

    def rewrite_history(self, new_version, history_file, format_, comment, date,
                        meta, delimiter):
        self.logger.info('Rewriting history file: %s' % new_version)
        h = AllHistory(self.core, self.read_history(history_file), meta,
                       delimiter, self.executor, logger=self.logger)
        h.add(new_version, comment, format_, date=date)
        h.write(history_file)
        self.commit_history(history_file)

    def set_current_version(self, new_version, msg='updating changelog'):
        self.logger.info('Tagging with new version: %s' % new_version)
        try:
            return self.vcs.tag(new_version, comment=msg)
        finally:
            if not self.profile.misc.dry_run:
                # Force a reload of the current_version if tagging fails:
                self._current_version = None


if __name__ == '__main__':  # pragma no cover
    pass