Add file loader and tests for compute_tab_table and load_cwd_guake_yml

Pipfile

@@ -34,7 +34,6 @@ pew = "*"
 black = "==21.8b0"
 flakehell = "*"
 toml = "*"
-dataclasses = {markers="python_version < '3.7'","version >" = "0.1.3"}
 
 [packages]
 pbr = "*"

guake/guake_app.py

@@ -19,7 +19,6 @@
 Boston, MA 02110-1301 USA
 """
 import json
-import yaml
 import logging
 import os
 import shutil
@@ -70,6 +69,7 @@
 from guake.theme import patch_gtk_theme
 from guake.theme import select_gtk_theme
 from guake.utils import BackgroundImageManager
+from guake.utils import FileManager
 from guake.utils import FullscreenManager
 from guake.utils import HidePrevention
 from guake.utils import RectCalculator
@@ -191,9 +191,8 @@ def load_schema():
         # FullscreenManager
         self.fullscreen_manager = FullscreenManager(self.settings, self.window, self)
 
-        # Hold a copy of guake_yaml
-        self._guake_yml = {}
-        self._guake_yml_load_monotonic = {}
+        # Start the file manager (only used by guake.yml so far).
+        self.fm = FileManager()
 
         # Workspace tracking
         self.notebook_manager = NotebookManager(
@@ -258,7 +257,6 @@ def window_event(*args):
         Keybinder.init()
         self.hotkeys = Keybinder
         Keybindings(self)
-
         self.load_config()
 
         if self.settings.general.get_boolean("start-fullscreen"):
@@ -1111,20 +1109,13 @@ def load_cwd_guake_yaml(self, vte) -> dict:
             return {}
 
         cwd = Path(vte.get_current_directory())
-        guake_yml = cwd.joinpath(".guake.yml")
-        content = {}
+        filename = str(cwd.joinpath(".guake.yml"))
 
-        if self._guake_yml_load_monotonic.get(guake_yml, 0.0) + 1.0 > pytime.monotonic():
-            content = self._guake_yml.get(guake_yml, {})
-        else:
-            try:
-                if guake_yml.is_file():
-                    with guake_yml.open(encoding="utf-8") as fd:
-                        content = yaml.safe_load(fd)
-                        self._guake_yml[guake_yml] = content
-                        self._guake_yml_load_monotonic[guake_yml] = pytime.monotonic()
-            except PermissionError:
-                log.debug("PermissionError on accessing .guake.yml")
+        try:
+            content = self.fm.read_yaml(filename)
+        except Exception:
+            log.debug("Unexpected error reading %s.", filename, exc_info=True)
+            content = {}
 
         if not isinstance(content, dict):
             content = {}

guake/keybindings.py

@@ -17,9 +17,10 @@
 Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
 Boston, MA 02110-1301 USA
 """
-from collections import defaultdict
 import logging
 
+from collections import defaultdict
+
 import gi
 
 gi.require_version("Gtk", "3.0")

guake/tests/__init__.py

@@ -1,4 +1,5 @@
 import builtins
+
 from locale import gettext
 
 builtins.__dict__["_"] = gettext

guake/tests/test_guake.py

@@ -189,3 +189,55 @@ def test_guake_hide_tab_bar_if_one_tab(mocker, g, fs):
     g.settings.general.set_boolean("hide-tabs-if-one-tab", True)
     assert g.get_notebook().get_n_pages() == 1
     assert g.get_notebook().get_property("show-tabs") is False
