diff --git a/connpy/_version.py b/connpy/_version.py index bb17a09..290678f 100644 --- a/connpy/_version.py +++ b/connpy/_version.py @@ -1,2 +1 @@ -__version__ = "5.0b4" - +__version__ = "5.0b5" diff --git a/connpy/configfile.py b/connpy/configfile.py index 44b8549..5a6b292 100755 --- a/connpy/configfile.py +++ b/connpy/configfile.py @@ -48,7 +48,7 @@ class configfile: ### Optional Parameters: - conf (str): Path/file to config file. If left empty default - path is ~/.config/conn/config.json + path is ~/.config/conn/config.yaml - key (str): Path/file to RSA key file. If left empty default path is ~/.config/conn/.osk @@ -87,13 +87,29 @@ class configfile: try: with open(legacy_file, 'r') as f: old_data = json.load(f) - with open(self.file, 'w') as f: - yaml.dump(old_data, f, default_flow_style=False, sort_keys=False) - with open(self.cachefile, 'w') as f: - json.dump(old_data, f) - shutil.move(legacy_file, legacy_file + ".backup") - printer.success(f"Migrated legacy config ({len(old_data.get('connections',{}))} folders/nodes) into YAML and Cache successfully!") + if not self._validate_config(old_data): + printer.warning(f"Legacy config {legacy_file} has invalid structure, skipping migration.") + else: + with open(self.file, 'w') as f: + yaml.dump(old_data, f, default_flow_style=False, sort_keys=False) + # Verify the written YAML can be read back correctly + with open(self.file, 'r') as f: + verify = yaml.safe_load(f) + if not self._validate_config(verify): + os.remove(self.file) + printer.warning("YAML verification failed after migration, keeping legacy config.") + else: + with open(self.cachefile, 'w') as f: + json.dump(old_data, f) + shutil.move(legacy_file, legacy_file + ".backup") + printer.success(f"Migrated legacy config ({len(old_data.get('connections',{}))} folders/nodes) into YAML and Cache successfully!") except Exception as e: + # Clean up partial YAML if it was created + if os.path.exists(self.file): + try: + os.remove(self.file) + except OSError: + pass printer.warning(f"Failed to migrate legacy config: {e}") else: self.file = conf @@ -122,6 +138,13 @@ class configfile: self._generate_nodes_cache() + def _validate_config(self, data): + """Verify config data has the required structure.""" + if not isinstance(data, dict): + return False + required = {"config", "connections", "profiles"} + return required.issubset(data.keys()) + def _loadconfig(self, conf): #Loads config file using dual cache cache_exists = os.path.exists(self.cachefile) @@ -131,6 +154,20 @@ class configfile: if not cache_exists or yaml_time > cache_time: with open(conf, 'r') as f: data = yaml.safe_load(f) + if not self._validate_config(data): + # YAML is broken, try to recover from cache + if cache_exists: + printer.warning("Config file appears corrupt, recovering from cache...") + with open(self.cachefile, 'r') as f: + data = json.load(f) + if self._validate_config(data): + # Re-write the YAML from good cache + with open(conf, 'w') as f: + yaml.dump(data, f, default_flow_style=False, sort_keys=False) + return data + # Both broken or no cache - create fresh + printer.error("Config file is corrupt and no valid cache exists. Creating default config.") + return self._createconfig(conf) try: with open(self.cachefile, 'w') as f: json.dump(data, f) @@ -139,39 +176,55 @@ class configfile: return data else: with open(self.cachefile, 'r') as f: - return json.load(f) + data = json.load(f) + if not self._validate_config(data): + # Cache broken, try yaml + with open(conf, 'r') as f: + data = yaml.safe_load(f) + if self._validate_config(data): + return data + # Both broken + printer.error("Both config and cache are corrupt. Creating default config.") + return self._createconfig(conf) + return data def _createconfig(self, conf): - #Create config file + #Create config file (always writes defaults, safe for recovery) defaultconfig = {'config': {'case': False, 'idletime': 30, 'fzf': False}, 'connections': {}, 'profiles': { "default": { "host":"", "protocol":"ssh", "port":"", "user":"", "password":"", "options":"", "logs":"", "tags": "", "jumphost":""}}} - if not os.path.exists(conf): - with open(conf, "w") as f: - yaml.dump(defaultconfig, f, default_flow_style=False, sort_keys=False) - os.chmod(conf, 0o600) - try: - with open(self.cachefile, 'w') as f: - json.dump(defaultconfig, f) - except Exception: - pass - with open(conf, 'r') as f: - jsondata = yaml.safe_load(f) - return jsondata + with open(conf, "w") as f: + yaml.dump(defaultconfig, f, default_flow_style=False, sort_keys=False) + os.chmod(conf, 0o600) + try: + with open(self.cachefile, 'w') as f: + json.dump(defaultconfig, f) + except Exception: + pass + return defaultconfig @MethodHook def _saveconfig(self, conf): - #Save config file + #Save config file atomically to prevent corruption newconfig = {"config":{}, "connections": {}, "profiles": {}} newconfig["config"] = self.config newconfig["connections"] = self.connections newconfig["profiles"] = self.profiles + tmpfile = conf + '.tmp' try: - with open(conf, "w") as f: + with open(tmpfile, "w") as f: yaml.dump(newconfig, f, default_flow_style=False, sort_keys=False) + # Atomic replace: only overwrite original if write succeeded + shutil.move(tmpfile, conf) with open(self.cachefile, "w") as f: json.dump(newconfig, f) self._generate_nodes_cache() except (IOError, OSError) as e: printer.error(f"Failed to save config: {e}") + # Clean up temp file if it exists + if os.path.exists(tmpfile): + try: + os.remove(tmpfile) + except OSError: + pass return 1 return 0 diff --git a/connpy/tests/conftest.py b/connpy/tests/conftest.py index 16dbaab..297b1b3 100644 --- a/connpy/tests/conftest.py +++ b/connpy/tests/conftest.py @@ -5,6 +5,7 @@ No test touches ~/.config/conn/ """ import pytest import json +import yaml import os from unittest.mock import patch, MagicMock from Crypto.PublicKey import RSA @@ -72,9 +73,9 @@ def tmp_config_dir(tmp_path): plugins_dir = config_dir / "plugins" plugins_dir.mkdir() - # Write config.json - config_file = config_dir / "config.json" - config_file.write_text(json.dumps(DEFAULT_CONFIG, indent=4)) + # Write config.yaml + config_file = config_dir / "config.yaml" + config_file.write_text(yaml.dump(DEFAULT_CONFIG, default_flow_style=False, sort_keys=False)) os.chmod(str(config_file), 0o600) # Write .folder (points to itself) @@ -94,7 +95,7 @@ def tmp_config_dir(tmp_path): def config(tmp_config_dir): """Create a configfile instance pointing to tmp directory.""" from connpy.configfile import configfile - conf_path = str(tmp_config_dir / "config.json") + conf_path = str(tmp_config_dir / "config.yaml") key_path = str(tmp_config_dir / ".osk") return configfile(conf=conf_path, key=key_path) @@ -102,13 +103,13 @@ def config(tmp_config_dir): @pytest.fixture def populated_config(tmp_config_dir): """Create a configfile with sample nodes/profiles pre-loaded.""" - config_file = tmp_config_dir / "config.json" + config_file = tmp_config_dir / "config.yaml" data = { "config": {"case": False, "idletime": 30, "fzf": False}, "connections": SAMPLE_CONNECTIONS, "profiles": SAMPLE_PROFILES } - config_file.write_text(json.dumps(data, indent=4)) + config_file.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False)) from connpy.configfile import configfile return configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk")) @@ -173,7 +174,7 @@ def mock_litellm(): @pytest.fixture def ai_config(tmp_config_dir): """Create a configfile with AI keys configured for AI tests.""" - config_file = tmp_config_dir / "config.json" + config_file = tmp_config_dir / "config.yaml" data = { "config": { "case": False, "idletime": 30, "fzf": False, @@ -187,6 +188,6 @@ def ai_config(tmp_config_dir): "connections": SAMPLE_CONNECTIONS, "profiles": SAMPLE_PROFILES } - config_file.write_text(json.dumps(data, indent=4)) + config_file.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False)) from connpy.configfile import configfile return configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk")) diff --git a/connpy/tests/test_configfile.py b/connpy/tests/test_configfile.py index 39c508b..1fde087 100644 --- a/connpy/tests/test_configfile.py +++ b/connpy/tests/test_configfile.py @@ -375,3 +375,210 @@ class TestGetAll: from connpy.configfile import configfile reloaded = configfile(conf=config.file, key=config.key) assert "test_node" in reloaded.connections + + +class TestValidateConfig: + def test_valid_config(self, config): + data = {"config": {}, "connections": {}, "profiles": {}} + assert config._validate_config(data) == True + + def test_none_data(self, config): + assert config._validate_config(None) == False + + def test_string_data(self, config): + assert config._validate_config("not a dict") == False + + def test_missing_key(self, config): + assert config._validate_config({"config": {}, "connections": {}}) == False + + def test_empty_dict(self, config): + assert config._validate_config({}) == False + + +class TestCorruptionRecovery: + def test_corrupt_yaml_recovers_from_cache(self, tmp_config_dir): + """If YAML is corrupt but cache is valid, recovers from cache.""" + config_file = tmp_config_dir / "config.yaml" + key_file = tmp_config_dir / ".osk" + + # Write valid config with router1 + valid_data = { + "config": {"case": False, "idletime": 30, "fzf": False}, + "connections": {"router1": {"host": "10.0.0.1", "type": "connection", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}, + "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}} + } + config_file.write_text(yaml.dump(valid_data, default_flow_style=False, sort_keys=False)) + + from connpy.configfile import configfile + conf = configfile(conf=str(config_file), key=str(key_file)) + # Save to populate cache at the real self.cachefile path + conf._saveconfig(conf.file) + cachefile_path = conf.cachefile + assert os.path.exists(cachefile_path) + + # Now corrupt the YAML + config_file.write_text("") + import time; time.sleep(0.05) # Ensure YAML is newer than cache + + # Reload - should recover from cache + conf2 = configfile(conf=str(config_file), key=str(key_file)) + assert "router1" in conf2.connections + assert conf2.connections["router1"]["host"] == "10.0.0.1" + + def test_corrupt_cache_uses_yaml(self, tmp_config_dir): + """If cache is corrupt but YAML is valid, uses YAML.""" + config_file = tmp_config_dir / "config.yaml" + key_file = tmp_config_dir / ".osk" + + valid_data = { + "config": {"case": False, "idletime": 30, "fzf": False}, + "connections": {}, + "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}} + } + config_file.write_text(yaml.dump(valid_data, default_flow_style=False, sort_keys=False)) + + from connpy.configfile import configfile + conf = configfile(conf=str(config_file), key=str(key_file)) + cachefile_path = conf.cachefile + + # Now corrupt the cache (valid JSON but invalid config structure) + from pathlib import Path + Path(cachefile_path).write_text(json.dumps({"garbage": True})) + # Make cache newer than YAML to force cache path + import time; time.sleep(0.05) + os.utime(cachefile_path, None) + + conf2 = configfile(conf=str(config_file), key=str(key_file)) + assert conf2.config["case"] == False + assert "default" in conf2.profiles + + def test_both_corrupt_creates_default(self, tmp_config_dir): + """If both YAML and cache are corrupt, creates fresh config.""" + config_file = tmp_config_dir / "config.yaml" + key_file = tmp_config_dir / ".osk" + + from connpy.configfile import configfile + conf = configfile(conf=str(config_file), key=str(key_file)) + cachefile_path = conf.cachefile + + # Corrupt YAML + config_file.write_text("") + # Corrupt cache + from pathlib import Path + Path(cachefile_path).write_text(json.dumps({"garbage": True})) + import time; time.sleep(0.05) + os.utime(str(config_file), None) + + conf2 = configfile(conf=str(config_file), key=str(key_file)) + + # Should get defaults, not crash + assert conf2.config is not None + assert "default" in conf2.profiles + assert isinstance(conf2.connections, dict) + + +class TestAtomicSave: + def test_save_creates_no_leftover_tmp(self, config): + """After successful save, no .tmp file remains.""" + config._connections_add( + id="test123", host="1.2.3.4", protocol="ssh", + port="", user="", password="", options="", + logs="", tags="", jumphost="" + ) + result = config._saveconfig(config.file) + assert result == 0 + assert not os.path.exists(config.file + '.tmp') + + def test_save_preserves_original_on_error(self, config): + """If save fails, original config file is not corrupted.""" + import unittest.mock as mock + + config._connections_add( + id="original_node", host="10.0.0.1", protocol="ssh", + port="", user="", password="", options="", + logs="", tags="", jumphost="" + ) + config._saveconfig(config.file) + + # Now add another node and make yaml.dump fail + config._connections_add( + id="new_node", host="10.0.0.2", protocol="ssh", + port="", user="", password="", options="", + logs="", tags="", jumphost="" + ) + + with mock.patch('connpy.configfile.yaml.dump', side_effect=IOError("disk full")): + result = config._saveconfig(config.file) + assert result == 1 + + # Original file should still be valid with original_node + from connpy.configfile import configfile + reloaded = configfile(conf=config.file, key=config.key) + assert "original_node" in reloaded.connections + + +class TestMigrationSafety: + def test_migration_validates_legacy_data(self, tmp_path): + """Migration skips invalid legacy JSON files.""" + from unittest.mock import patch + config_dir = tmp_path / ".config" / "conn" + config_dir.mkdir(parents=True) + (config_dir / "plugins").mkdir() + + # Write .folder + (config_dir / ".folder").write_text(str(config_dir)) + + # Generate RSA key + from Crypto.PublicKey import RSA + key = RSA.generate(2048) + key_file = config_dir / ".osk" + key_file.write_bytes(key.export_key("PEM")) + os.chmod(str(key_file), 0o600) + + # Write invalid JSON config (missing required keys) + legacy_file = config_dir / "config.json" + legacy_file.write_text(json.dumps({"garbage": True})) + + with patch("os.path.expanduser", return_value=str(tmp_path)): + from connpy.configfile import configfile + conf = configfile(key=str(key_file)) + + # Legacy file should NOT have been moved to .backup + assert legacy_file.exists() + assert not (config_dir / "config.json.backup").exists() + + def test_migration_verifies_written_yaml(self, tmp_path): + """Migration succeeds when legacy JSON is valid.""" + from unittest.mock import patch + config_dir = tmp_path / ".config" / "conn" + config_dir.mkdir(parents=True) + (config_dir / "plugins").mkdir() + + # Write .folder + (config_dir / ".folder").write_text(str(config_dir)) + + # Generate RSA key + from Crypto.PublicKey import RSA + key = RSA.generate(2048) + key_file = config_dir / ".osk" + key_file.write_bytes(key.export_key("PEM")) + os.chmod(str(key_file), 0o600) + + valid_data = { + "config": {"case": False, "idletime": 30, "fzf": False}, + "connections": {"r1": {"host": "1.2.3.4", "type": "connection", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}, + "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}} + } + legacy_file = config_dir / "config.json" + legacy_file.write_text(json.dumps(valid_data)) + + with patch("os.path.expanduser", return_value=str(tmp_path)): + from connpy.configfile import configfile + conf = configfile(key=str(key_file)) + + # Migration should have succeeded: YAML exists, JSON backed up + yaml_file = config_dir / "config.yaml" + assert yaml_file.exists() + assert (config_dir / "config.json.backup").exists() + assert not legacy_file.exists() + assert "r1" in conf.connections diff --git a/docs/connpy/index.html b/docs/connpy/index.html index 166e0ba..fced62f 100644 --- a/docs/connpy/index.html +++ b/docs/connpy/index.html @@ -2240,7 +2240,7 @@ class configfile: ### Optional Parameters: - conf (str): Path/file to config file. If left empty default - path is ~/.config/conn/config.json + path is ~/.config/conn/config.yaml - key (str): Path/file to RSA key file. If left empty default path is ~/.config/conn/.osk @@ -2279,13 +2279,29 @@ class configfile: try: with open(legacy_file, 'r') as f: old_data = json.load(f) - with open(self.file, 'w') as f: - yaml.dump(old_data, f, default_flow_style=False, sort_keys=False) - with open(self.cachefile, 'w') as f: - json.dump(old_data, f) - shutil.move(legacy_file, legacy_file + ".backup") - printer.success(f"Migrated legacy config ({len(old_data.get('connections',{}))} folders/nodes) into YAML and Cache successfully!") + if not self._validate_config(old_data): + printer.warning(f"Legacy config {legacy_file} has invalid structure, skipping migration.") + else: + with open(self.file, 'w') as f: + yaml.dump(old_data, f, default_flow_style=False, sort_keys=False) + # Verify the written YAML can be read back correctly + with open(self.file, 'r') as f: + verify = yaml.safe_load(f) + if not self._validate_config(verify): + os.remove(self.file) + printer.warning("YAML verification failed after migration, keeping legacy config.") + else: + with open(self.cachefile, 'w') as f: + json.dump(old_data, f) + shutil.move(legacy_file, legacy_file + ".backup") + printer.success(f"Migrated legacy config ({len(old_data.get('connections',{}))} folders/nodes) into YAML and Cache successfully!") except Exception as e: + # Clean up partial YAML if it was created + if os.path.exists(self.file): + try: + os.remove(self.file) + except OSError: + pass printer.warning(f"Failed to migrate legacy config: {e}") else: self.file = conf @@ -2314,6 +2330,13 @@ class configfile: self._generate_nodes_cache() + def _validate_config(self, data): + """Verify config data has the required structure.""" + if not isinstance(data, dict): + return False + required = {"config", "connections", "profiles"} + return required.issubset(data.keys()) + def _loadconfig(self, conf): #Loads config file using dual cache cache_exists = os.path.exists(self.cachefile) @@ -2323,6 +2346,20 @@ class configfile: if not cache_exists or yaml_time > cache_time: with open(conf, 'r') as f: data = yaml.safe_load(f) + if not self._validate_config(data): + # YAML is broken, try to recover from cache + if cache_exists: + printer.warning("Config file appears corrupt, recovering from cache...") + with open(self.cachefile, 'r') as f: + data = json.load(f) + if self._validate_config(data): + # Re-write the YAML from good cache + with open(conf, 'w') as f: + yaml.dump(data, f, default_flow_style=False, sort_keys=False) + return data + # Both broken or no cache - create fresh + printer.error("Config file is corrupt and no valid cache exists. Creating default config.") + return self._createconfig(conf) try: with open(self.cachefile, 'w') as f: json.dump(data, f) @@ -2331,39 +2368,55 @@ class configfile: return data else: with open(self.cachefile, 'r') as f: - return json.load(f) + data = json.load(f) + if not self._validate_config(data): + # Cache broken, try yaml + with open(conf, 'r') as f: + data = yaml.safe_load(f) + if self._validate_config(data): + return data + # Both broken + printer.error("Both config and cache are corrupt. Creating default config.") + return self._createconfig(conf) + return data def _createconfig(self, conf): - #Create config file + #Create config file (always writes defaults, safe for recovery) defaultconfig = {'config': {'case': False, 'idletime': 30, 'fzf': False}, 'connections': {}, 'profiles': { "default": { "host":"", "protocol":"ssh", "port":"", "user":"", "password":"", "options":"", "logs":"", "tags": "", "jumphost":""}}} - if not os.path.exists(conf): - with open(conf, "w") as f: - yaml.dump(defaultconfig, f, default_flow_style=False, sort_keys=False) - os.chmod(conf, 0o600) - try: - with open(self.cachefile, 'w') as f: - json.dump(defaultconfig, f) - except Exception: - pass - with open(conf, 'r') as f: - jsondata = yaml.safe_load(f) - return jsondata + with open(conf, "w") as f: + yaml.dump(defaultconfig, f, default_flow_style=False, sort_keys=False) + os.chmod(conf, 0o600) + try: + with open(self.cachefile, 'w') as f: + json.dump(defaultconfig, f) + except Exception: + pass + return defaultconfig @MethodHook def _saveconfig(self, conf): - #Save config file + #Save config file atomically to prevent corruption newconfig = {"config":{}, "connections": {}, "profiles": {}} newconfig["config"] = self.config newconfig["connections"] = self.connections newconfig["profiles"] = self.profiles + tmpfile = conf + '.tmp' try: - with open(conf, "w") as f: + with open(tmpfile, "w") as f: yaml.dump(newconfig, f, default_flow_style=False, sort_keys=False) + # Atomic replace: only overwrite original if write succeeded + shutil.move(tmpfile, conf) with open(self.cachefile, "w") as f: json.dump(newconfig, f) self._generate_nodes_cache() except (IOError, OSError) as e: printer.error(f"Failed to save config: {e}") + # Clean up temp file if it exists + if os.path.exists(tmpfile): + try: + os.remove(tmpfile) + except OSError: + pass return 1 return 0 @@ -2738,7 +2791,7 @@ class configfile:

