diff --git a/.gitignore b/.gitignore index 8a27892..5772024 100755 --- a/.gitignore +++ b/.gitignore @@ -100,3 +100,4 @@ _version.py tests/hdx/api/gsheet_auth.json .vscode/settings.json +.claude/ diff --git a/documentation/index.md b/documentation/index.md index 26ba2f3..12b12a1 100755 --- a/documentation/index.md +++ b/documentation/index.md @@ -33,6 +33,7 @@ upload your datasets to HDX. - [Showcase Management](#showcase-management) - [User Management](#user-management) - [Organization Management](#organization-management) + - [Location Management](#location-management) - [Vocabulary Management](#vocabulary-management) - [Pipeline State](#pipeline-state) - [Working Examples](#working-examples) @@ -1063,6 +1064,48 @@ You can delete a user from an organization: organization.delete_user("USER ID") +## Location Management + +The **Location** class wraps HDX locations (CKAN groups eg. countries, "world"). Reading +locations does not require any special permissions, so these are the operations you will +typically use. Creating, updating and deleting locations is restricted to HDX sysadmins +and works the same way as for other HDX objects (see +[Operations on HDX Objects](#operations-on-hdx-objects)). + +You can read a single location: + + location = Location.read_from_hdx("LOCATION_ID_OR_NAME") + +You can get all location names in HDX: + + location_names = Location.get_all_location_names(**kwargs) + +Passing `all_fields=True` (plus optionally `include_extras=True` etc.) returns the full +location dictionaries instead of just names. Various additional arguments (`**kwargs`) +can be supplied and are detailed in the API documentation. + +You can fetch the countries that are currently active on HDX's Data Grid (a live lookup, +since Data Grid membership changes over time, rather than a fixed list): + + country_codes = Location.get_data_grid_countries() + +This returns a sorted list of 3-letter location codes. + +You can get the datasets belonging to a location as follows: + + datasets = location.get_datasets(**kwargs) + +Various additional arguments (`**kwargs`) can be supplied. These are detailed in the API +documentation. + +You can autocomplete a location name: + + matches = Location.autocomplete("NAME") + +For simple name/code lookups against a cached list of valid locations (eg. converting +between a country name and its HDX code) rather than the full **Location** object, see +the **Locations** helper class instead. + ## Vocabulary Management The **Vocabulary** class enables you to manage CKAN vocabularies, creating, deleting and diff --git a/src/hdx/api/hdx_base_configuration.yaml b/src/hdx/api/hdx_base_configuration.yaml index 0dce103..d2142c7 100755 --- a/src/hdx/api/hdx_base_configuration.yaml +++ b/src/hdx/api/hdx_base_configuration.yaml @@ -72,6 +72,10 @@ organization: - name - title - description +group: + required_fields: + - name + - title "resource view": required_fields: - resource_id diff --git a/src/hdx/api/locations.py b/src/hdx/api/locations.py index 2264e0f..ffb9de0 100755 --- a/src/hdx/api/locations.py +++ b/src/hdx/api/locations.py @@ -3,6 +3,7 @@ from collections.abc import Sequence from hdx.api.configuration import Configuration +from hdx.data.location import Location class Locations: @@ -22,10 +23,8 @@ def validlocations(cls, configuration=None) -> list[dict]: A list of valid locations """ if cls._validlocations is None: - if configuration is None: - configuration = Configuration.read() - cls._validlocations = configuration.call_remoteckan( - "group_list", {"all_fields": True} + cls._validlocations = Location.get_all_location_names( + configuration=configuration, all_fields=True ) return cls._validlocations diff --git a/src/hdx/data/location.py b/src/hdx/data/location.py new file mode 100644 index 0000000..b2c791c --- /dev/null +++ b/src/hdx/data/location.py @@ -0,0 +1,237 @@ +"""Location class containing all logic for creating, checking, and updating locations.""" + +import logging +from collections.abc import Sequence +from typing import Any, Optional + +from hdx.api.configuration import Configuration +from hdx.data.hdxobject import HDXObject + +logger = logging.getLogger(__name__) + + +class Location(HDXObject): + """Location class containing all logic for creating, checking, and updating locations. + A location is a CKAN group. + + Args: + initial_data: Initial location metadata dictionary. Defaults to None. + configuration: HDX configuration. Defaults to global configuration. + """ + + def __init__( + self, + initial_data: dict | None = None, + configuration: Configuration | None = None, + ) -> None: + if not initial_data: + initial_data = {} + super().__init__(initial_data, configuration=configuration) + + @staticmethod + def actions() -> dict[str, str]: + """Dictionary of actions that can be performed on object + + Returns: + Dictionary of actions that can be performed on object + """ + return { + "show": "group_show", + "update": "group_update", + "create": "group_create", + "delete": "group_delete", + "list": "group_list", + "autocomplete": "group_autocomplete", + } + + @classmethod + def read_from_hdx( + cls, identifier: str, configuration: Configuration | None = None + ) -> Optional["Location"]: + """Reads the location given by identifier from HDX and returns Location object + + Args: + identifier: Identifier of location + configuration: HDX configuration. Defaults to global configuration. + + Returns: + Location object if successful read, None if not + """ + return cls._read_from_hdx_class("group", identifier, configuration) + + def check_required_fields(self, ignore_fields: Sequence[str] = ()) -> None: + """Check that metadata for location is complete. The parameter ignore_fields should + be set if required to any fields that should be ignored for the particular operation. + + Args: + ignore_fields: Fields to ignore. Default is (). + + Returns: + None + """ + self._check_required_fields("group", ignore_fields) + + def update_in_hdx(self, **kwargs: Any) -> None: + """Check if location exists in HDX and if so, update location + + Returns: + None + """ + self._update_in_hdx("group", "id", **kwargs) + + def create_in_hdx(self, **kwargs: Any) -> None: + """Check if location exists in HDX and if so, update it, otherwise create location + + Returns: + None + """ + self._create_in_hdx("group", "id", "name", **kwargs) + + def delete_from_hdx(self) -> None: + """Deletes a location from HDX. + + Returns: + None + """ + self._delete_from_hdx("group", "id") + + def get_datasets(self, query: str = "*:*", **kwargs: Any) -> list["Dataset"]: # noqa: F821 + """Get list of datasets in location + + Args: + query: Restrict datasets returned to this query (in Solr format). Defaults to '*:*'. + **kwargs: See below + sort (string): Sorting of the search results. Defaults to 'relevance asc, metadata_modified desc'. + rows (int): Number of matching rows to return. Defaults to all datasets (sys.maxsize). + start (int): Offset in the complete result for where the set of returned datasets should begin + facet (string): Whether to enable faceted results. Default to True. + facet.mincount (int): Minimum counts for facet fields should be included in the results + facet.limit (int): Maximum number of values the facet fields return (- = unlimited). Defaults to 50. + facet.field (list[str]): Fields to facet upon. Default is empty. + use_default_schema (bool): Use default package schema instead of custom schema. Defaults to False. + + Returns: + List of datasets in location + """ + import hdx.data.dataset # avoid circular import + + return hdx.data.dataset.Dataset.search_in_hdx( + query=query, + configuration=self.configuration, + fq=f"groups:{self.data['name']}", + **kwargs, + ) + + @staticmethod + def get_all_location_names( + configuration: Configuration | None = None, **kwargs: Any + ) -> list[str]: + """Get all location names in HDX + + Args: + configuration: HDX configuration. Defaults to global configuration. + **kwargs: See below + sort (str): Sort the search results according to field name and sort-order. Allowed fields are ‘name’, ‘package_count’ and ‘title’. Defaults to 'name asc'. + groups (list[str]): List of names of the groups to return. + all_fields (bool): Return group dictionaries instead of just names. Only core fields are returned - get some more using the include_* options. Defaults to False. + include_extras (bool): If all_fields, include the group extra fields. Defaults to False. + include_tags (bool): If all_fields, include the group tags. Defaults to False. + include_groups (bool): If all_fields, include the groups the groups are in. Defaults to False. + + Returns: + List of all location names in HDX + """ + location = Location(configuration=configuration) + + # Early return for the standard, non-paginated case + if not kwargs.get("all_fields"): + return location._write_to_hdx("list", kwargs) + + compiled_locations = {} + # Extract limit and offset to dictate our paging sizes, defaulting to 400 and 0 + page_size = kwargs.pop("limit", 400) + current_offset = kwargs.pop("offset", 0) + + # 1. Fetch in pages using the provided/default page size + while True: + page_kwargs = kwargs.copy() + page_kwargs["limit"] = page_size + page_kwargs["offset"] = current_offset + + page_data = location._write_to_hdx("list", page_kwargs) + + if not page_data: + break + + # Store by name to automatically deduplicate offset shifts + for loc in page_data: + compiled_locations[loc["name"]] = loc + + if len(page_data) < page_size: + break + + current_offset += page_size + + # 2. Establish Ground Truth LAST (without limit/offset constraints) + truth_kwargs = kwargs.copy() + truth_kwargs["all_fields"] = False + + ground_truth_names = location._write_to_hdx("list", truth_kwargs) + + # 3 & 4. Check for missing IDs, fetch individually, and implicitly prune deletes + final_locations = [] + for name in ground_truth_names: + if name in compiled_locations: + final_locations.append(compiled_locations[name]) + else: + try: + missing_location = location._write_to_hdx("show", {"id": name}) + if missing_location: + final_locations.append(missing_location) + except Exception: + pass + + return final_locations + + @staticmethod + def get_data_grid_countries( + configuration: Configuration | None = None, + ) -> list[str]: + """Fetch HDX's active Data Grid countries (3-letter group names). + + Treated as a live/short-TTL lookup rather than a frozen static list, since Data Grid + membership changes over time. + + Args: + configuration: HDX configuration. Defaults to global configuration. + + Returns: + Sorted list of active Data Grid country (group) names + """ + groups = Location.get_all_location_names( + configuration=configuration, all_fields=True, include_extras=True + ) + return sorted( + group["name"] + for group in groups + if len(group["name"]) == 3 and group.get("data_completeness") == "active" + ) + + @classmethod + def autocomplete( + cls, + name: str, + limit: int = 20, + configuration: Configuration | None = None, + ) -> list: + """Autocomplete a location name and return matches + + Args: + name: Name to autocomplete + limit: Maximum number of matches to return + configuration: HDX configuration. Defaults to global configuration. + + Returns: + Autocomplete matches + """ + return cls._autocomplete(name, limit, configuration) diff --git a/tests/fixtures/location/location_list_all_fields.yaml b/tests/fixtures/location/location_list_all_fields.yaml new file mode 100644 index 0000000..51f6a4c --- /dev/null +++ b/tests/fixtures/location/location_list_all_fields.yaml @@ -0,0 +1,31 @@ +- id: afg + name: afg + title: Afghanistan + display_name: Afghanistan + type: group + is_organization: false + state: active + data_completeness: active + description: "" + package_count: 480 + +- id: zmb + name: zmb + title: Zambia + display_name: Zambia + type: group + is_organization: false + state: active + data_completeness: inactive + description: "" + package_count: 120 + +- id: world + name: world + title: World + display_name: World + type: group + is_organization: false + state: active + description: "" + package_count: 12 diff --git a/tests/fixtures/location/location_show_results.yaml b/tests/fixtures/location/location_show_results.yaml new file mode 100644 index 0000000..7af6fee --- /dev/null +++ b/tests/fixtures/location/location_show_results.yaml @@ -0,0 +1,11 @@ +id: "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" +name: gnq +title: "Equatorial Guinea" +display_name: "Equatorial Guinea" +type: group +is_organization: false +state: active +data_completeness: active +description: "" +package_count: 5 +created: "2015-01-09T14:44:54.006612" diff --git a/tests/hdx/__init__.py b/tests/hdx/__init__.py index 372904a..82e0540 100755 --- a/tests/hdx/__init__.py +++ b/tests/hdx/__init__.py @@ -119,6 +119,11 @@ def json(self): "description": "We do humanitarian work", } +location_data = { + "name": "MyLocation1", + "title": "My Location", +} + user_data = { "name": "MyUser1", "email": "xxx@yyy.com", diff --git a/tests/hdx/api/test_configuration.py b/tests/hdx/api/test_configuration.py index ddde871..888ac8f 100755 --- a/tests/hdx/api/test_configuration.py +++ b/tests/hdx/api/test_configuration.py @@ -487,6 +487,12 @@ def test_project_configuration_dict(self, hdx_config_yaml): "description", ] }, + "group": { + "required_fields": [ + "name", + "title", + ] + }, "resource view": { "required_fields": [ "resource_id", @@ -617,6 +623,12 @@ def test_project_configuration_json(self, hdx_config_yaml, project_config_json): "description", ] }, + "group": { + "required_fields": [ + "name", + "title", + ] + }, "resource view": { "required_fields": [ "resource_id", @@ -739,6 +751,12 @@ def test_project_configuration_yaml(self, hdx_config_yaml, project_config_yaml): "description", ] }, + "group": { + "required_fields": [ + "name", + "title", + ] + }, "resource view": { "required_fields": [ "resource_id", diff --git a/tests/hdx/api/test_locations.py b/tests/hdx/api/test_locations.py index acff9cd..ccaa826 100755 --- a/tests/hdx/api/test_locations.py +++ b/tests/hdx/api/test_locations.py @@ -7,8 +7,10 @@ class MyConfiguration: - def call_remoteckan(self, a, b): - return [{"name": "zaf", "title": "South Africa"}] + def call_remoteckan(self, a, b, files=None): + if b.get("all_fields"): + return [{"name": "zaf", "title": "South Africa"}] + return ["zaf"] class TestHDXLocations: diff --git a/tests/hdx/data/test_location.py b/tests/hdx/data/test_location.py new file mode 100644 index 0000000..9231f86 --- /dev/null +++ b/tests/hdx/data/test_location.py @@ -0,0 +1,381 @@ +"""Location Tests""" + +import copy +import json +from pathlib import Path + +import pytest +from hdx.utilities.dictandlist import merge_two_dictionaries +from hdx.utilities.loader import load_yaml + +from .. import MockResponse, location_data +from hdx.api.configuration import Configuration +from hdx.data.hdxobject import HDXError +from hdx.data.location import Location + +resultdict = load_yaml( + Path("tests") / "fixtures" / "location" / "location_show_results.yaml" +) + +location_list = [ + "afg", + "bgd", + "cod", + "eth", + "irq", + "mli", + "ner", + "som", + "ssd", + "yem", +] +location_list_all_fields = load_yaml( + Path("tests") / "fixtures" / "location" / "location_list_all_fields.yaml" +) + +searchdict = load_yaml(Path("tests") / "fixtures" / "dataset_search_results.yaml") + +location_autocomplete = [ + { + "title": "Zambia", + "id": "zmb", + "name": "zmb", + } +] + + +def location_mockshow(url, datadict): + if "show" not in url: + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not show", "__type": "TEST ERROR: Not Show Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_show"}', + ) + result = json.dumps(resultdict) + if ( + datadict["id"] == "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + or datadict["id"] == "TEST1" + ): + return MockResponse( + 200, + f'{{"success": true, "result": {result}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_show"}}', + ) + if datadict["id"] == "TEST2": + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_show"}', + ) + if datadict["id"] == "TEST3": + return MockResponse( + 200, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_show"}', + ) + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_show"}', + ) + + +def mocklist(url): + if "list" not in url: + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not all", "__type": "TEST ERROR: Not All Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_list"}', + ) + return MockResponse( + 200, + f'{{"success": true, "result": {json.dumps(location_list)}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_list"}}', + ) + + +def mockgetdatasets(url, datadict): + if "search" not in url: + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not search", "__type": "TEST ERROR: Not Search Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=package_search"}', + ) + if datadict["fq"] == "groups:gnq": + return MockResponse( + 200, + f'{{"success": true, "result": {json.dumps(searchdict)}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=package_search"}}', + ) + + +class TestLocation: + @pytest.fixture(scope="function") + def read(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + datadict = json.loads(data.decode("utf-8")) + return location_mockshow(url, datadict) + + Configuration.read().remoteckan().session = MockSession() + + @pytest.fixture(scope="function") + def post_create(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + datadict = json.loads(data.decode("utf-8")) + if "show" in url: + return location_mockshow(url, datadict) + if "create" not in url: + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not create", "__type": "TEST ERROR: Not Create Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_create"}', + ) + + resultdictcopy = copy.deepcopy(resultdict) + resultdictcopy["state"] = datadict["state"] + result = json.dumps(resultdictcopy) + if datadict["name"] == "MyLocation1": + return MockResponse( + 200, + f'{{"success": true, "result": {result}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_create"}}', + ) + if datadict["name"] == "MyLocation2": + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_create"}', + ) + if datadict["name"] == "MyLocation3": + return MockResponse( + 200, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_create"}', + ) + + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_create"}', + ) + + Configuration.read().remoteckan().session = MockSession() + + @pytest.fixture(scope="function") + def post_update(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + datadict = json.loads(data.decode("utf-8")) + if "show" in url: + return location_mockshow(url, datadict) + if "update" not in url: + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not update", "__type": "TEST ERROR: Not Update Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_update"}', + ) + resultdictcopy = copy.deepcopy(resultdict) + merge_two_dictionaries(resultdictcopy, datadict) + + result = json.dumps(resultdictcopy) + if datadict["name"] == "MyLocation1": + return MockResponse( + 200, + f'{{"success": true, "result": {result}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_update"}}', + ) + if datadict["name"] == "MyLocation2": + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_update"}', + ) + if datadict["name"] == "MyLocation3": + return MockResponse( + 200, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_update"}', + ) + + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_update"}', + ) + + Configuration.read().remoteckan().session = MockSession() + + @pytest.fixture(scope="function") + def post_delete(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + decodedata = data.decode("utf-8") + datadict = json.loads(decodedata) + if "show" in url: + return location_mockshow(url, datadict) + if "delete" not in url: + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not delete", "__type": "TEST ERROR: Not Delete Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_delete"}', + ) + if datadict["id"] == "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a": + return MockResponse( + 200, + f'{{"success": true, "result": {decodedata}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_delete"}}', + ) + + return MockResponse( + 404, + '{"success": false, "error": {"message": "Not found", "__type": "Not Found Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_delete"}', + ) + + Configuration.read().remoteckan().session = MockSession() + + @pytest.fixture(scope="function") + def post_list(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + json.loads(data.decode("utf-8")) + return mocklist(url) + + Configuration.read().remoteckan().session = MockSession() + + @pytest.fixture(scope="function") + def post_all_fields(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + kwargs = json.loads(data.decode("utf-8")) + if "show" in url: + return location_mockshow(url, kwargs) + if kwargs["all_fields"]: + return MockResponse( + 200, + f'{{"success": true, "result": {json.dumps(location_list_all_fields)}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_list"}}', + ) + names = [x["name"] for x in location_list_all_fields] + names.append("TEST1") + return MockResponse( + 200, + f'{{"success": true, "result": {json.dumps(names)}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_list"}}', + ) + + Configuration.read().remoteckan().session = MockSession() + + @pytest.fixture(scope="function") + def datasets_get(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + datadict = json.loads(data.decode("utf-8")) + return mockgetdatasets(url, datadict) + + Configuration.read().remoteckan().session = MockSession() + + @pytest.fixture(scope="function") + def post_autocomplete(self): + class MockSession: + @staticmethod + def post(url, data, **kwargs): + decodedata = data.decode("utf-8") + datadict = json.loads(decodedata) + if "autocomplete" not in url or "zmb" not in datadict["q"]: + return MockResponse( + 404, + '{"success": false, "error": {"message": "TEST ERROR: Not autocomplete", "__type": "TEST ERROR: Not Autocomplete Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_autocomplete"}', + ) + result = json.dumps(location_autocomplete) + return MockResponse( + 200, + f'{{"success": true, "result": {result}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=group_autocomplete"}}', + ) + + Configuration.read().remoteckan().session = MockSession() + + def test_read_from_hdx(self, configuration, read, mocksmtp): + location = Location.read_from_hdx("f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a") + assert location["id"] == "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + assert location["name"] == "gnq" + location = Location.read_from_hdx("TEST2") + assert location is None + location = Location.read_from_hdx("TEST3") + assert location is None + + def test_create_in_hdx(self, configuration, post_create): + location = Location() + with pytest.raises(HDXError): + location.create_in_hdx() + location["id"] = "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + location["name"] = "LALA" + with pytest.raises(HDXError): + location.create_in_hdx() + + loc_data = copy.deepcopy(location_data) + location = Location(loc_data) + location.create_in_hdx() + assert location["id"] == "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + assert location["state"] == "active" + + loc_data["name"] = "MyLocation2" + location = Location(loc_data) + with pytest.raises(HDXError): + location.create_in_hdx() + + loc_data["name"] = "MyLocation3" + location = Location(loc_data) + with pytest.raises(HDXError): + location.create_in_hdx() + + def test_update_in_hdx(self, configuration, post_update): + location = Location() + location["id"] = "NOTEXIST" + with pytest.raises(HDXError): + location.update_in_hdx() + location["name"] = "LALA" + with pytest.raises(HDXError): + location.update_in_hdx() + + location = Location.read_from_hdx("f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a") + assert location["id"] == "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + assert location["name"] == "gnq" + + location["title"] = "New Title" + location["id"] = "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + location["name"] = "MyLocation1" + location.update_in_hdx() + assert location["id"] == "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + assert location["title"] == "New Title" + assert location["state"] == "active" + + location["id"] = "NOTEXIST" + with pytest.raises(HDXError): + location.update_in_hdx() + + del location["id"] + with pytest.raises(HDXError): + location.update_in_hdx() + + loc_data = copy.deepcopy(location_data) + loc_data["name"] = "MyLocation1" + loc_data["id"] = "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + location = Location(loc_data) + location.create_in_hdx() + assert location["id"] == "f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a" + assert location["name"] == "MyLocation1" + assert location["state"] == "active" + + def test_delete_from_hdx(self, configuration, post_delete): + location = Location.read_from_hdx("f8b0e58f-9be5-4b25-9d5f-1c8b6d8b6a0a") + location.delete_from_hdx() + del location["id"] + with pytest.raises(HDXError): + location.delete_from_hdx() + + def test_get_all_location_names(self, configuration, post_list): + locations = Location.get_all_location_names() + assert len(locations) == 10 + + def test_get_all_location_names_all_fields(self, configuration, post_all_fields): + locations = Location.get_all_location_names(all_fields=True) + assert len(locations) == 4 + + def test_get_datasets(self, configuration, datasets_get): + loc_data = copy.deepcopy(resultdict) + location = Location(loc_data) + datasets = location.get_datasets() + assert len(datasets) == 10 + + def test_autocomplete(self, configuration, post_autocomplete): + assert Location.autocomplete("zmb") == location_autocomplete + + def test_get_data_grid_countries(self, configuration, post_all_fields): + countries = Location.get_data_grid_countries() + assert countries == ["afg", "gnq"]