1#@!#!123s

: / lib / python3 / dist-packages / debian /
Filename : watch.py
back
#!/usr/bin/python3
# Copyright (C) 2019-2020 Jelmer Vernooij <jelmer@debian.org>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.


"""Functions for working with watch files."""

import re
from warnings import warn

try:
    # pylint: disable=unused-import
    from typing import (
        Iterable,
        Iterator,
        List,
        Optional,
        Sequence,
        TextIO,
        Tuple,
        )
except ImportError:
    # Lack of typing is not important at runtime
    pass

# The default watch file version to use for new files.
DEFAULT_VERSION = 4

# Standard substitutions applied by uscan as documented in uscan(1):
SUBSTITUTIONS = {
    # This is substituted by the legal upstream version regex (capturing).
    '@ANY_VERSION@': r'[-_]?(\d[\-+\.:\~\da-zA-Z]*)',
    # This is substituted by the typical archive file extension regex
    # (non-capturing).
    '@ARCHIVE_EXT@': r'(?i)\.(?:tar\.xz|tar\.bz2|tar\.gz|zip|tgz|tbz|txz)',
    # This is substituted by the typical signature file extension regex
    # (non-capturing).
    '@SIGNATURE_EXT@':
        r'(?i)\.(?:tar\.xz|tar\.bz2|tar\.gz|zip|tgz|tbz|txz)'
        r'\.(?:asc|pgp|gpg|sig|sign)',
    # This is substituted by the typical Debian extension regexp (capturing).
    '@DEB_EXT@': r'[\+~](debian|dfsg|ds|deb)(\.)?(\d+)?$',
}


class MissingVersion(Exception):
    """The version= line is missing."""


class WatchFileFormatError(ValueError):
    """Raised when the input is not valid.
    """


def expand(text, package):
    # type: (str, str) -> str
    """Apply substitutions to a string.

    :param text: text to apply substitutions to
    :param package: package name, as a string
    :return: text with subsitutions applied
    """
    substs = dict(SUBSTITUTIONS.items())
    # This is substituted with the source package name found in the first line
    # of the debian/changelog file.
    substs['@PACKAGE@'] = package
    for k, v in substs.items():
        text = text.replace(k, v)
    return text


def _complain(msg, strict):
    # type: (str, bool) -> None
    if strict:
        raise WatchFileFormatError(msg)
    warn(msg)


