path: root/src/debputy/lsp/vendoring/_deb822_repro/parsing.py

# -*- coding: utf-8 -*- vim: fileencoding=utf-8 :

import collections.abc
import contextlib
import sys
import textwrap
import weakref
from abc import ABC
from types import TracebackType
from weakref import ReferenceType

from ._util import (
    combine_into_replacement,
    BufferingIterator,
    len_check_iterator,
)
from .formatter import (
    FormatterContentToken,
    one_value_per_line_trailing_separator,
    format_field,
)
from .locatable import Locatable, START_POSITION, Position, Range
from .tokens import (
    Deb822Token,
    Deb822ValueToken,
    Deb822SemanticallySignificantWhiteSpace,
    Deb822SpaceSeparatorToken,
    Deb822CommentToken,
    Deb822WhitespaceToken,
    Deb822ValueContinuationToken,
    Deb822NewlineAfterValueToken,
    Deb822CommaToken,
    Deb822FieldNameToken,
    Deb822FieldSeparatorToken,
    Deb822ErrorToken,
    tokenize_deb822_file,
    comma_split_tokenizer,
    whitespace_split_tokenizer,
)
from .types import AmbiguousDeb822FieldKeyError, SyntaxOrParseError
from debian._util import (
    resolve_ref,
    LinkedList,
    LinkedListNode,
    OrderedSet,
    _strI,
    default_field_sort_key,
)

try:
    from typing import (
        Iterable,
        Iterator,
        List,
        Union,
        Dict,
        Optional,
        Callable,
        Any,
        Generic,
        Type,
        Tuple,
        IO,
        cast,
        overload,
        Mapping,
        TYPE_CHECKING,
        Sequence,
    )
    from debian._util import T

    # for some reason, pylint does not see that Commentish is used in typing
    from .types import (  # pylint: disable=unused-import
        ST,
        VE,
        TE,
        ParagraphKey,
        TokenOrElement,
        Commentish,
        ParagraphKeyBase,
        FormatterCallback,
    )

    if TYPE_CHECKING:
        StreamingValueParser = Callable[
            [Deb822Token, BufferingIterator[Deb822Token]], VE
        ]
        StrToValueParser = Callable[[str], Iterable[Union["Deb822Token", VE]]]
        KVPNode = LinkedListNode["Deb822KeyValuePairElement"]
    else:
        StreamingValueParser = None
        StrToValueParser = None
        KVPNode = None
except ImportError:
    if not TYPE_CHECKING:
        # pylint: disable=unnecessary-lambda-assignment
        cast = lambda t, v: v
        overload = lambda f: None


class ValueReference(Generic[TE]):
    """Reference to a value inside a Deb822 paragraph

    This is useful for cases where want to modify values "in-place" or maybe
    conditionally remove a value after looking at it.

    ValueReferences can be invalidated by various changes or actions performed
    to the underlying provider of the value reference.  As an example, sorting
    a list of values will generally invalidate all ValueReferences related to
    that list.

    The ValueReference will raise validity issues where it detects them but most
    of the time it will not notice.  As a means to this end,  the ValueReference
    will *not* keep a strong reference to the underlying value.  This enables it
    to detect when the container goes out of scope.  However, keep in mind that
    the timeliness of garbage collection is implementation defined (e.g., pypy
    does not use ref-counting).
    """

    __slots__ = (
        "_node",
        "_render",
        "_value_factory",
        "_removal_handler",
        "_mutation_notifier",
    )

    def __init__(
        self,
        node,  # type: LinkedListNode[TE]
        render,  # type: Callable[[TE], str]
        value_factory,  # type: Callable[[str], TE]
        removal_handler,  # type: Callable[[LinkedListNode[TokenOrElement]], None]
        mutation_notifier,  # type: Optional[Callable[[], None]]
    ):
        self._node = weakref.ref(
            node
        )  # type: Optional[ReferenceType[LinkedListNode[TE]]]
        self._render = render
        self._value_factory = value_factory
        self._removal_handler = removal_handler
        self._mutation_notifier = mutation_notifier

    def _resolve_node(self):
        # type: () -> LinkedListNode[TE]
        # NB: We check whether the "ref" itself is None (instead of the ref resolving to None)
        # This enables us to tell the difference between "known removal" vs. "garbage collected"
        if self._node is None:
            raise RuntimeError("Cannot use ValueReference after remove()")
        node = self._node()
        if node is None:
            raise RuntimeError("ValueReference is invalid (garbage collected)")
        return node

    @property
    def value(self):
        # type: () -> str
        """Resolve the reference into a str"""
        return self._render(self._resolve_node().value)

    @value.setter
    def value(self, new_value):
        # type: (str) -> None
        """Update the reference value

        Updating the value via this method will *not* invalidate the reference (or other
        references to the same container).

        This can raise an exception if the new value does not follow the requirements
        for the referenced values.  As an example, values in whitespace separated
        lists cannot contain spaces and would trigger an exception.
        """
        self._resolve_node().value = self._value_factory(new_value)
        if self._mutation_notifier is not None:
            self._mutation_notifier()

    @property
    def locatable(self):
        # type: () -> Locatable
        """Reference to a locatable that can be used to determine where this value is"""
        return self._resolve_node().value

    def remove(self):
        # type: () -> None
        """Remove the underlying value

        This will invalidate the ValueReference (and any other ValueReferences pointing
        to that exact value).  The validity of other ValueReferences to that container
        remains unaffected.
        """
        self._removal_handler(
            cast("LinkedListNode[TokenOrElement]", self._resolve_node())
        )
        self._node = None


if sys.version_info >= (3, 9) or TYPE_CHECKING:
    _Deb822ParsedTokenList_ContextManager = contextlib.AbstractContextManager[T]
else:
    # Python 3.5 - 3.8 compat - we are not allowed to subscript the abc.Iterator
    # - use this little hack to work around it
    # Note that Python 3.5 is so old that it does not have AbstractContextManager,
    # so we re-implement it here.
    class _Deb822ParsedTokenList_ContextManager(Generic[T]):

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            return None