Optional Parameters:

- conf (str): Path/file to config file. If left empty default
-              path is ~/.config/conn/config.json
+              path is ~/.config/conn/config.yaml
 
 - key  (str): Path/file to RSA key file. If left empty default
               path is ~/.config/conn/.osk
diff --git a/docs/connpy/tests/conftest.html b/docs/connpy/tests/conftest.html
index f9181aa..560bf76 100644
--- a/docs/connpy/tests/conftest.html
+++ b/docs/connpy/tests/conftest.html
@@ -58,7 +58,7 @@ No test touches ~/.config/conn/

 
@pytest.fixture
 def ai_config(tmp_config_dir):
     """Create a configfile with AI keys configured for AI tests."""
-    config_file = tmp_config_dir / "config.json"
+    config_file = tmp_config_dir / "config.yaml"
     data = {
         "config": {
             "case": False, "idletime": 30, "fzf": False,
@@ -72,7 +72,7 @@ def ai_config(tmp_config_dir):
         "connections": SAMPLE_CONNECTIONS,
         "profiles": SAMPLE_PROFILES
     }
-    config_file.write_text(json.dumps(data, indent=4))
+    config_file.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False))
     from connpy.configfile import configfile
     return configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))

 
@@ -90,7 +90,7 @@ def ai_config(tmp_config_dir):
 def config(tmp_config_dir):
     """Create a configfile instance pointing to tmp directory."""
     from connpy.configfile import configfile
