fix(config): implement atomic save, validation and recovery for config files and update pdoc
This commit is contained in:
+1
-2
@@ -1,2 +1 @@
|
||||
__version__ = "5.0b4"
|
||||
|
||||
__version__ = "5.0b5"
|
||||
|
||||
+76
-23
@@ -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
|
||||
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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
|
||||
|
||||
+77
-24
@@ -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:
|
||||
</code></pre>
|
||||
<h3 id="optional-parameters">Optional Parameters:</h3>
|
||||
<pre><code>- 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
|
||||
|
||||
@@ -58,7 +58,7 @@ No test touches ~/.config/conn/</p>
|
||||
<pre><code class="python">@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"))</code></pre>
|
||||
</details>
|
||||
@@ -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)</code></pre>
|
||||
</details>
|
||||
@@ -182,13 +182,13 @@ def mock_pexpect():
|
||||
<pre><code class="python">@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"))</code></pre>
|
||||
</details>
|
||||
@@ -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)
|
||||
|
||||
@@ -47,6 +47,116 @@ el.replaceWith(d);
|
||||
<section>
|
||||
<h2 class="section-title" id="header-classes">Classes</h2>
|
||||
<dl>
|
||||
<dt id="connpy.tests.test_configfile.TestAtomicSave"><code class="flex name class">
|
||||
<span>class <span class="ident">TestAtomicSave</span></span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
<h3>Methods</h3>
|
||||
<dl>
|
||||
<dt id="connpy.tests.test_configfile.TestAtomicSave.test_save_creates_no_leftover_tmp"><code class="name flex">
|
||||
<span>def <span class="ident">test_save_creates_no_leftover_tmp</span></span>(<span>self, config)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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')</code></pre>
|
||||
</details>
|
||||
<div class="desc"><p>After successful save, no .tmp file remains.</p></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestAtomicSave.test_save_preserves_original_on_error"><code class="name flex">
|
||||
<span>def <span class="ident">test_save_preserves_original_on_error</span></span>(<span>self, config)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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</code></pre>
|
||||
</details>
|
||||
<div class="desc"><p>If save fails, original config file is not corrupted.</p></div>
|
||||
</dd>
|
||||
</dl>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestCRUDNodes"><code class="flex name class">
|
||||
<span>class <span class="ident">TestCRUDNodes</span></span>
|
||||
</code></dt>
|
||||
@@ -565,6 +675,210 @@ el.replaceWith(d);
|
||||
</dd>
|
||||
</dl>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestCorruptionRecovery"><code class="flex name class">
|
||||
<span>class <span class="ident">TestCorruptionRecovery</span></span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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)</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
<h3>Methods</h3>
|
||||
<dl>
|
||||
<dt id="connpy.tests.test_configfile.TestCorruptionRecovery.test_both_corrupt_creates_default"><code class="name flex">
|
||||
<span>def <span class="ident">test_both_corrupt_creates_default</span></span>(<span>self, tmp_config_dir)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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)</code></pre>
|
||||
</details>
|
||||
<div class="desc"><p>If both YAML and cache are corrupt, creates fresh config.</p></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestCorruptionRecovery.test_corrupt_cache_uses_yaml"><code class="name flex">
|
||||
<span>def <span class="ident">test_corrupt_cache_uses_yaml</span></span>(<span>self, tmp_config_dir)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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</code></pre>
|
||||
</details>
|
||||
<div class="desc"><p>If cache is corrupt but YAML is valid, uses YAML.</p></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestCorruptionRecovery.test_corrupt_yaml_recovers_from_cache"><code class="name flex">
|
||||
<span>def <span class="ident">test_corrupt_yaml_recovers_from_cache</span></span>(<span>self, tmp_config_dir)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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"</code></pre>
|
||||
</details>
|
||||
<div class="desc"><p>If YAML is corrupt but cache is valid, recovers from cache.</p></div>
|
||||
</dd>
|
||||
</dl>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestEncryption"><code class="flex name class">
|
||||
<span>class <span class="ident">TestEncryption</span></span>
|
||||
</code></dt>
|
||||
@@ -1297,6 +1611,266 @@ el.replaceWith(d);
|
||||
</dd>
|
||||
</dl>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestMigrationSafety"><code class="flex name class">
|
||||
<span>class <span class="ident">TestMigrationSafety</span></span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
<h3>Methods</h3>
|
||||
<dl>
|
||||
<dt id="connpy.tests.test_configfile.TestMigrationSafety.test_migration_validates_legacy_data"><code class="name flex">
|
||||
<span>def <span class="ident">test_migration_validates_legacy_data</span></span>(<span>self, tmp_path)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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()</code></pre>
|
||||
</details>
|
||||
<div class="desc"><p>Migration skips invalid legacy JSON files.</p></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestMigrationSafety.test_migration_verifies_written_yaml"><code class="name flex">
|
||||
<span>def <span class="ident">test_migration_verifies_written_yaml</span></span>(<span>self, tmp_path)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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</code></pre>
|
||||
</details>
|
||||
<div class="desc"><p>Migration succeeds when legacy JSON is valid.</p></div>
|
||||
</dd>
|
||||
</dl>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestValidateConfig"><code class="flex name class">
|
||||
<span>class <span class="ident">TestValidateConfig</span></span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">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</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
<h3>Methods</h3>
|
||||
<dl>
|
||||
<dt id="connpy.tests.test_configfile.TestValidateConfig.test_empty_dict"><code class="name flex">
|
||||
<span>def <span class="ident">test_empty_dict</span></span>(<span>self, config)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">def test_empty_dict(self, config):
|
||||
assert config._validate_config({}) == False</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestValidateConfig.test_missing_key"><code class="name flex">
|
||||
<span>def <span class="ident">test_missing_key</span></span>(<span>self, config)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">def test_missing_key(self, config):
|
||||
assert config._validate_config({"config": {}, "connections": {}}) == False</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestValidateConfig.test_none_data"><code class="name flex">
|
||||
<span>def <span class="ident">test_none_data</span></span>(<span>self, config)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">def test_none_data(self, config):
|
||||
assert config._validate_config(None) == False</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestValidateConfig.test_string_data"><code class="name flex">
|
||||
<span>def <span class="ident">test_string_data</span></span>(<span>self, config)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">def test_string_data(self, config):
|
||||
assert config._validate_config("not a dict") == False</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
</dd>
|
||||
<dt id="connpy.tests.test_configfile.TestValidateConfig.test_valid_config"><code class="name flex">
|
||||
<span>def <span class="ident">test_valid_config</span></span>(<span>self, config)</span>
|
||||
</code></dt>
|
||||
<dd>
|
||||
<details class="source">
|
||||
<summary>
|
||||
<span>Expand source code</span>
|
||||
</summary>
|
||||
<pre><code class="python">def test_valid_config(self, config):
|
||||
data = {"config": {}, "connections": {}, "profiles": {}}
|
||||
assert config._validate_config(data) == True</code></pre>
|
||||
</details>
|
||||
<div class="desc"></div>
|
||||
</dd>
|
||||
</dl>
|
||||
</dd>
|
||||
</dl>
|
||||
</section>
|
||||
</article>
|
||||
@@ -1313,6 +1887,13 @@ el.replaceWith(d);
|
||||
<li><h3><a href="#header-classes">Classes</a></h3>
|
||||
<ul>
|
||||
<li>
|
||||
<h4><code><a title="connpy.tests.test_configfile.TestAtomicSave" href="#connpy.tests.test_configfile.TestAtomicSave">TestAtomicSave</a></code></h4>
|
||||
<ul class="">
|
||||
<li><code><a title="connpy.tests.test_configfile.TestAtomicSave.test_save_creates_no_leftover_tmp" href="#connpy.tests.test_configfile.TestAtomicSave.test_save_creates_no_leftover_tmp">test_save_creates_no_leftover_tmp</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestAtomicSave.test_save_preserves_original_on_error" href="#connpy.tests.test_configfile.TestAtomicSave.test_save_preserves_original_on_error">test_save_preserves_original_on_error</a></code></li>
|
||||
</ul>
|
||||
</li>
|
||||
<li>
|
||||
<h4><code><a title="connpy.tests.test_configfile.TestCRUDNodes" href="#connpy.tests.test_configfile.TestCRUDNodes">TestCRUDNodes</a></code></h4>
|
||||
<ul class="">
|
||||
<li><code><a title="connpy.tests.test_configfile.TestCRUDNodes.test_add_folder" href="#connpy.tests.test_configfile.TestCRUDNodes.test_add_folder">test_add_folder</a></code></li>
|
||||
@@ -1345,6 +1926,14 @@ el.replaceWith(d);
|
||||
</ul>
|
||||
</li>
|
||||
<li>
|
||||
<h4><code><a title="connpy.tests.test_configfile.TestCorruptionRecovery" href="#connpy.tests.test_configfile.TestCorruptionRecovery">TestCorruptionRecovery</a></code></h4>
|
||||
<ul class="">
|
||||
<li><code><a title="connpy.tests.test_configfile.TestCorruptionRecovery.test_both_corrupt_creates_default" href="#connpy.tests.test_configfile.TestCorruptionRecovery.test_both_corrupt_creates_default">test_both_corrupt_creates_default</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestCorruptionRecovery.test_corrupt_cache_uses_yaml" href="#connpy.tests.test_configfile.TestCorruptionRecovery.test_corrupt_cache_uses_yaml">test_corrupt_cache_uses_yaml</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestCorruptionRecovery.test_corrupt_yaml_recovers_from_cache" href="#connpy.tests.test_configfile.TestCorruptionRecovery.test_corrupt_yaml_recovers_from_cache">test_corrupt_yaml_recovers_from_cache</a></code></li>
|
||||
</ul>
|
||||
</li>
|
||||
<li>
|
||||
<h4><code><a title="connpy.tests.test_configfile.TestEncryption" href="#connpy.tests.test_configfile.TestEncryption">TestEncryption</a></code></h4>
|
||||
<ul class="">
|
||||
<li><code><a title="connpy.tests.test_configfile.TestEncryption.test_encrypt_decrypt_roundtrip" href="#connpy.tests.test_configfile.TestEncryption.test_encrypt_decrypt_roundtrip">test_encrypt_decrypt_roundtrip</a></code></li>
|
||||
@@ -1391,6 +1980,23 @@ el.replaceWith(d);
|
||||
<li><code><a title="connpy.tests.test_configfile.TestGetItem.test_getitems_multiple" href="#connpy.tests.test_configfile.TestGetItem.test_getitems_multiple">test_getitems_multiple</a></code></li>
|
||||
</ul>
|
||||
</li>
|
||||
<li>
|
||||
<h4><code><a title="connpy.tests.test_configfile.TestMigrationSafety" href="#connpy.tests.test_configfile.TestMigrationSafety">TestMigrationSafety</a></code></h4>
|
||||
<ul class="">
|
||||
<li><code><a title="connpy.tests.test_configfile.TestMigrationSafety.test_migration_validates_legacy_data" href="#connpy.tests.test_configfile.TestMigrationSafety.test_migration_validates_legacy_data">test_migration_validates_legacy_data</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestMigrationSafety.test_migration_verifies_written_yaml" href="#connpy.tests.test_configfile.TestMigrationSafety.test_migration_verifies_written_yaml">test_migration_verifies_written_yaml</a></code></li>
|
||||
</ul>
|
||||
</li>
|
||||
<li>
|
||||
<h4><code><a title="connpy.tests.test_configfile.TestValidateConfig" href="#connpy.tests.test_configfile.TestValidateConfig">TestValidateConfig</a></code></h4>
|
||||
<ul class="">
|
||||
<li><code><a title="connpy.tests.test_configfile.TestValidateConfig.test_empty_dict" href="#connpy.tests.test_configfile.TestValidateConfig.test_empty_dict">test_empty_dict</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestValidateConfig.test_missing_key" href="#connpy.tests.test_configfile.TestValidateConfig.test_missing_key">test_missing_key</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestValidateConfig.test_none_data" href="#connpy.tests.test_configfile.TestValidateConfig.test_none_data">test_none_data</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestValidateConfig.test_string_data" href="#connpy.tests.test_configfile.TestValidateConfig.test_string_data">test_string_data</a></code></li>
|
||||
<li><code><a title="connpy.tests.test_configfile.TestValidateConfig.test_valid_config" href="#connpy.tests.test_configfile.TestValidateConfig.test_valid_config">test_valid_config</a></code></li>
|
||||
</ul>
|
||||
</li>
|
||||
</ul>
|
||||
</li>
|
||||
</ul>
|
||||
|
||||
Reference in New Issue
Block a user