diff --git a/python/grass/temporal/abstract_map_dataset.py b/python/grass/temporal/abstract_map_dataset.py index f575d9810dd..ac31dc28f94 100644 --- a/python/grass/temporal/abstract_map_dataset.py +++ b/python/grass/temporal/abstract_map_dataset.py @@ -165,6 +165,10 @@ def get_map_id(self): """ return self.base.get_map_id() + def get_readonly(self) -> bool: + """Return True if this map is read-only (belongs to a foreign mapset)""" + return self.base.get_readonly() + @staticmethod def split_name(name: str, layer=None, mapset=None): """Convenient method to split a map name into three potentially @@ -356,7 +360,7 @@ def insert(self, dbif=None, execute: bool = True): :return: The SQL insert statement in case execute=False, or an empty string otherwise """ - if get_enable_timestamp_write(): + if get_enable_timestamp_write() and not self.get_readonly(): self.write_timestamp_to_grass() return AbstractDataset.insert(self, dbif=dbif, execute=execute) @@ -375,7 +379,7 @@ def update(self, dbif=None, execute: bool = True): :return: The SQL insert statement in case execute=False, or an empty string otherwise """ - if get_enable_timestamp_write(): + if get_enable_timestamp_write() and not self.get_readonly(): self.write_timestamp_to_grass() return AbstractDataset.update(self, dbif, execute) @@ -395,7 +399,7 @@ def update_all(self, dbif=None, execute: bool = True): empty string otherwise """ - if get_enable_timestamp_write(): + if get_enable_timestamp_write() and not self.get_readonly(): self.write_timestamp_to_grass() return AbstractDataset.update_all(self, dbif, execute) @@ -534,7 +538,7 @@ def update_absolute_time(self, start_time, end_time=None, dbif=None) -> None: if connection_state_changed: dbif.close() - if get_enable_timestamp_write(): + if get_enable_timestamp_write() and not self.get_readonly(): self.write_timestamp_to_grass() def set_relative_time(self, start_time, end_time, unit) -> bool: @@ -655,7 +659,7 @@ def update_relative_time(self, start_time, end_time, unit, dbif=None) -> None: if connection_state_changed: dbif.close() - if get_enable_timestamp_write(): + if get_enable_timestamp_write() and not self.get_readonly(): self.write_timestamp_to_grass() def set_temporal_extent(self, extent) -> None: diff --git a/python/grass/temporal/base.py b/python/grass/temporal/base.py index 3504844999f..2d786eeba72 100644 --- a/python/grass/temporal/base.py +++ b/python/grass/temporal/base.py @@ -256,10 +256,11 @@ def __init__(self, table=None, ident=None) -> None: self.ident = ident self.msgr = get_tgis_message_interface() - if self.ident and self.ident.find("@") >= 0: + if self.ident and "@" in self.ident: self.mapset = self.ident.split("@")[1] else: self.mapset = None + self._readonly = False def get_table_name(self): """Return the name of the table in which the internal @@ -631,12 +632,12 @@ class DatasetBase(SQLDatabaseInterface): def __init__( self, table=None, - ident=None, + ident: str | None = None, name: str | None = None, - mapset=None, - creator=None, - ctime=None, - ttype=None, + mapset: str | None = None, + creator: str | None = None, + ctime: datetime | None = None, + ttype: str | None = None, ) -> None: """Constructor @@ -660,15 +661,17 @@ def __init__( self.set_id(ident) if ident is not None and name is None and mapset is None: - if ident.find("@") >= 0: - name, mapset = ident.split("@") - if name.find(":") >= 0: - name, layer = ident.split(":") + if "@" in ident: + name, mapset = ident.split("@", 1) + if ":" in name: + name, layer = name.split(":", 1) self.set_name(name) self.set_mapset(mapset) self.set_creator(creator) self.set_ctime(ctime) self.set_ttype(ttype) + self._readonly = False + self.set_readonly() def set_id(self, ident) -> None: """Convenient method to set the unique identifier (primary key) @@ -682,14 +685,14 @@ def set_id(self, ident) -> None: name = "" if ident is not None: - if ident.find("@") >= 0: - name, mapset = ident.split("@") + if "@" in ident: + name, mapset = ident.split("@", 1) self.set_mapset(mapset) self.set_name(name) else: self.msgr.fatal(_("Wrong identifier, the mapset is missing")) - if name.find(":") >= 0: - name, layer = ident.split(":") + if ":" in name: + name, layer = name.split(":", 1) self.set_layer(layer) self.set_name(name) @@ -706,6 +709,7 @@ def set_mapset(self, mapset) -> None: :param mapset: The name of the mapset in which this dataset is stored """ self.D["mapset"] = mapset + self.set_readonly() def set_layer(self, layer) -> None: """Convenient method to set the layer of the map (part of primary key) @@ -745,16 +749,14 @@ def set_ttype(self, ttype) -> None: else: self.D["temporal_type"] = ttype - def get_id(self): + def get_id(self) -> str | None: """Convenient method to get the unique identifier (primary key) :return: None if not found """ - if "id" in self.D: - return self.D["id"] - return None + return self.D.get("id") - def get_map_id(self): + def get_map_id(self) -> str | None: """Convenient method to get the unique map identifier without layer information @@ -762,67 +764,65 @@ def get_map_id(self): or None in case the id was not set """ if self.id: - if self.id.find(":") >= 0: + if ":" in self.id: # Remove the layer identifier from the id - return self.id.split("@")[0].split(":")[0] + "@" + self.id.split("@")[1] + return f"{self.id.split('@')[0].split(':')[0]}@{self.id.split('@')[1]}" return self.id return None - def get_layer(self): + def get_layer(self) -> str | None: """Convenient method to get the layer of the map (part of primary key) Layer are currently supported for vector maps :return: None if not found """ - if "layer" in self.D: - return self.D["layer"] - return None + return self.D.get("layer") - def get_name(self): + def get_name(self) -> str | None: """Get the name of the dataset :return: None if not found """ - if "name" in self.D: - return self.D["name"] - return None + return self.D.get("name") - def get_mapset(self): + def get_mapset(self) -> str | None: """Get the name of mapset of this dataset :return: None if not found """ - if "mapset" in self.D: - return self.D["mapset"] - return None + return self.D.get("mapset") - def get_creator(self): + def get_creator(self) -> str | None: """Get the creator of the dataset :return: None if not found """ - if "creator" in self.D: - return self.D["creator"] - return None + return self.D.get("creator") - def get_ctime(self): + def get_ctime(self) -> datetime | None: """Get the creation time of the dataset, datatype is datetime :return: None if not found """ - if "creation_time" in self.D: - return self.D["creation_time"] - return None + return self.D.get("creation_time") - def get_ttype(self): + def get_ttype(self) -> str | None: """Get the temporal type of the map :return: None if not found """ - if "temporal_type" in self.D: - return self.D["temporal_type"] - return None + return self.D.get("temporal_type") + + def set_readonly(self) -> None: + """Set the readonly flag for this object""" + self._readonly = bool( + self.get_mapset() and self.get_mapset() != get_current_mapset() + ) + + def get_readonly(self) -> bool: + """Get the readonly flag for this object""" + return self._readonly # Properties of this class id = property(fget=get_id, fset=set_id) @@ -832,6 +832,7 @@ def get_ttype(self): ctime = property(fget=get_ctime, fset=set_ctime) ttype = property(fget=get_ttype, fset=set_ttype) creator = property(fget=get_creator, fset=set_creator) + readonly = property(fget=get_readonly) def print_info(self) -> None: """Print information about this class in human readable style""" @@ -1030,9 +1031,7 @@ def get_semantic_type(self): :return: None if not found """ - if "semantic_type" in self.D: - return self.D["semantic_type"] - return None + return self.D.get("semantic_type") def get_mtime(self): """Get the modification time of the space time dataset, datatype is @@ -1040,9 +1039,7 @@ def get_mtime(self): :return: None if not found """ - if "modification_time" in self.D: - return self.D["modification_time"] - return None + return self.D.get("modification_time") semantic_type = property(fget=get_semantic_type, fset=set_semantic_type) @@ -1202,9 +1199,7 @@ def get_id(self): :return: None if not found """ - if "id" in self.D: - return self.D["id"] - return None + return self.D.get("id") def get_registered_stds(self): """Get the comma separated list of space time datasets ids @@ -1212,9 +1207,7 @@ def get_registered_stds(self): :return: None if not found """ - if "registered_stds" in self.D: - return self.D["registered_stds"] - return None + return self.D.get("registered_stds") # Properties of this class id = property(fget=get_id, fset=set_id) diff --git a/python/grass/temporal/space_time_datasets.py b/python/grass/temporal/space_time_datasets.py index 4dc9bf6315d..6c21be6c587 100644 --- a/python/grass/temporal/space_time_datasets.py +++ b/python/grass/temporal/space_time_datasets.py @@ -309,6 +309,9 @@ def write_timestamp_to_grass(self) -> bool: :return: True if success, False on error """ + if self.get_readonly(): + return False + check = self.ciface.write_raster_timestamp( self.get_name(), self.get_mapset(), self._convert_timestamp() ) @@ -341,6 +344,9 @@ def remove_timestamp_from_grass(self) -> bool: :return: True if success, False on error """ + if self.get_readonly(): + return False + check = self.ciface.remove_raster_timestamp(self.get_name(), self.get_mapset()) if check == -1: @@ -376,10 +382,14 @@ def write_semantic_label_to_grass(self) -> bool: """Write the semantic label of this map into the map metadata in the GRASS file system based spatial database. - Internally the libgis API functions are used for writing + Internally the libgis API functions are used for writing. Writing is + only possible if the map is in the current mapset (not read-only). :return: True if success, False on error """ + if self.get_readonly(): + return False + check = self.ciface.write_raster_semantic_label( self.get_name(), self.get_mapset(), self.metadata.get_semantic_label() ) @@ -470,7 +480,8 @@ def set_semantic_label(self, semantic_label) -> None: :param str semantic_label: semantic label (eg. S2_1) """ self.metadata.set_semantic_label(semantic_label) - self.write_semantic_label_to_grass() + if not self.get_readonly(): + self.write_semantic_label_to_grass() ############################################################################### @@ -745,6 +756,9 @@ def write_timestamp_to_grass(self) -> bool: :return: True if success, False on error """ + if self.get_readonly(): + return False + check = self.ciface.write_raster3d_timestamp( self.get_name(), self.get_mapset(), self._convert_timestamp() ) @@ -774,6 +788,9 @@ def remove_timestamp_from_grass(self) -> bool: :return: True if success, False on error """ + if self.get_readonly(): + return False + check = self.ciface.remove_raster3d_timestamp( self.get_name(), self.get_mapset() ) @@ -1084,6 +1101,9 @@ def write_timestamp_to_grass(self) -> bool: Internally the libgis API functions are used for writing """ + if self.get_readonly(): + return False + check = self.ciface.write_vector_timestamp( self.get_name(), self.get_mapset(), @@ -1113,6 +1133,9 @@ def remove_timestamp_from_grass(self) -> bool: Internally the libgis API functions are used for removal """ + if self.get_readonly(): + return False + check = self.ciface.remove_vector_timestamp(self.get_name(), self.get_mapset()) if check == -1: diff --git a/python/grass/temporal/tests/grass_temporal_readonly_mapset_test.py b/python/grass/temporal/tests/grass_temporal_readonly_mapset_test.py new file mode 100644 index 00000000000..e587046f5b9 --- /dev/null +++ b/python/grass/temporal/tests/grass_temporal_readonly_mapset_test.py @@ -0,0 +1,182 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""Test that registering a map from another mapset does not +write file-based metadata (timestamps, semantic labels) to +the source mapset or create ghost metadata files in the current mapset. + +A map in a foreign mapset is read-only from the perspective of the current +mapset, so timestamps and semantic labels must only be stored in the temporal +database, not written back to the file-system metadata of that foreign map. +The C-level write functions ignore the mapset argument and always write into +the current mapset, so without the read-only guard a registration would create +ghost cell_misc entries in PERMANENT for a map that does not exist there. +""" + +from datetime import datetime +from pathlib import Path + +import pytest + +from grass.grassdb.create import create_mapset +import grass.script as gs +import grass.temporal as tgis +from grass.tools import Tools + + +@pytest.fixture(scope="module") +def cross_mapset_session(tmp_path_factory): + """Project with a *source* mapset and PERMANENT. + + source: raster map *foreign_map* (no timestamp, no semantic label) + PERMANENT: raster map *local_map* registered in *test_strds*; also + registers *foreign_map@source* with an absolute timestamp and a + semantic label via a file. + """ + project = tmp_path_factory.mktemp("data") / "project" + file_dir = tmp_path_factory.mktemp("files") + gs.create_project(project) + create_mapset(project / "source") + + with gs.setup.init(project / "source"): + tools = Tools() + tools.g_region(s=0, n=1, w=0, e=1, res=1) + tools.r_mapcalc(expression="foreign_map = 1") + + with gs.setup.init(project) as session: + tools = Tools(session=session) + tools.g_region(s=0, n=1, w=0, e=1, res=1) + tools.r_mapcalc(expression="local_map = 1") + + tools.t_create(output="test_strds", type="strds", title="T", description="T") + tools.t_register(input="test_strds", maps="local_map", start="2001-01-01") + + # Pipe in a timestamp and semantic label for the foreign map. + map_file = file_dir / "maps.txt" + map_file.write_text("foreign_map@source|2010-06-01|S2_1\n") + tools.t_register(input="test_strds", file=map_file) + + yield session + + +def test_local_map_gets_file_timestamp(cross_mapset_session): + """A map in the current mapset receives a file-based timestamp.""" + tgis.init() + rmap = tgis.RasterDataset("local_map@PERMANENT") + assert not rmap.get_readonly() + assert rmap.has_grass_timestamp(), ( + "local_map should have a file-based timestamp after t.register" + ) + + +def test_foreign_map_no_file_timestamp(cross_mapset_session): + """A map in another mapset must not receive a file-based timestamp.""" + tgis.init() + rmap = tgis.RasterDataset("foreign_map@source") + assert rmap.get_readonly() + assert not rmap.has_grass_timestamp(), ( + "foreign_map in source mapset must not have its file-based timestamp " + "written when registered from PERMANENT" + ) + + +def test_foreign_map_no_file_semantic_label(cross_mapset_session): + """A map in another mapset must not have a semantic label written to file.""" + tgis.init() + rmap = tgis.RasterDataset("foreign_map@source") + # read_semantic_label_from_grass returns False when no label is on disk + assert rmap.get_readonly() + assert not rmap.read_semantic_label_from_grass(), ( + "foreign_map in source mapset must not have its semantic label " + "written to the file system when registered from PERMANENT" + ) + + +def test_foreign_map_direct_metadata_writes_are_skipped( + cross_mapset_session, monkeypatch +): + """Direct file-metadata helpers must also respect a foreign mapset.""" + tgis.init() + rmap = tgis.RasterDataset("foreign_map@source") + rmap.set_absolute_time(datetime(2010, 6, 1)) + rmap.metadata.set_semantic_label("S2_1") + + def unexpected_write(*args): + pytest.fail("GRASS metadata write was called for a foreign map") + + monkeypatch.setattr(rmap.ciface, "write_raster_timestamp", unexpected_write) + monkeypatch.setattr(rmap.ciface, "write_raster_semantic_label", unexpected_write) + + assert not rmap.write_timestamp_to_grass() + assert not rmap.write_semantic_label_to_grass() + + +def test_foreign_map_direct_timestamp_removal_is_skipped(cross_mapset_session): + """Removing a foreign timestamp must not remove a current-mapset file.""" + tgis.init() + rmap = tgis.RasterDataset("foreign_map@source") + env = gs.gisenv() + project = Path(env["GISDBASE"]) / env["LOCATION_NAME"] + timestamp = project / "PERMANENT" / "cell_misc" / "foreign_map" / "timestamp" + timestamp.parent.mkdir(parents=True, exist_ok=True) + timestamp.write_text("local timestamp") + + try: + assert not rmap.remove_timestamp_from_grass() + assert timestamp.read_text() == "local timestamp" + finally: + timestamp.unlink(missing_ok=True) + + +def test_mapset_change_updates_readonly_state(cross_mapset_session): + """Changing a dataset mapset must update its file-write protection.""" + tgis.init() + rmap = tgis.RasterDataset("local_map@PERMANENT") + assert not rmap.get_readonly() + + rmap.base.set_mapset("source") + assert rmap.get_readonly() + + rmap.base.set_mapset("PERMANENT") + assert not rmap.get_readonly() + + +def test_foreign_map_has_db_timestamp(cross_mapset_session): + """The temporal DB entry for the foreign map has the registered time.""" + tgis.init() + dbif = tgis.SQLDatabaseInterfaceConnection() + dbif.connect() + try: + rmap = tgis.RasterDataset("foreign_map@source") + rmap.select(dbif=dbif, mapset=tgis.get_current_mapset()) + start, _ = rmap.get_absolute_time() + finally: + dbif.close() + assert rmap.get_readonly() + assert start is not None + assert start.year == 2010 + + +def test_no_ghost_timestamp_in_current_mapset(cross_mapset_session): + """Registering a foreign map must not create a ghost timestamp in PERMANENT. + + G_write_*_timestamp ignores the mapset argument and always writes into the + current mapset. Without the read-only guard this would create + cell_misc/foreign_map/timestamp in PERMANENT for a map that lives in source. + """ + env = gs.gisenv() + project = Path(env["GISDBASE"]) / env["LOCATION_NAME"] + ghost = project / "PERMANENT" / "cell_misc" / "foreign_map" / "timestamp" + assert not ghost.exists(), ( + "Registering foreign_map@source must not create a ghost timestamp " + "file in PERMANENT/cell_misc/foreign_map/" + ) + + +def test_no_ghost_semantic_label_in_current_mapset(cross_mapset_session): + """Registering a foreign map must not create a ghost semantic label in PERMANENT.""" + env = gs.gisenv() + project = Path(env["GISDBASE"]) / env["LOCATION_NAME"] + ghost = project / "PERMANENT" / "cell_misc" / "foreign_map" / "semantic_label" + assert not ghost.exists(), ( + "Registering foreign_map@source must not create a ghost semantic_label " + "file in PERMANENT/cell_misc/foreign_map/" + )