-    conf_path = str(tmp_config_dir / "config.json")
+    conf_path = str(tmp_config_dir / "config.yaml")
     key_path = str(tmp_config_dir / ".osk")
     return configfile(conf=conf_path, key=key_path)
@@ -182,13 +182,13 @@ def mock_pexpect(): 
@pytest.fixture
 def populated_config(tmp_config_dir):
     """Create a configfile with sample nodes/profiles pre-loaded."""
-    config_file = tmp_config_dir / "config.json"
+    config_file = tmp_config_dir / "config.yaml"
     data = {
         "config": {"case": False, "idletime": 30, "fzf": False},
         "connections": SAMPLE_CONNECTIONS,
         "profiles": SAMPLE_PROFILES
     }
-    config_file.write_text(json.dumps(data, indent=4))
+    config_file.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False))
     from connpy.configfile import configfile
     return configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))
@@ -210,9 +210,9 @@ def tmp_config_dir(tmp_path): plugins_dir = config_dir / "plugins" plugins_dir.mkdir() - # Write config.json - config_file = config_dir / "config.json" - config_file.write_text(json.dumps(DEFAULT_CONFIG, indent=4)) + # Write config.yaml + config_file = config_dir / "config.yaml" + config_file.write_text(yaml.dump(DEFAULT_CONFIG, default_flow_style=False, sort_keys=False)) os.chmod(str(config_file), 0o600) # Write .folder (points to itself) diff --git a/docs/connpy/tests/test_configfile.html b/docs/connpy/tests/test_configfile.html index eea4bc0..7134fb4 100644 --- a/docs/connpy/tests/test_configfile.html +++ b/docs/connpy/tests/test_configfile.html @@ -47,6 +47,116 @@ el.replaceWith(d);