+
+
+def test_load_cwd_guake_yml_not_found_error(g):
+    vte = g.get_notebook().get_current_terminal()
+    assert g.fm.read_yaml("/foo/.guake.yml") is None
+    assert g.load_cwd_guake_yaml(vte) == {}
+
+
+def test_load_cwd_guake_yml_encoding_error(g, mocker, fs):
+    vte = g.get_notebook().get_current_terminal()
+    mocker.patch.object(vte, "get_current_directory", return_value="/foo/")
+    fs.create_file("/foo/.guake.yml", contents=b"\xfe\xf0[\xb1\x0b\xc1\x18\xda")
+    assert g.fm.read_yaml("/foo/.guake.yml") is None
+    assert g.load_cwd_guake_yaml(vte) == {}
+
+
+def test_load_cwd_guake_yml_format_error(g, mocker, fs):
+    vte = g.get_notebook().get_current_terminal()
+    mocker.patch.object(vte, "get_current_directory", return_value="/foo/")
+    fs.create_file("/foo/.guake.yml", contents=b"[[as]")
+    assert g.fm.read_yaml("/foo/.guake.yml") is None
+    assert g.load_cwd_guake_yaml(vte) == {}
+
+
+def test_load_cwd_guake_yml(mocker, g, fs):
+    vte = g.get_notebook().get_current_terminal()
+    mocker.patch.object(vte, "get_current_directory", return_value="/foo/")
+
+    f = fs.create_file("/foo/.guake.yml", contents="title: bar")
+    assert g.load_cwd_guake_yaml(vte) == {"title": "bar"}
+
+    # Cache in action.
+    f.set_contents("title: foo")
+    assert g.load_cwd_guake_yaml(vte) == {"title": "bar"}
+    g.fm.clear()
+    assert g.load_cwd_guake_yaml(vte) == {"title": "foo"}
+
+
+def test_guake_compute_tab_title(mocker, g, fs):
+    vte = g.get_notebook().get_current_terminal()
+    mocker.patch.object(vte, "get_current_directory", return_value="/foo/")
+
+    # Original title.
+    assert g.compute_tab_title(vte) == "Terminal"
+
+    # Change title.
+    fs.create_file("/foo/.guake.yml", contents="title: bar")
+    assert g.compute_tab_title(vte) == "bar"
+
+    # Avoid loading the guake.yml
+    mocker.patch.object(g.settings.general, "get_boolean", return_value=False)
+    assert g.compute_tab_title(vte) == "Terminal"

guake/tests/test_utils.py

@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+# pylint: disable=redefined-outer-name
+
+from guake.utils import FileManager
+
+
+def test_file_manager(fs):
+    fs.create_file("/foo/bar", contents="test")
+    fm = FileManager()
+    assert fm.read("/foo/bar") == "test"
+
+
+def test_file_manager_hit(fs):
+    f = fs.create_file("/foo/bar", contents="test")
+
+    fm = FileManager(delta=9999)
+    assert fm.read("/foo/bar") == "test"
+    f.set_contents("changed")
+    assert fm.read("/foo/bar") == "test"
+
+
+def test_file_manager_miss(fs):
+    f = fs.create_file("/foo/bar", contents="test")
+
+    fm = FileManager(delta=0.0)
+    assert fm.read("/foo/bar") == "test"
+    f.set_contents("changed")
+    assert fm.read("/foo/bar") == "changed"
+
+
+def test_file_manager_clear(fs):
+    f = fs.create_file("/foo/bar", contents="test")
+
+    fm = FileManager(delta=9999)
+    assert fm.read("/foo/bar") == "test"
+    f.set_contents("changed")
+    assert fm.read("/foo/bar") == "test"
+    fm.clear()
+    assert fm.read("/foo/bar") == "changed"

guake/utils.py

@@ -21,9 +21,10 @@
 """
 import enum
 import logging
+import os
 import subprocess
 import time
-import os
+import yaml
 
 import cairo
 
@@ -109,6 +110,46 @@ def restore_preferences(filename):
         p.communicate(input=prefs)
 
 
+class FileManager:
+    def __init__(self, delta=1.0):
+        self._cache = {}
+        self._delta = max(0.0, delta)
+
+    def clear(self):
+        self._cache.clear()
+
+    def read_yaml(self, filename: str):
+
+        content = None
+
+        try:
+            content = self.read(filename)
+        except PermissionError:
+            log.debug("PermissionError while reading %s.", filename)
+        except FileNotFoundError:
+            log.debug("File %s does not exists.", filename)
+        except UnicodeDecodeError:
+            log.debug("Encoding error %s (we assume is utf-8).", filename)
+
+        if content is not None:
+            try:
+                content = yaml.safe_load(content)
+            except yaml.YAMLError:
+                log.debug("YAMLError reading %s.", filename)
+                content = None
+        return content
+
+    def read(self, filename: str) -> str:
+        # Return the content of a file from the fs or from cache.
+        if (
+            filename not in self._cache
+            or self._cache[filename]["time"] + self._delta < time.monotonic()
+        ):
+            with open(filename, mode="r", encoding="utf-8") as fd:
+                self._cache[filename] = {"time": time.monotonic(), "content": fd.read()}
+        return self._cache[filename]["content"]
+
+
 class TabNameUtils:
     @classmethod
     def shorten(cls, text, settings):

requirements-dev.txt

@@ -15,9 +15,8 @@ astroid
 autopep8
 black==21.8b0
 colorlog
-dataclasses
 fiximports>=0.1.18
-flake8
+flake8>=3.8.0,<4.0.0
 flakehell
 mock>=2.0.0
 pathlib2