class Deb822ParsedTokenList(
    Generic[VE, ST],
    _Deb822ParsedTokenList_ContextManager["Deb822ParsedTokenList[VE, ST]"],
):

    def __init__(
        self,
        kvpair_element,  # type: 'Deb822KeyValuePairElement'
        interpreted_value_element,  # type: Deb822InterpretationProxyElement
        vtype,  # type: Type[VE]
        stype,  # type: Type[ST]
        str2value_parser,  # type: StrToValueParser[VE]
        default_separator_factory,  # type: Callable[[], ST]
        render,  # type: Callable[[VE], str]
    ):
        # type: (...) -> None
        self._kvpair_element = kvpair_element
        self._proxy_element = interpreted_value_element
        self._token_list = LinkedList(interpreted_value_element.parts)
        self._vtype = vtype
        self._stype = stype
        self._str2value_parser = str2value_parser
        self._default_separator_factory = default_separator_factory
        self._value_factory = _parser_to_value_factory(str2value_parser, vtype)
        self._render = render
        self._format_preserve_original_formatting = True
        self._formatter = (
            one_value_per_line_trailing_separator
        )  # type: FormatterCallback
        self._changed = False
        self.__continuation_line_char = None  # type: Optional[str]
        assert self._token_list
        last_token = self._token_list.tail

        if last_token is not None and isinstance(
            last_token, Deb822NewlineAfterValueToken
        ):
            # We always remove the last newline (if present), because then
            # adding values will happen after the last value rather than on
            # a new line by default.
            #
            # On write, we always ensure the value ends on a newline (even
            # if it did not before).  This is simpler and should be a
            # non-issue in practise.
            self._token_list.pop()

    def __iter__(self):
        # type: () -> Iterator[str]
        yield from (self._render(v) for v in self.value_parts)

    def __bool__(self):
        # type: () -> bool
        return next(iter(self), None) is not None

    def __exit__(
        self,
        exc_type,  # type: Optional[Type[BaseException]]
        exc_val,  # type: Optional[BaseException]
        exc_tb,  # type: Optional[TracebackType]
    ):
        # type: (...) -> Optional[bool]
        if exc_type is None and self._changed:
            self._update_field()
        return super().__exit__(exc_type, exc_val, exc_tb)

    @property
    def value_parts(self):
        # type: () -> Iterator[VE]
        yield from (v for v in self._token_list if isinstance(v, self._vtype))

    def _mark_changed(self):
        # type: () -> None
        self._changed = True

    def iter_value_references(self):
        # type: () -> Iterator[ValueReference[VE]]
        """Iterate over all values in the list (as ValueReferences)

        This is useful for doing inplace modification of the values or even
        streaming removal of field values.  It is in general also more
        efficient when more than one value is updated or removed.
        """
        yield from (
            ValueReference(
                cast("LinkedListNode[VE]", n),
                self._render,
                self._value_factory,
                self._remove_node,
                self._mark_changed,
            )
            for n in self._token_list.iter_nodes()
            if isinstance(n.value, self._vtype)
        )

    def append_separator(self, space_after_separator=True):
        # type: (bool) -> None

        separator_token = self._default_separator_factory()
        if separator_token.is_whitespace:
            space_after_separator = False

        self._changed = True
        self._append_continuation_line_token_if_necessary()
        self._token_list.append(separator_token)

        if space_after_separator and not separator_token.is_whitespace:
            self._token_list.append(Deb822WhitespaceToken(" "))

    def replace(self, orig_value, new_value):
        # type: (str, str) -> None
        """Replace the first instance of a value with another

        This method will *not* affect the validity of ValueReferences.
        """
        vtype = self._vtype
        for node in self._token_list.iter_nodes():
            if isinstance(node.value, vtype) and self._render(node.value) == orig_value:
                node.value = self._value_factory(new_value)
                self._changed = True
                break
        else:
            raise ValueError("list.replace(x, y): x not in list")

    def remove(self, value):
        # type: (str) -> None
        """Remove the first instance of a value

        Removal will invalidate ValueReferences to the value being removed.
        ValueReferences to other values will be unaffected.
        """
        vtype = self._vtype
        for node in self._token_list.iter_nodes():
            if isinstance(node.value, vtype) and self._render(node.value) == value:
                node_to_remove = node
                break
        else:
            raise ValueError("list.remove(x): x not in list")

        return self._remove_node(node_to_remove)

    def _remove_node(self, node_to_remove):
        # type: (LinkedListNode[TokenOrElement]) -> None
        vtype = self._vtype
        self._changed = True

        # We naively want to remove the node and every thing to the left of it
        # until the previous value.  That is the basic idea for now (ignoring
        # special-cases for now).
        #
        # Example:
        #
        # """
        # Multiline-Keywords: bar[
        # # Comment about foo
        #                     foo]
        #                     baz
        # Keywords: bar[ foo] baz
        # Comma-List: bar[, foo], baz,
        # Multiline-Comma-List: bar[,
        # # Comment about foo
        #                       foo],
        #                       baz,
        # """
        #
        # Assuming we want to remove "foo" for the lists, the []-markers
        # show what we aim to remove.  This has the nice side-effect of
        # preserving whether nor not the value has a trailing separator.
        # Note that we do *not* attempt to repair missing separators but
        # it may fix duplicated separators by "accident".
        #
        # Now, there are two special cases to be aware of, where this approach
        # has short comings:
        #
        # 1) If foo is the only value (in which case, "delete everything"
        #    is the only option).
        # 2) If foo is the first value
        # 3) If foo is not the only value on the line and we see a comment
        #    inside the deletion range.
        #
        # For 2) + 3), we attempt to flip and range to delete and every
        # thing after it (up to but exclusion "baz") instead.  This
        # definitely fixes 3), but 2) has yet another corner case, namely:
        #
        # """
        # Multiline-Comma-List: foo,
        # # Remark about bar
        #                       bar,
        # Another-Case: foo
        # # Remark, also we use leading separator
        #             , bar
        # """
        #
        # The options include:
        #
        #  A) Discard the comment - brain-dead simple
        #  B) Hoist the comment up to a field comment, but then what if the
        #     field already has a comment?
        #  C) Clear the first value line leaving just the newline and
        #     replace the separator before "bar" (if present) with a space.
        #     (leaving you with the value of the form "\n# ...\n      bar")
        #

        first_value_on_lhs = None  # type: Optional[LinkedListNode[TokenOrElement]]
        first_value_on_rhs = None  # type: Optional[LinkedListNode[TokenOrElement]]
        comment_before_previous_value = False
        comment_before_next_value = False
        for past_node in node_to_remove.iter_previous(skip_current=True):
            past_token = past_node.value
            if isinstance(past_token, Deb822Token) and past_token.is_comment:
                comment_before_previous_value = True
                continue
            if isinstance(past_token, vtype):
                first_value_on_lhs = past_node
                break

        for future_node in node_to_remove.iter_next(skip_current=True):
            future_token = future_node.value
            if isinstance(future_token, Deb822Token) and future_token.is_comment:
                comment_before_next_value = True
                continue
            if isinstance(future_token, vtype):
                first_value_on_rhs = future_node
                break

        if first_value_on_rhs is None and first_value_on_lhs is None:
            # This was the last value, just remove everything.
            self._token_list.clear()
            return

        if first_value_on_lhs is not None and not comment_before_previous_value:
            # Delete left
            delete_lhs_of_node = True
        elif first_value_on_rhs is not None and not comment_before_next_value:
            # Delete right
            delete_lhs_of_node = False
        else:
            # There is a comment on either side (or no value on one and a
            # comment and the other). Keep it simple, we just delete to
            # one side (preferring deleting to left if possible).
            delete_lhs_of_node = first_value_on_lhs is not None

        if delete_lhs_of_node:
            first_remain_lhs = first_value_on_lhs
            first_remain_rhs = node_to_remove.next_node
        else:
            first_remain_lhs = node_to_remove.previous_node
            first_remain_rhs = first_value_on_rhs

        # Actual deletion - with some manual labour to update HEAD/TAIL of
        # the list in case we do a "delete everything left/right this node".
        if first_remain_lhs is None:
            self._token_list.head_node = first_remain_rhs
        if first_remain_rhs is None:
            self._token_list.tail_node = first_remain_lhs
        LinkedListNode.link_nodes(first_remain_lhs, first_remain_rhs)

    def append(self, value):
        # type: (str) -> None
        vt = self._value_factory(value)
        self.append_value(vt)

    def append_value(self, vt):
        # type: (VE) -> None
        value_parts = self._token_list
        if value_parts:
            needs_separator = False
            stype = self._stype
            vtype = self._vtype
            for t in reversed(value_parts):
                if isinstance(t, vtype):
                    needs_separator = True
                    break
                if isinstance(t, stype):
                    break

            if needs_separator:
                self.append_separator()
        else:
            # Looks nicer if there is a space before the very first value
            self._token_list.append(Deb822WhitespaceToken(" "))
        self._append_continuation_line_token_if_necessary()
        self._changed = True
        value_parts.append(vt)

    def _previous_is_newline(self):
        # type: () -> bool
        tail = self._token_list.tail
        return tail is not None and tail.convert_to_text().endswith("\n")

    def append_newline(self):
        # type: () -> None
        if self._previous_is_newline():
            raise ValueError(
                "Cannot add a newline after a token that ends on a newline"
            )
        self._token_list.append(Deb822NewlineAfterValueToken())

    def append_comment(self, comment_text):
        # type: (str) -> None
        tail = self._token_list.tail
        if tail is None or not tail.convert_to_text().endswith("\n"):
            self.append_newline()
        comment_token = Deb822CommentToken(_format_comment(comment_text))
        self._token_list.append(comment_token)

    @property
    def _continuation_line_char(self):
        # type: () -> str
        char = self.__continuation_line_char
        if char is None:
            # Use ' ' by default but match the existing field if possible.
            char = " "
            for token in self._token_list:
                if isinstance(token, Deb822ValueContinuationToken):
                    char = token.text
                    break
            self.__continuation_line_char = char
        return char

    def _append_continuation_line_token_if_necessary(self):
        # type: () -> None
        tail = self._token_list.tail
        if tail is not None and tail.convert_to_text().endswith("\n"):
            self._token_list.append(
                Deb822ValueContinuationToken(self._continuation_line_char)
            )

    def reformat_when_finished(self):
        # type: () -> None
        self._enable_reformatting()
        self._changed = True

    def _enable_reformatting(self):
        # type: () -> None
        self._format_preserve_original_formatting = False

    def no_reformatting_when_finished(self):
        # type: () -> None
        self._format_preserve_original_formatting = True

    def value_formatter(
        self,
        formatter,  # type: FormatterCallback
        force_reformat=False,  # type: bool
    ):
        # type: (...) -> None
        """Use a custom formatter when formatting the value

        :param formatter: A formatter (see debian._deb822_repro.formatter.format_field
          for details)
        :param force_reformat: If True, always reformat the field even if there are
          no (other) changes performed.  By default, fields are only reformatted if
          they are changed.
        """
        self._formatter = formatter
        self._format_preserve_original_formatting = False
        if force_reformat:
            self._changed = True

    def clear(self):
        # type: () -> None
        """Like list.clear() - removes all content (including comments and spaces)"""
        if self._token_list:
            self._changed = True
        self._token_list.clear()

    def _iter_content_as_tokens(self):
        # type: () -> Iterable[Deb822Token]
        for te in self._token_list:
            if isinstance(te, Deb822Element):
                yield from te.iter_tokens()
            else:
                yield te

    def _generate_reformatted_field_content(self):
        # type: () -> str
        separator_token = self._default_separator_factory()
        vtype = self._vtype
        stype = self._stype
        token_list = self._token_list

        def _token_iter():
            # type: () -> Iterator[FormatterContentToken]
            text = ""  # type: str
            for te in token_list:
                if isinstance(te, Deb822Token):
                    if te.is_comment:
                        yield FormatterContentToken.comment_token(te.text)
                    elif isinstance(te, stype):
                        text = te.text
                        yield FormatterContentToken.separator_token(text)
                else:
                    assert isinstance(te, vtype)
                    text = te.convert_to_text()
                    yield FormatterContentToken.value_token(text)

        return format_field(
            self._formatter,
            self._kvpair_element.field_name,
            FormatterContentToken.separator_token(separator_token.text),
            _token_iter(),
        )

    def _generate_field_content(self):
        # type: () -> str
        return "".join(t.text for t in self._iter_content_as_tokens())

    def _update_field(self):
        # type: () -> None
        kvpair_element = self._kvpair_element
        field_name = kvpair_element.field_name
        token_list = self._token_list
        tail = token_list.tail
        had_tokens = False

        for t in self._iter_content_as_tokens():
            had_tokens = True
            if not t.is_comment and not t.is_whitespace:
                break
        else:
            if had_tokens:
                raise ValueError(
                    "Field must be completely empty or have content "
                    "(i.e. non-whitespace and non-comments)"
                )
        if tail is not None:
            if isinstance(tail, Deb822Token) and tail.is_comment:
                raise ValueError("Fields must not end on a comment")
            if not tail.convert_to_text().endswith("\n"):
                # Always end on a newline
                self.append_newline()

            if self._format_preserve_original_formatting:
                value_text = self._generate_field_content()
                text = ":".join((field_name, value_text))
            else:
                text = self._generate_reformatted_field_content()

            new_content = text.splitlines(keepends=True)
        else:
            # Special-case for the empty list which will be mapped to
            # an empty field.  Always end on a newline (avoids errors
            # if there is a field after this)
            new_content = [field_name + ":\n"]

        # As absurd as it might seem, it is easier to just use the parser to
        # construct the AST correctly
        deb822_file = parse_deb822_file(iter(new_content))
        error_token = deb822_file.find_first_error_element()
        if error_token:
            # _print_ast(deb822_file)
            raise ValueError("Syntax error in new field value for " + field_name)
        paragraph = next(iter(deb822_file))
        assert isinstance(paragraph, Deb822NoDuplicateFieldsParagraphElement)
        new_kvpair_element = paragraph.get_kvpair_element(field_name)
        assert new_kvpair_element is not None
        kvpair_element.value_element = new_kvpair_element.value_element
        self._changed = False

    def sort_elements(
        self,
        *,
        key=None,  # type: Optional[Callable[[VE], Any]]
        reverse=False,  # type: bool
    ):
        # type: (...) -> None
        """Sort the elements (abstract values) in this list.

        This method will sort the logical values of the list. It will
        attempt to preserve comments associated with a given value where
        possible.  Whether space and separators are preserved depends on
        the contents of the field as well as the formatting settings.

        Sorting (without reformatting) is likely to leave you with "awkward"
        whitespace. Therefore, you almost always want to apply reformatting
        such as the reformat_when_finished() method.

        Sorting will invalidate all ValueReferences.
        """
        comment_start_node = None
        vtype = self._vtype
        stype = self._stype

        def key_func(x):
            # type: (Tuple[VE, List[TokenOrElement]]) -> Any
            if key:
                return key(x[0])
            return x[0].convert_to_text()

        parts = []

        for node in self._token_list.iter_nodes():
            value = node.value
            if isinstance(value, Deb822Token) and value.is_comment:
                if comment_start_node is None:
                    comment_start_node = node
                continue

            if isinstance(value, vtype):
                comments = []
                if comment_start_node is not None:
                    for keep_node in comment_start_node.iter_next(skip_current=False):
                        if keep_node is node:
                            break
                        comments.append(keep_node.value)
                parts.append((value, comments))
                comment_start_node = None

        parts.sort(key=key_func, reverse=reverse)

        self._changed = True
        self._token_list.clear()
        first_value = True

        separator_is_space = self._default_separator_factory().is_whitespace

        for value, comments in parts:
            if first_value:
                first_value = False
                if comments:
                    # While unlikely, there could be a separator between the comments.
                    # It would be in the way and we remove it.
                    comments = [x for x in comments if not isinstance(x, stype)]
                    # Comments cannot start the field, so inject a newline to
                    # work around that
                    self.append_newline()
            else:
                if not separator_is_space and not any(
                    isinstance(x, stype) for x in comments
                ):
                    # While unlikely, you can hide a comma between two comments and expect
                    # us to preserve it.  However, the more common case is that the separator
                    # appeared before the comments and was thus omitted (leaving us to re-add
                    # it here).
                    self.append_separator(space_after_separator=False)
                if comments:
                    self.append_newline()
                else:
                    self._token_list.append(Deb822WhitespaceToken(" "))

            self._token_list.extend(comments)
            self.append_value(value)

    def sort(
        self,
        *,
        key=None,  # type: Optional[Callable[[str], Any]]
        **kwargs,  # type: Any
    ):
        # type: (...) -> None
        """Sort the values (rendered as str) in this list.

        This method will sort the logical values of the list. It will
        attempt to preserve comments associated with a given value where
        possible.  Whether space and separators are preserved depends on
        the contents of the field as well as the formatting settings.

        Sorting (without reformatting) is likely to leave you with "awkward"
        whitespace. Therefore, you almost always want to apply reformatting
        such as the reformat_when_finished() method.

        Sorting will invalidate all ValueReferences.
        """
        if key is not None:
            render = self._render
            kwargs["key"] = lambda vt: key(render(vt))
        self.sort_elements(**kwargs)


class Interpretation(Generic[T]):

    def interpret(
        self,
        kvpair_element,  # type: Deb822KeyValuePairElement
        discard_comments_on_read=True,  # type: bool
    ):
        # type: (...) -> T
        raise NotImplementedError  # pragma: no cover


class GenericContentBasedInterpretation(Interpretation[T], Generic[T, VE]):

    def __init__(
        self,
        tokenizer,  # type: Callable[[str], Iterable['Deb822Token']]
        value_parser,  # type: StreamingValueParser[VE]
    ):
        # type: (...) -> None
        super().__init__()
        self._tokenizer = tokenizer
        self._value_parser = value_parser

    def _high_level_interpretation(
        self,
        kvpair_element,  # type: Deb822KeyValuePairElement
        proxy_element,  # type: Deb822InterpretationProxyElement
        discard_comments_on_read=True,  # type: bool
    ):
        # type: (...) -> T
        raise NotImplementedError  # pragma: no cover

    def _parse_stream(
        self, buffered_iterator  # type: BufferingIterator[Deb822Token]
    ):
        # type: (...) -> Iterable[Union[Deb822Token, VE]]

        value_parser = self._value_parser
        for token in buffered_iterator:
            if isinstance(token, Deb822ValueToken):
                yield value_parser(token, buffered_iterator)
            else:
                yield token

    def _parse_kvpair(
        self, kvpair  # type: Deb822KeyValuePairElement
    ):
        # type: (...) -> Deb822InterpretationProxyElement
        value_element = kvpair.value_element
        content = value_element.convert_to_text()
        token_list = []  # type: List['TokenOrElement']
        token_list.extend(self._parse_str(content))
        return Deb822InterpretationProxyElement(value_element, token_list)

    def _parse_str(self, content):
        # type: (str) -> Iterable[Union[Deb822Token, VE]]
        content_len = len(content)
        biter = BufferingIterator(
            len_check_iterator(
                content,
                self._tokenizer(content),
                content_len=content_len,
            )
        )
        yield from len_check_iterator(
            content,
            self._parse_stream(biter),
            content_len=content_len,
        )

    def interpret(
        self,
        kvpair_element,  # type: Deb822KeyValuePairElement
        discard_comments_on_read=True,  # type: bool
    ):
        # type: (...) -> T
        proxy_element = self._parse_kvpair(kvpair_element)
        return self._high_level_interpretation(
            kvpair_element,
            proxy_element,
            discard_comments_on_read=discard_comments_on_read,
        )


def _parser_to_value_factory(
    parser,  # type: StrToValueParser[VE]
    vtype,  # type: Type[VE]
):
    # type: (...) -> Callable[[str], VE]
    def _value_factory(v):
        # type: (str) -> VE
        if v == "":
            raise ValueError("The empty string is not a value")
        token_iter = iter(parser(v))
        t1 = next(token_iter, None)  # type: Optional[Union[TokenOrElement]]
        t2 = next(token_iter, None)
        assert t1 is not None, (
            'Bad parser - it returned None (or no TE) for "' + v + '"'
        )
        if t2 is not None:
            msg = textwrap.dedent(
                """\
            The input "{v}" should have been exactly one element, but the parser provided at
             least two.  This can happen with unnecessary leading/trailing whitespace
             or including commas the value for a comma list.
            """
            ).format(v=v)
            raise ValueError(msg)
        if not isinstance(t1, vtype):
            if isinstance(t1, Deb822Token) and (t1.is_comment or t1.is_whitespace):
                raise ValueError(
                    'The input "{v}" is whitespace or a comment: Expected a value'
                )
            msg = (
                'The input "{v}" should have produced a element of type {vtype_name}, but'
                " instead it produced {t1}"
            )
            raise ValueError(msg.format(v=v, vtype_name=vtype.__name__, t1=t1))

        assert len(t1.convert_to_text()) == len(v), (
            "Bad tokenizer - the token did not cover the input text"
            " exactly ({t1_len} != {v_len}".format(
                t1_len=len(t1.convert_to_text()), v_len=len(v)
            )
        )
        return t1

    return _value_factory


class ListInterpretation(
    GenericContentBasedInterpretation[Deb822ParsedTokenList[VE, ST], VE]
):

    def __init__(
        self,
        tokenizer,  # type: Callable[[str], Iterable['Deb822Token']]
        value_parser,  # type: StreamingValueParser[VE]
        vtype,  # type: Type[VE]
        stype,  # type: Type[ST]
        default_separator_factory,  # type: Callable[[], ST]
        render_factory,  # type: Callable[[bool], Callable[[VE], str]]
    ):
        # type: (...) -> None
        super().__init__(tokenizer, value_parser)
        self._vtype = vtype
        self._stype = stype
        self._default_separator_factory = default_separator_factory
        self._render_factory = render_factory

    def _high_level_interpretation(
        self,
        kvpair_element,  # type: Deb822KeyValuePairElement
        proxy_element,  # type: Deb822InterpretationProxyElement
        discard_comments_on_read=True,  # type: bool
    ):
        # type: (...) -> Deb822ParsedTokenList[VE, ST]
        return Deb822ParsedTokenList(
            kvpair_element,
            proxy_element,
            self._vtype,
            self._stype,
            self._parse_str,
            self._default_separator_factory,
            self._render_factory(discard_comments_on_read),
        )


def _parse_whitespace_list_value(token, _):
    # type: (Deb822Token, BufferingIterator[Deb822Token]) -> Deb822ParsedValueElement
    return Deb822ParsedValueElement([token])


def _is_comma_token(v):
    # type: (TokenOrElement) -> bool
    # Consume tokens until the next comma
    return isinstance(v, Deb822CommaToken)


