Apply small code updates

Signed-off-by: Philippe Ombredanne <[email protected]>

Author: pombredanne

etc/scripts/utils_requirements.py

@@ -57,21 +57,25 @@ def get_required_name_version(requirement, with_unpinned=False):
     >>> assert get_required_name_version("fooA==1.2.3.DEV1") == ("fooa", "1.2.3.dev1")
     >>> assert get_required_name_version("foo==1.2.3", with_unpinned=False) == ("foo", "1.2.3")
     >>> assert get_required_name_version("foo", with_unpinned=True) == ("foo", "")
-    >>> assert get_required_name_version("foo>=1.2", with_unpinned=True) == ("foo", ""), get_required_name_version("foo>=1.2")
+    >>> expected = ("foo", ""), get_required_name_version("foo>=1.2")
+    >>> assert get_required_name_version("foo>=1.2", with_unpinned=True) == expected
     >>> try:
     ...   assert not get_required_name_version("foo", with_unpinned=False)
     ... except Exception as e:
     ...   assert "Requirement version must be pinned" in str(e)
     """
     requirement = requirement and "".join(requirement.lower().split())
-    assert requirement, f"specifier is required is empty:{requirement!r}"
+    if not requirement:
+        raise ValueError(f"specifier is required is empty:{requirement!r}")
     name, operator, version = split_req(requirement)
-    assert name, f"Name is required: {requirement}"
+    if not name:
+        raise ValueError(f"Name is required: {requirement}")
     is_pinned = operator == "=="
     if with_unpinned:
         version = ""
     else:
-        assert is_pinned and version, f"Requirement version must be pinned: {requirement}"
+        if not is_pinned and version:
+            raise ValueError(f"Requirement version must be pinned: {requirement}")
     return name, version
 
 
@@ -120,7 +124,7 @@ def get_installed_reqs(site_packages_dir):
     # setuptools, pip
     args = ["pip", "freeze", "--exclude-editable",
             "--all", "--path", site_packages_dir]
-    return subprocess.check_output(args, encoding="utf-8")
+    return subprocess.check_output(args, encoding="utf-8")  # noqa: S603
 
 
 comparators = (
@@ -150,9 +154,11 @@ def split_req(req):
     >>> assert split_req("foo >= 1.2.3 ") == ("foo", ">=", "1.2.3"), split_req("foo >= 1.2.3 ")
     >>> assert split_req("foo>=1.2") == ("foo", ">=", "1.2"), split_req("foo>=1.2")
     """
-    assert req
+    if not req:
+        raise ValueError("req is required")
     # do not allow multiple constraints and tags
-    assert not any(c in req for c in ",;")
+    if not any(c in req for c in ",;"):
+        raise Exception(f"complex requirements with : or ; not supported: {req}")
     req = "".join(req.split())
     if not any(c in req for c in comparators):
         return req, "", ""

etc/scripts/utils_thirdparty.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 #
 # Copyright (c) nexB Inc. and others. All rights reserved.
 # ScanCode is a trademark of nexB Inc.
@@ -559,7 +558,8 @@ def download(self, dest_dir=THIRDPARTY_DIR):
         Download this distribution into `dest_dir` directory.
         Return the fetched filename.
         """
-        assert self.filename
+        if not self.filename:
+            raise ValueError(f"self.filename has no value but is required: {self.filename!r}")
         if TRACE_DEEP:
             print(
                 f"Fetching distribution of {self.name}=={self.version}:",
@@ -829,10 +829,9 @@ def fetch_license_files(self, dest_dir=THIRDPARTY_DIR, use_cached_index=False):
         urls = LinksRepository.from_url(
             use_cached_index=use_cached_index).links
         errors = []
-        extra_lic_names = [l.get("file")
-                           for l in self.extra_data.get("licenses", {})]
+        extra_lic_names = [lic.get("file") for lic in self.extra_data.get("licenses", {})]
         extra_lic_names += [self.extra_data.get("license_file")]
-        extra_lic_names = [ln for ln in extra_lic_names if ln]
+        extra_lic_names = [eln for eln in extra_lic_names if eln]
         lic_names = [f"{key}.LICENSE" for key in self.get_license_keys()]
         for filename in lic_names + extra_lic_names:
             floc = os.path.join(dest_dir, filename)
@@ -853,7 +852,7 @@ def fetch_license_files(self, dest_dir=THIRDPARTY_DIR, use_cached_index=False):
                 if TRACE:
                     print(f"Fetched license from remote: {lic_url}")
 
-            except:
+            except Exception:
                 try:
                     # try licensedb second
                     lic_url = f"{LICENSEDB_API_URL}/{filename}"
@@ -866,8 +865,9 @@ def fetch_license_files(self, dest_dir=THIRDPARTY_DIR, use_cached_index=False):
                     if TRACE:
                         print(f"Fetched license from licensedb: {lic_url}")
 
-                except:
-                    msg = f'No text for license {filename} in expression "{self.license_expression}" from {self}'
+                except Exception:
+                    msg = f"No text for license {filename} in expression "
+                    f"{self.license_expression!r} from {self}"
                     print(msg)
                     errors.append(msg)
 
@@ -1009,7 +1009,7 @@ def get_license_link_for_filename(filename, urls):
     exception if no link is found or if there are more than one link for that
     file name.
     """
