Fix the remaining latent bugs found by the type checker

faust/__init__.py

@@ -19,7 +19,6 @@
 #  faust/agents.py         - Agents use all of the above.
 # --- ~~~~~ ~ ~  ~           ~             ~   ~                   ~
 import os
-import re
 import sys
 import typing
 
@@ -30,6 +29,8 @@
 
 from typing import Any, Mapping, NamedTuple, Optional, Sequence, Tuple
 
+from packaging.version import InvalidVersion, Version
+
 __version__ = version("faust-streaming")
 __author__ = "Robinhood Markets, Inc."
 __contact__ = "schrohm@gmail.com, vpatki@wayfair.com, williambbarnhart@gmail.com"
@@ -50,23 +51,43 @@ class VersionInfo(NamedTuple):
 version_info_t = VersionInfo  # XXX compat
 
 
-# bumpversion can only search for {current_version}
-# so we have to parse the version here.
-_match = re.match(r"^(?P<prefix>v)?(?P<version>[^\+]+)(?P<suffix>.*)?$", __version__)
-if _match is None:  # pragma: no cover
-    raise RuntimeError("THIS IS A BROKEN RELEASE!")
-_temp = _match.groups()
-# XXX This is broken and so is the public ``faust.version_info``: the regex
-# yields the strings ``(prefix, version, suffix)``, which land positionally in
-# the ``(major, minor, micro)`` int fields.  So ``.major`` is the ``'v'``
-# prefix or :const:`None`, ``.minor`` is the entire version string and
-# ``.micro`` is the suffix -- e.g. ``VersionInfo(major=None,
-# minor='0.11.5', micro='')`` instead of ``(0, 11, 5)``.
-# Left as-is because fixing it changes what ``faust.VERSION`` holds.
-VERSION = version_info = VersionInfo(*_temp)  # type: ignore[arg-type]
-del _match
-del _temp
-del re
+def _parse_version(version_string: str) -> VersionInfo:
+    """Parse a version string into a :class:`VersionInfo` tuple.
+
+    The PEP 440 grammar is delegated to :mod:`packaging`, the same parser
+    pip and setuptools use, rather than maintained here.  ``release``
+    supplies ``major``/``minor``/``micro`` (missing components default to
+    ``0``), and a pre/dev/post segment supplies ``releaselevel`` and
+    ``serial`` separately -- ``0.11.5rc1`` gives ``releaselevel='rc'``,
+    ``serial='1'`` -- mirroring what those field names mean on
+    :data:`sys.version_info`.
+
+    Only the parts :class:`VersionInfo` has fields for are kept.  A local
+    segment (the ``+g1234`` of ``0.11.5.dev1+g1234``) and any component
+    past the third are dropped; :data:`faust.__version__` remains the
+    full, unmodified string.
+
+    Never raises: a version :mod:`packaging` cannot parse ends up as the
+    ``releaselevel`` of ``VersionInfo(0, 0, 0)``.
+    """
+    releaselevel: Optional[str]
+    serial: Optional[str]
+    try:
+        parsed = Version(version_string)
+    except InvalidVersion:
+        return VersionInfo(0, 0, 0, version_string or None)
+    if parsed.pre is not None:  # 1.0rc1 / 1.0a2 / 1.0b3
+        releaselevel, serial = parsed.pre[0], str(parsed.pre[1])
+    elif parsed.dev is not None:  # 1.0.dev1
+        releaselevel, serial = "dev", str(parsed.dev)
+    elif parsed.post is not None:  # 1.0.post2
+        releaselevel, serial = "post", str(parsed.post)
+    else:
+        releaselevel, serial = None, None
+    return VersionInfo(parsed.major, parsed.minor, parsed.micro, releaselevel, serial)
+
+
+VERSION = version_info = _parse_version(__version__)
 
 
 # This is here to support setting the --datadir argument
@@ -307,5 +328,6 @@ def __dir__(self) -> Sequence[str]:
         "version_info_t": version_info_t,
         "version_info": version_info,
         "VERSION": VERSION,
+        "_parse_version": _parse_version,
     }
 )

faust/auth.py

@@ -130,15 +130,10 @@ def __init__(
         cadata: Optional[str] = None,
     ) -> None:
         if context is None:
+            if purpose is None:
+                purpose = ssl.Purpose.SERVER_AUTH
             context = ssl.create_default_context(
-                # XXX ``purpose`` defaults to None here, but
-                # ``ssl.create_default_context`` requires an ``ssl.Purpose``
-                # and raises ``TypeError`` on None -- so ``SSLCredentials()``
-                # with no explicit ``purpose`` cannot build a context at all.
-                # Real bug, kept as-is: fixing it changes the default TLS
-                # purpose of a security-relevant public API, which is out of
-                # scope for a typing pass.
-                purpose=purpose,  # type: ignore[arg-type]
+                purpose=purpose,
                 cafile=cafile,
                 capath=capath,
                 cadata=cadata,

faust/cli/base.py

@@ -797,7 +797,10 @@ class AppCommand(Command):
 
     abstract: ClassVar[bool] = True
 
-    app: AppT
+    #: The app this command runs for, or :const:`None` when the command
+    #: sets ``require_app = False`` and was invoked without ``-A``.
+    #: Use the :attr:`app` property to access it when an app is required.
+    _app: Optional[AppT] = None
 
     require_app = True
 
@@ -844,25 +847,44 @@ def __init__(
     ) -> None:
         super().__init__(ctx)
 
-        self.app = self._finalize_app(getattr(ctx.find_root(), "app", None))
+        self._app = self._finalize_app(getattr(ctx.find_root(), "app", None))
         self.args = args
         self.kwargs = kwargs
-        self.key_serializer = key_serializer or self.app.conf.key_serializer
-        self.value_serializer = value_serializer or self.app.conf.value_serializer
+        if self._app is not None:
+            conf = self._app.conf
+            self.key_serializer = key_serializer or conf.key_serializer
+            self.value_serializer = value_serializer or conf.value_serializer
+        else:
+            # ``require_app = False`` command invoked without ``-A``:
+            # there is no app to take the default codecs from.
+            self.key_serializer = key_serializer
+            self.value_serializer = value_serializer
+
+    @property
+    def app(self) -> AppT:
+        """Return the app this command runs for.
+
+        Raises:
+            UsageError: if the command was invoked without an app.
+                Only commands setting ``require_app = False`` can get
+                this far without one.
+        """
+        app = self._app
+        if app is None:
+            raise self.UsageError("Need to specify app using -A parameter")
+        return app
+
+    @app.setter
+    def app(self, app: AppT) -> None:
+        self._app = app
 
-    def _finalize_app(self, app: Optional[AppT]) -> AppT:
+    def _finalize_app(self, app: Optional[AppT]) -> Optional[AppT]:
         if app is not None:
             return self._finalize_concrete_app(app)
         else:
-            # XXX ``_app_from_str`` returns None for ``require_app = False``
-            # commands (see faust/cli/completion.py) that were invoked without
-            # ``-A``, so this really can hand back None -- and
-            # ``AppCommand.__init__`` dereferences ``self.app.conf``
-            # unconditionally right after, raising ``AttributeError: 'NoneType'
-            # object has no attribute 'conf'``.  Real bug; fixing it means
-            # deciding what such commands should do without an app, which is
-            # out of scope for a typing pass.
-            return self._app_from_str(self.state.app)  # type: ignore[return-value]
+            # Returns None for ``require_app = False`` commands
+            # (see faust/cli/completion.py) invoked without ``-A``.
+            return self._app_from_str(self.state.app)
 
     def _app_from_str(self, appstr: Optional[str] = None) -> Optional[AppT]:
         if appstr:
@@ -899,7 +921,9 @@ def _detect_main_package(self, argv: List[str]) -> str:  # pragma: no cover
     async def on_stop(self) -> None:
         """Call after command executed."""
         await super().on_stop()
-        app = cast(_App, self.app)
+        if self._app is None:
+            return  # command ran without an app: nothing to clean up.
+        app = cast(_App, self._app)
         # If command started the producer, we should also stop that
         #   - this will flush any buffers before exiting.
         if app._producer is not None and app._producer.started:
@@ -1014,7 +1038,10 @@ def abbreviate_fqdn(self, name: str, *, prefix: str = "") -> str:
     @property
     def blocking_timeout(self) -> float:
         """Return the blocking timeout used for this command."""
-        return self._blocking_timeout or self.app.conf.blocking_timeout
+        if self._app is None:
+            # command ran without an app: use the Command default.
+            return self._blocking_timeout or 0.0
+        return self._blocking_timeout or self._app.conf.blocking_timeout
 
     @blocking_timeout.setter
     def blocking_timeout(self, timeout: float) -> None:

faust/tables/base.py

@@ -388,23 +388,10 @@ async def _del_old_keys(self) -> None:
             while timestamps and window.stale(timestamps[0], time.time()):
                 timestamp = heappop(timestamps)
                 triggered_windows = [
-                    # XXX bug: this lookup can never hit.
                     # ``_partition_timestamp_keys`` is keyed by
-                    # ``(partition, range_end)`` -- a ``(int, float)`` pair,
-                    # written that way in ``_maybe_set_key_ttl`` and read that
-                    # way in ``_maybe_del_key_ttl``.  Here it is looked up by
-                    # ``(partition, window_range)`` where ``window_range`` is
-                    # the ``(start, end)`` tuple, so no key ever matches and
-                    # ``triggered_windows`` is always ``[None, ...]``.  The
-                    # consequence is that ``window_data`` stays empty and
-                    # ``on_window_close`` never receives the aggregated window
-                    # data, only the raw per-key value.  The correct key is
-                    # ``(partition, window_range[1])``; that is a behaviour
-                    # change, so it is not made here and the type error is
-                    # only silenced.
-                    self._partition_timestamp_keys.get(
-                        (partition, window_range)  # type: ignore[arg-type]
-                    )
+                    # ``(partition, range_end)``, so look up the end of
+                    # each ``(start, end)`` range.
+                    self._partition_timestamp_keys.get((partition, window_range[1]))
                     for window_range in self._window_ranges(timestamp)
                 ]
                 keys_to_remove = self._partition_timestamp_keys.pop(
@@ -420,9 +407,22 @@ async def _del_old_keys(self) -> None:
                                 # it is not related to window's timestamp
                                 # windows are in format:
                                 # (key, (window_start, window_end))
-                                window_data.setdefault(processed_window[0], []).extend(
-                                    self.data.get(processed_window, [])
-                                )
+                                #
+                                # Only list values are aggregated across the
+                                # overlapping windows: a list of events is the
+                                # shape this exists to accumulate, and it is
+                                # the only one that can be concatenated without
+                                # inventing semantics.  A counter (``default=int``)
+                                # is not iterable and would raise here; a string
+                                # is iterable and would be shredded into
+                                # characters.  Those keys are left out of
+                                # ``window_data`` and delivered below as the raw
+                                # per-key value instead.
+                                window_value = self.data.get(processed_window)
+                                if isinstance(window_value, list):
+                                    window_data.setdefault(
+                                        processed_window[0], []
+                                    ).extend(window_value)
 
                     for key_to_remove in keys_to_remove:
                         value = self.data.pop(key_to_remove, None)

faust/tables/recovery.py

@@ -811,18 +811,10 @@ async def detect_aborted_tx() -> None:
                     and offsets[tp] is not None
                     and offsets[tp] < highwater
                 ):
-                    # XXX bug: ConsumerT.position is declared
-                    # ``Optional[int]`` and really does return None when the
-                    # partition has no position yet, so this comparison raises
-                    # TypeError on that path (swallowed by the ``except
-                    # Exception`` in the caller's loop, which then skips the
-                    # aborted-tx fixup).  Guarding for None would change which
-                    # events reach the caller, so the behaviour is kept and
-                    # only the type error is silenced.
-                    if (
-                        await self.app.consumer.position(tp)  # type: ignore[operator]
-                        >= highwater
-                    ):
+                    # The partition may have no position yet, in which case
+                    # there is nothing to compare against and we skip it.
+                    position = await self.app.consumer.position(tp)
+                    if position is not None and position >= highwater:
                         logger.info(f"Aborted tx until highwater for {tp}")
                         offsets[tp] = highwater
 
@@ -848,58 +840,52 @@ async def detect_aborted_tx() -> None:
                 tp = message.tp
                 offset = message.offset
                 logger.debug(f"Recovery message topic {tp} offset {offset}")
-                offsets: Counter[TP]
-                bufsize = buffer_sizes.get(tp)
-                is_active = False
-                if tp in active_tps:
-                    is_active = True
-                    table = tp_to_table[tp]
-                    offsets = active_offsets
-                    if bufsize is None:
-                        bufsize = buffer_sizes[tp] = table.recovery_buffer_size
-                    active_events_received_at[tp] = now
-                elif tp in standby_tps:
-                    table = tp_to_table[tp]
-                    offsets = standby_offsets
-                    if bufsize is None:
-                        bufsize = buffer_sizes[tp] = table.standby_buffer_size
-                        standby_events_received_at[tp] = now
-                else:
+                if tp not in active_tps and tp not in standby_tps:
+                    # Not a partition we are recovering: skip the event,
+                    # but keep the bookkeeping below running.
                     logger.warning(f"recovery unknown topic {tp} offset {offset}")
-
-                seen_offset = offsets.get(tp, None)
-                logger.debug(
-                    f"seen offset for {tp} is {seen_offset} message offset {offset}"
-                )
-                if seen_offset is None or offset > seen_offset:
-                    offsets[tp] = offset
-                    buf = buffers[table]
-                    buf.append(event)
-                    await table.on_changelog_event(event)
-                    # XXX bug: the ``else`` branch above only logs a warning
-                    # and falls through, so an event for a TP that is neither
-                    # active nor standby reaches here with ``table``,
-                    # ``offsets`` and ``bufsize`` still holding the PREVIOUS
-                    # iteration's values -- the event is then applied to an
-                    # unrelated table (or raises UnboundLocalError if it is
-                    # the first event of the loop).  That is why ``bufsize``
-                    # is still ``Optional[int]`` here and the comparison can
-                    # raise TypeError.  Fixing it means skipping the untracked
-                    # TP, which is a behaviour change, so it is left as is.
-                    if len(buf) >= bufsize:  # type: ignore[operator]
-                        table.apply_changelog_batch(buf)
-                        buf.clear()
-                        self._last_flush_at = now
-                    now_after = monotonic()
-
-                    if is_active:
-                        last_processed_at = self._last_active_event_processed_at
-                        if last_processed_at is not None:
-                            processing_times.append(now_after - last_processed_at)
-                            max_samples = self.num_samples_required_for_estimate
-                            if len(processing_times) > max_samples:
-                                processing_times.popleft()
-                        self._last_active_event_processed_at = now_after
+                else:
+                    offsets: Counter[TP]
+                    bufsize = buffer_sizes.get(tp)
+                    is_active = False
+                    if tp in active_tps:
+                        is_active = True
+                        table = tp_to_table[tp]
+                        offsets = active_offsets
+                        if bufsize is None:
+                            bufsize = buffer_sizes[tp] = table.recovery_buffer_size
+                        active_events_received_at[tp] = now
+                    else:
+                        table = tp_to_table[tp]
+                        offsets = standby_offsets
+                        if bufsize is None:
+                            bufsize = buffer_sizes[tp] = table.standby_buffer_size
+                            standby_events_received_at[tp] = now
+
+                    seen_offset = offsets.get(tp, None)
+                    logger.debug(
+                        f"seen offset for {tp} is {seen_offset} "
+                        f"message offset {offset}"
+                    )
+                    if seen_offset is None or offset > seen_offset:
+                        offsets[tp] = offset
+                        buf = buffers[table]
+                        buf.append(event)
+                        await table.on_changelog_event(event)
+                        if len(buf) >= bufsize:
+                            table.apply_changelog_batch(buf)
+                            buf.clear()
+                            self._last_flush_at = now
+                        now_after = monotonic()
+
+                        if is_active:
+                            last_processed_at = self._last_active_event_processed_at
+                            if last_processed_at is not None:
+                                processing_times.append(now_after - last_processed_at)
+                                max_samples = self.num_samples_required_for_estimate
+                                if len(processing_times) > max_samples:
+                                    processing_times.popleft()
+                            self._last_active_event_processed_at = now_after
 
                 await _maybe_signal_recovery_end()