def _parse_comma_list_value(token, buffered_iterator):
    # type: (Deb822Token, BufferingIterator[Deb822Token]) -> Deb822ParsedValueElement
    comma_offset = buffered_iterator.peek_find(_is_comma_token)
    value_parts = [token]
    if comma_offset is not None:
        # The value is followed by a comma and now we know where it ends
        value_parts.extend(buffered_iterator.peek_many(comma_offset - 1))
    else:
        # The value is the last value there is.  Consume all remaining tokens
        # and then trim from the right.
        value_parts.extend(buffered_iterator.peek_buffer())
    while value_parts and not isinstance(value_parts[-1], Deb822ValueToken):
        value_parts.pop()

    buffered_iterator.consume_many(len(value_parts) - 1)
    return Deb822ParsedValueElement(value_parts)


def _parse_uploaders_list_value(token, buffered_iterator):
    # type: (Deb822Token, BufferingIterator[Deb822Token]) -> Deb822ParsedValueElement

    # This is similar to _parse_comma_list_value *except* that there is an extra special
    # case.  Namely comma only counts as a true separator if it follows ">"
    value_parts = [token]
    comma_offset = -1  # type: Optional[int]
    while comma_offset is not None:
        comma_offset = buffered_iterator.peek_find(_is_comma_token)
        if comma_offset is not None:
            # The value is followed by a comma.  Verify that this is a terminating
            # comma (comma may appear in the name or email)
            #
            # We include value_parts[-1] to easily cope with the common case of
            # "foo <a@b.com>," where we will have 0 peeked element to examine.
            peeked_elements = [value_parts[-1]]
            peeked_elements.extend(buffered_iterator.peek_many(comma_offset - 1))
            comma_was_separator = False
            i = len(peeked_elements) - 1
            while i >= 0:
                token = peeked_elements[i]
                if isinstance(token, Deb822ValueToken):
                    if token.text.endswith(">"):
                        # The comma terminates the value
                        value_parts.extend(buffered_iterator.consume_many(i))
                        assert isinstance(
                            value_parts[-1], Deb822ValueToken
                        ) and value_parts[-1].text.endswith(">"), "Got: " + str(
                            value_parts
                        )
                        comma_was_separator = True
                    break
                i -= 1
            if comma_was_separator:
                break
            value_parts.extend(buffered_iterator.consume_many(comma_offset))
            assert isinstance(value_parts[-1], Deb822CommaToken)
        else:
            # The value is the last value there is.  Consume all remaining tokens
            # and then trim from the right.
            remaining_part = buffered_iterator.peek_buffer()
            consume_elements = len(remaining_part)
            value_parts.extend(remaining_part)
            while value_parts and not isinstance(value_parts[-1], Deb822ValueToken):
                value_parts.pop()
                consume_elements -= 1
            buffered_iterator.consume_many(consume_elements)

    return Deb822ParsedValueElement(value_parts)


class Deb822Element(Locatable):
    """Composite elements (consists of 1 or more tokens)"""

    __slots__ = ("_parent_element", "_full_size_cache", "__weakref__")

    def __init__(self):
        # type: () -> None
        self._parent_element = None  # type: Optional[ReferenceType['Deb822Element']]
        self._full_size_cache = None  # type: Optional[Range]

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        raise NotImplementedError  # pragma: no cover

    def iter_parts_of_type(self, only_element_or_token_type):
        # type: (Type[TE]) -> Iterable[TE]
        for part in self.iter_parts():
            if isinstance(part, only_element_or_token_type):
                yield part

    def iter_tokens(self):
        # type: () -> Iterable[Deb822Token]
        for part in self.iter_parts():
            # Control check to catch bugs early
            assert part._parent_element is not None
            if isinstance(part, Deb822Element):
                yield from part.iter_tokens()
            else:
                yield part

    def iter_recurse(
        self, *, only_element_or_token_type=None  # type: Optional[Type[TE]]
    ):
        # type: (...) -> Iterable[TE]
        for part in self.iter_parts():
            if only_element_or_token_type is None or isinstance(
                part, only_element_or_token_type
            ):
                yield cast("TE", part)
            if isinstance(part, Deb822Element):
                yield from part.iter_recurse(
                    only_element_or_token_type=only_element_or_token_type
                )

    @property
    def is_error(self):
        # type: () -> bool
        return False

    @property
    def is_comment(self):
        # type: () -> bool
        return False

    @property
    def parent_element(self):
        # type: () -> Optional[Deb822Element]
        return resolve_ref(self._parent_element)

    @parent_element.setter
    def parent_element(self, new_parent):
        # type: (Optional[Deb822Element]) -> None
        self._parent_element = (
            weakref.ref(new_parent) if new_parent is not None else None
        )

    def _init_parent_of_parts(self):
        # type: () -> None
        for part in self.iter_parts():
            part.parent_element = self

    # Deliberately not a "text" property, to signal that it is not necessary cheap.
    def convert_to_text(self):
        # type: () -> str
        return "".join(t.text for t in self.iter_tokens())

    def clear_parent_if_parent(self, parent):
        # type: (Deb822Element) -> None
        if parent is self.parent_element:
            self._parent_element = None

    def size(self, *, skip_leading_comments: bool = True) -> Range:
        size_cache = self._full_size_cache
        if size_cache is None:
            size_cache = Range.from_position_and_sizes(
                START_POSITION,
                (p.size(skip_leading_comments=False) for p in self.iter_parts()),
            )
            self._full_size_cache = size_cache
        return size_cache


class Deb822InterpretationProxyElement(Deb822Element):

    __slots__ = ("parts",)

    def __init__(
        self, real_element: Deb822Element, parts: List[TokenOrElement]
    ) -> None:
        super().__init__()
        self.parent_element = real_element
        self.parts = parts
        for p in parts:
            p.parent_element = self

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        return iter(self.parts)

    def position_in_parent(self, *, skip_leading_comments: bool = True) -> Position:
        parent = self.parent_element
        if parent is None:
            raise RuntimeError("parent was garbage collected")
        return parent.position_in_parent()

    def position_in_file(self, *, skip_leading_comments: bool = True) -> Position:
        parent = self.parent_element
        if parent is None:
            raise RuntimeError("parent was garbage collected")
        return parent.position_in_file()

    def size(self, *, skip_leading_comments: bool = True) -> Range:
        # Same as parent except we never use a cache.
        sizes = (p.size(skip_leading_comments=False) for p in self.iter_parts())
        return Range.from_position_and_sizes(START_POSITION, sizes)


class Deb822ErrorElement(Deb822Element):
    """Element representing elements or tokens that are out of place

    Commonly, it will just be instances of Deb822ErrorToken, but it can be other
    things.  As an example if a parser discovers out of order elements/tokens,
    it can bundle them in a Deb822ErrorElement to signal that the sequence of
    elements/tokens are invalid (even if the tokens themselves are valid).
    """

    __slots__ = ("_parts",)

    def __init__(self, parts):
        # type: (Sequence[TokenOrElement]) -> None
        super().__init__()
        self._parts = tuple(parts)
        self._init_parent_of_parts()

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        yield from self._parts

    @property
    def is_error(self):
        # type: () -> bool
        return True


class Deb822ValueLineElement(Deb822Element):
    """Consists of one "line" of a value"""

    __slots__ = (
        "_comment_element",
        "_continuation_line_token",
        "_leading_whitespace_token",
        "_value_tokens",
        "_trailing_whitespace_token",
        "_newline_token",
    )

    def __init__(
        self,
        comment_element,  # type: Optional[Deb822CommentElement]
        continuation_line_token,  # type: Optional[Deb822ValueContinuationToken]
        leading_whitespace_token,  # type: Optional[Deb822WhitespaceToken]
        value_parts,  # type: List[TokenOrElement]
        trailing_whitespace_token,  # type: Optional[Deb822WhitespaceToken]
        # only optional if it is the last line of the file and the file does not
        # end with a newline.
        newline_token,  # type: Optional[Deb822WhitespaceToken]
    ):
        # type: (...) -> None
        super().__init__()
        if comment_element is not None and continuation_line_token is None:
            raise ValueError("Only continuation lines can have comments")
        self._comment_element = comment_element  # type: Optional[Deb822CommentElement]
        self._continuation_line_token = continuation_line_token
        self._leading_whitespace_token = (
            leading_whitespace_token
        )  # type: Optional[Deb822WhitespaceToken]
        self._value_tokens = value_parts  # type: List[TokenOrElement]
        self._trailing_whitespace_token = trailing_whitespace_token
        self._newline_token = newline_token  # type: Optional[Deb822WhitespaceToken]
        self._init_parent_of_parts()

    @property
    def comment_element(self):
        # type: () -> Optional[Deb822CommentElement]
        return self._comment_element

    @property
    def continuation_line_token(self):
        # type: () -> Optional[Deb822ValueContinuationToken]
        return self._continuation_line_token

    @property
    def newline_token(self):
        # type: () -> Optional[Deb822WhitespaceToken]
        return self._newline_token

    def add_newline_if_missing(self):
        # type: () -> bool
        if self._newline_token is None:
            self._newline_token = Deb822NewlineAfterValueToken()
            self._newline_token.parent_element = self
            self._full_size_cache = None
            return True
        return False

    def _iter_content_parts(self):
        # type: () -> Iterable[TokenOrElement]
        if self._leading_whitespace_token:
            yield self._leading_whitespace_token
        yield from self._value_tokens
        if self._trailing_whitespace_token:
            yield self._trailing_whitespace_token

    def _iter_content_tokens(self):
        # type: () -> Iterable[Deb822Token]
        for part in self._iter_content_parts():
            if isinstance(part, Deb822Element):
                yield from part.iter_tokens()
            else:
                yield part

    def convert_content_to_text(self):
        # type: () -> str
        if (
            len(self._value_tokens) == 1
            and not self._leading_whitespace_token
            and not self._trailing_whitespace_token
            and isinstance(self._value_tokens[0], Deb822Token)
        ):
            # By default, we get a single value spanning the entire line
            # (minus continuation line and newline, but we are supposed to
            # exclude those)
            return self._value_tokens[0].text

        return "".join(t.text for t in self._iter_content_tokens())

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        if self._comment_element:
            yield self._comment_element
        if self._continuation_line_token:
            yield self._continuation_line_token
        yield from self._iter_content_parts()
        if self._newline_token:
            yield self._newline_token

    def size(self, *, skip_leading_comments: bool = True) -> Range:
        if skip_leading_comments:
            return Range.from_position_and_sizes(
                START_POSITION,
                (
                    p.size(skip_leading_comments=False)
                    for p in self.iter_parts()
                    if not p.is_comment
                ),
            )
        return super().size(skip_leading_comments=skip_leading_comments)

    def position_in_parent(self, *, skip_leading_comments: bool = True) -> Position:
        base_pos = super().position_in_parent(skip_leading_comments=False)
        if skip_leading_comments:
            for p in self.iter_parts():
                if p.is_comment:
                    continue
                non_comment_pos = p.position_in_parent(skip_leading_comments=False)
                base_pos = non_comment_pos.relative_to(base_pos)
        return base_pos


class Deb822ValueElement(Deb822Element):
    __slots__ = ("_value_entry_elements",)

    def __init__(self, value_entry_elements):
        # type: (Sequence[Deb822ValueLineElement]) -> None
        super().__init__()
        # Split over two lines due to line length issues
        v = tuple(value_entry_elements)
        self._value_entry_elements = v  # type: Sequence[Deb822ValueLineElement]
        self._init_parent_of_parts()

    @property
    def value_lines(self):
        # type: () -> Sequence[Deb822ValueLineElement]
        """Read-only list of value entries"""
        return self._value_entry_elements

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        yield from self._value_entry_elements

    def add_final_newline_if_missing(self):
        # type: () -> bool
        if self._value_entry_elements:
            changed = self._value_entry_elements[-1].add_newline_if_missing()
            if changed:
                self._full_size_cache = None
            return changed
        return False


class Deb822ParsedValueElement(Deb822Element):

    __slots__ = ("_text_cached", "_text_no_comments_cached", "_token_list")

    def __init__(self, tokens):
        # type: (List[Deb822Token]) -> None
        super().__init__()
        self._token_list = tokens
        self._init_parent_of_parts()
        if not isinstance(tokens[0], Deb822ValueToken) or not isinstance(
            tokens[-1], Deb822ValueToken
        ):
            raise ValueError(
                self.__class__.__name__ + " MUST start and end on a Deb822ValueToken"
            )
        if len(tokens) == 1:
            token = tokens[0]
            self._text_cached = token.text  # type: Optional[str]
            self._text_no_comments_cached = token.text  # type: Optional[str]
        else:
            self._text_cached = None
            self._text_no_comments_cached = None

    def convert_to_text(self):
        # type: () -> str
        if self._text_no_comments_cached is None:
            self._text_no_comments_cached = super().convert_to_text()
        return self._text_no_comments_cached

    def convert_to_text_without_comments(self):
        # type: () -> str
        if self._text_no_comments_cached is None:
            self._text_no_comments_cached = "".join(
                t.text for t in self.iter_tokens() if not t.is_comment
            )
        return self._text_no_comments_cached

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        yield from self._token_list


class Deb822CommentElement(Deb822Element):
    __slots__ = ("_comment_tokens",)

    def __init__(self, comment_tokens):
        # type: (Sequence[Deb822CommentToken]) -> None
        super().__init__()
        self._comment_tokens = tuple(
            comment_tokens
        )  # type: Sequence[Deb822CommentToken]
        if not comment_tokens:  # pragma: no cover
            raise ValueError("Comment elements must have at least one comment token")
        self._init_parent_of_parts()

    @property
    def is_comment(self):
        # type: () -> bool
        return True

    def __len__(self):
        # type: () -> int
        return len(self._comment_tokens)

    def __getitem__(self, item):
        # type: (int) -> Deb822CommentToken
        return self._comment_tokens[item]

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        yield from self._comment_tokens