Classes

+
+class TestAtomicSave +
+
+
+ +Expand source code + +
class TestAtomicSave:
+    def test_save_creates_no_leftover_tmp(self, config):
+        """After successful save, no .tmp file remains."""
+        config._connections_add(
+            id="test123", host="1.2.3.4", protocol="ssh",
+            port="", user="", password="", options="",
+            logs="", tags="", jumphost=""
+        )
+        result = config._saveconfig(config.file)
+        assert result == 0
+        assert not os.path.exists(config.file + '.tmp')
+
+    def test_save_preserves_original_on_error(self, config):
+        """If save fails, original config file is not corrupted."""
+        import unittest.mock as mock
+
+        config._connections_add(
+            id="original_node", host="10.0.0.1", protocol="ssh",
+            port="", user="", password="", options="",
+            logs="", tags="", jumphost=""
+        )
+        config._saveconfig(config.file)
+
+        # Now add another node and make yaml.dump fail
+        config._connections_add(
+            id="new_node", host="10.0.0.2", protocol="ssh",
+            port="", user="", password="", options="",
+            logs="", tags="", jumphost=""
+        )
+
+        with mock.patch('connpy.configfile.yaml.dump', side_effect=IOError("disk full")):
+            result = config._saveconfig(config.file)
+            assert result == 1
+
+        # Original file should still be valid with original_node
+        from connpy.configfile import configfile
+        reloaded = configfile(conf=config.file, key=config.key)
+        assert "original_node" in reloaded.connections
+
+
+

