[AT-3970] enable pylint for utils and macros (#559)

* fix utils and macros

* PR comments

* update pylintrc

* address PR comment

CHANGELOG.md

@@ -30,6 +30,13 @@ You can check your current version with the following command:
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
 
+## 0.37.0a9
+
+**March 13, 2024**
+
+- Adjusted `macros.py`, `utils.py`, and `test_utils.py` in accordance with Pylint checks.
+
+
 ## 0.37.0a8
 
 **March 13, 2024**

docs/macros.py

@@ -12,7 +12,7 @@
 Docs: https://mkdocs-macros-plugin.readthedocs.io/en/latest/macros/
 """
 
-from types import ModuleType
+import types
 from typing import Callable, List, Optional, Union
 
 import up42
@@ -27,7 +27,7 @@ def indent(input_string: Optional[str]) -> Optional[str]:
         return input_string and input_string.replace("\n", "\n\t")
 
     def get_methods(
-        c: Union[Callable, ModuleType],
+        c: Union[Callable, types.ModuleType],
         exclude: Optional[List[str]] = None,
         exclude_viztools=False,
     ) -> List[str]:

pylintrc

@@ -15,7 +15,7 @@ ignore=
 
 # Files or directories matching the regex patterns are skipped. The regex
 # matches against base names, not paths.
-ignore-patterns=.*init.*,.*asset.*.py,.*initialization.py,.*job.*.py,.*http_adapter.py,.*main.py,macros.py,.*project.py,.*stac_client.py,.*utils.py,.*vis.*.py,.*viz.*.py,.*webhook.*.py,.*workflow.py,.*e2e.*.py
+ignore-patterns=.*init.*,.*asset.*.py,.*initialization.py,.*job.*.py,.*http_adapter.py,.*main.py,.*project.py,.*stac_client.py,.*vis.*.py,.*viz.*.py,.*webhook.*.py,.*workflow.py,.*e2e.*.py
 
 
 # Pickle collected data for later comparisons.

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "0.37.0a8"
+version = "0.37.0a9"
 description = "Python SDK for UP42, the geospatial marketplace and developer platform."
 authors = ["UP42 GmbH <[email protected]>"]
 license = "https://github.com/up42/up42-py/blob/master/LICENSE"

tests/test_utils.py

@@ -1,43 +1,33 @@
 import json
+import pathlib
 import tempfile
-from pathlib import Path
 from typing import Dict
-from unittest.mock import Mock, patch
+from unittest import mock
 
-import geopandas as gpd  # type: ignore
-import pandas as pd
+import geopandas  # type: ignore
+import pandas
 import pytest
-from dateutil.parser import parse
-from geopandas import GeoDataFrame
-from shapely.geometry import LinearRing, Polygon  # type: ignore
-
-from up42.utils import (
-    any_vector_to_fc,
-    autocomplete_order_parameters,
-    download_from_gcs_unpack,
-    fc_to_query_geometry,
-    filter_jobs_on_mode,
-    format_time,
-    get_up42_py_version,
-    read_json,
-)
+from dateutil import parser
+from shapely import geometry  # type: ignore
+
+from up42 import utils
 
-POLY = Polygon([(0, 0), (1, 1), (1, 0)])
+POLY = geometry.Polygon([(0, 0), (1, 1), (1, 0)])
 
 
 @pytest.mark.parametrize(
     "date,set_end_of_day,result_time",
     [
         ("2014-01-01", False, "2014-01-01T00:00:00Z"),
         ("2015-01-01T00:00:00", False, "2015-01-01T00:00:00Z"),
-        (parse("2016-01-01"), False, "2016-01-01T00:00:00Z"),
+        (parser.parse("2016-01-01"), False, "2016-01-01T00:00:00Z"),
         ("2017-01-01", True, "2017-01-01T23:59:59Z"),
         ("2018-01-01T00:10:00", True, "2018-01-01T00:10:00Z"),
-        (parse("2019-01-01"), True, "2019-01-01T00:00:00Z"),
+        (parser.parse("2019-01-01"), True, "2019-01-01T00:00:00Z"),
     ],
 )
 def test_format_time(date, set_end_of_day, result_time):
-    formatted_time = format_time(date=date, set_end_of_day=set_end_of_day)
+    formatted_time = utils.format_time(date=date, set_end_of_day=set_end_of_day)
     assert isinstance(formatted_time, str)
     assert formatted_time == result_time
 
@@ -48,16 +38,16 @@ def test_format_time(date, set_end_of_day, result_time):
         (1, POLY),
         (
             1,
-            gpd.GeoDataFrame(
-                pd.DataFrame([0], columns=["id"]),
+            geopandas.GeoDataFrame(
+                pandas.DataFrame([0], columns=["id"]),
                 crs={"init": "epsg:4326"},
                 geometry=[POLY],
             ),
         ),
         (
             2,
-            gpd.GeoDataFrame(
-                pd.DataFrame([0, 1], columns=["id"]),
+            geopandas.GeoDataFrame(
+                pandas.DataFrame([0, 1], columns=["id"]),
                 crs={"init": "epsg:4326"},
                 geometry=[POLY, POLY],
             ),
@@ -125,7 +115,7 @@ def test_format_time(date, set_end_of_day, result_time):
     ],
 )
 def test_any_vector_to_fc(len_fc, in_vector):
-    fc = any_vector_to_fc(in_vector)
+    fc = utils.any_vector_to_fc(in_vector)
     assert isinstance(fc, dict)
     assert fc["type"] == "FeatureCollection"
     assert fc.get("bbox") is not None
@@ -134,22 +124,22 @@ def test_any_vector_to_fc(len_fc, in_vector):
     assert len(fc["features"]) == len_fc
     assert fc["features"][0]["geometry"].get("coordinates") is not None
 
-    df = any_vector_to_fc(in_vector, as_dataframe=True)
-    assert isinstance(df, GeoDataFrame)
+    df = utils.any_vector_to_fc(in_vector, as_dataframe=True)
+    assert isinstance(df, geopandas.GeoDataFrame)
     assert df.crs.to_string() == "EPSG:4326"
 
 
 def test_any_vector_to_fc_raises_with_unaccepted_geometry_type():
-    ring = LinearRing([(0, 0), (1, 1), (1, 0)])
+    ring = geometry.LinearRing([(0, 0), (1, 1), (1, 0)])
     with pytest.raises(ValueError):
-        any_vector_to_fc(ring)
+        utils.any_vector_to_fc(ring)
 
 
 def test_fc_to_query_geometry_single_intersects():
-    fp = Path(__file__).resolve().parent / "mock_data/aoi_berlin.geojson"
-    with open(fp) as json_file:
+    fp = pathlib.Path(__file__).resolve().parent / "mock_data/aoi_berlin.geojson"
+    with open(fp, encoding="utf-8") as json_file:
         fc = json.load(json_file)
-    query_geometry = fc_to_query_geometry(fc=fc, geometry_operation="intersects")
+    query_geometry = utils.fc_to_query_geometry(fc=fc, geometry_operation="intersects")
     assert isinstance(query_geometry, dict)
     assert query_geometry["type"] == "Polygon"
     assert query_geometry["coordinates"] == [
@@ -164,67 +154,67 @@ def test_fc_to_query_geometry_single_intersects():
 
 
 def test_fc_to_query_geometry_single_bbox():
-    fp = Path(__file__).resolve().parent / "mock_data/aoi_berlin.geojson"
-    with open(fp) as json_file:
+    fp = pathlib.Path(__file__).resolve().parent / "mock_data/aoi_berlin.geojson"
+    with open(fp, encoding="utf-8") as json_file:
         fc = json.load(json_file)
-    query_geometry = fc_to_query_geometry(fc=fc, geometry_operation="bbox")
+    query_geometry = utils.fc_to_query_geometry(fc=fc, geometry_operation="bbox")
     assert isinstance(query_geometry, list)
     assert len(query_geometry) == 4
     assert query_geometry == [13.375966, 52.515068, 13.378314, 52.516639]
 
 
 def test_fc_to_query_geometry_multiple_raises():
-    fp = Path(__file__).resolve().parent / "mock_data/search_results_limited_columns.geojson"
-    with open(fp) as json_file:
+    fp = pathlib.Path(__file__).resolve().parent / "mock_data/search_results_limited_columns.geojson"
+    with open(fp, encoding="utf-8") as json_file:
         fc = json.load(json_file)
 
     with pytest.raises(ValueError) as e:
-        fc_to_query_geometry(fc=fc, geometry_operation="intersects")
+        utils.fc_to_query_geometry(fc=fc, geometry_operation="intersects")
     assert str(e.value) == "UP42 only accepts single geometries, the provided geometry contains multiple geometries."
 
     with pytest.raises(ValueError) as e:
-        fc_to_query_geometry(fc=fc, geometry_operation="bbox")
+        utils.fc_to_query_geometry(fc=fc, geometry_operation="bbox")
     assert str(e.value) == "UP42 only accepts single geometries, the provided geometry contains multiple geometries."
 
 
 def test_fc_to_query_geometry_multipolygon_raises():
-    fp = Path(__file__).resolve().parent / "mock_data/multipolygon.geojson"
-    with open(fp) as json_file:
+    fp = pathlib.Path(__file__).resolve().parent / "mock_data/multipolygon.geojson"
+    with open(fp, encoding="utf-8") as json_file:
         fc = json.load(json_file)
 
     with pytest.raises(ValueError) as e:
-        fc_to_query_geometry(fc=fc, geometry_operation="intersects")
+        utils.fc_to_query_geometry(fc=fc, geometry_operation="intersects")
     assert str(e.value) == "UP42 only accepts single geometries, the provided geometry is a MultiPolygon."
 
 
 def test_download_gcs_unpack_tgz(requests_mock):
     cloud_storage_url = "http://clouddownload.api.com/abcdef"
 
-    out_tgz = Path(__file__).resolve().parent / "mock_data/result_tif.tgz"
+    out_tgz = pathlib.Path(__file__).resolve().parent / "mock_data/result_tif.tgz"
     with open(out_tgz, "rb") as src_tgz:
         out_tgz_file = src_tgz.read()
     requests_mock.get(
         url=cloud_storage_url,
         content=out_tgz_file,
     )
     with tempfile.TemporaryDirectory() as tempdir:
-        with open(Path(tempdir) / "dummy.txt", "w") as fp:
+        with open(pathlib.Path(tempdir) / "dummy.txt", "w", encoding="utf-8") as fp:
             fp.write("dummy")
-        out_files = download_from_gcs_unpack(
+        out_files = utils.download_from_gcs_unpack(
             download_url=cloud_storage_url,
             output_directory=tempdir,
         )
 
         for file in out_files:
-            assert Path(file).exists()
+            assert pathlib.Path(file).exists()
         assert len(out_files) == 2
-        assert not (Path(tempdir) / "output").exists()
+        assert not (pathlib.Path(tempdir) / "output").exists()
 
 
 def test_download_gcs_unpack_zip(requests_mock):
     cloud_storage_url = "http://clouddownload.api.com/abcdef"
 
-    out_zip = Path(__file__).resolve().parent / "mock_data/result_tif.zip"
+    out_zip = pathlib.Path(__file__).resolve().parent / "mock_data/result_tif.zip"
     with open(out_zip, "rb") as src_zip:
         out_zip_file = src_zip.read()
 
@@ -233,23 +223,23 @@ def test_download_gcs_unpack_zip(requests_mock):
         content=out_zip_file,
     )
     with tempfile.TemporaryDirectory() as tempdir:
-        with open(Path(tempdir) / "dummy.txt", "w") as fp:
+        with open(pathlib.Path(tempdir) / "dummy.txt", "w", encoding="utf-8") as fp:
             fp.write("dummy")
-        out_files = download_from_gcs_unpack(
+        out_files = utils.download_from_gcs_unpack(
             download_url=cloud_storage_url,
             output_directory=tempdir,
         )
 
         for file in out_files:
-            assert Path(file).exists()
+            assert pathlib.Path(file).exists()
         assert len(out_files) == 2
-        assert not (Path(tempdir) / "output").exists()
+        assert not (pathlib.Path(tempdir) / "output").exists()
 
 
 def test_download_gcs_not_unpack(requests_mock):
     cloud_storage_url = "http://clouddownload.api.com/abcdef"
 
-    out_zip = Path(__file__).resolve().parent / "mock_data/aoi_berlin.geojson"
+    out_zip = pathlib.Path(__file__).resolve().parent / "mock_data/aoi_berlin.geojson"
     with open(out_zip, "rb") as src_zip:
         out_zip_file = src_zip.read()
 
@@ -258,35 +248,39 @@ def test_download_gcs_not_unpack(requests_mock):
         content=out_zip_file,
     )
     with tempfile.TemporaryDirectory() as tempdir:
-        with open(Path(tempdir) / "dummy.txt", "w") as fp:
+        with open(pathlib.Path(tempdir) / "dummy.txt", "w", encoding="utf-8") as fp:
             fp.write("dummy")
         with pytest.raises(ValueError):
-            download_from_gcs_unpack(
+            utils.download_from_gcs_unpack(
                 download_url=cloud_storage_url,
                 output_directory=tempdir,
             )
 
 
 def test_filter_jobs_on_mode():
     job_json = [{"mode": "DEFAULT"}, {"mode": "DRY_RUN"}]
-    r = filter_jobs_on_mode(job_json)
+    r = utils.filter_jobs_on_mode(job_json)
     assert len(r) == 2
-    r = filter_jobs_on_mode(job_json, test_jobs=False, real_jobs=True)
+    r = utils.filter_jobs_on_mode(job_json, test_jobs=False, real_jobs=True)
     assert len(r) == 1
-    r = filter_jobs_on_mode(job_json, test_jobs=True, real_jobs=False)
+    r = utils.filter_jobs_on_mode(job_json, test_jobs=True, real_jobs=False)
     assert len(r) == 1
     with pytest.raises(ValueError):
-        filter_jobs_on_mode(job_json, test_jobs=False, real_jobs=False)
+        utils.filter_jobs_on_mode(job_json, test_jobs=False, real_jobs=False)
 
 
 def test_autocomplete_order_parameters():
-    with open(Path(__file__).resolve().parent / "mock_data/data_product_spot_schema.json") as json_file:
+    with open(
+        pathlib.Path(__file__).resolve().parent / "mock_data/data_product_spot_schema.json", encoding="utf-8"
+    ) as json_file:
         json_data_product_schema = json.load(json_file)
     order_parameters = {
         "dataProduct": "123",
         "params": {"existing_param1": "abc"},
     }
-    order_parameters = autocomplete_order_parameters(order_parameters=order_parameters, schema=json_data_product_schema)
+    order_parameters = utils.autocomplete_order_parameters(
+        order_parameters=order_parameters, schema=json_data_product_schema
+    )
 
     assert "dataProduct" in order_parameters
     assert isinstance(order_parameters["params"], Dict)
@@ -295,38 +289,38 @@ def test_autocomplete_order_parameters():
     assert order_parameters["params"]["acquisitionMode"] is None
 
 
-@patch("importlib.metadata.version", return_value="some_version")
-def test_get_up42_py_version(version: Mock):
-    assert get_up42_py_version() == "some_version"
+@mock.patch("importlib.metadata.version", return_value="some_version")
+def test_get_up42_py_version(version: mock.Mock):
+    assert utils.get_up42_py_version() == "some_version"
     version.assert_called_with("up42-py")
 
 
 def test_read_json_should_skip_reading_if_path_is_none():
-    assert not read_json(path_or_dict=None)
+    assert not utils.read_json(path_or_dict=None)
 
 
 def test_read_json_should_skip_reading_if_dict_is_given():
     value = {"some": "data"}
-    assert read_json(path_or_dict=value) == value
+    assert utils.read_json(path_or_dict=value) == value
 
 
-@pytest.fixture(params=[True, False], ids=["str", "path"])
-def str_or_path(tmp_path, request):
+@pytest.fixture(params=[True, False], ids=["str", "path"], name="str_or_path")
+def _str_or_path(tmp_path, request):
     is_str = request.param
     result = tmp_path / "data.json"
     return str(result) if is_str else result
 
 
 def test_should_read_json(str_or_path):
     value = {"some": "data"}
-    with open(str_or_path, "w") as file:
+    with open(str_or_path, "w", encoding="utf-8") as file:
         json.dump(value, file)
 
-    assert read_json(path_or_dict=str_or_path) == value
+    assert utils.read_json(path_or_dict=str_or_path) == value
 
 
 def test_read_json_fails_if_path_not_found(str_or_path):
     with pytest.raises(ValueError) as ex:
-        assert read_json(path_or_dict=str_or_path)
+        assert utils.read_json(path_or_dict=str_or_path)
 
     assert str(str_or_path) in str(ex.value)