-    path_or_url = [l for l in urls if l.endswith(f"/{filename}")]
+    path_or_url = [url for url in urls if url.endswith(f"/{filename}")]
     if not path_or_url:
         raise Exception(f"Missing link to file: {filename}")
     if not len(path_or_url) == 1:
@@ -1140,7 +1140,6 @@ def to_filename(self):
 
 @attr.attributes
 class Wheel(Distribution):
-
     """
     Represents a wheel file.
 
@@ -1301,7 +1300,7 @@ def is_pure(self):
 def is_pure_wheel(filename):
     try:
         return Wheel.from_filename(filename).is_pure()
-    except:
+    except Exception:
         return False
 
 
@@ -1489,8 +1488,7 @@ def dists_from_paths_or_urls(cls, paths_or_urls):
                     )
             except InvalidDistributionFilename:
                 if TRACE_DEEP:
-                    print(
-                        f"     Skipping invalid distribution from: {path_or_url}")
+                    print(f"     Skipping invalid distribution from: {path_or_url}")
                 continue
         return dists
 
@@ -1500,8 +1498,7 @@ def get_distributions(self):
         """
         if self.sdist:
             yield self.sdist
-        for wheel in self.wheels:
-            yield wheel
+        yield from self.wheels
 
     def get_url_for_filename(self, filename):
         """
@@ -1632,7 +1629,8 @@ class PypiSimpleRepository:
         type=dict,
         default=attr.Factory(lambda: defaultdict(dict)),
         metadata=dict(
-            help="Mapping of {name: {version: PypiPackage, version: PypiPackage, etc} available in this repo"
+            help="Mapping of {name: {version: PypiPackage, version: PypiPackage, etc} "
+            "available in this repo"
         ),
     )
 
@@ -1647,7 +1645,8 @@ class PypiSimpleRepository:
         type=bool,
         default=False,
         metadata=dict(
-            help="If True, use any existing on-disk cached PyPI index files. Otherwise, fetch and cache."
+            help="If True, use any existing on-disk cached PyPI index files. "
+            "Otherwise, fetch and cache."
         ),
     )
 
@@ -1656,7 +1655,8 @@ def _get_package_versions_map(self, name):
         Return a mapping of all available PypiPackage version for this package name.
         The mapping may be empty. It is ordered by version from oldest to newest
         """
-        assert name
+        if not name:
+            raise ValueError(f"name is required: {name!r}")
         normalized_name = NameVer.normalize_name(name)
         versions = self.packages[normalized_name]
         if not versions and normalized_name not in self.fetched_package_normalized_names:
@@ -1713,7 +1713,7 @@ def fetch_links(self, normalized_name):
         )
         links = collect_urls(text)
         # TODO: keep sha256
-        links = [l.partition("#sha256=") for l in links]
+        links = [link.partition("#sha256=") for link in links]
         links = [url for url, _, _sha256 in links]
         return links
 
@@ -1936,7 +1936,7 @@ def get_remote_file_content(
     # several redirects and that we can ignore content there. A HEAD request may
     # not get us this last header
     print(f"    DOWNLOADING: {url}")
-    with requests.get(url, allow_redirects=True, stream=True, headers=headers) as response:
+    with requests.get(url, allow_redirects=True, stream=True, headers=headers) as response: # noqa: S113
         status = response.status_code
         if status != requests.codes.ok:  # NOQA
             if status == 429 and _delay < 20:
@@ -2161,7 +2161,7 @@ def call(args, verbose=TRACE):
     """
     if TRACE_DEEP:
         print("Calling:", " ".join(args))
-    with subprocess.Popen(
+    with subprocess.Popen( # noqa: S603
         args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
     ) as process:
 
@@ -2227,7 +2227,7 @@ def download_wheels_with_pip(
         cli_args.extend(["--requirement", req_file])
 
     if TRACE:
-        print(f"Downloading wheels using command:", " ".join(cli_args))
+        print("Downloading wheels using command:", " ".join(cli_args))
 
     existing = set(os.listdir(dest_dir))
     error = False
@@ -2260,7 +2260,7 @@ def download_wheels_with_pip(
 
 def check_about(dest_dir=THIRDPARTY_DIR):
     try:
-        subprocess.check_output(f"venv/bin/about check {dest_dir}".split())
+        subprocess.check_output(f"venv/bin/about check {dest_dir}".split()) # noqa: S603
     except subprocess.CalledProcessError as cpe:
         print()
         print("Invalid ABOUT files:")
@@ -2312,5 +2312,5 @@ def get_license_expression(declared_licenses):
         return get_only_expression_from_extracted_license(declared_licenses)
     except ImportError:
         # Scancode is not installed, clean and join all the licenses
-        lics = [python_safe_name(l).lower() for l in declared_licenses]
+        lics = [python_safe_name(lic).lower() for lic in declared_licenses]
         return " AND ".join(lics).lower()