class Deb822KeyValuePairElement(Deb822Element):
    __slots__ = (
        "_comment_element",
        "_field_token",
        "_separator_token",
        "_value_element",
    )

    def __init__(
        self,
        comment_element,  # type: Optional[Deb822CommentElement]
        field_token,  # type: Deb822FieldNameToken
        separator_token,  # type: Deb822FieldSeparatorToken
        value_element,  # type: Deb822ValueElement
    ):
        # type: (...) -> None
        super().__init__()
        self._comment_element = comment_element  # type: Optional[Deb822CommentElement]
        self._field_token = field_token  # type: Deb822FieldNameToken
        self._separator_token = separator_token  # type: Deb822FieldSeparatorToken
        self._value_element = value_element  # type: Deb822ValueElement
        self._init_parent_of_parts()

    @property
    def field_name(self):
        # type: () -> _strI
        return self.field_token.text

    @property
    def field_token(self):
        # type: () -> Deb822FieldNameToken
        return self._field_token

    @property
    def value_element(self):
        # type: () -> Deb822ValueElement
        return self._value_element

    @value_element.setter
    def value_element(self, new_value):
        # type: (Deb822ValueElement) -> None
        self._full_size_cache = None
        self._value_element.clear_parent_if_parent(self)
        self._value_element = new_value
        new_value.parent_element = self

    def interpret_as(
        self,
        interpreter,  # type: Interpretation[T]
        discard_comments_on_read=True,  # type: bool
    ):
        # type: (...) -> T
        return interpreter.interpret(
            self, discard_comments_on_read=discard_comments_on_read
        )

    @property
    def comment_element(self):
        # type: () -> Optional[Deb822CommentElement]
        return self._comment_element

    @comment_element.setter
    def comment_element(self, value):
        # type: (Optional[Deb822CommentElement]) -> None
        self._full_size_cache = None
        if value is not None:
            if not value[-1].text.endswith("\n"):
                raise ValueError("Field comments must end with a newline")
        if self._comment_element:
            self._comment_element.clear_parent_if_parent(self)
        if value is not None:
            value.parent_element = self
        self._comment_element = value

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        if self._comment_element:
            yield self._comment_element
        yield self._field_token
        yield self._separator_token
        yield self._value_element

    def position_in_parent(
        self,
        *,
        skip_leading_comments: bool = True,
    ) -> Position:
        position = super().position_in_parent(skip_leading_comments=False)
        if skip_leading_comments:
            if self._comment_element:
                field_pos = self._field_token.position_in_parent()
                position = field_pos.relative_to(position)
        return position

    def size(self, *, skip_leading_comments: bool = True) -> Range:
        if skip_leading_comments:
            return Range.from_position_and_sizes(
                START_POSITION,
                (
                    p.size(skip_leading_comments=False)
                    for p in self.iter_parts()
                    if not p.is_comment
                ),
            )
        return super().size(skip_leading_comments=False)


def _format_comment(c):
    # type: (str) -> str
    if c == "":
        # Special-case: Empty strings are mapped to an empty comment line
        return "#\n"
    if "\n" in c[:-1]:
        raise ValueError("Comment lines must not have embedded newlines")
    if not c.endswith("\n"):
        c = c.rstrip() + "\n"
    if not c.startswith("#"):
        c = "# " + c.lstrip()
    return c


def _unpack_key(
    item,  # type: ParagraphKey
    raise_if_indexed=False,  # type: bool
):
    # type: (...) -> Tuple[_strI, Optional[int], Optional[Deb822FieldNameToken]]
    index = None  # type: Optional[int]
    name_token = None  # type: Optional[Deb822FieldNameToken]
    if isinstance(item, tuple):
        key, index = item
        if raise_if_indexed:
            # Fudge "(key, 0)" into a "key" callers to defensively support
            # both paragraph styles with the same key.
            if index != 0:
                msg = 'Cannot resolve key "{key}" with index {index}. The key is not indexed'
                raise KeyError(msg.format(key=key, index=index))
            index = None
        key = _strI(key)
    else:
        index = None
        if isinstance(item, Deb822FieldNameToken):
            name_token = item
            key = name_token.text
        else:
            key = _strI(item)

    return key, index, name_token


def _convert_value_lines_to_lines(
    value_lines,  # type: Iterable[Deb822ValueLineElement]
    strip_comments,  # type: bool
):
    # type: (...) -> Iterable[str]
    if not strip_comments:
        yield from (v.convert_to_text() for v in value_lines)
    else:
        for element in value_lines:
            yield "".join(x.text for x in element.iter_tokens() if not x.is_comment)


if sys.version_info >= (3, 9) or TYPE_CHECKING:
    _ParagraphMapping_Base = collections.abc.Mapping[ParagraphKey, T]
else:
    # Python 3.5 - 3.8 compat - we are not allowed to subscript the abc.Iterator
    # - use this little hack to work around it
    class _ParagraphMapping_Base(collections.abc.Mapping, Generic[T], ABC):
        pass


# Deb822ParagraphElement uses this Mixin (by having `_paragraph` return self).
# Therefore, the Mixin needs to call the "proper" methods on the paragraph to
# avoid doing infinite recursion.
class AutoResolvingMixin(Generic[T], _ParagraphMapping_Base[T]):

    @property
    def _auto_resolve_ambiguous_fields(self):
        # type: () -> bool
        return True

    @property
    def _paragraph(self):
        # type: () -> Deb822ParagraphElement
        raise NotImplementedError  # pragma: no cover

    def __len__(self):
        # type: () -> int
        return self._paragraph.kvpair_count

    def __contains__(self, item):
        # type: (object) -> bool
        return self._paragraph.contains_kvpair_element(item)

    def __iter__(self):
        # type: () -> Iterator[ParagraphKey]
        return iter(self._paragraph.iter_keys())

    def __getitem__(self, item):
        # type: (ParagraphKey) -> T
        if self._auto_resolve_ambiguous_fields and isinstance(item, str):
            v = self._paragraph.get_kvpair_element((item, 0))
        else:
            v = self._paragraph.get_kvpair_element(item)
        assert v is not None
        return self._interpret_value(item, v)

    def __delitem__(self, item):
        # type: (ParagraphKey) -> None
        self._paragraph.remove_kvpair_element(item)

    def _interpret_value(self, key, value):
        # type: (ParagraphKey, Deb822KeyValuePairElement) -> T
        raise NotImplementedError  # pragma: no cover


# Deb822ParagraphElement uses this Mixin (by having `_paragraph` return self).
# Therefore, the Mixin needs to call the "proper" methods on the paragraph to
# avoid doing infinite recursion.
class Deb822ParagraphToStrWrapperMixin(AutoResolvingMixin[str], ABC):

    @property
    def _auto_map_initial_line_whitespace(self):
        # type: () -> bool
        return True

    @property
    def _discard_comments_on_read(self):
        # type: () -> bool
        return True

    @property
    def _auto_map_final_newline_in_multiline_values(self):
        # type: () -> bool
        return True

    @property
    def _preserve_field_comments_on_field_updates(self):
        # type: () -> bool
        return True

    def _convert_value_to_str(self, kvpair_element):
        # type: (Deb822KeyValuePairElement) -> str
        value_element = kvpair_element.value_element
        value_entries = value_element.value_lines
        if len(value_entries) == 1:
            # Special case single line entry (e.g. "Package: foo") as they never
            # have comments and we can do some parts more efficient.
            value_entry = value_entries[0]
            t = value_entry.convert_to_text()
            if self._auto_map_initial_line_whitespace:
                t = t.strip()
            return t

        if self._auto_map_initial_line_whitespace or self._discard_comments_on_read:
            converter = _convert_value_lines_to_lines(
                value_entries,
                self._discard_comments_on_read,
            )

            auto_map_space = self._auto_map_initial_line_whitespace

            # Because we know there are more than one line, we can unconditionally inject
            # the newline after the first line
            as_text = "".join(
                line.strip() + "\n" if auto_map_space and i == 1 else line
                for i, line in enumerate(converter, start=1)
            )
        else:
            # No rewrite necessary.
            as_text = value_element.convert_to_text()

        if self._auto_map_final_newline_in_multiline_values and as_text[-1] == "\n":
            as_text = as_text[:-1]
        return as_text

    def __setitem__(self, item, value):
        # type: (ParagraphKey, str) -> None
        keep_comments = (
            self._preserve_field_comments_on_field_updates
        )  # type: Optional[bool]
        comment = None
        if keep_comments and self._auto_resolve_ambiguous_fields:
            # For ambiguous fields, we have to resolve the original field as
            # the set_field_* methods do not cope with ambiguous fields.  This
            # means we might as well clear the keep_comments flag as we have
            # resolved the comment.
            keep_comments = None
            key_lookup = item
            if isinstance(item, str):
                key_lookup = (item, 0)
            orig_kvpair = self._paragraph.get_kvpair_element(key_lookup, use_get=True)
            if orig_kvpair is not None:
                comment = orig_kvpair.comment_element

        if self._auto_map_initial_line_whitespace:
            try:
                idx = value.index("\n")
            except ValueError:
                idx = -1
            if idx == -1 or idx == len(value):
                self._paragraph.set_field_to_simple_value(
                    item,
                    value.strip(),
                    preserve_original_field_comment=keep_comments,
                    field_comment=comment,
                )
                return
            # Regenerate the first line with normalized whitespace if necessary
            first_line, rest = value.split("\n", 1)
            if first_line and first_line[:1] not in ("\t", " "):
                value = "".join((" ", first_line.strip(), "\n", rest))
            else:
                value = "".join((first_line, "\n", rest))
        if not value.endswith("\n"):
            if not self._auto_map_final_newline_in_multiline_values:
                raise ValueError(
                    "Values must end with a newline (or be single line"
                    " values and use the auto whitespace mapping feature)"
                )
            value += "\n"
        self._paragraph.set_field_from_raw_string(
            item,
            value,
            preserve_original_field_comment=keep_comments,
            field_comment=comment,
        )

    def _interpret_value(self, key, value):
        # type: (ParagraphKey, Deb822KeyValuePairElement) -> str
        # mypy is a bit dense and cannot see that T == str
        return self._convert_value_to_str(value)


class AbstractDeb822ParagraphWrapper(AutoResolvingMixin[T], ABC):

    def __init__(
        self,
        paragraph,  # type: Deb822ParagraphElement
        *,
        auto_resolve_ambiguous_fields=False,  # type: bool
        discard_comments_on_read=True,  # type: bool
    ):
        # type: (...) -> None
        self.__paragraph = paragraph
        self.__auto_resolve_ambiguous_fields = auto_resolve_ambiguous_fields
        self.__discard_comments_on_read = discard_comments_on_read

    @property
    def _paragraph(self):
        # type: () -> Deb822ParagraphElement
        return self.__paragraph

    @property
    def _discard_comments_on_read(self):
        # type: () -> bool
        return self.__discard_comments_on_read

    @property
    def _auto_resolve_ambiguous_fields(self):
        # type: () -> bool
        return self.__auto_resolve_ambiguous_fields


class Deb822InterpretingParagraphWrapper(AbstractDeb822ParagraphWrapper[T]):

    def __init__(
        self,
        paragraph,  # type: Deb822ParagraphElement
        interpretation,  # type: Interpretation[T]
        *,
        auto_resolve_ambiguous_fields=False,  # type: bool
        discard_comments_on_read=True,  # type: bool
    ):
        # type: (...) -> None
        super().__init__(
            paragraph,
            auto_resolve_ambiguous_fields=auto_resolve_ambiguous_fields,
            discard_comments_on_read=discard_comments_on_read,
        )
        self._interpretation = interpretation

    def _interpret_value(self, key, value):
        # type: (ParagraphKey, Deb822KeyValuePairElement) -> T
        return self._interpretation.interpret(value)


class Deb822DictishParagraphWrapper(
    AbstractDeb822ParagraphWrapper[str], Deb822ParagraphToStrWrapperMixin
):

    def __init__(
        self,
        paragraph,  # type: Deb822ParagraphElement
        *,
        discard_comments_on_read=True,  # type: bool
        auto_map_initial_line_whitespace=True,  # type: bool
        auto_resolve_ambiguous_fields=False,  # type: bool
        preserve_field_comments_on_field_updates=True,  # type: bool
        auto_map_final_newline_in_multiline_values=True,  # type: bool
    ):
        # type: (...) -> None
        super().__init__(
            paragraph,
            auto_resolve_ambiguous_fields=auto_resolve_ambiguous_fields,
            discard_comments_on_read=discard_comments_on_read,
        )
        self.__auto_map_initial_line_whitespace = auto_map_initial_line_whitespace
        self.__preserve_field_comments_on_field_updates = (
            preserve_field_comments_on_field_updates
        )
        self.__auto_map_final_newline_in_multiline_values = (
            auto_map_final_newline_in_multiline_values
        )

    @property
    def _auto_map_initial_line_whitespace(self):
        # type: () -> bool
        return self.__auto_map_initial_line_whitespace

    @property
    def _preserve_field_comments_on_field_updates(self):
        # type: () -> bool
        return self.__preserve_field_comments_on_field_updates

    @property
    def _auto_map_final_newline_in_multiline_values(self):
        # type: () -> bool
        return self.__auto_map_final_newline_in_multiline_values


