Enable typeguard during build time and tests, fix issues found

It is also possible to use typeguard in osc-wrapper.py
by setting OSC_TYPEGUARD=1.

Author: dmach

.github/workflows/tests.yaml

@@ -151,6 +151,7 @@ jobs:
             sudo apt-get -y --no-install-recommends install libwww-perl
             # obs-scm-bridge is not available as a package at the moment, install it from github
             sudo pip3 config set global.break-system-packages 1
+            sudo pip3 install typeguard
             sudo pip3 install git+https://github.com/openSUSE/obs-scm-bridge
             sudo chmod a+x /usr/local/lib/*/*/obs_scm_bridge
             sudo mkdir -p /usr/lib/obs/service
@@ -173,4 +174,4 @@ jobs:
       - name: "Run tests"
         run: |
           cd behave
-          behave -Dosc=../osc-wrapper.py -Dgit-obs=../git-obs.py -Dgit-osc-precommit-hook=../git-osc-precommit-hook.py -Dpodman_max_containers=2
+          OSC_TYPEGUARD=1 behave -Dosc=../osc-wrapper.py -Dgit-obs=../git-obs.py -Dgit-osc-precommit-hook=../git-osc-precommit-hook.py -Dpodman_max_containers=2

contrib/osc.spec

@@ -35,6 +35,13 @@
 %bcond_with fdupes
 %endif
 
+# use typeguard during build on distros where typeguard is available
+%if (0%{?suse_version} > 1500 || 0%{?fedora} >= 37)
+%bcond_without typeguard
+%else
+%bcond_with typeguard
+%endif
+
 # the macro exists only on openSUSE based distros
 %if %{undefined python3_fix_shebang}
 %define python3_fix_shebang %nil
@@ -87,6 +94,9 @@ BuildRequires:  %{use_python_pkg}-cryptography
 BuildRequires:  %{use_python_pkg}-devel >= 3.6
 BuildRequires:  %{use_python_pkg}-rpm
 BuildRequires:  %{use_python_pkg}-setuptools
+%if %{with typeguard}
+BuildRequires:  %{use_python_pkg}-typeguard
+%endif
 BuildRequires:  %{use_python_pkg}-urllib3
 BuildRequires:  %{yaml_pkg}
 BuildRequires:  diffstat

osc-wrapper.py

@@ -4,6 +4,29 @@
 This wrapper allows osc to be called from the source directory during development.
 """
 
+
+import os
+
+
+USE_TYPEGUARD = os.environ.get("OSC_TYPEGUARD", "1").lower() in ("1", "true", "on")
+
+if USE_TYPEGUARD:
+    try:
+        from typeguard import install_import_hook
+    except ImportError:
+        install_import_hook = None
+
+    if install_import_hook is None:
+        try:
+            from typeguard.importhook import install_import_hook
+        except ImportError:
+            install_import_hook = None
+
+    if install_import_hook:
+        # install typeguard import hook only if available
+        install_import_hook("osc")
+
+
 import osc.babysitter
 
 osc.babysitter.main()

osc/commandline.py

@@ -4988,6 +4988,8 @@ def do_rdiff(self, subcmd, opts, *args):
                     rev2 = -rev - 1
                 else:
                     return
+                rev1 = str(rev1)
+                rev2 = str(rev2)
             except:
                 print(f'Revision \'{opts.change}\' not an integer', file=sys.stderr)
                 return

osc/core.py

@@ -1339,7 +1339,7 @@ def show_package_trigger_reason(apiurl: str, prj: str, pac: str, repo: str, arch
         raise
 
 
-def show_package_meta(apiurl: str, prj: str, pac: str, meta=False, blame=None):
+def show_package_meta(apiurl: str, prj: str, pac: str, meta=False, blame=None) -> List[bytes]:
     query: Dict[str, Union[str, int]] = {}
     if meta:
         query['meta'] = 1
@@ -2948,12 +2948,12 @@ def get_source_file_diff(dir, filename, rev, oldfilename=None, olddir=None, orig
 
 def server_diff(
     apiurl: str,
-    old_project: str,
-    old_package: str,
-    old_revision: str,
+    old_project: Optional[str],
+    old_package: Optional[str],
+    old_revision: Optional[str],
     new_project: str,
     new_package: str,
-    new_revision: str,
+    new_revision: Optional[str],
     unified=False,
     missingok=False,
     meta=False,
@@ -3014,12 +3014,12 @@ def server_diff(
 
 def server_diff_noex(
     apiurl: str,
-    old_project: str,
-    old_package: str,
-    old_revision: str,
+    old_project: Optional[str],
+    old_package: Optional[str],
+    old_revision: Optional[str],
     new_project: str,
     new_package: str,
-    new_revision: str,
+    new_revision: Optional[str],
     unified=False,
     missingok=False,
     meta=False,
@@ -3317,9 +3317,9 @@ def checkout_package(
 
 
 def replace_pkg_meta(
-    pkgmeta, new_name: str, new_prj: str, keep_maintainers=False, dst_userid=None, keep_develproject=False,
+    pkgmeta: List[bytes], new_name: str, new_prj: str, keep_maintainers=False, dst_userid=None, keep_develproject=False,
     keep_lock: bool = False, keep_scmsync: bool = True,
-):
+) -> str:
     """
     update pkgmeta with new new_name and new_prj and set calling user as the
     only maintainer (unless keep_maintainers is set). Additionally remove the