Methods

+
+
+def test_save_creates_no_leftover_tmp(self, config) +
+
+
+ +Expand source code + +
def test_save_creates_no_leftover_tmp(self, config):
+    """After successful save, no .tmp file remains."""
+    config._connections_add(
+        id="test123", host="1.2.3.4", protocol="ssh",
+        port="", user="", password="", options="",
+        logs="", tags="", jumphost=""
+    )
+    result = config._saveconfig(config.file)
+    assert result == 0
+    assert not os.path.exists(config.file + '.tmp')
+
+

After successful save, no .tmp file remains.

+
+
+def test_save_preserves_original_on_error(self, config) +
+
+
+ +Expand source code + +
def test_save_preserves_original_on_error(self, config):
+    """If save fails, original config file is not corrupted."""
+    import unittest.mock as mock
+
+    config._connections_add(
+        id="original_node", host="10.0.0.1", protocol="ssh",
+        port="", user="", password="", options="",
+        logs="", tags="", jumphost=""
+    )
+    config._saveconfig(config.file)
+
+    # Now add another node and make yaml.dump fail
+    config._connections_add(
+        id="new_node", host="10.0.0.2", protocol="ssh",
+        port="", user="", password="", options="",
+        logs="", tags="", jumphost=""
+    )
+
+    with mock.patch('connpy.configfile.yaml.dump', side_effect=IOError("disk full")):
+        result = config._saveconfig(config.file)
+        assert result == 1
+
+    # Original file should still be valid with original_node
+    from connpy.configfile import configfile
+    reloaded = configfile(conf=config.file, key=config.key)
+    assert "original_node" in reloaded.connections
+
+