class Deb822ParagraphElement(Deb822Element, Deb822ParagraphToStrWrapperMixin, ABC):

    @classmethod
    def new_empty_paragraph(cls):
        # type: () -> Deb822ParagraphElement
        return Deb822NoDuplicateFieldsParagraphElement([], OrderedSet())

    @classmethod
    def from_dict(cls, mapping):
        # type: (Mapping[str, str]) -> Deb822ParagraphElement
        paragraph = cls.new_empty_paragraph()
        for k, v in mapping.items():
            paragraph[k] = v
        return paragraph

    @classmethod
    def from_kvpairs(cls, kvpair_elements):
        # type: (List[Deb822KeyValuePairElement]) -> Deb822ParagraphElement
        if not kvpair_elements:
            raise ValueError(
                "A paragraph must consist of at least one field/value pair"
            )
        kvpair_order = OrderedSet(kv.field_name for kv in kvpair_elements)
        if len(kvpair_order) == len(kvpair_elements):
            # Each field occurs at most once, which is good because that
            # means it is a valid paragraph and we can use the optimized
            # implementation.
            return Deb822NoDuplicateFieldsParagraphElement(
                kvpair_elements, kvpair_order
            )
        # Fallback implementation, that can cope with the repeated field names
        # at the cost of complexity.
        return Deb822DuplicateFieldsParagraphElement(kvpair_elements)

    @property
    def has_duplicate_fields(self):
        # type: () -> bool
        """Tell whether this paragraph has duplicate fields"""
        return False

    def as_interpreted_dict_view(
        self,
        interpretation,  # type: Interpretation[T]
        *,
        auto_resolve_ambiguous_fields=True,  # type: bool
    ):
        # type: (...) -> Deb822InterpretingParagraphWrapper[T]
        r"""Provide a Dict-like view of the paragraph

        This method returns a dict-like object representing this paragraph and
        is useful for accessing fields in a given interpretation. It is possible
        to use multiple versions of this dict-like view with different interpretations
        on the same paragraph at the same time (for different fields).

            >>> example_deb822_paragraph = '''
            ... Package: foo
            ... # Field comment (because it becomes just before a field)
            ... Architecture: amd64
            ... # Inline comment (associated with the next line)
            ...               i386
            ... # We also support arm
            ...               arm64
            ...               armel
            ... '''
            >>> dfile = parse_deb822_file(example_deb822_paragraph.splitlines())
            >>> paragraph = next(iter(dfile))
            >>> list_view = paragraph.as_interpreted_dict_view(LIST_SPACE_SEPARATED_INTERPRETATION)
            >>> # With the defaults, you only deal with the semantic values
            >>> # - no leading or trailing whitespace on the first part of the value
            >>> list(list_view["Package"])
            ['foo']
            >>> with list_view["Architecture"] as arch_list:
            ...     orig_arch_list = list(arch_list)
            ...     arch_list.replace('i386', 'kfreebsd-amd64')
            >>> orig_arch_list
            ['amd64', 'i386', 'arm64', 'armel']
            >>> list(list_view["Architecture"])
            ['amd64', 'kfreebsd-amd64', 'arm64', 'armel']
            >>> print(paragraph.dump(), end='')
            Package: foo
            # Field comment (because it becomes just before a field)
            Architecture: amd64
            # Inline comment (associated with the next line)
                          kfreebsd-amd64
            # We also support arm
                          arm64
                          armel
            >>> # Format preserved and architecture replaced
            >>> with list_view["Architecture"] as arch_list:
            ...     # Prettify the result as sorting will cause awkward whitespace
            ...     arch_list.reformat_when_finished()
            ...     arch_list.sort()
            >>> print(paragraph.dump(), end='')
            Package: foo
            # Field comment (because it becomes just before a field)
            Architecture: amd64
            # We also support arm
                          arm64
                          armel
            # Inline comment (associated with the next line)
                          kfreebsd-amd64
            >>> list(list_view["Architecture"])
            ['amd64', 'arm64', 'armel', 'kfreebsd-amd64']
            >>> # Format preserved and architecture values sorted

        :param interpretation: Decides how the field values are interpreted.  As an example,
          use LIST_SPACE_SEPARATED_INTERPRETATION for fields such as Architecture in the
          debian/control file.
        :param auto_resolve_ambiguous_fields: This parameter is only relevant for paragraphs
          that contain the same field multiple times (these are generally invalid).  If the
          caller requests an ambiguous field from an invalid paragraph via a plain field name,
          the return dict-like object will refuse to resolve the field (not knowing which
          version to pick).  This parameter (if set to True) instead changes the error into
          assuming the caller wants the *first* variant.
        """
        return Deb822InterpretingParagraphWrapper(
            self,
            interpretation,
            auto_resolve_ambiguous_fields=auto_resolve_ambiguous_fields,
        )

    def configured_view(
        self,
        *,
        discard_comments_on_read=True,  # type: bool
        auto_map_initial_line_whitespace=True,  # type: bool
        auto_resolve_ambiguous_fields=True,  # type: bool
        preserve_field_comments_on_field_updates=True,  # type: bool
        auto_map_final_newline_in_multiline_values=True,  # type: bool
    ):
        # type: (...) -> Deb822DictishParagraphWrapper
        r"""Provide a Dict[str, str]-like view of this paragraph with non-standard parameters

        This method returns a dict-like object representing this paragraph that is
        optionally configured differently from the default view.

            >>> example_deb822_paragraph = '''
            ... Package: foo
            ... # Field comment (because it becomes just before a field)
            ... Depends: libfoo,
            ... # Inline comment (associated with the next line)
            ...          libbar,
            ... '''
            >>> dfile = parse_deb822_file(example_deb822_paragraph.splitlines())
            >>> paragraph = next(iter(dfile))
            >>> # With the defaults, you only deal with the semantic values
            >>> # - no leading or trailing whitespace on the first part of the value
            >>> paragraph["Package"]
            'foo'
            >>> # - no inline comments in multiline values (but whitespace will be present
            >>> #   subsequent lines.)
            >>> print(paragraph["Depends"])
            libfoo,
                     libbar,
            >>> paragraph['Foo'] = 'bar'
            >>> paragraph.get('Foo')
            'bar'
            >>> paragraph.get('Unknown-Field') is None
            True
            >>> # But you get asymmetric behaviour with set vs. get
            >>> paragraph['Foo'] = ' bar\n'
            >>> paragraph['Foo']
            'bar'
            >>> paragraph['Bar'] = '     bar\n#Comment\n another value\n'
            >>> # Note that the whitespace on the first line has been normalized.
            >>> print("Bar: " + paragraph['Bar'])
            Bar: bar
             another value
            >>> # The comment is present (in case you where wondering)
            >>> print(paragraph.get_kvpair_element('Bar').convert_to_text(), end='')
            Bar:     bar
            #Comment
             another value
            >>> # On the other hand, you can choose to see the values as they are
            >>> # - We will just reset the paragraph as a "nothing up my sleeve"
            >>> dfile = parse_deb822_file(example_deb822_paragraph.splitlines())
            >>> paragraph = next(iter(dfile))
            >>> nonstd_dictview = paragraph.configured_view(
            ...     discard_comments_on_read=False,
            ...     auto_map_initial_line_whitespace=False,
            ...     # For paragraphs with duplicate fields, you can choose to get an error
            ...     # rather than the dict picking the first value available.
            ...     auto_resolve_ambiguous_fields=False,
            ...     auto_map_final_newline_in_multiline_values=False,
            ... )
            >>> # Because we have reset the state, Foo and Bar are no longer there.
            >>> 'Bar' not in paragraph and 'Foo' not in paragraph
            True
            >>> # We can now see the comments (discard_comments_on_read=False)
            >>> # (The leading whitespace in front of "libfoo" is due to
            >>> #  auto_map_initial_line_whitespace=False)
            >>> print(nonstd_dictview["Depends"], end='')
             libfoo,
            # Inline comment (associated with the next line)
                     libbar,
            >>> # And all the optional whitespace on the first value line
            >>> # (auto_map_initial_line_whitespace=False)
            >>> nonstd_dictview["Package"] == ' foo\n'
            True
            >>> # ... which will give you symmetric behaviour with set vs. get
            >>> nonstd_dictview['Foo'] = '  bar \n'
            >>> nonstd_dictview['Foo']
            '  bar \n'
            >>> nonstd_dictview['Bar'] = '  bar \n#Comment\n another value\n'
            >>> nonstd_dictview['Bar']
            '  bar \n#Comment\n another value\n'
            >>> # But then you get no help either.
            >>> try:
            ...     nonstd_dictview["Baz"] = "foo"
            ... except ValueError:
            ...     print("Rejected")
            Rejected
            >>> # With auto_map_initial_line_whitespace=False, you have to include minimum a newline
            >>> nonstd_dictview["Baz"] = "foo\n"
            >>> # The absence of leading whitespace gives you the terse variant at the expensive
            >>> # readability
            >>> paragraph.get_kvpair_element('Baz').convert_to_text()
            'Baz:foo\n'
            >>> # But because they are views, changes performed via one view is visible in the other
            >>> paragraph['Foo']
            'bar'
            >>> # The views show the values according to their own rules. Therefore, there is an
            >>> # asymmetric between paragraph['Foo'] and nonstd_dictview['Foo']
            >>> # Nevertheless, you can read or write the fields via either - enabling you to use
            >>> # the view that best suit your use-case for the given field.
            >>> 'Baz' in paragraph and nonstd_dictview.get('Baz') is not None
            True
            >>> # Deletion via the view also works
            >>> del nonstd_dictview['Baz']
            >>> 'Baz' not in paragraph and nonstd_dictview.get('Baz') is None
            True


        :param discard_comments_on_read: When getting a field value from the dict,
          this parameter decides how in-line comments are handled.  When setting
          the value, inline comments are still allowed and will be retained.
          However, keep in mind that this option makes getter and setter assymetric
          as a "get" following a "set" with inline comments will omit the comments
          even if they are there (see the code example).
        :param auto_map_initial_line_whitespace: Special-case the first value line
          by trimming unnecessary whitespace leaving only the value. For single-line
          values, all space including newline is pruned. For multi-line values, the
          newline is preserved / needed to distinguish the first line from the
          following lines.  When setting a value, this option normalizes the
          whitespace of the initial line of the value field.
          When this option is set to True makes the dictionary behave more like the
          original Deb822 module.
        :param preserve_field_comments_on_field_updates: Whether to preserve the field
          comments when mutating the field.
        :param auto_resolve_ambiguous_fields: This parameter is only relevant for paragraphs
          that contain the same field multiple times (these are generally invalid).  If the
          caller requests an ambiguous field from an invalid paragraph via a plain field name,
          the return dict-like object will refuse to resolve the field (not knowing which
          version to pick).  This parameter (if set to True) instead changes the error into
          assuming the caller wants the *first* variant.
        :param auto_map_final_newline_in_multiline_values: This parameter controls whether
          a multiline field with have / need a trailing newline. If True, the trailing
          newline is hidden on get and automatically added in set (if missing).
          When this option is set to True makes the dictionary behave more like the
          original Deb822 module.
        """
        return Deb822DictishParagraphWrapper(
            self,
            discard_comments_on_read=discard_comments_on_read,
            auto_map_initial_line_whitespace=auto_map_initial_line_whitespace,
            auto_resolve_ambiguous_fields=auto_resolve_ambiguous_fields,
            preserve_field_comments_on_field_updates=preserve_field_comments_on_field_updates,
            auto_map_final_newline_in_multiline_values=auto_map_final_newline_in_multiline_values,
        )

    @property
    def _paragraph(self):
        # type: () -> Deb822ParagraphElement
        return self

    def order_last(self, field):
        # type: (ParagraphKey) -> None
        """Re-order the given field so it is "last" in the paragraph"""
        raise NotImplementedError  # pragma: no cover

    def order_first(self, field):
        # type: (ParagraphKey) -> None
        """Re-order the given field so it is "first" in the paragraph"""
        raise NotImplementedError  # pragma: no cover

    def order_before(self, field, reference_field):
        # type: (ParagraphKey, ParagraphKey) -> None
        """Re-order the given field so appears directly after the reference field in the paragraph

        The reference field must be present."""
        raise NotImplementedError  # pragma: no cover

    def order_after(self, field, reference_field):
        # type: (ParagraphKey, ParagraphKey) -> None
        """Re-order the given field so appears directly before the reference field in the paragraph

        The reference field must be present.
        """
        raise NotImplementedError  # pragma: no cover

    @property
    def kvpair_count(self):
        # type: () -> int
        raise NotImplementedError  # pragma: no cover

    def iter_keys(self):
        # type: () -> Iterable[ParagraphKey]
        raise NotImplementedError  # pragma: no cover

    def contains_kvpair_element(self, item):
        # type: (object) -> bool
        raise NotImplementedError  # pragma: no cover

    def get_kvpair_element(
        self,
        item,  # type: ParagraphKey
        use_get=False,  # type: bool
    ):
        # type: (...) -> Optional[Deb822KeyValuePairElement]
        raise NotImplementedError  # pragma: no cover

    def set_kvpair_element(self, key, value):
        # type: (ParagraphKey, Deb822KeyValuePairElement) -> None
        raise NotImplementedError  # pragma: no cover

    def remove_kvpair_element(self, key):
        # type: (ParagraphKey) -> None
        raise NotImplementedError  # pragma: no cover

    def sort_fields(
        self, key=None  # type: Optional[Callable[[str], Any]]
    ):
        # type: (...) -> None
        """Re-order all fields

        :param key: Provide a key function (same semantics as for sorted).  Keep in mind that
          the module preserve the cases for field names - in generally, callers are recommended
          to use "lower()" to normalize the case.
        """
        raise NotImplementedError  # pragma: no cover

    def set_field_to_simple_value(
        self,
        item,  # type: ParagraphKey
        simple_value,  # type: str
        *,
        preserve_original_field_comment=None,  # type: Optional[bool]
        field_comment=None,  # type: Optional[Commentish]
    ):
        # type: (...) -> None
        r"""Sets a field in this paragraph to a simple "word" or "phrase"

        In many cases, it is better for callers to just use the paragraph as
        if it was a dictionary.  However, this method does enable to you choose
        the field comment (if any), which can be a reason for using it.

        This is suitable for "simple" fields like "Package".  Example:

            >>> example_deb822_paragraph = '''
            ... Package: foo
            ... '''
            >>> dfile = parse_deb822_file(example_deb822_paragraph.splitlines())
            >>> p = next(iter(dfile))
            >>> p.set_field_to_simple_value("Package", "mscgen")
            >>> p.set_field_to_simple_value("Architecture", "linux-any kfreebsd-any",
            ...                             field_comment=['Only ported to linux and kfreebsd'])
            >>> p.set_field_to_simple_value("Priority", "optional")
            >>> print(p.dump(), end='')
            Package: mscgen
            # Only ported to linux and kfreebsd
            Architecture: linux-any kfreebsd-any
            Priority: optional
            >>> # Values are formatted nicely by default, but it does not work with
            >>> # multi-line values
            >>> p.set_field_to_simple_value("Foo", "bar\nbin\n")
            Traceback (most recent call last):
                ...
            ValueError: Cannot use set_field_to_simple_value for values with newlines

        :param item: Name of the field to set.  If the paragraph already
          contains the field, then it will be replaced.  If the field exists,
          then it will preserve its order in the paragraph.  Otherwise, it is
          added to the end of the paragraph.
          Note this can be a "paragraph key", which enables you to control
          *which* instance of a field is being replaced (in case of duplicate
          fields).
        :param simple_value: The text to use as the value.  The value must not
          contain newlines.  Leading and trailing will be stripped but space
          within the value is preserved.  The value cannot contain comments
          (i.e. if the "#" token appears in the value, then it is considered
          a value rather than "start of a comment)
        :param preserve_original_field_comment: See the description for the
          parameter with the same name in the set_field_from_raw_string method.
        :param field_comment: See the description for the parameter with the same
          name in the set_field_from_raw_string method.
        """
        if "\n" in simple_value:
            raise ValueError(
                "Cannot use set_field_to_simple_value for values with newlines"
            )

        # Reformat it with a leading space and trailing newline. The latter because it is
        # necessary if there are any fields after it and the former because it looks nicer so
        # have single space after the field separator
        stripped = simple_value.strip()
        if stripped:
            raw_value = " " + stripped + "\n"
        else:
            # Special-case for empty values
            raw_value = "\n"
        self.set_field_from_raw_string(
            item,
            raw_value,
            preserve_original_field_comment=preserve_original_field_comment,
            field_comment=field_comment,
        )

    def set_field_from_raw_string(
        self,
        item,  # type: ParagraphKey
        raw_string_value,  # type: str
        *,
        preserve_original_field_comment=None,  # type: Optional[bool]
        field_comment=None,  # type: Optional[Commentish]
    ):
        # type: (...) -> None
        """Sets a field in this paragraph to a given text value

        In many cases, it is better for callers to just use the paragraph as
        if it was a dictionary.  However, this method does enable to you choose
        the field comment (if any) and lets to have a higher degree of control
        over whitespace (on the first line), which can be a reason for using it.

        Example usage:

            >>> example_deb822_paragraph = '''
            ... Package: foo
            ... '''
            >>> dfile = parse_deb822_file(example_deb822_paragraph.splitlines())
            >>> p = next(iter(dfile))
            >>> raw_value = '''
            ... Build-Depends: debhelper-compat (= 12),
            ...                some-other-bd,
            ... # Comment
            ...                another-bd,
            ... '''.lstrip()  # Remove leading newline, but *not* the trailing newline
            >>> fname, new_value = raw_value.split(':', 1)
            >>> p.set_field_from_raw_string(fname, new_value)
            >>> print(p.dump(), end='')
            Package: foo
            Build-Depends: debhelper-compat (= 12),
                           some-other-bd,
            # Comment
                           another-bd,
            >>> # Format preserved

        :param item: Name of the field to set.  If the paragraph already
          contains the field, then it will be replaced.  Otherwise, it is
          added to the end of the paragraph.
          Note this can be a "paragraph key", which enables you to control
          *which* instance of a field is being replaced (in case of duplicate
          fields).
        :param raw_string_value: The text to use as the value.  The text must
          be valid deb822 syntax and is used *exactly* as it is given.
          Accordingly, multi-line values must include mandatory leading space
          on continuation lines, newlines after the value, etc. On the
          flip-side, any optional space or comments will be included.

          Note that the first line will *never* be read as a comment (if the
          first line of the value starts with a "#" then it will result
          in "Field-Name:#..." which is parsed as a value starting with "#"
          rather than a comment).
        :param preserve_original_field_comment: If True, then if there is an
          existing field and that has a comment, then the comment will remain
          after this operation.  This is the default is the `field_comment`
          parameter is omitted.
          Note that if the parameter is True and the item is ambiguous, this
          will raise an AmbiguousDeb822FieldKeyError.  When the parameter is
          omitted, the ambiguity is resolved automatically and if the resolved
          field has a comment then that will be preserved (assuming
          field_comment is None).
        :param field_comment: If not None, add or replace the comment for
          the field.  Each string in the list will become one comment
          line (inserted directly before the field name). Will appear in the
          same order as they do in the list.

          If you want complete control over the formatting of the comments,
          then ensure that each line start with "#" and end with "\\n" before
          the call.  Otherwise, leading/trailing whitespace is normalized
          and the missing "#"/"\\n" character is inserted.
        """

        new_content = []  # type: List[str]
        if preserve_original_field_comment is not None:
            if field_comment is not None:
                raise ValueError(
                    'The "preserve_original_field_comment" conflicts with'
                    ' "field_comment" parameter'
                )
        elif field_comment is not None:
            if not isinstance(field_comment, Deb822CommentElement):
                new_content.extend(_format_comment(x) for x in field_comment)
                field_comment = None
            preserve_original_field_comment = False

        field_name, _, _ = _unpack_key(item)

        cased_field_name = field_name
        try:
            original = self.get_kvpair_element(item, use_get=True)
        except AmbiguousDeb822FieldKeyError:
            if preserve_original_field_comment:
                # If we were asked to preserve the original comment, then we
                # require a strict lookup
                raise
            original = self.get_kvpair_element((field_name, 0), use_get=True)

        if preserve_original_field_comment is None:
            # We simplify preserve_original_field_comment after the lookup of the field.
            # Otherwise, we can get ambiguous key errors when updating an ambiguous field
            # when the caller did not explicitly ask for that behaviour.
            preserve_original_field_comment = True

        if original:
            # If we already have the field, then preserve the original case
            cased_field_name = original.field_name
        raw = ":".join((cased_field_name, raw_string_value))
        raw_lines = raw.splitlines(keepends=True)
        for i, line in enumerate(raw_lines, start=1):
            if not line.endswith("\n"):
                raise ValueError(
                    "Line {i} in new value was missing trailing newline".format(i=i)
                )
            if i != 1 and line[0] not in (" ", "\t", "#"):
                msg = (
                    "Line {i} in new value was invalid.  It must either start"
                    ' with " " space (continuation line) or "#" (comment line).'
                    ' The line started with "{line}"'
                )
                raise ValueError(msg.format(i=i, line=line[0]))
        if len(raw_lines) > 1 and raw_lines[-1].startswith("#"):
            raise ValueError("The last line in a value field cannot be a comment")
        new_content.extend(raw_lines)
        # As absurd as it might seem, it is easier to just use the parser to
        # construct the AST correctly
        deb822_file = parse_deb822_file(iter(new_content))
        error_token = deb822_file.find_first_error_element()
        if error_token:
            raise ValueError("Syntax error in new field value for " + field_name)
        paragraph = next(iter(deb822_file))
        assert isinstance(paragraph, Deb822NoDuplicateFieldsParagraphElement)
        value = paragraph.get_kvpair_element(field_name)
        assert value is not None
        if preserve_original_field_comment:
            if original:
                value.comment_element = original.comment_element
                original.comment_element = None
        elif field_comment is not None:
            value.comment_element = field_comment
        self.set_kvpair_element(item, value)

    @overload
    def dump(
        self, fd  # type: IO[bytes]
    ):
        # type: (...) -> None
        pass

    @overload
    def dump(self):
        # type: () -> str
        pass

    def dump(
        self, fd=None  # type: Optional[IO[bytes]]
    ):
        # type: (...) -> Optional[str]
        if fd is None:
            return "".join(t.text for t in self.iter_tokens())
        for token in self.iter_tokens():
            fd.write(token.text.encode("utf-8"))
        return None