@@ -3562,7 +3562,7 @@ def aggregate_pac(
 
     if meta_change:
         src_meta = show_package_meta(apiurl, src_project, src_package_meta)
-        dst_meta = replace_pkg_meta(src_meta, dst_package_meta, dst_project)
+        dst_meta = replace_pkg_meta(src_meta, dst_package_meta, dst_project).split("\n")
         meta_change = True
 
     if disable_publish:
@@ -4853,25 +4853,26 @@ def get_commitlog(
             # revision is srcmd5
             revision_list = [i for i in revision_list if i.srcmd5 == revision]
         else:
-            revision = int(revision)
+            assert revision is not None
+            revision_int = int(revision)
             if revision_is_empty(revision_upper):
-                revision_list = [i for i in revision_list if i.rev == revision]
+                revision_list = [i for i in revision_list if i.rev == revision_int]
             else:
-                revision_upper = int(revision_upper)
-                revision_list = [i for i in revision_list if i.rev <= revision_upper and i.rev >= revision]
+                revision_upper_int = int(revision_upper)
+                revision_list = [i for i in revision_list if i.rev <= revision_upper_int and i.rev >= revision_int]
 
     if format == "csv":
         f = io.StringIO()
         writer = csv.writer(f, dialect="unix")
-        for revision in reversed(revision_list):
+        for i in reversed(revision_list):
             writer.writerow(
                 (
-                    revision.rev,
-                    revision.user,
-                    revision.get_time_str(),
-                    revision.srcmd5,
-                    revision.comment,
-                    revision.requestid,
+                    i.rev,
+                    i.user,
+                    i.get_time_str(),
+                    i.srcmd5,
+                    i.comment,
+                    i.requestid,
                 )
             )
         f.seek(0)
@@ -4880,42 +4881,42 @@ def get_commitlog(
 
     if format == "xml":
         root = ET.Element("log")
-        for revision in reversed(revision_list):
+        for i in reversed(revision_list):
             entry = ET.SubElement(root, "logentry")
-            entry.attrib["revision"] = str(revision.rev)
-            entry.attrib["srcmd5"] = revision.srcmd5
-            ET.SubElement(entry, "author").text = revision.user
-            ET.SubElement(entry, "date").text = revision.get_time_str()
-            ET.SubElement(entry, "requestid").text = str(revision.requestid) if revision.requestid else ""
-            ET.SubElement(entry, "msg").text = revision.comment or ""
+            entry.attrib["revision"] = str(i.rev)
+            entry.attrib["srcmd5"] = i.srcmd5
+            ET.SubElement(entry, "author").text = i.user
+            ET.SubElement(entry, "date").text = i.get_time_str()
+            ET.SubElement(entry, "requestid").text = str(i.requestid) if i.requestid else ""
+            ET.SubElement(entry, "msg").text = i.comment or ""
         xmlindent(root)
         yield from ET.tostring(root, encoding="utf-8").decode("utf-8").splitlines()
         return
 
     if format == "text":
-        for revision in reversed(revision_list):
+        for i in reversed(revision_list):
             entry = (
-                f"r{revision.rev}",
-                revision.user,
-                revision.get_time_str(),
-                revision.srcmd5,
-                revision.version,
-                f"rq{revision.requestid}" if revision.requestid else ""
+                f"r{i.rev}",
+                i.user,
+                i.get_time_str(),
+                i.srcmd5,
+                i.version,
+                f"rq{i.requestid}" if i.requestid else ""
             )
             yield 76 * "-"
             yield " | ".join(entry)
             yield ""
-            yield revision.comment or "<no message>"
+            yield i.comment or "<no message>"
             yield ""
             if patch:
                 rdiff = server_diff(
                     apiurl,
                     prj,
                     package,
-                    revision.rev - 1,
+                    str(i.rev - 1),
                     prj,
                     package,
-                    revision.rev,
+                    str(i.rev),
                     meta=meta,
                 )
                 yield highlight_diff(rdiff).decode("utf-8", errors="replace")

osc/meter.py

@@ -102,7 +102,6 @@ def create_text_meter(*args, **kwargs) -> TextMeterBase:
 
     use_pb_fallback = kwargs.pop("use_pb_fallback", False)
 
-    meter_class: TextMeterBase
     if config.quiet:
         meter_class = NoTextMeter
     elif not have_pb_module or not config.show_download_progress or not sys.stdout.isatty() or use_pb_fallback:

osc/output/output.py

@@ -5,7 +5,9 @@
 import subprocess
 import sys
 import tempfile
+from typing import BinaryIO
 from typing import Dict
+from typing import Generator
 from typing import List
 from typing import Optional
 from typing import TextIO
@@ -137,7 +139,7 @@ def safe_print(*args, **kwargs):
     print(*args, **kwargs)
 
 
-def safe_write(file: TextIO, text: Union[str, bytes], *, add_newline: bool = False):
+def safe_write(file: Union[BinaryIO, TextIO], text: Union[str, bytes], *, add_newline: bool = False):
     """
     Run sanitize_text() on ``text`` and write it to ``file``.
 
@@ -211,7 +213,7 @@ def run_pager(message: Union[bytes, str], tmp_suffix: str = ""):
         run_external(*cmd, env=env)
 
 
-def pipe_to_pager(lines: Union[List[bytes], List[str]], *, add_newlines=False):
+def pipe_to_pager(lines: Union[List[bytes], List[str], Generator[bytes, None, None], Generator[str, None, None]], *, add_newlines=False):
     """
     Pipe ``lines`` to the pager.
     If running in a non-interactive terminal, print the data instead.

osc/util/ar.py

@@ -37,7 +37,7 @@ def __str__(self):
 class ArHdr:
     """Represents an ar header entry"""
 
-    def __init__(self, fn: bytes, date: bytes, uid: bytes, gid: bytes, mode: bytes, size: bytes, fmag: bytes, off: bytes):
+    def __init__(self, fn: bytes, date: bytes, uid: bytes, gid: bytes, mode: bytes, size: bytes, fmag: bytes, off: int):
         self.file = fn.strip()
         self.date = date.strip()
         self.uid = uid.strip()

osc/util/models.py

@@ -834,7 +834,7 @@ def xml_request(
         apiurl: str,
         path: List[str],
         query: Optional[dict] = None,
-        headers: Optional[str] = None,
+        headers: Optional[dict] = None,
         data: Optional[str] = None,
     ) -> urllib3.response.HTTPResponse:
         from ..connection import http_request

osc/util/safewriter.py

@@ -1,22 +1,39 @@
+import io
+
 # be careful when debugging this code:
 # don't add print statements when setting sys.stdout = SafeWriter(sys.stdout)...
-class SafeWriter:
+class SafeWriter(io.TextIOBase):
     """
     Safely write an (unicode) str. In case of an "UnicodeEncodeError" the
     the str is encoded with the "encoding" encoding.
     All getattr, setattr calls are passed through to the "writer" instance.
     """
 
     def __init__(self, writer, encoding='unicode_escape'):
+        super().__init__()
         self._writer = writer
         self._encoding = encoding
 
+    # TextIOBase requires overriding the following stub methods: detach, read, readline, and write
+
+    def detach(self, *args, **kwargs):
+        return self._writer.detach(*args, **kwargs)
+
+    def read(self, *args, **kwargs):
+        return self._writer.read(args, **kwargs)
+
+    def readline(self, *args, **kwargs):
+        return self._writer.readline(args, **kwargs)
+
     def write(self, s):
         try:
             self._writer.write(s)
         except UnicodeEncodeError as e:
             self._writer.write(s.encode(self._encoding))
 
+    def fileno(self, *args, **kwargs):
+        return self._writer.fileno(*args, **kwargs)
+
     def __getattr__(self, name):
         return getattr(self._writer, name)