If save fails, original config file is not corrupted.

+
+
+
class TestCRUDNodes
@@ -565,6 +675,210 @@ el.replaceWith(d);
+
+class TestCorruptionRecovery +
+
+
+ +Expand source code + +
class TestCorruptionRecovery:
+    def test_corrupt_yaml_recovers_from_cache(self, tmp_config_dir):
+        """If YAML is corrupt but cache is valid, recovers from cache."""
+        config_file = tmp_config_dir / "config.yaml"
+        key_file = tmp_config_dir / ".osk"
+
+        # Write valid config with router1
+        valid_data = {
+            "config": {"case": False, "idletime": 30, "fzf": False},
+            "connections": {"router1": {"host": "10.0.0.1", "type": "connection", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}},
+            "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}
+        }
+        config_file.write_text(yaml.dump(valid_data, default_flow_style=False, sort_keys=False))
+
+        from connpy.configfile import configfile
+        conf = configfile(conf=str(config_file), key=str(key_file))
+        # Save to populate cache at the real self.cachefile path
+        conf._saveconfig(conf.file)
+        cachefile_path = conf.cachefile
+        assert os.path.exists(cachefile_path)
+
+        # Now corrupt the YAML
+        config_file.write_text("")
+        import time; time.sleep(0.05)  # Ensure YAML is newer than cache
+
+        # Reload - should recover from cache
+        conf2 = configfile(conf=str(config_file), key=str(key_file))
+        assert "router1" in conf2.connections
+        assert conf2.connections["router1"]["host"] == "10.0.0.1"
+
+    def test_corrupt_cache_uses_yaml(self, tmp_config_dir):
+        """If cache is corrupt but YAML is valid, uses YAML."""
+        config_file = tmp_config_dir / "config.yaml"
+        key_file = tmp_config_dir / ".osk"
+
+        valid_data = {
+            "config": {"case": False, "idletime": 30, "fzf": False},
+            "connections": {},
+            "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}
+        }
+        config_file.write_text(yaml.dump(valid_data, default_flow_style=False, sort_keys=False))
+
+        from connpy.configfile import configfile
+        conf = configfile(conf=str(config_file), key=str(key_file))
+        cachefile_path = conf.cachefile
+
+        # Now corrupt the cache (valid JSON but invalid config structure)
+        from pathlib import Path
+        Path(cachefile_path).write_text(json.dumps({"garbage": True}))
+        # Make cache newer than YAML to force cache path
+        import time; time.sleep(0.05)
+        os.utime(cachefile_path, None)
+
+        conf2 = configfile(conf=str(config_file), key=str(key_file))
+        assert conf2.config["case"] == False
+        assert "default" in conf2.profiles
+
+    def test_both_corrupt_creates_default(self, tmp_config_dir):
+        """If both YAML and cache are corrupt, creates fresh config."""
+        config_file = tmp_config_dir / "config.yaml"
+        key_file = tmp_config_dir / ".osk"
+
+        from connpy.configfile import configfile
+        conf = configfile(conf=str(config_file), key=str(key_file))
+        cachefile_path = conf.cachefile
+
+        # Corrupt YAML
+        config_file.write_text("")
+        # Corrupt cache
+        from pathlib import Path
+        Path(cachefile_path).write_text(json.dumps({"garbage": True}))
+        import time; time.sleep(0.05)
+        os.utime(str(config_file), None)
+
+        conf2 = configfile(conf=str(config_file), key=str(key_file))
+
+        # Should get defaults, not crash
+        assert conf2.config is not None
+        assert "default" in conf2.profiles
+        assert isinstance(conf2.connections, dict)
+
+
+

Methods

+
+
+def test_both_corrupt_creates_default(self, tmp_config_dir) +
+
+
+ +Expand source code + +
def test_both_corrupt_creates_default(self, tmp_config_dir):
+    """If both YAML and cache are corrupt, creates fresh config."""
+    config_file = tmp_config_dir / "config.yaml"
+    key_file = tmp_config_dir / ".osk"
+
+    from connpy.configfile import configfile
+    conf = configfile(conf=str(config_file), key=str(key_file))
+    cachefile_path = conf.cachefile
+
+    # Corrupt YAML
+    config_file.write_text("")
+    # Corrupt cache
+    from pathlib import Path
+    Path(cachefile_path).write_text(json.dumps({"garbage": True}))
+    import time; time.sleep(0.05)
+    os.utime(str(config_file), None)
+
+    conf2 = configfile(conf=str(config_file), key=str(key_file))
+
+    # Should get defaults, not crash
+    assert conf2.config is not None
+    assert "default" in conf2.profiles
+    assert isinstance(conf2.connections, dict)
+
+