class Deb822NoDuplicateFieldsParagraphElement(Deb822ParagraphElement):
    """Paragraph implementation optimized for valid deb822 files

    When there are no duplicated fields, we can use simpler and faster
    datastructures for common operations.
    """

    def __init__(
        self,
        kvpair_elements,  # type: List[Deb822KeyValuePairElement]
        kvpair_order,  # type: OrderedSet
    ):
        # type: (...) -> None
        super().__init__()
        self._kvpair_elements = {kv.field_name: kv for kv in kvpair_elements}
        self._kvpair_order = kvpair_order
        self._init_parent_of_parts()

    @property
    def kvpair_count(self):
        # type: () -> int
        return len(self._kvpair_elements)

    def order_last(self, field):
        # type: (ParagraphKey) -> None
        """Re-order the given field so it is "last" in the paragraph"""
        unpacked_field, _, _ = _unpack_key(field, raise_if_indexed=True)
        self._kvpair_order.order_last(unpacked_field)

    def order_first(self, field):
        # type: (ParagraphKey) -> None
        """Re-order the given field so it is "first" in the paragraph"""
        unpacked_field, _, _ = _unpack_key(field, raise_if_indexed=True)
        self._kvpair_order.order_first(unpacked_field)

    def order_before(self, field, reference_field):
        # type: (ParagraphKey, ParagraphKey) -> None
        """Re-order the given field so appears directly after the reference field in the paragraph

        The reference field must be present."""
        unpacked_field, _, _ = _unpack_key(field, raise_if_indexed=True)
        unpacked_ref_field, _, _ = _unpack_key(reference_field, raise_if_indexed=True)
        self._kvpair_order.order_before(unpacked_field, unpacked_ref_field)

    def order_after(self, field, reference_field):
        # type: (ParagraphKey, ParagraphKey) -> None
        """Re-order the given field so appears directly before the reference field in the paragraph

        The reference field must be present.
        """
        unpacked_field, _, _ = _unpack_key(field, raise_if_indexed=True)
        unpacked_ref_field, _, _ = _unpack_key(reference_field, raise_if_indexed=True)
        self._kvpair_order.order_after(unpacked_field, unpacked_ref_field)

    # Overload to narrow the type to just str.
    def __iter__(self):
        # type: () -> Iterator[str]
        return iter(str(k) for k in self._kvpair_order)

    def iter_keys(self):
        # type: () -> Iterable[str]
        yield from (str(k) for k in self._kvpair_order)

    def remove_kvpair_element(self, key):
        # type: (ParagraphKey) -> None
        self._full_size_cache = None
        key, _, _ = _unpack_key(key, raise_if_indexed=True)
        del self._kvpair_elements[key]
        self._kvpair_order.remove(key)

    def contains_kvpair_element(self, item):
        # type: (object) -> bool
        if not isinstance(item, (str, tuple, Deb822FieldNameToken)):
            return False
        item = cast("ParagraphKey", item)
        key, _, _ = _unpack_key(item, raise_if_indexed=True)
        return key in self._kvpair_elements

    def get_kvpair_element(
        self,
        item,  # type: ParagraphKey
        use_get=False,  # type: bool
    ):
        # type: (...) -> Optional[Deb822KeyValuePairElement]
        item, _, _ = _unpack_key(item, raise_if_indexed=True)
        if use_get:
            return self._kvpair_elements.get(item)
        return self._kvpair_elements[item]

    def set_kvpair_element(self, key, value):
        # type: (ParagraphKey, Deb822KeyValuePairElement) -> None
        key, _, _ = _unpack_key(key, raise_if_indexed=True)
        if isinstance(key, Deb822FieldNameToken):
            if key is not value.field_token:
                raise ValueError(
                    "Key is a Deb822FieldNameToken, but not *the* Deb822FieldNameToken"
                    " for the value"
                )
            key = value.field_name
        else:
            if key != value.field_name:
                raise ValueError(
                    "Cannot insert value under a different field value than field name"
                    " from its Deb822FieldNameToken implies"
                )
            # Use the string from the Deb822FieldNameToken as we need to keep that in memory either
            # way
            key = value.field_name
        original_value = self._kvpair_elements.get(key)
        self._full_size_cache = None
        self._kvpair_elements[key] = value
        self._kvpair_order.append(key)
        if original_value is not None:
            original_value.parent_element = None
        value.parent_element = self

    def sort_fields(self, key=None):
        # type: (Optional[Callable[[str], Any]]) -> None
        """Re-order all fields

        :param key: Provide a key function (same semantics as for sorted).  Keep in mind that
          the module preserve the cases for field names - in generally, callers are recommended
          to use "lower()" to normalize the case.
        """
        for last_field_name in reversed(self._kvpair_order):
            last_kvpair = self._kvpair_elements[cast("_strI", last_field_name)]
            if last_kvpair.value_element.add_final_newline_if_missing():
                self._full_size_cache = None
            break

        if key is None:
            key = default_field_sort_key

        self._kvpair_order = OrderedSet(sorted(self._kvpair_order, key=key))

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        yield from (
            self._kvpair_elements[x]
            for x in cast("Iterable[_strI]", self._kvpair_order)
        )


