@@ -100,12 +112,15 @@ el.replaceWith(d);
connpy.tests.conftest
connpy.tests.test_ai
connpy.tests.test_api
+connpy.tests.test_capture
connpy.tests.test_completion
connpy.tests.test_configfile
+connpy.tests.test_context
connpy.tests.test_core
connpy.tests.test_hooks
connpy.tests.test_plugins
connpy.tests.test_printer
+connpy.tests.test_sync
diff --git a/docs/connpy/tests/test_capture.html b/docs/connpy/tests/test_capture.html
new file mode 100644
index 0000000..db68f63
--- /dev/null
+++ b/docs/connpy/tests/test_capture.html
@@ -0,0 +1,235 @@
+
+
+
+
+
+
+connpy.tests.test_capture API documentation
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+Module connpy.tests.test_capture
+
+
+Tests for connpy.core_plugins.capture
+
+
+
+
+
+
+
+def mock_connapp()
+-
+
+
+Expand source code
+
+@pytest.fixture
+def mock_connapp():
+ app = MagicMock()
+ app.nodes_list = ["test_node"]
+ app.config.getitem.return_value = {"host": "127.0.0.1", "protocol": "ssh"}
+ mock_node = MagicMock()
+ mock_node.protocol = "ssh"
+ mock_node.unique = "test_node"
+ app.node.return_value = mock_node
+ app.config.config = {"wireshark_path": "/fake/ws"}
+ return app
+
+
+
+
+
+
+
+
+
+class TestRemoteCapture
+-
+
+
+Expand source code
+
+class TestRemoteCapture:
+ def test_init_node_not_found(self, mock_connapp):
+ # Attempt to capture a node not in nodes_list
+ mock_connapp.nodes_list = ["other_node"]
+ with pytest.raises(SystemExit) as exc:
+ RemoteCapture(mock_connapp, "test_node", "eth0")
+ assert exc.value.code == 2
+
+ def test_init_success(self, mock_connapp):
+ rc = RemoteCapture(mock_connapp, "test_node", "eth0")
+ assert rc.node_name == "test_node"
+ assert rc.interface == "eth0"
+ assert rc.wireshark_path == "/fake/ws"
+
+ @patch("connpy.core_plugins.capture.socket")
+ def test_is_port_in_use(self, mock_socket, mock_connapp):
+ rc = RemoteCapture(mock_connapp, "test_node", "eth0")
+ mock_sock_instance = MagicMock()
+ mock_socket.socket.return_value.__enter__.return_value = mock_sock_instance
+
+ mock_sock_instance.connect_ex.return_value = 0
+ assert rc._is_port_in_use(8080) is True
+
+ mock_sock_instance.connect_ex.return_value = 1
+ assert rc._is_port_in_use(8080) is False
+
+ @patch.object(RemoteCapture, "_is_port_in_use")
+ def test_find_free_port(self, mock_is_in_use, mock_connapp):
+ rc = RemoteCapture(mock_connapp, "test_node", "eth0")
+ # First 2 ports in use, 3rd is free
+ mock_is_in_use.side_effect = [True, True, False]
+ port = rc._find_free_port(20000, 30000)
+ assert 20000 <= port <= 30000
+ assert mock_is_in_use.call_count == 3
+
+
+Methods
+
+
+def test_find_free_port(self, mock_is_in_use, mock_connapp)
+-
+
+
+Expand source code
+
+@patch.object(RemoteCapture, "_is_port_in_use")
+def test_find_free_port(self, mock_is_in_use, mock_connapp):
+ rc = RemoteCapture(mock_connapp, "test_node", "eth0")
+ # First 2 ports in use, 3rd is free
+ mock_is_in_use.side_effect = [True, True, False]
+ port = rc._find_free_port(20000, 30000)
+ assert 20000 <= port <= 30000
+ assert mock_is_in_use.call_count == 3
+
+
+
+
+def test_init_node_not_found(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_init_node_not_found(self, mock_connapp):
+ # Attempt to capture a node not in nodes_list
+ mock_connapp.nodes_list = ["other_node"]
+ with pytest.raises(SystemExit) as exc:
+ RemoteCapture(mock_connapp, "test_node", "eth0")
+ assert exc.value.code == 2
+
+
+
+
+def test_init_success(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_init_success(self, mock_connapp):
+ rc = RemoteCapture(mock_connapp, "test_node", "eth0")
+ assert rc.node_name == "test_node"
+ assert rc.interface == "eth0"
+ assert rc.wireshark_path == "/fake/ws"
+
+
+
+
+def test_is_port_in_use(self, mock_socket, mock_connapp)
+-
+
+
+Expand source code
+
+@patch("connpy.core_plugins.capture.socket")
+def test_is_port_in_use(self, mock_socket, mock_connapp):
+ rc = RemoteCapture(mock_connapp, "test_node", "eth0")
+ mock_sock_instance = MagicMock()
+ mock_socket.socket.return_value.__enter__.return_value = mock_sock_instance
+
+ mock_sock_instance.connect_ex.return_value = 0
+ assert rc._is_port_in_use(8080) is True
+
+ mock_sock_instance.connect_ex.return_value = 1
+ assert rc._is_port_in_use(8080) is False
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/docs/connpy/tests/test_configfile.html b/docs/connpy/tests/test_configfile.html
index 164ba3c..eea4bc0 100644
--- a/docs/connpy/tests/test_configfile.html
+++ b/docs/connpy/tests/test_configfile.html
@@ -383,9 +383,9 @@ el.replaceWith(d);
class TestConfigfileInit:
def test_creates_default_config(self, tmp_config_dir):
- """Creates config.json with defaults when it doesn't exist."""
- config_file = tmp_config_dir / "config.json"
- config_file.unlink() # Remove existing
+ """Creates config.yaml with defaults when it doesn't exist."""
+ config_file = tmp_config_dir / "config.yaml"
+ config_file.unlink(missing_ok=True) # Remove existing
key_file = tmp_config_dir / ".osk"
from connpy.configfile import configfile
@@ -402,7 +402,7 @@ el.replaceWith(d);
key_file.unlink() # Remove existing
from connpy.configfile import configfile
- conf = configfile(conf=str(tmp_config_dir / "config.json"), key=str(key_file))
+ conf = configfile(conf=str(tmp_config_dir / "config.yaml"), key=str(key_file))
assert key_file.exists()
assert conf.privatekey is not None
@@ -416,8 +416,8 @@ el.replaceWith(d);
def test_config_file_permissions(self, tmp_config_dir):
"""Config is created with 0o600 permissions."""
- config_file = tmp_config_dir / "config.json"
- config_file.unlink()
+ config_file = tmp_config_dir / "config.yaml"
+ config_file.unlink(missing_ok=True)
from connpy.configfile import configfile
configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))
@@ -437,7 +437,7 @@ el.replaceWith(d);
(dot_folder / ".folder").write_text(str(config_dir))
(dot_folder / "plugins").mkdir(exist_ok=True)
- conf_path = str(config_dir / "my_config.json")
+ conf_path = str(config_dir / "my_config.yaml")
key_path = str(config_dir / "my_key")
from connpy.configfile import configfile
@@ -459,8 +459,8 @@ el.replaceWith(d);
def test_config_file_permissions(self, tmp_config_dir):
"""Config is created with 0o600 permissions."""
- config_file = tmp_config_dir / "config.json"
- config_file.unlink()
+ config_file = tmp_config_dir / "config.yaml"
+ config_file.unlink(missing_ok=True)
from connpy.configfile import configfile
configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))
@@ -479,9 +479,9 @@ el.replaceWith(d);
Expand source code
def test_creates_default_config(self, tmp_config_dir):
- """Creates config.json with defaults when it doesn't exist."""
- config_file = tmp_config_dir / "config.json"
- config_file.unlink() # Remove existing
+ """Creates config.yaml with defaults when it doesn't exist."""
+ config_file = tmp_config_dir / "config.yaml"
+ config_file.unlink(missing_ok=True) # Remove existing
key_file = tmp_config_dir / ".osk"
from connpy.configfile import configfile
@@ -492,7 +492,7 @@ el.replaceWith(d);
assert conf.config["idletime"] == 30
assert "default" in conf.profiles
-Creates config.json with defaults when it doesn't exist.
Creates config.yaml with defaults when it doesn't exist.
def test_creates_rsa_key(self, tmp_config_dir)
@@ -508,7 +508,7 @@ el.replaceWith(d);
key_file.unlink() # Remove existing
from connpy.configfile import configfile
- conf = configfile(conf=str(tmp_config_dir / "config.json"), key=str(key_file))
+ conf = configfile(conf=str(tmp_config_dir / "config.yaml"), key=str(key_file))
assert key_file.exists()
assert conf.privatekey is not None
@@ -536,7 +536,7 @@ el.replaceWith(d);
(dot_folder / ".folder").write_text(str(config_dir))
(dot_folder / "plugins").mkdir(exist_ok=True)
- conf_path = str(config_dir / "my_config.json")
+ conf_path = str(config_dir / "my_config.yaml")
key_path = str(config_dir / "my_key")
from connpy.configfile import configfile
@@ -846,7 +846,7 @@ el.replaceWith(d);
def test_profileused(self, tmp_config_dir):
"""Detects nodes using a specific profile."""
- config_file = tmp_config_dir / "config.json"
+ config_file = tmp_config_dir / "config.yaml"
data = {
"config": {"case": False, "idletime": 30, "fzf": False},
"connections": {
@@ -872,7 +872,7 @@ el.replaceWith(d);
"options": "", "logs": "", "tags": "", "jumphost": ""}
}
}
- 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
conf = configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))
@@ -1014,7 +1014,7 @@ el.replaceWith(d);
def test_profileused(self, tmp_config_dir):
"""Detects nodes using a specific profile."""
- config_file = tmp_config_dir / "config.json"
+ config_file = tmp_config_dir / "config.yaml"
data = {
"config": {"case": False, "idletime": 30, "fzf": False},
"connections": {
@@ -1040,7 +1040,7 @@ el.replaceWith(d);
"options": "", "logs": "", "tags": "", "jumphost": ""}
}
}
- 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
conf = configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))
@@ -1111,7 +1111,7 @@ el.replaceWith(d);
def test_getitem_with_profile_extraction(self, tmp_config_dir):
"""extract=True resolves @profile references."""
- config_file = tmp_config_dir / "config.json"
+ config_file = tmp_config_dir / "config.yaml"
data = {
"config": {"case": False, "idletime": 30, "fzf": False},
"connections": {
@@ -1131,7 +1131,7 @@ el.replaceWith(d);
"options": "", "logs": "", "tags": "", "jumphost": ""}
}
}
- 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
conf = configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))
@@ -1235,7 +1235,7 @@ el.replaceWith(d);
def test_getitem_with_profile_extraction(self, tmp_config_dir):
"""extract=True resolves @profile references."""
- config_file = tmp_config_dir / "config.json"
+ config_file = tmp_config_dir / "config.yaml"
data = {
"config": {"case": False, "idletime": 30, "fzf": False},
"connections": {
@@ -1255,7 +1255,7 @@ el.replaceWith(d);
"options": "", "logs": "", "tags": "", "jumphost": ""}
}
}
- 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
conf = configfile(conf=str(config_file), key=str(tmp_config_dir / ".osk"))
diff --git a/docs/connpy/tests/test_context.html b/docs/connpy/tests/test_context.html
new file mode 100644
index 0000000..bc4ca50
--- /dev/null
+++ b/docs/connpy/tests/test_context.html
@@ -0,0 +1,475 @@
+
+
+
+
+
+
+connpy.tests.test_context API documentation
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+Module connpy.tests.test_context
+
+
+Tests for connpy.core_plugins.context
+
+
+
+
+
+
+
+def mock_connapp()
+-
+
+
+Expand source code
+
+@pytest.fixture
+def mock_connapp():
+ connapp = MagicMock()
+ connapp.config.config = {
+ "contexts": {"all": [".*"]},
+ "current_context": "all"
+ }
+ return connapp
+
+
+
+
+
+
+
+
+
+class TestContextManager
+-
+
+
+Expand source code
+
+class TestContextManager:
+ def test_init(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ assert cm.contexts == {"all": [".*"]}
+ assert cm.current_context == "all"
+ assert len(cm.regex) == 1
+
+ def test_add_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.add_context("prod", ["^prod_.*"])
+ assert "prod" in cm.contexts
+ mock_connapp._change_settings.assert_called_with("contexts", cm.contexts)
+
+ def test_add_context_invalid_name(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.add_context("prod-env", ["Regex"])
+ assert exc.value.code == 1
+
+ def test_add_context_already_exists(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.add_context("all", ["Regex"])
+ assert exc.value.code == 2
+
+ def test_modify_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.add_context("prod", ["old"])
+ cm.modify_context("prod", ["new"])
+ assert cm.contexts["prod"] == ["new"]
+
+ def test_modify_context_all(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.modify_context("all", ["new"])
+ assert exc.value.code == 3
+
+ def test_modify_context_not_exists(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.modify_context("fake", ["new"])
+ assert exc.value.code == 4
+
+ def test_delete_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.add_context("prod", ["old"])
+ cm.delete_context("prod")
+ assert "prod" not in cm.contexts
+
+ def test_delete_context_all(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.delete_context("all")
+ assert exc.value.code == 3
+
+ def test_delete_context_current(self, mock_connapp):
+ mock_connapp.config.config["current_context"] = "prod"
+ mock_connapp.config.config["contexts"]["prod"] = [".*"]
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.delete_context("prod")
+ assert exc.value.code == 5
+
+ def test_set_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.contexts["prod"] = [".*"]
+ cm.set_context("prod")
+ mock_connapp._change_settings.assert_called_with("current_context", "prod")
+
+ def test_set_context_already_set(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.set_context("all")
+ assert exc.value.code == 0
+
+ def test_match_regexp(self, mock_connapp):
+ mock_connapp.config.config["contexts"]["all"] = ["^prod", "^test"]
+ cm = context_manager(mock_connapp)
+ assert cm.match_any_regex("prod_node", cm.regex) is True
+ assert cm.match_any_regex("test_node", cm.regex) is True
+ assert cm.match_any_regex("dev_node", cm.regex) is False
+
+ def test_modify_node_list(self, mock_connapp):
+ mock_connapp.config.config["contexts"]["all"] = ["^prod"]
+ cm = context_manager(mock_connapp)
+ nodes = ["prod_1", "dev_1", "prod_2"]
+ result = cm.modify_node_list(result=nodes)
+ assert result == ["prod_1", "prod_2"]
+
+ def test_modify_node_dict(self, mock_connapp):
+ mock_connapp.config.config["contexts"]["all"] = ["^prod"]
+ cm = context_manager(mock_connapp)
+ nodes = {"prod_1": {}, "dev_1": {}, "prod_2": {}}
+ result = cm.modify_node_dict(result=nodes)
+ assert set(result.keys()) == {"prod_1", "prod_2"}
+
+
+Methods
+
+
+def test_add_context_already_exists(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_add_context_already_exists(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.add_context("all", ["Regex"])
+ assert exc.value.code == 2
+
+
+
+
+def test_add_context_invalid_name(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_add_context_invalid_name(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.add_context("prod-env", ["Regex"])
+ assert exc.value.code == 1
+
+
+
+
+def test_add_context_success(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_add_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.add_context("prod", ["^prod_.*"])
+ assert "prod" in cm.contexts
+ mock_connapp._change_settings.assert_called_with("contexts", cm.contexts)
+
+
+
+
+def test_delete_context_all(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_delete_context_all(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.delete_context("all")
+ assert exc.value.code == 3
+
+
+
+
+def test_delete_context_current(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_delete_context_current(self, mock_connapp):
+ mock_connapp.config.config["current_context"] = "prod"
+ mock_connapp.config.config["contexts"]["prod"] = [".*"]
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.delete_context("prod")
+ assert exc.value.code == 5
+
+
+
+
+def test_delete_context_success(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_delete_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.add_context("prod", ["old"])
+ cm.delete_context("prod")
+ assert "prod" not in cm.contexts
+
+
+
+
+def test_init(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_init(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ assert cm.contexts == {"all": [".*"]}
+ assert cm.current_context == "all"
+ assert len(cm.regex) == 1
+
+
+
+
+def test_match_regexp(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_match_regexp(self, mock_connapp):
+ mock_connapp.config.config["contexts"]["all"] = ["^prod", "^test"]
+ cm = context_manager(mock_connapp)
+ assert cm.match_any_regex("prod_node", cm.regex) is True
+ assert cm.match_any_regex("test_node", cm.regex) is True
+ assert cm.match_any_regex("dev_node", cm.regex) is False
+
+
+
+
+def test_modify_context_all(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_modify_context_all(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.modify_context("all", ["new"])
+ assert exc.value.code == 3
+
+
+
+
+def test_modify_context_not_exists(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_modify_context_not_exists(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.modify_context("fake", ["new"])
+ assert exc.value.code == 4
+
+
+
+
+def test_modify_context_success(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_modify_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.add_context("prod", ["old"])
+ cm.modify_context("prod", ["new"])
+ assert cm.contexts["prod"] == ["new"]
+
+
+
+
+def test_modify_node_dict(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_modify_node_dict(self, mock_connapp):
+ mock_connapp.config.config["contexts"]["all"] = ["^prod"]
+ cm = context_manager(mock_connapp)
+ nodes = {"prod_1": {}, "dev_1": {}, "prod_2": {}}
+ result = cm.modify_node_dict(result=nodes)
+ assert set(result.keys()) == {"prod_1", "prod_2"}
+
+
+
+
+def test_modify_node_list(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_modify_node_list(self, mock_connapp):
+ mock_connapp.config.config["contexts"]["all"] = ["^prod"]
+ cm = context_manager(mock_connapp)
+ nodes = ["prod_1", "dev_1", "prod_2"]
+ result = cm.modify_node_list(result=nodes)
+ assert result == ["prod_1", "prod_2"]
+
+
+
+
+def test_set_context_already_set(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_set_context_already_set(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ with pytest.raises(SystemExit) as exc:
+ cm.set_context("all")
+ assert exc.value.code == 0
+
+
+
+
+def test_set_context_success(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_set_context_success(self, mock_connapp):
+ cm = context_manager(mock_connapp)
+ cm.contexts["prod"] = [".*"]
+ cm.set_context("prod")
+ mock_connapp._change_settings.assert_called_with("current_context", "prod")
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/docs/connpy/tests/test_sync.html b/docs/connpy/tests/test_sync.html
new file mode 100644
index 0000000..d1e0b08
--- /dev/null
+++ b/docs/connpy/tests/test_sync.html
@@ -0,0 +1,396 @@
+
+
+
+
+
+
+connpy.tests.test_sync API documentation
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+Module connpy.tests.test_sync
+
+
+Tests for connpy.core_plugins.sync
+
+
+
+
+
+
+
+def mock_connapp()
+-
+
+
+Expand source code
+
+@pytest.fixture
+def mock_connapp():
+ app = MagicMock()
+ app.config.defaultdir = "/fake/dir"
+ app.config.file = "/fake/dir/config.yaml"
+ app.config.key = "/fake/dir/.osk"
+ app.config.config = {"sync": True}
+ return app
+
+
+
+
+
+
+
+
+
+class TestSyncPlugin
+-
+
+
+Expand source code
+
+class TestSyncPlugin:
+ def test_init(self, mock_connapp):
+ s = sync(mock_connapp)
+ assert s.sync is True
+ assert s.file == "/fake/dir/config.yaml"
+ assert s.token_file == "/fake/dir/gtoken.json"
+
+ @patch("connpy.core_plugins.sync.os.path.exists")
+ @patch("connpy.core_plugins.sync.Credentials")
+ def test_get_credentials_success(self, MockCreds, mock_exists, mock_connapp):
+ mock_exists.return_value = True
+ mock_cred_instance = MagicMock()
+ mock_cred_instance.valid = True
+ MockCreds.from_authorized_user_file.return_value = mock_cred_instance
+
+ s = sync(mock_connapp)
+ creds = s.get_credentials()
+ assert creds == mock_cred_instance
+
+ @patch("connpy.core_plugins.sync.os.path.exists")
+ def test_get_credentials_not_found(self, mock_exists, mock_connapp):
+ mock_exists.return_value = False
+ s = sync(mock_connapp)
+ assert s.get_credentials() == 0
+
+ @patch("connpy.core_plugins.sync.zipfile.ZipFile")
+ @patch("connpy.core_plugins.sync.os.path.basename")
+ def test_compress_specific_files(self, mock_basename, MockZipFile, mock_connapp):
+ mock_basename.return_value = "config.yaml"
+ s = sync(mock_connapp)
+ zip_mock = MagicMock()
+ MockZipFile.return_value.__enter__.return_value = zip_mock
+
+ s.compress_specific_files("/fake/zip.zip")
+ zip_mock.write.assert_any_call(s.file, "config.yaml")
+ zip_mock.write.assert_any_call(s.key, ".osk")
+
+ @patch("connpy.core_plugins.sync.zipfile.ZipFile")
+ @patch("connpy.core_plugins.sync.os.path.dirname")
+ def test_decompress_zip_yaml(self, mock_dirname, MockZipFile, mock_connapp):
+ mock_dirname.return_value = "/fake/dir"
+ s = sync(mock_connapp)
+ zip_mock = MagicMock()
+ zip_mock.namelist.return_value = ["config.yaml", ".osk"]
+ MockZipFile.return_value.__enter__.return_value = zip_mock
+
+ assert s.decompress_zip("/fake/zip.zip") == 0
+ zip_mock.extract.assert_any_call("config.yaml", "/fake/dir")
+ zip_mock.extract.assert_any_call(".osk", "/fake/dir")
+
+ @patch("connpy.core_plugins.sync.zipfile.ZipFile")
+ @patch("connpy.core_plugins.sync.os.path.dirname")
+ def test_decompress_zip_json_fallback(self, mock_dirname, MockZipFile, mock_connapp):
+ mock_dirname.return_value = "/fake/dir"
+ s = sync(mock_connapp)
+ zip_mock = MagicMock()
+ zip_mock.namelist.return_value = ["config.json", ".osk"]
+ MockZipFile.return_value.__enter__.return_value = zip_mock
+
+ assert s.decompress_zip("/fake/old_zip.zip") == 0
+ zip_mock.extract.assert_any_call("config.json", "/fake/dir")
+
+ @patch.object(sync, "get_credentials")
+ @patch("connpy.core_plugins.sync.build")
+ def test_get_appdata_files(self, mock_build, mock_get_credentials, mock_connapp):
+ mock_get_credentials.return_value = MagicMock()
+ mock_service = MagicMock()
+ mock_build.return_value = mock_service
+
+ mock_service.files().list().execute.return_value = {
+ "files": [
+ {"id": "1", "name": "backup1.zip", "appProperties": {"timestamp": "1000", "date": "2024"}}
+ ]
+ }
+
+ s = sync(mock_connapp)
+ files = s.get_appdata_files()
+ assert len(files) == 1
+ assert files[0]["id"] == "1"
+ assert files[0]["timestamp"] == "1000"
+
+ @patch.object(sync, "get_credentials")
+ @patch("connpy.core_plugins.sync.build")
+ @patch("connpy.core_plugins.sync.MediaFileUpload")
+ @patch("connpy.core_plugins.sync.os.path.basename")
+ def test_backup_file_to_drive(self, mock_basename, mock_media, mock_build, mock_get_credentials, mock_connapp):
+ mock_get_credentials.return_value = MagicMock()
+ mock_basename.return_value = "backup.zip"
+ mock_service = MagicMock()
+ mock_build.return_value = mock_service
+
+ s = sync(mock_connapp)
+ assert s.backup_file_to_drive("/fake/backup.zip", 1234567890000) == 0
+ mock_service.files().create.assert_called_once()
+
+
+Methods
+
+
+def test_backup_file_to_drive(self, mock_basename, mock_media, mock_build, mock_get_credentials, mock_connapp)
+-
+
+
+Expand source code
+
+@patch.object(sync, "get_credentials")
+@patch("connpy.core_plugins.sync.build")
+@patch("connpy.core_plugins.sync.MediaFileUpload")
+@patch("connpy.core_plugins.sync.os.path.basename")
+def test_backup_file_to_drive(self, mock_basename, mock_media, mock_build, mock_get_credentials, mock_connapp):
+ mock_get_credentials.return_value = MagicMock()
+ mock_basename.return_value = "backup.zip"
+ mock_service = MagicMock()
+ mock_build.return_value = mock_service
+
+ s = sync(mock_connapp)
+ assert s.backup_file_to_drive("/fake/backup.zip", 1234567890000) == 0
+ mock_service.files().create.assert_called_once()
+
+
+
+
+def test_compress_specific_files(self, mock_basename, MockZipFile, mock_connapp)
+-
+
+
+Expand source code
+
+@patch("connpy.core_plugins.sync.zipfile.ZipFile")
+@patch("connpy.core_plugins.sync.os.path.basename")
+def test_compress_specific_files(self, mock_basename, MockZipFile, mock_connapp):
+ mock_basename.return_value = "config.yaml"
+ s = sync(mock_connapp)
+ zip_mock = MagicMock()
+ MockZipFile.return_value.__enter__.return_value = zip_mock
+
+ s.compress_specific_files("/fake/zip.zip")
+ zip_mock.write.assert_any_call(s.file, "config.yaml")
+ zip_mock.write.assert_any_call(s.key, ".osk")
+
+
+
+
+def test_decompress_zip_json_fallback(self, mock_dirname, MockZipFile, mock_connapp)
+-
+
+
+Expand source code
+
+@patch("connpy.core_plugins.sync.zipfile.ZipFile")
+@patch("connpy.core_plugins.sync.os.path.dirname")
+def test_decompress_zip_json_fallback(self, mock_dirname, MockZipFile, mock_connapp):
+ mock_dirname.return_value = "/fake/dir"
+ s = sync(mock_connapp)
+ zip_mock = MagicMock()
+ zip_mock.namelist.return_value = ["config.json", ".osk"]
+ MockZipFile.return_value.__enter__.return_value = zip_mock
+
+ assert s.decompress_zip("/fake/old_zip.zip") == 0
+ zip_mock.extract.assert_any_call("config.json", "/fake/dir")
+
+
+
+
+def test_decompress_zip_yaml(self, mock_dirname, MockZipFile, mock_connapp)
+-
+
+
+Expand source code
+
+@patch("connpy.core_plugins.sync.zipfile.ZipFile")
+@patch("connpy.core_plugins.sync.os.path.dirname")
+def test_decompress_zip_yaml(self, mock_dirname, MockZipFile, mock_connapp):
+ mock_dirname.return_value = "/fake/dir"
+ s = sync(mock_connapp)
+ zip_mock = MagicMock()
+ zip_mock.namelist.return_value = ["config.yaml", ".osk"]
+ MockZipFile.return_value.__enter__.return_value = zip_mock
+
+ assert s.decompress_zip("/fake/zip.zip") == 0
+ zip_mock.extract.assert_any_call("config.yaml", "/fake/dir")
+ zip_mock.extract.assert_any_call(".osk", "/fake/dir")
+
+
+
+
+def test_get_appdata_files(self, mock_build, mock_get_credentials, mock_connapp)
+-
+
+
+Expand source code
+
+@patch.object(sync, "get_credentials")
+@patch("connpy.core_plugins.sync.build")
+def test_get_appdata_files(self, mock_build, mock_get_credentials, mock_connapp):
+ mock_get_credentials.return_value = MagicMock()
+ mock_service = MagicMock()
+ mock_build.return_value = mock_service
+
+ mock_service.files().list().execute.return_value = {
+ "files": [
+ {"id": "1", "name": "backup1.zip", "appProperties": {"timestamp": "1000", "date": "2024"}}
+ ]
+ }
+
+ s = sync(mock_connapp)
+ files = s.get_appdata_files()
+ assert len(files) == 1
+ assert files[0]["id"] == "1"
+ assert files[0]["timestamp"] == "1000"
+
+
+
+
+def test_get_credentials_not_found(self, mock_exists, mock_connapp)
+-
+
+
+Expand source code
+
+@patch("connpy.core_plugins.sync.os.path.exists")
+def test_get_credentials_not_found(self, mock_exists, mock_connapp):
+ mock_exists.return_value = False
+ s = sync(mock_connapp)
+ assert s.get_credentials() == 0
+
+
+
+
+def test_get_credentials_success(self, MockCreds, mock_exists, mock_connapp)
+-
+
+
+Expand source code
+
+@patch("connpy.core_plugins.sync.os.path.exists")
+@patch("connpy.core_plugins.sync.Credentials")
+def test_get_credentials_success(self, MockCreds, mock_exists, mock_connapp):
+ mock_exists.return_value = True
+ mock_cred_instance = MagicMock()
+ mock_cred_instance.valid = True
+ MockCreds.from_authorized_user_file.return_value = mock_cred_instance
+
+ s = sync(mock_connapp)
+ creds = s.get_credentials()
+ assert creds == mock_cred_instance
+
+
+
+
+def test_init(self, mock_connapp)
+-
+
+
+Expand source code
+
+def test_init(self, mock_connapp):
+ s = sync(mock_connapp)
+ assert s.sync is True
+ assert s.file == "/fake/dir/config.yaml"
+ assert s.token_file == "/fake/dir/gtoken.json"
+
+
+
+
+
+
+
+
+
+
+
+
+