diff --git a/cmem_plugin_base/dataintegration/description.py b/cmem_plugin_base/dataintegration/description.py
index 01fefe1..8654493 100644
--- a/cmem_plugin_base/dataintegration/description.py
+++ b/cmem_plugin_base/dataintegration/description.py
@@ -7,7 +7,7 @@
from inspect import _empty
from mimetypes import guess_type
from pkgutil import get_data
-from typing import Any
+from typing import Any, ClassVar
from cmem_plugin_base.dataintegration.plugins import TransformPlugin, WorkflowPlugin
from cmem_plugin_base.dataintegration.types import (
@@ -83,7 +83,7 @@ def __init__( # noqa: PLR0913
label: str = "",
description: str = "",
param_type: ParameterType | None = None,
- default_value: Any | None = None,
+ default_value: Any | None = None, # noqa: ANN401
advanced: bool = False,
visible: bool = True,
) -> None:
@@ -110,7 +110,7 @@ class PluginDescription:
def __init__( # noqa: PLR0913
self,
- plugin_class,
+ plugin_class: type,
label: str,
plugin_id: str | None = None,
description: str = "",
@@ -126,7 +126,7 @@ def __init__( # noqa: PLR0913
elif issubclass(plugin_class, TransformPlugin):
self.plugin_type = "TransformPlugin"
else:
- raise ValueError(
+ raise TypeError(
f"Class {plugin_class.__name__} does not implement a supported "
f"plugin base class (e.g., WorkflowPlugin)."
)
@@ -231,7 +231,7 @@ class Plugin:
:param icon: Optional custom plugin icon.
"""
- plugins: list[PluginDescription] = []
+ plugins: ClassVar[list[PluginDescription]] = []
def __init__( # noqa: PLR0913
self,
@@ -257,7 +257,7 @@ def __init__( # noqa: PLR0913
else:
self.parameters = parameters
- def __call__(self, func):
+ def __call__(self, func: type):
"""Allow to call the instance"""
plugin_desc = PluginDescription(
plugin_class=func,
diff --git a/cmem_plugin_base/dataintegration/discovery.py b/cmem_plugin_base/dataintegration/discovery.py
index 81e135f..cb3b472 100644
--- a/cmem_plugin_base/dataintegration/discovery.py
+++ b/cmem_plugin_base/dataintegration/discovery.py
@@ -17,7 +17,7 @@
)
-def get_packages():
+def get_packages() -> object:
"""Get installed python packages.
Returns a list of dict with the following keys:
@@ -25,7 +25,7 @@ def get_packages():
- version - package version
"""
return json.loads(
- check_output(["pip", "list", "--format", "json"], shell=False) # nosec
+ check_output(["pip", "list", "--format", "json"], shell=False) # noqa: S603, S607
)
diff --git a/cmem_plugin_base/dataintegration/entity.py b/cmem_plugin_base/dataintegration/entity.py
index 5b39f10..1bd0f38 100644
--- a/cmem_plugin_base/dataintegration/entity.py
+++ b/cmem_plugin_base/dataintegration/entity.py
@@ -27,7 +27,7 @@ def __repr__(self) -> str:
}
return f"EntityPath({obj})"
- def __eq__(self, other) -> bool:
+ def __eq__(self, other: object) -> bool:
"""Compare"""
return (
isinstance(other, EntityPath)
@@ -43,7 +43,7 @@ class EntitySchema:
:param type_uri: The entity type
:param paths: Ordered list of paths
:param path_to_root: Specifies a path which defines where this schema is located
- in the schema tree
+ in the schema tree. Empty by default.
:param sub_schemata: Nested entity schemata
"""
@@ -51,12 +51,15 @@ def __init__(
self,
type_uri: str,
paths: Sequence[EntityPath],
- path_to_root: EntityPath = EntityPath(""),
+ path_to_root: EntityPath | None = None,
sub_schemata: Sequence["EntitySchema"] | None = None,
) -> None:
self.type_uri = type_uri
self.paths = paths
- self.path_to_root = path_to_root
+ if path_to_root is None:
+ self.path_to_root = EntityPath("")
+ else:
+ self.path_to_root = path_to_root
self.sub_schemata = sub_schemata
def __repr__(self) -> str:
@@ -64,7 +67,7 @@ def __repr__(self) -> str:
obj = {"type_uri": self.type_uri, "paths": self.paths, "path_to_root": self.path_to_root}
return f"EntitySchema({obj})"
- def __eq__(self, other) -> bool:
+ def __eq__(self, other: object) -> bool:
"""Compare"""
return (
isinstance(other, EntitySchema)
diff --git a/cmem_plugin_base/dataintegration/parameter/choice.py b/cmem_plugin_base/dataintegration/parameter/choice.py
index 9951fe5..8c7b88a 100644
--- a/cmem_plugin_base/dataintegration/parameter/choice.py
+++ b/cmem_plugin_base/dataintegration/parameter/choice.py
@@ -20,13 +20,13 @@ def __init__(self, choice_list: collections.OrderedDict[str, str]):
self.choice_list = choice_list
def label(
- self, value: str, depend_on_parameter_values: list[Any], context: PluginContext # noqa: ARG002
+ self, value: str, depend_on_parameter_values: list[Any], context: PluginContext
) -> str | None:
"""Return the label for the given choice value."""
return self.choice_list[value]
def autocomplete(
- self, query_terms: list[str], depend_on_parameter_values: list[Any], context: PluginContext, # noqa: ARG002
+ self, query_terms: list[str], depend_on_parameter_values: list[Any], context: PluginContext,
) -> list[Autocompletion]:
"""Autocompletion request - Returns all results that match ALL provided query terms."""
result = []
@@ -37,6 +37,6 @@ def autocomplete(
result.append(Autocompletion(value=identifier, label=label))
for term in query_terms:
if term.lower() in label.lower():
- result.append(Autocompletion(value=identifier, label=label))
- result.sort(key=lambda x: x.label) # type: ignore
+ result.append(Autocompletion(value=identifier, label=label)) # noqa: PERF401
+ result.sort(key=lambda x: x.label)
return list(set(result))
diff --git a/cmem_plugin_base/dataintegration/parameter/dataset.py b/cmem_plugin_base/dataintegration/parameter/dataset.py
index bc821b5..46d013e 100644
--- a/cmem_plugin_base/dataintegration/parameter/dataset.py
+++ b/cmem_plugin_base/dataintegration/parameter/dataset.py
@@ -26,7 +26,7 @@ def __init__(self, dataset_type: str | None = None):
def label(
self, value: str, depend_on_parameter_values: list[Any], context: PluginContext
) -> str | None:
- """Returns the label for the given dataset."""
+ """Return the label for the given dataset."""
setup_cmempy_user_access(context.user)
task_label = str(get_task(project=context.project_id, task=value)["metadata"]["label"])
return f"{task_label}"
@@ -37,6 +37,7 @@ def autocomplete(
depend_on_parameter_values: list[Any],
context: PluginContext,
) -> list[Autocompletion]:
+ """Autocompletion request - Returns all results that match all provided query terms."""
setup_cmempy_user_access(context.user)
datasets = list_items(item_type="dataset", project=context.project_id)["results"]
@@ -50,9 +51,9 @@ def autocomplete(
continue
for term in query_terms:
if term.lower() in label.lower():
- result.append(Autocompletion(value=identifier, label=label))
+ result.append(Autocompletion(value=identifier, label=label)) # noqa: PERF401
if len(query_terms) == 0:
# add any dataset to list if no search terms are given
result.append(Autocompletion(value=identifier, label=label))
- result.sort(key=lambda x: x.label) # type: ignore
+ result.sort(key=lambda x: x.label)
return list(set(result))
diff --git a/cmem_plugin_base/dataintegration/parameter/graph.py b/cmem_plugin_base/dataintegration/parameter/graph.py
index 2f99c94..bc90b10 100644
--- a/cmem_plugin_base/dataintegration/parameter/graph.py
+++ b/cmem_plugin_base/dataintegration/parameter/graph.py
@@ -88,5 +88,5 @@ def autocomplete(
if term.lower() in label.lower():
result.append(Autocompletion(value=iri, label=label))
continue
- result.sort(key=lambda x: x.label) # type: ignore
+ result.sort(key=lambda x: x.label)
return list(set(result))
diff --git a/cmem_plugin_base/dataintegration/parameter/password.py b/cmem_plugin_base/dataintegration/parameter/password.py
index 3b06f99..e923379 100644
--- a/cmem_plugin_base/dataintegration/parameter/password.py
+++ b/cmem_plugin_base/dataintegration/parameter/password.py
@@ -12,7 +12,7 @@ def __init__(self, encrypted_value: str, system: SystemContext):
self.system = system
def decrypt(self) -> str:
- """Returns the decrypted value"""
+ """Return the decrypted value"""
return self.system.decrypt(self.encrypted_value)
@@ -26,7 +26,8 @@ class PasswordParameterType(ParameterType[Password]):
"""Prefix to identify already encrypted values."""
def from_string(self, value: str, context: PluginContext) -> Password:
- """Parses strings into parameter values.
+ """Parse strings into parameter values.
+
Decrypts the password if the encryption preamble is present
"""
if value is None or value == "":
@@ -38,7 +39,8 @@ def from_string(self, value: str, context: PluginContext) -> Password:
return Password(encrypted_value, context.system)
def to_string(self, value: Password) -> str:
- """Converts parameter values into their string representation.
+ """Convert parameter values into their string representation.
+
Encrypts the password so that it won't be stored verbatim.
"""
if value.encrypted_value == "":
diff --git a/cmem_plugin_base/dataintegration/parameter/resource.py b/cmem_plugin_base/dataintegration/parameter/resource.py
index 74def09..d79d3ac 100644
--- a/cmem_plugin_base/dataintegration/parameter/resource.py
+++ b/cmem_plugin_base/dataintegration/parameter/resource.py
@@ -22,6 +22,7 @@ def autocomplete(
depend_on_parameter_values: list[Any],
context: PluginContext,
) -> list[Autocompletion]:
+ """Autocompletion request - Returns all results that match ALL provided query terms."""
setup_cmempy_user_access(context.user)
resources = get_resources(context.project_id)
result = [
diff --git a/cmem_plugin_base/dataintegration/types.py b/cmem_plugin_base/dataintegration/types.py
index 5131d2e..b5d2917 100644
--- a/cmem_plugin_base/dataintegration/types.py
+++ b/cmem_plugin_base/dataintegration/types.py
@@ -4,7 +4,7 @@
from dataclasses import dataclass
from enum import Enum
from inspect import Parameter
-from typing import Any, Generic, Optional, Type, TypeVar
+from typing import Any, ClassVar, Generic, TypeVar
from cmem_plugin_base.dataintegration.context import PluginContext
@@ -42,21 +42,21 @@ class ParameterType(Generic[T]):
"""Signals that the auto-completed values have labels that must be
displayed to the user."""
- autocompletion_depends_on_parameters: list[str] = []
+ autocompletion_depends_on_parameters: ClassVar[list[str]] = []
"""The other plugin parameters the auto-completion depends on.
Without those values given no auto-completion is possible.
The values of all parameters specified here will be provided
to the autocomplete function."""
- def get_type(self):
- """Retrieves the type that is supported by a given instance."""
+ def get_type(self) -> type:
+ """Retrieve the type that is supported by a given instance."""
return self.__orig_bases__[0].__args__[0]
def from_string(self, value: str, context: PluginContext) -> T:
- """Parses strings into parameter values."""
+ """Parse strings into parameter values."""
def to_string(self, value: T) -> str:
- """Converts parameter values into their string representation."""
+ """Convert parameter values into their string representation."""
return str(value)
def autocomplete(
@@ -77,8 +77,8 @@ def autocomplete(
def label(
self, value: str, depend_on_parameter_values: list[Any], context: PluginContext
- ) -> Optional[str]:
- """Returns the label if exists for the given value.
+ ) -> str | None:
+ """Return the label if exists for the given value.
:param value: The value for which a label should be generated.
:param depend_on_parameter_values The values of the parameters specified
@@ -88,8 +88,11 @@ def label(
return None
def autocompletion_enabled(self) -> bool:
- """True, if autocompletion should be enabled on this type.
- By default, checks if the type implements its own autocomplete method."""
+ """Enable autocompletion.
+
+ True, if autocompletion should be enabled on this type.
+ By default, checks if the type implements its own autocomplete method.
+ """
return type(self).autocomplete != ParameterType.autocomplete
@@ -99,6 +102,7 @@ class StringParameterType(ParameterType[str]):
name = "string"
def from_string(self, value: str, context: PluginContext) -> str:
+ """Return the string."""
return value
@@ -108,6 +112,7 @@ class IntParameterType(ParameterType[int]):
name = "Long"
def from_string(self, value: str, context: PluginContext) -> int:
+ """Parse string into int."""
return int(value)
@@ -117,6 +122,7 @@ class FloatParameterType(ParameterType[float]):
name = "double"
def from_string(self, value: str, context: PluginContext) -> float:
+ """Parse string into float."""
return float(value)
@@ -125,7 +131,7 @@ class BoolParameterType(ParameterType[bool]):
name = "boolean"
- def from_string(self, value: str, context: PluginContext) -> bool: # noqa: ARG002
+ def from_string(self, value: str, context: PluginContext) -> bool:
"""Get boolean value from string"""
lower = value.lower()
if lower in ("true", "1"):
@@ -147,9 +153,11 @@ class PluginContextParameterType(ParameterType[PluginContext]):
name = "PluginContext"
def from_string(self, value: str, context: PluginContext) -> PluginContext:
+ """Return the plugin context."""
return context
def to_string(self, value: PluginContext) -> str:
+ """Return an empty string, since from_string will always return the plugin context."""
return ""
@@ -160,11 +168,12 @@ class EnumParameterType(ParameterType[Enum]):
allow_only_autocompleted_values = True
- def __init__(self, enum_type: Type[Enum]):
+ def __init__(self, enum_type: type[Enum]):
super().__init__()
self.enum_type = enum_type
def from_string(self, value: str, context: PluginContext) -> Enum:
+ """Parse string into the corresponding enum value."""
values = self.enum_type.__members__
if not value:
raise ValueError("Empty value is not allowed.")
@@ -174,6 +183,7 @@ def from_string(self, value: str, context: PluginContext) -> Enum:
return values[value]
def to_string(self, value: Enum) -> str:
+ """Convert an enum into its string value."""
return str(value.name)
def autocomplete(
@@ -182,6 +192,7 @@ def autocomplete(
depend_on_parameter_values: list[str],
context: PluginContext,
) -> list[Autocompletion]:
+ """Autocompletion request - Returns all results that match all provided query terms."""
values = self.enum_type.__members__.keys()
return list(self.find_matches(query_terms, values))
@@ -189,14 +200,14 @@ def autocomplete(
def find_matches(
lower_case_terms: list[str], values: Iterable[str]
) -> Iterable[Autocompletion]:
- """Finds auto completions in a list of values"""
+ """Find auto completions in a list of values"""
for value in values:
if EnumParameterType.matches_search_term(lower_case_terms, value.lower()):
yield Autocompletion(value, value)
@staticmethod
def matches_search_term(lower_case_terms: list[str], search_in: str) -> bool:
- """Tests if a string contains a list of (lower case) search terms."""
+ """Test if a string contains a list of (lower case) search terms."""
lower_case_text = search_in.lower()
return all(search_term in lower_case_text for search_term in lower_case_terms)
@@ -204,7 +215,7 @@ def matches_search_term(lower_case_terms: list[str], search_in: str) -> bool:
class ParameterTypes:
"""Manages the available parameter types."""
- registered_types: list[ParameterType] = [
+ registered_types: ClassVar[list[ParameterType]] = [
StringParameterType(),
BoolParameterType(),
IntParameterType(),
@@ -214,18 +225,20 @@ class ParameterTypes:
@staticmethod
def register_type(param_type: ParameterType) -> None:
- """Registers a new custom parameter type. All registered types will be detected
+ """Register a new custom parameter type.
+
+ All registered types will be detected
in plugin constructors. If a type with an existing name is registered, it will
- overwrite the previous one."""
+ overwrite the previous one.
+ """
ParameterTypes.registered_types = [
t for t in ParameterTypes.registered_types if t.name != param_type.name
]
ParameterTypes.registered_types.append(param_type)
@staticmethod
- def get_type(param_type: Type) -> ParameterType:
- """Retrieves the ParameterType instance for a given type."""
-
+ def get_type(param_type: type) -> ParameterType:
+ """Retrieve the ParameterType instance for a given type."""
if issubclass(param_type, Enum):
return EnumParameterType(param_type)
found_type = next(
@@ -233,7 +246,7 @@ def get_type(param_type: Type) -> ParameterType:
None,
)
if found_type is None:
- mapped = map(lambda t: str(t.get_type().__name__), ParameterTypes.registered_types)
+ mapped = [str(t.get_type().__name__) for t in ParameterTypes.registered_types]
raise ValueError(
f"Parameter has an unsupported type {param_type.__name__}. "
"Supported types are: Enum, "
@@ -243,8 +256,7 @@ def get_type(param_type: Type) -> ParameterType:
@staticmethod
def get_param_type(param: Parameter) -> ParameterType:
- """Retrieves the ParameterType instance for a given parameter."""
-
+ """Retrieve the ParameterType instance for a given parameter."""
if param.annotation == Parameter.empty:
# If there is no type annotation, DI should send the parameter as a string
return StringParameterType()
diff --git a/cmem_plugin_base/dataintegration/utils/__init__.py b/cmem_plugin_base/dataintegration/utils/__init__.py
index eaa9dc2..665f616 100644
--- a/cmem_plugin_base/dataintegration/utils/__init__.py
+++ b/cmem_plugin_base/dataintegration/utils/__init__.py
@@ -2,7 +2,6 @@
import os
import re
-from typing import Optional
from cmem.cmempy.workspace.projects.datasets.dataset import post_resource
diff --git a/cmem_plugin_base/dataintegration/utils/entity_builder.py b/cmem_plugin_base/dataintegration/utils/entity_builder.py
index 74ca2d8..8f25eee 100644
--- a/cmem_plugin_base/dataintegration/utils/entity_builder.py
+++ b/cmem_plugin_base/dataintegration/utils/entity_builder.py
@@ -5,7 +5,7 @@
from cmem_plugin_base.dataintegration.entity import Entities, Entity, EntityPath, EntitySchema
-def merge_path_values(paths_map1, paths_map2):
+def merge_path_values(paths_map1: dict, paths_map2: dict) -> dict:
"""Merge two dictionaries representing paths and values.
This function takes two dictionaries, `paths_map1` and `paths_map2`,
@@ -29,9 +29,8 @@ def merge_path_values(paths_map1, paths_map2):
return paths_map1
-def generate_paths_from_data(data, path="root"):
- """Generate a dictionary representing paths and data types from a nested JSON
- structure.
+def generate_paths_from_data(data: dict | list, path: str | None = "root") -> dict:
+ """Generate a dictionary representing paths and data types from a nested JSON structure.
This function recursively traverses a nested JSON structure ('data') and builds
a dictionary ('paths_map') where keys are paths and values are dictionaries
@@ -70,7 +69,7 @@ def generate_paths_from_data(data, path="root"):
return paths_map
-def _get_schema(data: dict | list):
+def _get_schema(data: dict | list) -> dict | None:
"""Get the schema of an entity."""
if not data:
return None
@@ -94,7 +93,7 @@ def _get_schema(data: dict | list):
return path_to_schema_map
-def extend_path_list(path_to_entities, sub_path_to_entities) -> None:
+def extend_path_list(path_to_entities: dict, sub_path_to_entities: dict) -> None:
"""Extend a dictionary of paths to entities by merging with another.
This function takes two dictionaries, `path_to_entities` and `sub_path_to_entities`,
@@ -118,10 +117,10 @@ def extend_path_list(path_to_entities, sub_path_to_entities) -> None:
def _get_entity(
- path_from_root,
- path_to_schema_map,
- data,
-):
+ path_from_root: str,
+ path_to_schema_map: dict,
+ data: dict,
+) -> dict:
"""Get an entity based on the schema and data."""
path_to_entities = {}
entity_uri = f"urn:x-ulid:{ULID()}"
diff --git a/tests/conftest.py b/tests/conftest.py
index c2040de..89df83c 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -34,7 +34,7 @@ def _json_dataset() -> Generator[dict, None, None]:
@pytest.fixture(name="json_resource", scope="module")
-def _json_resource():
+def _json_resource() -> object:
"""Set up json resource"""
_project_name = "resource_test_project"
_resource_name = "sample_test.json"
diff --git a/tests/parameter_types/__init__.py b/tests/parameter_types/__init__.py
index 0e9f173..30ea453 100644
--- a/tests/parameter_types/__init__.py
+++ b/tests/parameter_types/__init__.py
@@ -1 +1 @@
-"""tests"""
\ No newline at end of file
+"""tests"""
diff --git a/tests/parameter_types/test_code.py b/tests/parameter_types/test_code.py
index d336b99..e46211c 100644
--- a/tests/parameter_types/test_code.py
+++ b/tests/parameter_types/test_code.py
@@ -30,16 +30,16 @@ def test__detection(self) -> None:
class MyTransformPlugin(TransformPlugin):
"""Test My Transform Plugin"""
- def __init__( # pylint: disable=too-many-arguments
+ def __init__( # pylint: disable=too-many-arguments # noqa: PLR0913
self,
- xml: XmlCode = XmlCode(""),
- json: JsonCode = JsonCode("{}"),
- jinja: JinjaCode = JinjaCode(""),
- sql: SqlCode = SqlCode(""),
- yaml: YamlCode = YamlCode(""),
- sparql: SparqlCode = SparqlCode(""),
- turtle: TurtleCode = TurtleCode(""),
- python: PythonCode = PythonCode(""),
+ xml: XmlCode = XmlCode(""), # noqa: B008
+ json: JsonCode = JsonCode("{}"), # noqa: B008
+ jinja: JinjaCode = JinjaCode(""), # noqa: B008
+ sql: SqlCode = SqlCode(""), # noqa: B008
+ yaml: YamlCode = YamlCode(""), # noqa: B008
+ sparql: SparqlCode = SparqlCode(""), # noqa: B008
+ turtle: TurtleCode = TurtleCode(""), # noqa: B008
+ python: PythonCode = PythonCode(""), # noqa: B008
) -> None:
self.xml = xml
self.json = json
diff --git a/tests/parameter_types/test_password.py b/tests/parameter_types/test_password.py
index 13f3d0b..12aeb60 100644
--- a/tests/parameter_types/test_password.py
+++ b/tests/parameter_types/test_password.py
@@ -25,7 +25,7 @@ class MyTransformPlugin(TransformPlugin):
def __init__(self, password: Password) -> None:
self.password = password
- def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]: # noqa: ARG002
+ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
"""Test transform"""
return []
diff --git a/tests/test_icon.py b/tests/test_icon.py
index b305f74..8508699 100644
--- a/tests/test_icon.py
+++ b/tests/test_icon.py
@@ -17,7 +17,7 @@
class MyWorkflowPlugin(WorkflowPlugin):
"""My Workflow Plugin Class"""
- def execute(self, inputs: Sequence[Entities], context: ExecutionContext) -> None: # noqa: ARG002
+ def execute(self, inputs: Sequence[Entities], context: ExecutionContext) -> None:
"""Execute the workflow plugin on a given collection of entities."""
return
diff --git a/tests/test_output_only_plugin.py b/tests/test_output_only_plugin.py
index 283a5d5..fb5cfd1 100644
--- a/tests/test_output_only_plugin.py
+++ b/tests/test_output_only_plugin.py
@@ -1,5 +1,7 @@
"""test file."""
+from collections.abc import Sequence
+from cmem_plugin_base.dataintegration.context import ExecutionContext
from cmem_plugin_base.dataintegration.description import Plugin, PluginParameter
from cmem_plugin_base.dataintegration.entity import (
Entities,
@@ -32,7 +34,8 @@ class OutputOnlyPlugin(WorkflowPlugin):
def __init__(self, param1: str) -> None:
self.param1 = param1
- def execute(self, inputs=(), context=()) -> Entities:
+ def execute(self, inputs: Sequence[Entities] = (), context: ExecutionContext = ()) -> Entities:
+ """Execute the workflow plugin on a given collection of entities."""
entity1 = Entity(uri="urn:my:1", values=(["value1"], ["value2"]))
entity2 = Entity(uri="urn:my:2", values=(["value3"], ["value4"]))
schema = EntitySchema(
diff --git a/tests/test_utils_build_entities_from_data.py b/tests/test_utils_build_entities_from_data.py
index ad6a350..c7c7218 100644
--- a/tests/test_utils_build_entities_from_data.py
+++ b/tests/test_utils_build_entities_from_data.py
@@ -2,11 +2,12 @@
import json
-from cmem_plugin_base.dataintegration.entity import EntityPath, EntitySchema, Entities
+from cmem_plugin_base.dataintegration.entity import Entities, EntityPath, EntitySchema
from cmem_plugin_base.dataintegration.utils.entity_builder import build_entities_from_data
def build_entities_from_json(json_data: str) -> Entities:
+ """Build entities from a test JSON string."""
data = json.loads(json_data)
entities = build_entities_from_data(data)
assert entities is not None
@@ -36,9 +37,7 @@ def test_single_object() -> None:
def test_single_object_one_level() -> None:
- """Test generation of entities and schema for a JSON object with one level of
- hierarchy
- """
+ """Test generation of entities and schema for a JSON object with one level of hierarchy."""
test_data = """
{
"name": "sai",
@@ -79,8 +78,9 @@ def test_single_object_one_level() -> None:
def test_single_object_one_level_array() -> None:
- """Test generation of entities and schema for a JSON object with array object in
- first level of hierarchy
+ """Test generation of entities and schema for a JSON object.
+
+ Test with array object in first level of hierarchy.
"""
test_data = """
{
@@ -126,8 +126,9 @@ def test_single_object_one_level_array() -> None:
def test_single_object_two_level_array() -> None:
- """Test generation of entities and schema for a JSON object with two levels of
- hierarchy
+ """Test generation of entities and schema.
+
+ Tests a JSON object with two levels of hierarchy.
"""
test_data = """
{
diff --git a/tests/test_utils_write_to_dataset.py b/tests/test_utils_write_to_dataset.py
index 74a1319..8173751 100644
--- a/tests/test_utils_write_to_dataset.py
+++ b/tests/test_utils_write_to_dataset.py
@@ -13,7 +13,7 @@
@needs_cmem
-def test_write_to_json_dataset(json_dataset) -> None:
+def test_write_to_json_dataset(json_dataset: dict) -> None:
"""Test write to json dataset"""
project_name = json_dataset["project"]
dataset_name = json_dataset["id"]
diff --git a/tests/utils.py b/tests/utils.py
index 8d88234..3740ea8 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -8,6 +8,7 @@
from cmem.cmempy.api import get_token
from cmem_plugin_base.dataintegration.context import PluginContext, UserContext
+from cmem_plugin_base.dataintegration.types import ParameterType
needs_cmem = pytest.mark.skipif(
"CMEM_BASE_URI" not in os.environ, reason="Needs CMEM configuration"
@@ -33,13 +34,17 @@ class TestPluginContext(PluginContext):
def __init__(
self,
project_id: str = "dummyProject",
- user: UserContext | None = TestUserContext(),
+ user: UserContext | None = None,
):
self.project_id = project_id
- self.user = user
+ if user is None:
+ self.user = TestUserContext()
+ else:
+ self.user = user
-def get_autocomplete_values(parameter, query_terms, context):
+def get_autocomplete_values(parameter: ParameterType, query_terms: list[str],
+ context: PluginContext) -> list[str]:
"""Get autocomplete values"""
return [
x.value