If both YAML and cache are corrupt, creates fresh config.

+
+
+def test_corrupt_cache_uses_yaml(self, tmp_config_dir) +
+
+
+ +Expand source code + +
def test_corrupt_cache_uses_yaml(self, tmp_config_dir):
+    """If cache is corrupt but YAML is valid, uses YAML."""
+    config_file = tmp_config_dir / "config.yaml"
+    key_file = tmp_config_dir / ".osk"
+
+    valid_data = {
+        "config": {"case": False, "idletime": 30, "fzf": False},
+        "connections": {},
+        "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}
+    }
+    config_file.write_text(yaml.dump(valid_data, default_flow_style=False, sort_keys=False))
+
+    from connpy.configfile import configfile
+    conf = configfile(conf=str(config_file), key=str(key_file))
+    cachefile_path = conf.cachefile
+
+    # Now corrupt the cache (valid JSON but invalid config structure)
+    from pathlib import Path
+    Path(cachefile_path).write_text(json.dumps({"garbage": True}))
+    # Make cache newer than YAML to force cache path
+    import time; time.sleep(0.05)
+    os.utime(cachefile_path, None)
+
+    conf2 = configfile(conf=str(config_file), key=str(key_file))
+    assert conf2.config["case"] == False
+    assert "default" in conf2.profiles
+
+

If cache is corrupt but YAML is valid, uses YAML.

+
+
+def test_corrupt_yaml_recovers_from_cache(self, tmp_config_dir) +
+
+
+ +Expand source code + +
def test_corrupt_yaml_recovers_from_cache(self, tmp_config_dir):
+    """If YAML is corrupt but cache is valid, recovers from cache."""
+    config_file = tmp_config_dir / "config.yaml"
+    key_file = tmp_config_dir / ".osk"
+
+    # Write valid config with router1
+    valid_data = {
+        "config": {"case": False, "idletime": 30, "fzf": False},
+        "connections": {"router1": {"host": "10.0.0.1", "type": "connection", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}},
+        "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}
+    }
+    config_file.write_text(yaml.dump(valid_data, default_flow_style=False, sort_keys=False))
+
+    from connpy.configfile import configfile
+    conf = configfile(conf=str(config_file), key=str(key_file))
+    # Save to populate cache at the real self.cachefile path
+    conf._saveconfig(conf.file)
+    cachefile_path = conf.cachefile
+    assert os.path.exists(cachefile_path)
+
+    # Now corrupt the YAML
+    config_file.write_text("")
+    import time; time.sleep(0.05)  # Ensure YAML is newer than cache
+
+    # Reload - should recover from cache
+    conf2 = configfile(conf=str(config_file), key=str(key_file))
+    assert "router1" in conf2.connections
+    assert conf2.connections["router1"]["host"] == "10.0.0.1"
+
+

If YAML is corrupt but cache is valid, recovers from cache.

+
+
+
class TestEncryption
@@ -1297,6 +1611,266 @@ el.replaceWith(d); +
+class TestMigrationSafety +
+
+
+ +Expand source code + +
class TestMigrationSafety:
+    def test_migration_validates_legacy_data(self, tmp_path):
+        """Migration skips invalid legacy JSON files."""
+        from unittest.mock import patch
+        config_dir = tmp_path / ".config" / "conn"
+        config_dir.mkdir(parents=True)
+        (config_dir / "plugins").mkdir()
+
+        # Write .folder
+        (config_dir / ".folder").write_text(str(config_dir))
+
+        # Generate RSA key
+        from Crypto.PublicKey import RSA
+        key = RSA.generate(2048)
+        key_file = config_dir / ".osk"
+        key_file.write_bytes(key.export_key("PEM"))
+        os.chmod(str(key_file), 0o600)
+
+        # Write invalid JSON config (missing required keys)
+        legacy_file = config_dir / "config.json"
+        legacy_file.write_text(json.dumps({"garbage": True}))
+
+        with patch("os.path.expanduser", return_value=str(tmp_path)):
+            from connpy.configfile import configfile
+            conf = configfile(key=str(key_file))
+
+        # Legacy file should NOT have been moved to .backup
+        assert legacy_file.exists()
+        assert not (config_dir / "config.json.backup").exists()
+
+    def test_migration_verifies_written_yaml(self, tmp_path):
+        """Migration succeeds when legacy JSON is valid."""
+        from unittest.mock import patch
+        config_dir = tmp_path / ".config" / "conn"
+        config_dir.mkdir(parents=True)
+        (config_dir / "plugins").mkdir()
+
+        # Write .folder
+        (config_dir / ".folder").write_text(str(config_dir))
+
+        # Generate RSA key
+        from Crypto.PublicKey import RSA
+        key = RSA.generate(2048)
+        key_file = config_dir / ".osk"
+        key_file.write_bytes(key.export_key("PEM"))
+        os.chmod(str(key_file), 0o600)
+
+        valid_data = {
+            "config": {"case": False, "idletime": 30, "fzf": False},
+            "connections": {"r1": {"host": "1.2.3.4", "type": "connection", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}},
+            "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}
+        }
+        legacy_file = config_dir / "config.json"
+        legacy_file.write_text(json.dumps(valid_data))
+
+        with patch("os.path.expanduser", return_value=str(tmp_path)):
+            from connpy.configfile import configfile
+            conf = configfile(key=str(key_file))
+
+        # Migration should have succeeded: YAML exists, JSON backed up
+        yaml_file = config_dir / "config.yaml"
+        assert yaml_file.exists()
+        assert (config_dir / "config.json.backup").exists()
+        assert not legacy_file.exists()
+        assert "r1" in conf.connections
+
+
+

