Merge pull request #43 from Deric-W/Issue#42

fix Issue #42

Author: Deric-W

pyhp/backends/caches/timestamped/files.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 import os
+import sys
 import base64
 import pickle
 from typing import TypeVar, Any, Mapping, Iterator
@@ -33,12 +34,16 @@
 
 
 __all__ = (
+    "FILE_EXTENSION",
     "FileCacheSource",
     "FileCache"
 )
 
 S = TypeVar("S", bound=TimestampedCodeSource)
 
+# use cache tag in extension to prevent errors with different python versions
+FILE_EXTENSION = f".{sys.implementation.cache_tag}.pickle"
+
 
 class FileCacheSource(CacheSource[S]):
     """
@@ -139,9 +144,10 @@ def __eq__(self, other: object) -> bool:
 
     def __getitem__(self, name: str) -> FileCacheSource[S]:
         """get FileCacheSource with path as returned by .path()"""
+        path = self.path(name)  # dont leak sources if self.path fails
         return FileCacheSource(
             self.source_container[name],
-            self.path(name),
+            path,
             self.ttl
         )
 
@@ -194,20 +200,20 @@ def clear(self) -> None:
                 pass
 
     def path(self, name: str) -> str:
-        """return directory_name/<base32 encoded name>.pickle"""
+        """return directory_name/<base32 encoded name>FILE_EXTENSION"""
         return os.path.join(
             self.directory_name,
-            base64.b32encode(name.encode("utf8")).decode("utf8") + ".pickle"
+            base64.b32encode(name.encode("utf8")).decode("utf8") + FILE_EXTENSION
         )   # use base32 because of case-insensitive file systems and forbidden characters
 
     def reconstruct_name(self, path: str) -> str:
         """reconstruct the name from a file cache path"""
-        name, _, _ = os.path.basename(path).rpartition(".")
+        name, _, _ = os.path.basename(path).partition(".")
         return base64.b32decode(name.encode("utf8"), casefold=True).decode("utf8")
 
     def paths(self) -> Iterator[str]:
         """return a iterator yielding all paths currently in use (including outdated ones)"""
         with os.scandir(self.directory_name) as directory:
             for entry in directory:
-                if entry.name.endswith(".pickle") and entry.is_file():
+                if entry.name.endswith(FILE_EXTENSION) and entry.is_file():
                     yield entry.path

tests/backends/caches/timestamped/test_files.py

@@ -14,7 +14,8 @@
 from pyhp.backends.caches import NotCachedException
 from pyhp.backends.caches.timestamped.files import (
     FileCacheSource,
-    FileCache
+    FileCache,
+    FILE_EXTENSION
 )
 from pyhp.compiler.parsers import RegexParser
 from pyhp.compiler.generic import GenericCodeBuilder
@@ -250,3 +251,48 @@ def test_path(self) -> None:
             self.assertNotEqual(cache.path("test"), cache.path("test2"))
             self.assertNotIn("|", cache.path("|"))
             self.assertEqual(cache.reconstruct_name(cache.path("test2")), "test2")
+
+    def test_paths(self) -> None:
+        """test FileCache.paths"""
+        with tempfile.TemporaryDirectory() as directory, \
+                FileCache(Directory("tests/embedding", compiler), directory) as cache:
+            tmp_path = os.path.join(directory, "test.txt")
+            dir_path = os.path.join(directory, "test" + FILE_EXTENSION)
+            with cache["syntax.pyhp"] as source:
+                source.fetch()
+            with cache["shebang.pyhp"] as source:
+                source.fetch()
+            try:
+                open(tmp_path, "xb").close()
+                try:
+                    os.mkdir(dir_path)
+                    try:
+                        paths = list(cache.paths())
+                        self.assertEqual(len(paths), 2)
+                        self.assertEqual(   # order not important, use sets
+                            set(paths),
+                            {
+                                cache.path("syntax.pyhp"),
+                                cache.path("shebang.pyhp")
+                            }
+                        )
+                    finally:
+                        os.rmdir(dir_path)
+                finally:
+                    os.unlink(tmp_path)
+            finally:
+                cache.clear()
+
+    def test_reconstruct_name(self) -> None:
+        """test FileCache.reconstruct_name"""
+        names = [
+            "test",
+            "test/42/9",
+            "äöü|<",
+            "\n\n\n"
+        ]
+        with FileCache(Directory("tests/embedding", compiler), "tmp") as cache:
+            for name in names:
+                path = cache.path(name)
+                self.assertEqual(cache.reconstruct_name(path.lower()), name)
+                self.assertEqual(cache.reconstruct_name(path.upper()), name)