From 6899fe00036ed506828e71075bcdcaba9216fd15 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:33:48 +0900 Subject: [PATCH 1/2] feat: path_of(store, key), public file path of a key through wrapped file stores (#158) --- dol/__init__.py | 1 + dol/filesys.py | 44 ++++++++++++++++++++++++++++++++++++--- dol/tests/test_filesys.py | 38 ++++++++++++++++++++++++++++----- 3 files changed, 75 insertions(+), 8 deletions(-) diff --git a/dol/__init__.py b/dol/__init__.py index 22d310f3..f9930f71 100644 --- a/dol/__init__.py +++ b/dol/__init__.py @@ -106,6 +106,7 @@ def ihead(store, n=1): resolve_dir, # to get a full path (resolve ~ and .) and ensure it is a directory DirReader, # recursive read-only access to directories, temp_dir, # make a temporary directory, + path_of, # file-system path of a key in a (possibly wrapped) file store PickleFiles, # CRUD access to pickled files JsonFiles, # CRUD access to jsob files, Jsons, # Same as JsonFiles, but with added .json extension handling diff --git a/dol/filesys.py b/dol/filesys.py index 9572605e..46235bcc 100644 --- a/dol/filesys.py +++ b/dol/filesys.py @@ -524,9 +524,9 @@ def __init__( if max_levels is None: max_levels = inf subpath_implied_min_levels = len(subpath.split(os.path.sep)) - 1 - assert max_levels >= subpath_implied_min_levels, ( - f"max_levels is {max_levels}, but subpath {subpath} would imply at least {subpath_implied_min_levels}" - ) + assert ( + max_levels >= subpath_implied_min_levels + ), f"max_levels is {max_levels}, but subpath {subpath} would imply at least {subpath_implied_min_levels}" pattern_for_field = pattern_for_field or {} self.rootdir = ensure_slash_suffix(rootdir) self.subpath = subpath @@ -1050,6 +1050,44 @@ def _missing_dir_of_key(store, k): return dirname +def path_of(store, key) -> str: + """The file-system path of ``key`` in a file-backed ``store``, however it is wrapped. + + The key is followed through every layer's ``_id_of_key`` (the same walk as + ``dol.inner_most_key``) down to the innermost file-system store, which must accept + the result: a key leaving ``rootdir`` raises ``KeyValidationError`` (a ``KeyError``), + as a read or write of it would. A store with no file-system layer at its core + raises ``TypeError``. Nothing is read or created. + + >>> import tempfile + >>> from dol import Files, KeyCodecs + >>> root = tempfile.mkdtemp() + >>> path_of(Files(root), 'a/b.txt') == os.path.join(root, 'a', 'b.txt') + True + >>> s = KeyCodecs.suffixed('.json')(Files(root)) # wrapped: the outer key is mapped + >>> path_of(s, 'a').endswith('a.json') + True + >>> path_of(Files(root), '../x') + Traceback (most recent call last): + ... + dol.filesys.KeyValidationError: 'Key not valid (usually because does not exist or access not permitted): ...' + >>> path_of({}, 'a') + Traceback (most recent call last): + ... + TypeError: ... + """ + from dol.dig import inner_most_key + + innermost = _innermost_layer(store) + if not isinstance(innermost, FileSysCollection): + raise TypeError( + f"{type(innermost).__name__} is not a file-system store: no path for {key!r}" + ) + path = os.fspath(inner_most_key(store, key)) + innermost.validate_key(path) + return path + + def _innermost_layer(store): """The deepest layer of ``store``'s ``.store`` chain, walked the way ``inner_most_key`` walks it (only through layers that define ``_id_of_key``).""" diff --git a/dol/tests/test_filesys.py b/dol/tests/test_filesys.py index 6cf49f78..38495809 100644 --- a/dol/tests/test_filesys.py +++ b/dol/tests/test_filesys.py @@ -232,7 +232,10 @@ def test_subfolder_stores(): # Testing folder1 folder1_store = stores["folder1"] assert isinstance(folder1_store, Files) - assert set(folder1_store.keys()) == {"day.doc", os.path.join("subfolder", "apple.p")} + assert set(folder1_store.keys()) == { + "day.doc", + os.path.join("subfolder", "apple.p"), + } assert folder1_store["day.doc"] == b"time" # Testing folder1/subfolder @@ -252,7 +255,8 @@ def test_subfolder_stores(): def test_is_valid_key_sees_the_inner_key(tmpdir): """``Files`` is ``mk_relative_path_store(FileBytesPersister)``, so ``is_valid_key`` was reached through ``Store.__getattr__`` with the RELATIVE key while the leaf matched it - against a pattern built from the absolute path -- returning False for keys that exist.""" + against a pattern built from the absolute path -- returning False for keys that exist. + """ import os from dol import Files, TextFiles @@ -484,9 +488,10 @@ def test_symlink_then_dotdot_is_refused_by_every_containment_user(tmp_path): s[os.path.join("lnk", "..", "made", "x.txt")] = b"pwned" assert not (tmp_path / "outside" / "made").exists() # allow_escape=True is the explicit opt-out - assert Files(str(root), allow_escape=True)[ - os.path.join("lnk", "..", "secret.txt") - ] == b"secret" + assert ( + Files(str(root), allow_escape=True)[os.path.join("lnk", "..", "secret.txt")] + == b"secret" + ) def test_rootdir_with_dotdot_after_a_symlink_lists_and_reads(tmp_path): @@ -563,3 +568,26 @@ def test_bytes_persister_rejects_non_bytes_before_open(tmp_path): assert s["c"] == b"keep" s["c"] = bytearray(b"ok") assert s["c"] == b"ok" + + +def test_path_of_follows_wrappers_and_validates(tmp_path): + import os + + import pytest + + from dol import Files, KeyCodecs, path_of, wrap_kvs + from dol.filesys import KeyValidationError + + root = str(tmp_path) + assert path_of(Files(root), "a/b.txt") == os.path.join(root, "a", "b.txt") + suffixed = KeyCodecs.suffixed(".json")(Files(root)) + assert path_of(suffixed, "a") == os.path.join(root, "a.json") + upper = wrap_kvs(Files(root), id_of_key=str.lower, key_of_id=str.upper) + assert path_of(upper, "ABC") == os.path.join(root, "abc") + with pytest.raises(KeyError): + path_of(Files(root), "../x") + with pytest.raises(KeyValidationError): + path_of(suffixed, "../x") + with pytest.raises(TypeError): + path_of({}, "a") + assert os.listdir(root) == [] # nothing created From 6b189e7b9474bf7cc54d577b7265ada0964f4c75 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:38:59 +0900 Subject: [PATCH 2/2] path_of: abspath result, bare leaf, shared with _missing_dir_of_key, more tests (review) --- dol/filesys.py | 45 ++++++++++++++----------------- dol/tests/test_filesys.py | 56 ++++++++++++++++++++++++++++----------- 2 files changed, 61 insertions(+), 40 deletions(-) diff --git a/dol/filesys.py b/dol/filesys.py index 46235bcc..5d439da8 100644 --- a/dol/filesys.py +++ b/dol/filesys.py @@ -524,9 +524,9 @@ def __init__( if max_levels is None: max_levels = inf subpath_implied_min_levels = len(subpath.split(os.path.sep)) - 1 - assert ( - max_levels >= subpath_implied_min_levels - ), f"max_levels is {max_levels}, but subpath {subpath} would imply at least {subpath_implied_min_levels}" + assert max_levels >= subpath_implied_min_levels, ( + f"max_levels is {max_levels}, but subpath {subpath} would imply at least {subpath_implied_min_levels}" + ) pattern_for_field = pattern_for_field or {} self.rootdir = ensure_slash_suffix(rootdir) self.subpath = subpath @@ -1030,24 +1030,12 @@ def _missing_dir_of_key(store, k): relative path (a relative ``rootdir``) is made relative to the CWD, which is where the write will put it. """ - from dol.dig import inner_most_key - - innermost = _innermost_layer(store) - if not isinstance(innermost, FileSysCollection): - return None - _id = inner_most_key(store, k, default=None) - if isinstance(_id, os.PathLike): - _id = os.fspath(_id) - if not isinstance(_id, str): - return None - dirname = os.path.dirname(_id) - if not dirname or os.path.isdir(dirname): - return None try: - innermost.validate_key(_id) - except KeyValidationError: - return None # the write will raise for this key; make no dirs for it (#107) - return dirname + path = path_of(store, k) + except (TypeError, KeyError): # not a file store / key not followable / key rejected + return None + dirname = os.path.dirname(path) + return None if os.path.isdir(dirname) else dirname def path_of(store, key) -> str: @@ -1056,13 +1044,19 @@ def path_of(store, key) -> str: The key is followed through every layer's ``_id_of_key`` (the same walk as ``dol.inner_most_key``) down to the innermost file-system store, which must accept the result: a key leaving ``rootdir`` raises ``KeyValidationError`` (a ``KeyError``), - as a read or write of it would. A store with no file-system layer at its core - raises ``TypeError``. Nothing is read or created. + as a read or write of it would. A store with no file-system layer at its core, or + with a layer that maps keys without defining ``_id_of_key`` (so the key cannot be + followed), raises ``TypeError``. Nothing is read or created. The result is an + absolute, normalized path (relative roots resolve against the CWD). + + Limit: like ``dol.dig.key_in_scope``, this reads what the layers declare through + ``_id_of_key``; a layer that maps keys by other means (``add_path_access``, + ``flatten``) is not followed, and a ``filt_iter`` scope is not enforced. >>> import tempfile >>> from dol import Files, KeyCodecs >>> root = tempfile.mkdtemp() - >>> path_of(Files(root), 'a/b.txt') == os.path.join(root, 'a', 'b.txt') + >>> path_of(Files(root), os.path.join('a', 'b.txt')) == os.path.join(root, 'a', 'b.txt') True >>> s = KeyCodecs.suffixed('.json')(Files(root)) # wrapped: the outer key is mapped >>> path_of(s, 'a').endswith('a.json') @@ -1083,9 +1077,10 @@ def path_of(store, key) -> str: raise TypeError( f"{type(innermost).__name__} is not a file-system store: no path for {key!r}" ) - path = os.fspath(inner_most_key(store, key)) + path = key if innermost is store else inner_most_key(store, key) + path = os.fspath(path) innermost.validate_key(path) - return path + return os.path.abspath(path) def _innermost_layer(store): diff --git a/dol/tests/test_filesys.py b/dol/tests/test_filesys.py index 38495809..a64bba87 100644 --- a/dol/tests/test_filesys.py +++ b/dol/tests/test_filesys.py @@ -232,10 +232,7 @@ def test_subfolder_stores(): # Testing folder1 folder1_store = stores["folder1"] assert isinstance(folder1_store, Files) - assert set(folder1_store.keys()) == { - "day.doc", - os.path.join("subfolder", "apple.p"), - } + assert set(folder1_store.keys()) == {"day.doc", os.path.join("subfolder", "apple.p")} assert folder1_store["day.doc"] == b"time" # Testing folder1/subfolder @@ -255,8 +252,7 @@ def test_subfolder_stores(): def test_is_valid_key_sees_the_inner_key(tmpdir): """``Files`` is ``mk_relative_path_store(FileBytesPersister)``, so ``is_valid_key`` was reached through ``Store.__getattr__`` with the RELATIVE key while the leaf matched it - against a pattern built from the absolute path -- returning False for keys that exist. - """ + against a pattern built from the absolute path -- returning False for keys that exist.""" import os from dol import Files, TextFiles @@ -488,10 +484,9 @@ def test_symlink_then_dotdot_is_refused_by_every_containment_user(tmp_path): s[os.path.join("lnk", "..", "made", "x.txt")] = b"pwned" assert not (tmp_path / "outside" / "made").exists() # allow_escape=True is the explicit opt-out - assert ( - Files(str(root), allow_escape=True)[os.path.join("lnk", "..", "secret.txt")] - == b"secret" - ) + assert Files(str(root), allow_escape=True)[ + os.path.join("lnk", "..", "secret.txt") + ] == b"secret" def test_rootdir_with_dotdot_after_a_symlink_lists_and_reads(tmp_path): @@ -575,19 +570,50 @@ def test_path_of_follows_wrappers_and_validates(tmp_path): import pytest - from dol import Files, KeyCodecs, path_of, wrap_kvs - from dol.filesys import KeyValidationError + from dol import Files, Jsons, KeyCodecs, Pipe, mk_dirs_if_missing, path_of, wrap_kvs + from dol.filesys import FileBytesPersister, KeyValidationError root = str(tmp_path) - assert path_of(Files(root), "a/b.txt") == os.path.join(root, "a", "b.txt") + assert path_of(Files(root), os.path.join("a", "b.txt")) == os.path.join( + root, "a", "b.txt" + ) suffixed = KeyCodecs.suffixed(".json")(Files(root)) assert path_of(suffixed, "a") == os.path.join(root, "a.json") upper = wrap_kvs(Files(root), id_of_key=str.lower, key_of_id=str.upper) assert path_of(upper, "ABC") == os.path.join(root, "abc") + stacked = Pipe(KeyCodecs.suffixed(".x"), mk_dirs_if_missing)(Files(root)) + assert path_of(stacked, "k") == os.path.join(root, "k.x") + assert path_of(Jsons(root), "k") == os.path.join(root, "k.json") + # a bare leaf: keys already are paths + leaf = FileBytesPersister(root) + assert path_of(leaf, os.path.join(root, "z")) == os.path.join(root, "z") with pytest.raises(KeyError): - path_of(Files(root), "../x") + path_of(Files(root), os.path.join("..", "x")) + with pytest.raises(KeyValidationError): + path_of(suffixed, os.path.join("..", "x")) with pytest.raises(KeyValidationError): - path_of(suffixed, "../x") + path_of(leaf, os.path.join(root, "..", "x")) with pytest.raises(TypeError): path_of({}, "a") assert os.listdir(root) == [] # nothing created + + +def test_path_of_relative_root_is_absolute(tmp_path, monkeypatch): + import os + + from dol import Files, path_of + + monkeypatch.chdir(tmp_path) + got = path_of(Files("sub"), "a") + assert os.path.isabs(got) and got == os.path.abspath(os.path.join("sub", "a")) + + +def test_path_of_allow_escape(tmp_path): + import os + + from dol import Files, path_of + + root = tmp_path / "r" + root.mkdir() + s = Files(str(root), allow_escape=True) + assert path_of(s, os.path.join("..", "x")) == str(tmp_path / "x")