diff --git a/CHANGELOG.md b/CHANGELOG.md index 1062517..36dcdd1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ Types of changes ## [Unreleased] ### Added - persistent processing +- Add create_actinia_pc_item function to utils ## [0.3.2] - 2023-12-21 diff --git a/src/actinia/utils.py b/src/actinia/utils.py index 8a10a6c..6411512 100644 --- a/src/actinia/utils.py +++ b/src/actinia/utils.py @@ -52,6 +52,7 @@ def request_and_check(url, auth, status_code=200): def set_job_names(name, default_name="unknown_job"): + """Function to set the date/time to the job name""" now = datetime.now() if name is None: orig_name = default_name @@ -60,3 +61,89 @@ def set_job_names(name, default_name="unknown_job"): orig_name = name name += f"_{now.strftime('%Y%d%m_%H%M%S')}" return orig_name, name + + +def create_actinia_pc_item( + id, + module, + inputs=None, + outputs=None, + flags=None, + stdin=None, + stdout=None, + overwrite=False, + superquiet=False, + verbose=False, + interface_description=False, +): + """ + Creates a list item for an actinia process chain + + Parameters + ---------- + id: str + unique id for this item + module: str + some valid GRASS or actinia module + inputs: list or dict + list of input parameters with values in the form + [{"param": key1, "value": value1}, {"param": key2, "value": value2}, + ...] + shorter alternative as dict + {"key1": value1, "key2": value2, ...} + outputs: list or dict + list of output parameters with values in the form + [{"param": key1, "value": value1}, {"param": key2, "value": value2}, + ...] + shorter alternative as dict + {"key1": value1, "key2": value2, ...} + flags: str + optional flags for the module + stdin: dict + options to read stdin + stdout: dict + options to write to stdout + must be of the form + {"id": value1, "format": value2, "delimiter": value3} + overwrite: bool + optional, set to True to allow overwriting existing data + superquiet: bool + optional, set to True to suppress all messages but errors + verbose: bool + optional, set to True to allow verbose messages + interface_description: bool + optional, set to True to create an interface_description + """ + pc_item = {"id": str(id), "module": module} + if inputs: + if isinstance(inputs, list): + pc_item["inputs"] = inputs + elif isinstance(inputs, dict): + tmplist = [] + for k, v in inputs.items(): + tmplist.append({"param": k, "value": v}) + pc_item["inputs"] = tmplist + if outputs: + if isinstance(outputs, list): + pc_item["outputs"] = outputs + elif isinstance(outputs, dict): + tmplist = [] + for k, v in outputs.items(): + tmplist.append({"param": k, "value": v}) + pc_item["outputs"] = tmplist + if flags: + pc_item["flags"] = flags + if stdin: + pc_item["stdin"] = stdin + if stdout: + pc_item["stdout"] = stdout + if overwrite is True: + pc_item["overwrite"] = True + if superquiet is True: + pc_item["superquiet"] = True + if verbose is True: + pc_item["verbose"] = True + if interface_description is True: + pc_item["interface_description"] = True + + return pc_item diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..6db354b --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +####### +# actinia-python-client is a python client for actinia - an open source REST +# API for scalable, distributed, high performance processing of geographical +# data that uses GRASS GIS for computational tasks. +# +# Copyright (c) 2023 mundialis GmbH & Co. KG +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +####### + +__license__ = "GPLv3" +__author__ = "Anika Weinmann, Markus Metz" +__copyright__ = "Copyright 2023, mundialis GmbH & Co. KG" +__maintainer__ = "Anika Weinmann" + +import pytest + +from actinia.utils import ( + create_actinia_pc_item, + request_and_check, + set_job_names, +) +from .actinia_config import ACTINIA_BASEURL, ACTINIA_VERSION, ACTINIA_AUTH + + +class TestActiniaUtils(object): + def test_request_and_check(self): + """Test request_and_check utils function.""" + url = f"{ACTINIA_BASEURL}api/{ACTINIA_VERSION}/version" + resp = request_and_check(url, ACTINIA_AUTH, status_code=200) + assert "version" in resp + + def test_request_and_check_wrong_url(self): + """Test request_and_check utils function with .""" + url = f"{ACTINIA_BASEURL}api/{ACTINIA_VERSION}/version_fail" + err_msg = "The requested URL was not found on the server." + with pytest.raises(Exception) as excinfo: + request_and_check(url, ACTINIA_AUTH, status_code=200) + assert err_msg in str(excinfo.value) + + def test_request_and_check_wrong_auth(self): + """Test request_and_check utils function with wrong auth.""" + url = f"{ACTINIA_BASEURL}api/{ACTINIA_VERSION}/locations" + err_msg = "Unauthorized Access" + wrong_auth = ("actinia-gdi", "wrong_pw") + with pytest.raises(Exception) as excinfo: + request_and_check(url, wrong_auth, status_code=200) + assert err_msg in str(excinfo.value) + + def test_set_job_names(self): + """Test set_job_names utils function.""" + def_name = "def_job_name" + name_to_set = "test" + orig_name, name = set_job_names(name_to_set, default_name=def_name) + assert name_to_set == orig_name + assert name_to_set in name + + def test_set_job_names_using_own_default(self): + """Test set_job_names utils function with using own default name.""" + def_name = "def_job_name" + name_to_set = None + orig_name, name = set_job_names(name_to_set, default_name=def_name) + assert def_name == orig_name + assert "job" in name + + def test_set_job_names_using_default(self): + """Test set_job_names utils function with using default name.""" + name_to_set = None + orig_name, name = set_job_names(name_to_set) + assert "unknown_job" == orig_name + assert "job" in name + + def test_create_actinia_pc_item(self): + """Test set_job_names utils function with using default name.""" + module = "g.region" + id = "pc_item_ide" + inputs = { + "raster": "elevation@PERMANENT", + "res": "5000", + } + pc_item = create_actinia_pc_item( + id=id, + module="g.region", + inputs=inputs, + flags="g", + ) + assert "module" in pc_item + assert "id" in pc_item + assert "inputs" in pc_item + assert pc_item["module"] == module + assert pc_item["id"] == id + assert isinstance(pc_item["inputs"], list) + for inp in pc_item["inputs"]: + param = inp["param"] + value = inp["value"] + assert param in inputs + assert inputs[param] == value