class Deb822DuplicateFieldsParagraphElement(Deb822ParagraphElement):

    def __init__(self, kvpair_elements):
        # type: (List[Deb822KeyValuePairElement]) -> None
        super().__init__()
        self._kvpair_order = LinkedList()  # type: LinkedList[Deb822KeyValuePairElement]
        self._kvpair_elements = {}  # type: Dict[_strI, List[KVPNode]]
        self._init_kvpair_fields(kvpair_elements)
        self._init_parent_of_parts()

    @property
    def has_duplicate_fields(self):
        # type: () -> bool
        # Most likely, the answer is "True" but if the caller "fixes" the problem
        # then this can return "False"
        return len(self._kvpair_order) > len(self._kvpair_elements)

    def _init_kvpair_fields(self, kvpairs):
        # type: (Iterable[Deb822KeyValuePairElement]) -> None
        assert not self._kvpair_order
        assert not self._kvpair_elements
        for kv in kvpairs:
            field_name = kv.field_name
            node = self._kvpair_order.append(kv)
            if field_name not in self._kvpair_elements:
                self._kvpair_elements[field_name] = [node]
            else:
                self._kvpair_elements[field_name].append(node)

    def _nodes_being_relocated(self, field):
        # type: (ParagraphKey) -> Tuple[List[KVPNode], List[KVPNode]]
        key, index, name_token = _unpack_key(field)
        nodes = self._kvpair_elements[key]
        nodes_being_relocated = []

        if name_token is not None or index is not None:
            single_node = self._resolve_to_single_node(nodes, key, index, name_token)
            assert single_node is not None
            nodes_being_relocated.append(single_node)
        else:
            nodes_being_relocated = nodes
        return nodes, nodes_being_relocated

    def order_last(self, field):
        # type: (ParagraphKey) -> None
        """Re-order the given field so it is "last" in the paragraph"""
        nodes, nodes_being_relocated = self._nodes_being_relocated(field)
        assert len(nodes_being_relocated) == 1 or len(nodes) == len(
            nodes_being_relocated
        )

        kvpair_order = self._kvpair_order
        for node in nodes_being_relocated:
            if kvpair_order.tail_node is node:
                # Special case for relocating a single node that happens to be the last.
                continue
            kvpair_order.remove_node(node)
            # assertion for mypy
            assert kvpair_order.tail_node is not None
            kvpair_order.insert_node_after(node, kvpair_order.tail_node)

        if (
            len(nodes_being_relocated) == 1
            and nodes_being_relocated[0] is not nodes[-1]
        ):
            single_node = nodes_being_relocated[0]
            nodes.remove(single_node)
            nodes.append(single_node)

    def order_first(self, field):
        # type: (ParagraphKey) -> None
        """Re-order the given field so it is "first" in the paragraph"""
        nodes, nodes_being_relocated = self._nodes_being_relocated(field)
        assert len(nodes_being_relocated) == 1 or len(nodes) == len(
            nodes_being_relocated
        )

        kvpair_order = self._kvpair_order
        for node in nodes_being_relocated:
            if kvpair_order.head_node is node:
                # Special case for relocating a single node that happens to be the first.
                continue
            kvpair_order.remove_node(node)
            # assertion for mypy
            assert kvpair_order.head_node is not None
            kvpair_order.insert_node_before(node, kvpair_order.head_node)

        if len(nodes_being_relocated) == 1 and nodes_being_relocated[0] is not nodes[0]:
            single_node = nodes_being_relocated[0]
            nodes.remove(single_node)
            nodes.insert(0, single_node)

    def order_before(self, field, reference_field):
        # type: (ParagraphKey, ParagraphKey) -> None
        """Re-order the given field so appears directly after the reference field in the paragraph

        The reference field must be present."""
        nodes, nodes_being_relocated = self._nodes_being_relocated(field)
        assert len(nodes_being_relocated) == 1 or len(nodes) == len(
            nodes_being_relocated
        )
        # For "before" we always use the "first" variant as reference in case of doubt
        _, reference_nodes = self._nodes_being_relocated(reference_field)
        reference_node = reference_nodes[0]
        if reference_node in nodes_being_relocated:
            raise ValueError("Cannot re-order a field relative to itself")

        kvpair_order = self._kvpair_order
        for node in nodes_being_relocated:
            kvpair_order.remove_node(node)
            kvpair_order.insert_node_before(node, reference_node)

        if len(nodes_being_relocated) == 1 and len(nodes) > 1:
            # Regenerate the (new) relative field order.
            field_name = nodes_being_relocated[0].value.field_name
            self._regenerate_relative_kvapir_order(field_name)

    def order_after(self, field, reference_field):
        # type: (ParagraphKey, ParagraphKey) -> None
        """Re-order the given field so appears directly before the reference field in the paragraph

        The reference field must be present.
        """
        nodes, nodes_being_relocated = self._nodes_being_relocated(field)
        assert len(nodes_being_relocated) == 1 or len(nodes) == len(
            nodes_being_relocated
        )
        _, reference_nodes = self._nodes_being_relocated(reference_field)
        # For "after" we always use the "last" variant as reference in case of doubt
        reference_node = reference_nodes[-1]
        if reference_node in nodes_being_relocated:
            raise ValueError("Cannot re-order a field relative to itself")

        kvpair_order = self._kvpair_order
        # Use "reversed" to preserve the relative order of the nodes assuming a bulk reorder
        for node in reversed(nodes_being_relocated):
            kvpair_order.remove_node(node)
            kvpair_order.insert_node_after(node, reference_node)

        if len(nodes_being_relocated) == 1 and len(nodes) > 1:
            # Regenerate the (new) relative field order.
            field_name = nodes_being_relocated[0].value.field_name
            self._regenerate_relative_kvapir_order(field_name)

    def _regenerate_relative_kvapir_order(self, field_name):
        # type: (_strI) -> None
        nodes = []
        for node in self._kvpair_order.iter_nodes():
            if node.value.field_name == field_name:
                nodes.append(node)
        self._kvpair_elements[field_name] = nodes

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        yield from self._kvpair_order

    @property
    def kvpair_count(self):
        # type: () -> int
        return len(self._kvpair_order)

    def iter_keys(self):
        # type: () -> Iterable[ParagraphKey]
        yield from (kv.field_name for kv in self._kvpair_order)

    def _resolve_to_single_node(
        self,
        nodes,  # type: List[KVPNode]
        key,  # type: str
        index,  # type: Optional[int]
        name_token,  # type: Optional[Deb822FieldNameToken]
        use_get=False,  # type: bool
    ):
        # type: (...) -> Optional[KVPNode]
        if index is None:
            if len(nodes) != 1:
                if name_token is not None:
                    node = self._find_node_via_name_token(name_token, nodes)
                    if node is not None:
                        return node
                msg = (
                    "Ambiguous key {key} - the field appears {res_len} times. Use"
                    " ({key}, index) to denote which instance of the field you want.  (Index"
                    " can be 0..{res_len_1} or e.g. -1 to denote the last field)"
                )
                raise AmbiguousDeb822FieldKeyError(
                    msg.format(key=key, res_len=len(nodes), res_len_1=len(nodes) - 1)
                )
            index = 0
        try:
            return nodes[index]
        except IndexError:
            if use_get:
                return None
            msg = 'Field "{key}" was present but the index "{index}" was invalid.'
            raise KeyError(msg.format(key=key, index=index))

    def get_kvpair_element(
        self,
        item,  # type: ParagraphKey
        use_get=False,  # type: bool
    ):
        # type: (...) -> Optional[Deb822KeyValuePairElement]
        key, index, name_token = _unpack_key(item)
        if use_get:
            nodes = self._kvpair_elements.get(key)
            if nodes is None:
                return None
        else:
            nodes = self._kvpair_elements[key]
        node = self._resolve_to_single_node(
            nodes, key, index, name_token, use_get=use_get
        )
        if node is not None:
            return node.value
        return None

    @staticmethod
    def _find_node_via_name_token(
        name_token,  # type: Deb822FieldNameToken
        elements,  # type: Iterable[KVPNode]
    ):
        # type: (...) -> Optional[KVPNode]
        # if we are given a name token, then it is non-ambiguous if we have exactly
        # that name token in our list of nodes.  It will be an O(n) lookup but we
        # probably do not have that many duplicate fields (and even if do, it is not
        # exactly a valid file, so there little reason to optimize for it)
        for node in elements:
            if name_token is node.value.field_token:
                return node
        return None

    def contains_kvpair_element(self, item):
        # type: (object) -> bool
        if not isinstance(item, (str, tuple, Deb822FieldNameToken)):
            return False
        item = cast("ParagraphKey", item)
        try:
            return self.get_kvpair_element(item, use_get=True) is not None
        except AmbiguousDeb822FieldKeyError:
            return True

    def set_kvpair_element(self, key, value):
        # type: (ParagraphKey, Deb822KeyValuePairElement) -> None
        key, index, name_token = _unpack_key(key)
        if name_token:
            if name_token is not value.field_token:
                original_nodes = self._kvpair_elements.get(value.field_name)
                original_node = None
                if original_nodes is not None:
                    original_node = self._find_node_via_name_token(
                        name_token, original_nodes
                    )

                if original_node is None:
                    raise ValueError(
                        "Key is a Deb822FieldNameToken, but not *the*"
                        " Deb822FieldNameToken for the value nor the"
                        " Deb822FieldNameToken for an existing field in the paragraph"
                    )
                # Primarily for mypy's sake
                assert original_nodes is not None
                # Rely on the index-based code below to handle update.
                index = original_nodes.index(original_node)
            key = value.field_name
        else:
            if key != value.field_name:
                raise ValueError(
                    "Cannot insert value under a different field value than field name"
                    " from its Deb822FieldNameToken implies"
                )
            # Use the string from the Deb822FieldNameToken as it is a _strI and has the same value
            # (memory optimization)
            key = value.field_name
        self._full_size_cache = None
        original_nodes = self._kvpair_elements.get(key)
        if original_nodes is None or not original_nodes:
            if index is not None and index != 0:
                msg = (
                    "Cannot replace field ({key}, {index}) as the field does not exist"
                    " in the first place.  Please index-less key or ({key}, 0) if you"
                    " want to add the field."
                )
                raise KeyError(msg.format(key=key, index=index))
            node = self._kvpair_order.append(value)
            if key not in self._kvpair_elements:
                self._kvpair_elements[key] = [node]
            else:
                self._kvpair_elements[key].append(node)
            return

        replace_all = False
        if index is None:
            replace_all = True
            node = original_nodes[0]
            if len(original_nodes) != 1:
                self._kvpair_elements[key] = [node]
        else:
            # We insist on there being an original node, which as a side effect ensures
            # you cannot add additional copies of the field.  This means that you cannot
            # make the problem worse.
            node = original_nodes[index]

        # Replace the value of the existing node plus do a little dance
        # for the parent element part.
        node.value.parent_element = None
        value.parent_element = self
        node.value = value

        if replace_all and len(original_nodes) != 1:
            # If we were in a replace-all mode, discard any remaining nodes
            for n in original_nodes[1:]:
                n.value.parent_element = None
                self._kvpair_order.remove_node(n)

    def remove_kvpair_element(self, key):
        # type: (ParagraphKey) -> None
        key, idx, name_token = _unpack_key(key)
        field_list = self._kvpair_elements[key]

        if name_token is None and idx is None:
            self._full_size_cache = None
            # Remove all case
            for node in field_list:
                node.value.parent_element = None
                self._kvpair_order.remove_node(node)
            del self._kvpair_elements[key]
            return

        if name_token is not None:
            # Indirection between original_node and node for mypy's sake
            original_node = self._find_node_via_name_token(name_token, field_list)
            if original_node is None:
                msg = 'The field "{key}" is present but key used to access it is not.'
                raise KeyError(msg.format(key=key))
            node = original_node
        else:
            assert idx is not None
            try:
                node = field_list[idx]
            except KeyError:
                msg = 'The field "{key}" is present, but the index "{idx}" was invalid.'
                raise KeyError(msg.format(key=key, idx=idx))

        self._full_size_cache = None
        if len(field_list) == 1:
            del self._kvpair_elements[key]
        else:
            field_list.remove(node)
        node.value.parent_element = None
        self._kvpair_order.remove_node(node)

    def sort_fields(self, key=None):
        # type: (Optional[Callable[[str], Any]]) -> None
        """Re-order all fields

        :param key: Provide a key function (same semantics as for sorted).   Keep in mind that
          the module preserve the cases for field names - in generally, callers are recommended
          to use "lower()" to normalize the case.
        """

        if key is None:
            key = default_field_sort_key

        # Work around mypy that cannot seem to shred the Optional notion
        # without this little indirection
        key_impl = key

        def _actual_key(kvpair):
            # type: (Deb822KeyValuePairElement) -> Any
            return key_impl(kvpair.field_name)

        for last_kvpair in reversed(self._kvpair_order):
            if last_kvpair.value_element.add_final_newline_if_missing():
                self._full_size_cache = None
            break

        sorted_kvpair_list = sorted(self._kvpair_order, key=_actual_key)
        self._kvpair_order = LinkedList()
        self._kvpair_elements = {}
        self._init_kvpair_fields(sorted_kvpair_list)


class Deb822FileElement(Deb822Element):
    """Represents the entire deb822 file"""

    def __init__(self, token_and_elements):
        # type: (LinkedList[TokenOrElement]) -> None
        super().__init__()
        self._token_and_elements = token_and_elements
        self._init_parent_of_parts()

    @classmethod
    def new_empty_file(cls):
        # type: () -> Deb822FileElement
        """Creates a new Deb822FileElement with no contents

        Note that a deb822 file must be non-empty to be considered valid
        """
        return cls(LinkedList())

    @property
    def is_valid_file(self):
        # type: () -> bool
        """Returns true if the file is valid

        Invalid elements include error elements (Deb822ErrorElement) but also
        issues such as paragraphs with duplicate fields or "empty" files
        (a valid deb822 file contains at least one paragraph).
        """
        had_paragraph = False
        for paragraph in self:
            had_paragraph = True
            if not paragraph or paragraph.has_duplicate_fields:
                return False

        if not had_paragraph:
            return False

        return self.find_first_error_element() is None

    def find_first_error_element(self):
        # type: () -> Optional[Deb822ErrorElement]
        """Returns the first Deb822ErrorElement (or None) in the file"""
        return next(
            iter(self.iter_recurse(only_element_or_token_type=Deb822ErrorElement)), None
        )

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

    def iter_parts(self):
        # type: () -> Iterable[TokenOrElement]
        yield from self._token_and_elements

    def insert(self, idx, para):
        # type: (int, Deb822ParagraphElement) -> None
        """Inserts a paragraph into the file at the given "index" of paragraphs

        Note that if the index is between two paragraphs containing a "free
        floating" comment (e.g. paragrah/start-of-file, empty line, comment,
        empty line, paragraph) then it is unspecified which "side" of the
        comment the new paragraph will appear and this may change between
        versions of python-debian.


        >>> original = '''
        ... Package: libfoo-dev
        ... Depends: libfoo1 (= ${binary:Version}), ${shlib:Depends}, ${misc:Depends}
        ... '''.lstrip()
        >>> deb822_file = parse_deb822_file(original.splitlines())
        >>> para1 = Deb822ParagraphElement.new_empty_paragraph()
        >>> para1["Source"] = "foo"
        >>> para1["Build-Depends"] = "debhelper-compat (= 13)"
        >>> para2 = Deb822ParagraphElement.new_empty_paragraph()
        >>> para2["Package"] = "libfoo1"
        >>> para2["Depends"] = "${shlib:Depends}, ${misc:Depends}"
        >>> deb822_file.insert(0, para1)
        >>> deb822_file.insert(1, para2)
        >>> expected = '''
        ... Source: foo
        ... Build-Depends: debhelper-compat (= 13)
        ...
        ... Package: libfoo1
        ... Depends: ${shlib:Depends}, ${misc:Depends}
        ...
        ... Package: libfoo-dev
        ... Depends: libfoo1 (= ${binary:Version}), ${shlib:Depends}, ${misc:Depends}
        ... '''.lstrip()
        >>> deb822_file.dump() == expected
        True
        """

        anchor_node = None
        needs_newline = True
        self._full_size_cache = None
        if idx == 0:
            # Special-case, if idx is 0, then we insert it before everything else.
            # This is mostly a cosmetic choice for corner cases involving free-floating
            # comments in the file.
            if not self._token_and_elements:
                self.append(para)
                return
            anchor_node = self._token_and_elements.head_node
            needs_newline = bool(self._token_and_elements)
        else:
            i = 0
            for node in self._token_and_elements.iter_nodes():
                entry = node.value
                if isinstance(entry, Deb822ParagraphElement):
                    i += 1
                if idx == i - 1:
                    anchor_node = node
                    break

        if anchor_node is None:
            # Empty list or idx after the last paragraph both degenerate into append
            self.append(para)
        else:
            if needs_newline:
                # Remember to inject the "separating" newline between two paragraphs
                nl_token = self._set_parent(Deb822WhitespaceToken("\n"))
                anchor_node = self._token_and_elements.insert_before(
                    nl_token, anchor_node
                )
            self._token_and_elements.insert_before(self._set_parent(para), anchor_node)

    def append(self, paragraph):
        # type: (Deb822ParagraphElement) -> None
        """Appends a paragraph to the file

        >>> deb822_file = Deb822FileElement.new_empty_file()
        >>> para1 = Deb822ParagraphElement.new_empty_paragraph()
        >>> para1["Source"] = "foo"
        >>> para1["Build-Depends"] = "debhelper-compat (= 13)"
        >>> para2 = Deb822ParagraphElement.new_empty_paragraph()
        >>> para2["Package"] = "foo"
        >>> para2["Depends"] = "${shlib:Depends}, ${misc:Depends}"
        >>> deb822_file.append(para1)
        >>> deb822_file.append(para2)
        >>> expected = '''
        ... Source: foo
        ... Build-Depends: debhelper-compat (= 13)
        ...
        ... Package: foo
        ... Depends: ${shlib:Depends}, ${misc:Depends}
        ... '''.lstrip()
        >>> deb822_file.dump() == expected
        True
        """
        tail_element = self._token_and_elements.tail
        if paragraph.parent_element is not None:
            if paragraph.parent_element is self:
                raise ValueError("Paragraph is already a part of this file")
            raise ValueError("Paragraph is already part of another Deb822File")

        self._full_size_cache = None
        # We need a separating newline if there is not a whitespace token at the end of the file.
        # Note the special case where the file ends on a comment; here we insert a whitespace too
        # to be sure.  Otherwise, we would have to check that there is an empty line before that
        # comment and that is too much effort.
        if tail_element and not isinstance(tail_element, Deb822WhitespaceToken):
            self._token_and_elements.append(
                self._set_parent(Deb822WhitespaceToken("\n"))
            )
        self._token_and_elements.append(self._set_parent(paragraph))

    def remove(self, paragraph):
        # type: (Deb822ParagraphElement) -> None
        if paragraph.parent_element is not self:
            raise ValueError("Paragraph is part of a different file")
        node = None
        for node in self._token_and_elements.iter_nodes():
            if node.value is paragraph:
                break
        if node is None:
            raise RuntimeError("unable to find paragraph")
        self._full_size_cache = None
        previous_node = node.previous_node
        next_node = node.next_node
        self._token_and_elements.remove_node(node)
        if next_node is None:
            if previous_node and isinstance(previous_node.value, Deb822WhitespaceToken):
                self._token_and_elements.remove_node(previous_node)
        else:
            if isinstance(next_node.value, Deb822WhitespaceToken):
                self._token_and_elements.remove_node(next_node)
        paragraph.parent_element = None

    def _set_parent(self, t):
        # type: (TE) -> TE
        t.parent_element = self
        return t

    def position_in_parent(self, *, skip_leading_comments: bool = True) -> Position:
        # Recursive base-case
        return START_POSITION

    def position_in_file(self, *, skip_leading_comments: bool = True) -> Position:
        # By definition
        return START_POSITION

    @overload
    def dump(
        self, fd  # type: IO[bytes]
    ):
        # type: (...) -> None
        pass

    @overload
    def dump(self):
        # type: () -> str
        pass

    def dump(
        self, fd=None  # type: Optional[IO[bytes]]
    ):
        # type: (...) -> Optional[str]
        if fd is None:
            return "".join(t.text for t in self.iter_tokens())
        for token in self.iter_tokens():
            fd.write(token.text.encode("utf-8"))
        return None