Methods

+
+
+def test_migration_validates_legacy_data(self, tmp_path) +
+
+
+ +Expand source code + +
def test_migration_validates_legacy_data(self, tmp_path):
+    """Migration skips invalid legacy JSON files."""
+    from unittest.mock import patch
+    config_dir = tmp_path / ".config" / "conn"
+    config_dir.mkdir(parents=True)
+    (config_dir / "plugins").mkdir()
+
+    # Write .folder
+    (config_dir / ".folder").write_text(str(config_dir))
+
+    # Generate RSA key
+    from Crypto.PublicKey import RSA
+    key = RSA.generate(2048)
+    key_file = config_dir / ".osk"
+    key_file.write_bytes(key.export_key("PEM"))
+    os.chmod(str(key_file), 0o600)
+
+    # Write invalid JSON config (missing required keys)
+    legacy_file = config_dir / "config.json"
+    legacy_file.write_text(json.dumps({"garbage": True}))
+
+    with patch("os.path.expanduser", return_value=str(tmp_path)):
+        from connpy.configfile import configfile
+        conf = configfile(key=str(key_file))
+
+    # Legacy file should NOT have been moved to .backup
+    assert legacy_file.exists()
+    assert not (config_dir / "config.json.backup").exists()
+
+

Migration skips invalid legacy JSON files.

+
+
+def test_migration_verifies_written_yaml(self, tmp_path) +
+
+
+ +Expand source code + +
def test_migration_verifies_written_yaml(self, tmp_path):
+    """Migration succeeds when legacy JSON is valid."""
+    from unittest.mock import patch
+    config_dir = tmp_path / ".config" / "conn"
+    config_dir.mkdir(parents=True)
+    (config_dir / "plugins").mkdir()
+
+    # Write .folder
+    (config_dir / ".folder").write_text(str(config_dir))
+
+    # Generate RSA key
+    from Crypto.PublicKey import RSA
+    key = RSA.generate(2048)
+    key_file = config_dir / ".osk"
+    key_file.write_bytes(key.export_key("PEM"))
+    os.chmod(str(key_file), 0o600)
+
+    valid_data = {
+        "config": {"case": False, "idletime": 30, "fzf": False},
+        "connections": {"r1": {"host": "1.2.3.4", "type": "connection", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}},
+        "profiles": {"default": {"host": "", "protocol": "ssh", "port": "", "user": "", "password": "", "options": "", "logs": "", "tags": "", "jumphost": ""}}
+    }
+    legacy_file = config_dir / "config.json"
+    legacy_file.write_text(json.dumps(valid_data))
+
+    with patch("os.path.expanduser", return_value=str(tmp_path)):
+        from connpy.configfile import configfile
+        conf = configfile(key=str(key_file))
+
+    # Migration should have succeeded: YAML exists, JSON backed up
+    yaml_file = config_dir / "config.yaml"
+    assert yaml_file.exists()
+    assert (config_dir / "config.json.backup").exists()
+    assert not legacy_file.exists()
+    assert "r1" in conf.connections
+
+

Migration succeeds when legacy JSON is valid.

+
+
+
+
+class TestValidateConfig +
+
+
+ +Expand source code + +
class TestValidateConfig:
+    def test_valid_config(self, config):
+        data = {"config": {}, "connections": {}, "profiles": {}}
+        assert config._validate_config(data) == True
+
+    def test_none_data(self, config):
+        assert config._validate_config(None) == False
+
+    def test_string_data(self, config):
+        assert config._validate_config("not a dict") == False
+
+    def test_missing_key(self, config):
+        assert config._validate_config({"config": {}, "connections": {}}) == False
+
+    def test_empty_dict(self, config):
+        assert config._validate_config({}) == False
+
+
+

Methods

+
+
+def test_empty_dict(self, config) +
+
+
+ +Expand source code + +
def test_empty_dict(self, config):
+    assert config._validate_config({}) == False
+
+
+
+
+def test_missing_key(self, config) +
+
+
+ +Expand source code + +
def test_missing_key(self, config):
+    assert config._validate_config({"config": {}, "connections": {}}) == False
+
+
+
+
+def test_none_data(self, config) +
+
+
+ +Expand source code + +
def test_none_data(self, config):
+    assert config._validate_config(None) == False
+
+
+
+
+def test_string_data(self, config) +
+
+
+ +Expand source code + +
def test_string_data(self, config):
+    assert config._validate_config("not a dict") == False
+
+
+
+
+def test_valid_config(self, config) +
+
+
+ +Expand source code + +
def test_valid_config(self, config):
+    data = {"config": {}, "connections": {}, "profiles": {}}
+    assert config._validate_config(data) == True
+
+
+
+
+
@@ -1313,6 +1887,13 @@ el.replaceWith(d);

  • Classes