class WatchFile(object):
    """A Debian watch file.

    :ivar entries: list of Watch entries
    :ivar options: optional list of global options, applied to all Watch
        entries
    :ivar version: watch file version
    """

    def __init__(self,
                 entries=None,              # type: Optional[Sequence[Watch]]
                 options=None,              # type: Optional[Sequence[str]]
                 version=DEFAULT_VERSION,   # type: Optional[int]
                 ):
        self.version = version
        if entries is None:
            entries = []
        self.entries = entries
        if options is None:
            options = []
        self.options = options

    def __iter__(self):
        # type: () -> Iterator[Watch]
        return iter(self.entries)

    def dump(self, f):
        # type: (TextIO) -> None
        """Write the contents of a watch file to a file-like object.

        Note that this will not preserve the formatting of the original file,
        and thus it is currently not possible to use this function to
        parse and reserialize a file and end up with the same contents.

        :param f: File-like object to write to
        """
        def serialize_options(opts):
            # type: (Sequence[str]) -> str
            s = ','.join(opts)
            if ' ' in s or '\t' in s:
                return 'opts="' + s + '"'
            return 'opts=' + s
        if self.version is not None:
            f.write('version=%d\n' % self.version)
        if self.options:
            f.write(serialize_options(self.options) + '\n')
        for entry in self.entries:
            if entry.options:
                f.write(serialize_options(entry.options) + ' ')
            f.write(entry.url)
            if entry.matching_pattern:
                f.write(' ' + entry.matching_pattern)
            if entry.version:
                f.write(' ' + entry.version)
            if entry.script:
                f.write(' ' + entry.script)
            f.write('\n')

    @classmethod
    def from_lines(cls, lines, strict=False):
        # type: (Iterable[str], bool) -> Optional[WatchFile]
        """Parse from the contents that make up a watch file.

        :param lines: watch file lines to parse
        :return: instance or None if there are no non-comment lines in the file
        :raise MissingVersion: if there is no version number declared
        :raise ValueError: when syntax errors are encountered
        """
        joined_lines = []   # type: List[List[str]]
        continued = []   # type: List[str]
        for line in lines:
            if line.startswith('#'):
                continue
            if not line.strip():
                continue
            if line.rstrip('\n').endswith('\\'):
                continued.append(line.rstrip('\n\\'))
            else:
                continued.append(line)
                joined_lines.append(continued)
                continued = []
        if continued:
            # Hmm, broken line?
            _complain('watchfile ended with \\; skipping last line', strict)
            joined_lines.append(continued)
        if not joined_lines:
            return None
        firstline = ''.join(joined_lines.pop(0))
        try:
            key, value = firstline.split('=', 1)
        except ValueError:
            raise MissingVersion()
        if key.strip() != 'version':
            raise MissingVersion()
        version = int(value.strip())
        persistent_options = []
        entries = []
        for chunked in joined_lines:
            if version > 3:
                # Leading whitespace is stripped in version
                # 4 and up.
                chunked = [chunk.lstrip() for chunk in chunked]
            line = ''.join(chunked).strip()
            if not line:
                continue
            if line.startswith('opts='):
                if line[5] == '"':
                    optend = line.index('"', 6)
                    if optend == -1:
                        raise ValueError('Not matching " in %r' % line)
                    opts_str = line[6:optend]
                    line = line[optend+1:]
                else:
                    try:
                        (opts_str, line) = line[5:].split(None, 1)
                    except ValueError:
                        opts_str = line[5:]
                        line = ''
                opts = opts_str.split(',')
            else:
                opts = []
            if line:
                try:
                    url, line = line.split(None, 1)
                except ValueError:
                    url = line
                    line = ''
                m = re.findall(r'/([^/]*\([^/]*\)[^/]*)$', url)
                if m:
                    parts = (str(m[0]), ) + tuple(line.split(None, 1))
                    url = url[:-len(m[0])-1]
                else:
                    parts = tuple(line.split(None, 2))
                entries.append(Watch(url, *parts, opts=opts))  # type: ignore
            else:
                persistent_options.extend(opts)
        return cls(
            entries=entries, options=persistent_options, version=version)


class Watch(object):
    """Watch line entry.

    This will contain the attributes documented in uscan(1):

    :ivar url: The URL (possibly including the filename regex)
    :ivar matching_pattern: a filename regex, optional
    :ivar version: version policy, optional
    :ivar script: script to run, optional
    :ivar opts: a list of options, as strings
    """

    def __init__(self,
                 url,                    # type: str
                 matching_pattern=None,  # type: Optional[str]
                 version=None,           # type: Optional[str]
                 script=None,            # type: Optional[str]
                 opts=None,              # type: Optional[Sequence[str]]
                 ):
        self.url = url
        self.matching_pattern = matching_pattern
        self.version = version
        self.script = script
        if opts is None:
            opts = []
        self.options = opts

    def __repr__(self):
        # type: () -> str
        return (
            "%s(%r, matching_pattern=%r, version=%r, script=%r, opts=%r)" % (
                self.__class__.__name__, self.url, self.matching_pattern,
                self.version, self.script, self.options))

    def __eq__(self, other):
        # type: (object) -> bool
        if not isinstance(other, Watch):
            return False
        return (other.url == self.url and
                other.matching_pattern == self.matching_pattern and
                other.version == self.version and
                other.script == self.script and
                other.options == self.options)