_combine_error_tokens_into_elements = combine_into_replacement(
    Deb822ErrorToken, Deb822ErrorElement
)
_combine_comment_tokens_into_elements = combine_into_replacement(
    Deb822CommentToken, Deb822CommentElement
)
_combine_vl_elements_into_value_elements = combine_into_replacement(
    Deb822ValueLineElement, Deb822ValueElement
)
_combine_kvp_elements_into_paragraphs = combine_into_replacement(
    Deb822KeyValuePairElement,
    Deb822ParagraphElement,
    constructor=Deb822ParagraphElement.from_kvpairs,
)


def _parsed_value_render_factory(discard_comments):
    # type: (bool) -> Callable[[Deb822ParsedValueElement], str]
    return (
        Deb822ParsedValueElement.convert_to_text_without_comments
        if discard_comments
        else Deb822ParsedValueElement.convert_to_text
    )


LIST_SPACE_SEPARATED_INTERPRETATION = ListInterpretation(
    whitespace_split_tokenizer,
    _parse_whitespace_list_value,
    Deb822ParsedValueElement,
    Deb822SemanticallySignificantWhiteSpace,
    lambda: Deb822SpaceSeparatorToken(" "),
    _parsed_value_render_factory,
)
LIST_COMMA_SEPARATED_INTERPRETATION = ListInterpretation(
    comma_split_tokenizer,
    _parse_comma_list_value,
    Deb822ParsedValueElement,
    Deb822CommaToken,
    Deb822CommaToken,
    _parsed_value_render_factory,
)
LIST_UPLOADERS_INTERPRETATION = ListInterpretation(
    comma_split_tokenizer,
    _parse_uploaders_list_value,
    Deb822ParsedValueElement,
    Deb822CommaToken,
    Deb822CommaToken,
    _parsed_value_render_factory,
)


def _non_end_of_line_token(v):
    # type: (TokenOrElement) -> bool
    # Consume tokens until the newline
    return not isinstance(v, Deb822WhitespaceToken) or v.text != "\n"


def _build_value_line(
    token_stream,  # type: Iterable[Union[TokenOrElement, Deb822CommentElement]]
):
    # type: (...) -> Iterable[Union[TokenOrElement, Deb822ValueLineElement]]
    """Parser helper - consumes tokens part of a Deb822ValueEntryElement and turns them into one"""
    buffered_stream = BufferingIterator(token_stream)

    # Deb822ValueLineElement is a bit tricky because of how we handle whitespace
    # and comments.
    #
    # In relation to comments, then only continuation lines can have comments.
    # If there is a comment before a "K: V" line, then the comment is associated
    # with the field rather than the value.
    #
    # On the whitespace front, then we separate syntactical mandatory whitespace
    # from optional whitespace.  As an example:
    #
    # """
    # # some comment associated with the Depends field
    # Depends:_foo_$
    # # some comment associated with the line containing "bar"
    # !________bar_$
    # """
    #
    # Where "$" and "!" represents mandatory whitespace (the newline and the first
    # space are required for the file to be parsed correctly), where as "_" is
    # "optional" whitespace (from a syntactical point of view).
    #
    # This distinction enable us to facilitate APIs for easy removal/normalization
    # of redundant whitespaces without having programmers worry about trashing
    # the file.
    #
    #

    comment_element = None
    continuation_line_token = None
    token = None  # type: Optional[TokenOrElement]

    for token in buffered_stream:
        start_of_value_entry = False
        if isinstance(token, Deb822ValueContinuationToken):
            continuation_line_token = token
            start_of_value_entry = True
            token = None
        elif isinstance(token, Deb822FieldSeparatorToken):
            start_of_value_entry = True
        elif isinstance(token, Deb822CommentElement):
            next_token = buffered_stream.peek()
            # If the next token is a continuation line token, then this comment
            # belong to a value and we might as well just start the value
            # parsing now.
            #
            # Note that we rely on this behaviour to avoid emitting the comment
            # token (failing to do so would cause the comment to appear twice
            # in the file).
            if isinstance(next_token, Deb822ValueContinuationToken):
                start_of_value_entry = True
                comment_element = token
                token = None
                # Use next with None to avoid raising StopIteration inside a generator
                # It won't happen, but pylint cannot see that, so we do this instead.
                continuation_line_token = cast(
                    "Deb822ValueContinuationToken", next(buffered_stream, None)
                )
                assert continuation_line_token is not None

        if token is not None:
            yield token
        if start_of_value_entry:
            tokens_in_value = list(buffered_stream.takewhile(_non_end_of_line_token))
            eol_token = cast("Deb822WhitespaceToken", next(buffered_stream, None))
            assert eol_token is None or eol_token.text == "\n"
            leading_whitespace = None
            trailing_whitespace = None
            # "Depends:\n foo" would cause tokens_in_value to be empty for the
            # first "value line" (the empty part between ":" and "\n")
            if tokens_in_value:
                # Another special-case, "Depends: \n foo" (i.e. space after colon)
                # should not introduce an IndexError
                if isinstance(tokens_in_value[-1], Deb822WhitespaceToken):
                    trailing_whitespace = cast(
                        "Deb822WhitespaceToken", tokens_in_value.pop()
                    )
                if tokens_in_value and isinstance(
                    tokens_in_value[-1], Deb822WhitespaceToken
                ):
                    leading_whitespace = cast(
                        "Deb822WhitespaceToken", tokens_in_value[0]
                    )
                    tokens_in_value = tokens_in_value[1:]
            yield Deb822ValueLineElement(
                comment_element,
                continuation_line_token,
                leading_whitespace,
                tokens_in_value,
                trailing_whitespace,
                eol_token,
            )
            comment_element = None
            continuation_line_token = None


def _build_field_with_value(
    token_stream,  # type: Iterable[Union[TokenOrElement, Deb822ValueElement]]
):
    # type: (...) -> Iterable[Union[TokenOrElement, Deb822KeyValuePairElement]]
    buffered_stream = BufferingIterator(token_stream)
    for token_or_element in buffered_stream:
        start_of_field = False
        comment_element = None
        if isinstance(token_or_element, Deb822FieldNameToken):
            start_of_field = True
        elif isinstance(token_or_element, Deb822CommentElement):
            comment_element = token_or_element
            next_token = buffered_stream.peek()
            start_of_field = isinstance(next_token, Deb822FieldNameToken)
            if start_of_field:
                # Remember to consume the field token
                try:
                    token_or_element = next(buffered_stream)
                except StopIteration:  # pragma: no cover
                    raise AssertionError

        if start_of_field:
            field_name = token_or_element
            separator = next(buffered_stream, None)
            value_element = next(buffered_stream, None)
            if separator is None or value_element is None:
                # Early EOF - should not be possible with how the tokenizer works
                # right now, but now it is future-proof.
                if comment_element:
                    yield comment_element
                error_elements = [field_name]
                if separator is not None:
                    error_elements.append(separator)
                yield Deb822ErrorElement(error_elements)
                return

            if isinstance(separator, Deb822FieldSeparatorToken) and isinstance(
                value_element, Deb822ValueElement
            ):
                yield Deb822KeyValuePairElement(
                    comment_element,
                    cast("Deb822FieldNameToken", field_name),
                    separator,
                    value_element,
                )
            else:
                # We had a parse error, consume until the newline.
                error_tokens = [token_or_element]  # type: List[TokenOrElement]
                error_tokens.extend(buffered_stream.takewhile(_non_end_of_line_token))
                nl = buffered_stream.peek()
                # Take the newline as well if present
                if nl and isinstance(nl, Deb822NewlineAfterValueToken):
                    next(buffered_stream, None)
                    error_tokens.append(nl)
                yield Deb822ErrorElement(error_tokens)
        else:
            # Token is not part of a field, emit it as-is
            yield token_or_element


def _abort_on_error_tokens(sequence):
    # type: (Iterable[TokenOrElement]) -> Iterable[TokenOrElement]
    line_no = 1
    for token in sequence:
        # We are always called while the sequence consists entirely of tokens
        if token.is_error:
            error_as_text = token.convert_to_text().replace("\n", "\\n")
            raise SyntaxOrParseError(
                'Syntax or Parse error on or near line {line_no}: "{error_as_text}"'.format(
                    error_as_text=error_as_text, line_no=line_no
                )
            )
        line_no += token.convert_to_text().count("\n")
        yield token


def parse_deb822_file(
    sequence,  # type: Union[Iterable[Union[str, bytes]], str]
    *,
    accept_files_with_error_tokens=False,  # type: bool
    accept_files_with_duplicated_fields=False,  # type: bool
    encoding="utf-8",  # type: str
):
    # type: (...) -> Deb822FileElement
    """

    :param sequence: An iterable over lines of str or bytes (an open file for
      reading will do).  If line endings are provided in the input, then they
      must be present on every line (except the last) will be preserved as-is.
      If omitted and the content is at least 2 lines, then parser will assume
      implicit newlines.
    :param accept_files_with_error_tokens: If True, files with critical syntax
      or parse errors will be returned as "successfully" parsed. Usually,
      working on files with this kind of errors are not desirable as it is
      hard to make sense of such files (and they might in fact not be a deb822
      file at all).  When set to False (the default) a ValueError is raised if
      there is a critical syntax or parse error.
      Note that duplicated fields in a paragraph is not considered a critical
      parse error by this parser as the implementation can gracefully cope
      with these. Use accept_files_with_duplicated_fields to determine if
      such files should be accepted.
    :param accept_files_with_duplicated_fields: If True, then
      files containing paragraphs with duplicated fields will be returned as
      "successfully" parsed even though they are invalid according to the
      specification.  The paragraphs will prefer the first appearance of the
      field unless caller explicitly requests otherwise (e.g., via
      Deb822ParagraphElement.configured_view).  If False, then this method
      will raise a ValueError if any duplicated fields are seen inside any
      paragraph.
    :param encoding: The encoding to use (this is here to support Deb822-like
       APIs, new code should not use this parameter).
    """

    if isinstance(sequence, (str, bytes)):
        # Match the deb822 API.
        sequence = sequence.splitlines(True)

    # The order of operations are important here.  As an example,
    # _build_value_line assumes that all comment tokens have been merged
    # into comment elements.  Likewise, _build_field_and_value assumes
    # that value tokens (along with their comments) have been combined
    # into elements.
    tokens = tokenize_deb822_file(
        sequence, encoding=encoding
    )  # type: Iterable[TokenOrElement]
    if not accept_files_with_error_tokens:
        tokens = _abort_on_error_tokens(tokens)
    tokens = _combine_comment_tokens_into_elements(tokens)
    tokens = _build_value_line(tokens)
    tokens = _combine_vl_elements_into_value_elements(tokens)
    tokens = _build_field_with_value(tokens)
    tokens = _combine_kvp_elements_into_paragraphs(tokens)
    # Combine any free-floating error tokens into error elements.  We do
    # this last as it enables other parts of the parser to include error
    # tokens in their error elements if they discover something is wrong.
    tokens = _combine_error_tokens_into_elements(tokens)

    deb822_file = Deb822FileElement(LinkedList(tokens))

    if not accept_files_with_duplicated_fields:
        for no, paragraph in enumerate(deb822_file):
            if isinstance(paragraph, Deb822DuplicateFieldsParagraphElement):
                field_names = set()
                dup_field = None
                for field in paragraph.keys():
                    field_name, _, _ = _unpack_key(field)
                    # assert for mypy
                    assert isinstance(field_name, str)
                    if field_name in field_names:
                        dup_field = field_name
                        break
                    field_names.add(field_name)
                if dup_field is not None:
                    msg = 'Duplicate field "{dup_field}" in paragraph number {no}'
                    raise ValueError(msg.format(dup_field=dup_field, no=no))

    return deb822_file


if __name__ == "__main__":  # pragma: no cover
    import doctest

    doctest.testmod()