Module connpy.tests.test_connapp

Functions

def app(populated_config)

Expand source code

@pytest.fixture
def app(populated_config):
    """Returns an instance of connapp initialized with the mock config."""
    return connapp(populated_config)

Returns an instance of connapp initialized with the mock config.

def test_ai(mock_status, mock_ask, app)

Expand source code

@patch("connpy.services.ai_service.AIService.ask")
@patch("connpy.connapp.console.status")
def test_ai(mock_status, mock_ask, app):
    mock_ask.return_value = {"response": "AI output", "usage": {"total": 10, "input": 5, "output": 5}}
    
    app.start(["ai", "--engineer-api-key", "testkey", "how are you"])
    mock_ask.assert_called_once()

def test_ai_list(mock_list_sessions, app)

Expand source code

@patch("connpy.services.ai_service.AIService.list_sessions")
def test_ai_list(mock_list_sessions, app):
    mock_list_sessions.return_value = [{"id": "1", "title": "t", "created_at": "now", "model": "m"}]
    app.start(["ai", "--list"])
    mock_list_sessions.assert_called_once()

def test_api_debug(mock_status, app)

Expand source code

@patch("connpy.services.system_service.SystemService.get_api_status")
def test_api_debug(mock_status, app):
    mock_status.return_value = {"running": False}
    app.services.system.debug_api = MagicMock()
    app.start(["api", "-d", "8080"])
    app.services.system.debug_api.assert_called_once_with(port=8080)

def test_api_start(mock_status, app)

Expand source code

@patch("connpy.services.system_service.SystemService.get_api_status")
def test_api_start(mock_status, app):
    mock_status.return_value = {"running": False}
    app.services.system.start_api = MagicMock()
    app.start(["api", "-s", "8080"])
    app.services.system.start_api.assert_called_once_with(port=8080)

def test_api_stop(mock_status, app)

Expand source code

@patch("connpy.services.system_service.SystemService.get_api_status")
def test_api_stop(mock_status, app):
    mock_status.return_value = {"running": True, "pid": "1234"}
    app.services.system.stop_api = MagicMock(return_value=True)
    app.start(["api", "-x"])
    app.services.system.stop_api.assert_called_once()

def test_bulk(mock_bulk_add, mock_q_bulk, app)

Expand source code

@patch("connpy.cli.forms.Forms.questions_bulk")
@patch("connpy.services.node_service.NodeService.bulk_add")
def test_bulk(mock_bulk_add, mock_q_bulk, app):
    mock_q_bulk.return_value = {"ids": "node1", "host": "host1", "location": ""}
    mock_bulk_add.return_value = 1
    app.start(["bulk"])
    mock_bulk_add.assert_called_once()

def test_config(mock_update_setting, app)

Expand source code

@patch("connpy.services.config_service.ConfigService.update_setting")
def test_config(mock_update_setting, app):
    app.start(["config", "--allow-uppercase", "true"])
    mock_update_setting.assert_called_with("case", True)

def test_config_folder(mock_set_config_folder, app)

Expand source code

@patch("connpy.services.config_service.ConfigService.set_config_folder")
def test_config_folder(mock_set_config_folder, app):
    app.start(["config", "--configfolder", "/new/path"])
    mock_set_config_folder.assert_called_once_with("/new/path")

def test_config_various(mock_update_setting, app)

Expand source code

@patch("connpy.services.config_service.ConfigService.update_setting")
def test_config_various(mock_update_setting, app):
    app.start(["config", "--fzf", "true"])
    mock_update_setting.assert_called_with("fzf", True)
    app.start(["config", "--keepalive", "60"])
    mock_update_setting.assert_called_with("idletime", 60)

def test_connapp_init(app, populated_config)

Expand source code

def test_connapp_init(app, populated_config):
    """Test that connapp initializes correctly with config."""
    assert app.config == populated_config
    assert app.case == populated_config.config.get("case", False)

Test that connapp initializes correctly with config.

def test_copy(mock_move_node, app)

Expand source code

@patch("connpy.services.node_service.NodeService.move_node")
def test_copy(mock_move_node, app):
    app.start(["copy", "src_node", "dst_node"])
    mock_move_node.assert_called_once_with("src_node", "dst_node", copy=True)

def test_export(mock_export, app)

Expand source code

@patch("connpy.services.import_export_service.ImportExportService.export_to_file")
def test_export(mock_export, app):
    with pytest.raises(SystemExit):
        app.start(["export", "file.yml", "@folder1"])
    mock_export.assert_called_once_with("file.yml", folders=["@folder1"])

def test_import(mock_import, mock_prompt, mock_exists, app)

Expand source code

@patch("os.path.exists")
@patch("inquirer.prompt")
@patch("connpy.services.import_export_service.ImportExportService.import_from_file")
def test_import(mock_import, mock_prompt, mock_exists, app):
    mock_exists.return_value = True
    mock_prompt.return_value = {"import": True}
    app.start(["import", "file.yml"])
    mock_import.assert_called_once_with("file.yml")

def test_list_folders(mock_list_folders, app)

Expand source code

@patch("connpy.services.node_service.NodeService.list_folders")
def test_list_folders(mock_list_folders, app):
    mock_list_folders.return_value = ["folder1"]
    app.start(["list", "folders"])
    # Called during init and during the list command
    assert mock_list_folders.call_count >= 2

def test_move(mock_move_node, app)

Expand source code

@patch("connpy.services.node_service.NodeService.move_node")
def test_move(mock_move_node, app):
    app.start(["move", "src_node", "dst_node"])
    mock_move_node.assert_called_once_with("src_node", "dst_node", copy=False)

def test_node_add(mock_func_node, app)

Expand source code

@patch("connpy.cli.node_handler.NodeHandler.dispatch")
def test_node_add(mock_func_node, app):
    """Test that 'node -a' command correctly parses."""
    app.start(["node", "-a", "new_router"])
    mock_func_node.assert_called_once()
    args = mock_func_node.call_args[0][0]
    assert args.data == "new_router"
    assert args.action == "add"

Test that 'node -a' command correctly parses.

def test_node_default(mock_func_node, app)

Expand source code

@patch("connpy.cli.node_handler.NodeHandler.dispatch")
def test_node_default(mock_func_node, app):
    """Test that default 'node' command correctly parses and calls _func_node."""
    app.start(["node", "router1"])
    mock_func_node.assert_called_once()
    args = mock_func_node.call_args[0][0]
    assert args.data == "router1"
    assert args.action == "connect"

Test that default 'node' command correctly parses and calls _func_node.

def test_node_del(mock_prompt, mock_delete_node, mock_list_nodes, app)

Expand source code

@patch("connpy.services.node_service.NodeService.list_nodes")
@patch("connpy.services.node_service.NodeService.delete_node")
@patch("inquirer.prompt")
def test_node_del(mock_prompt, mock_delete_node, mock_list_nodes, app):
    mock_list_nodes.return_value = ["router1"]
    mock_prompt.return_value = {"delete": True}
    app.start(["node", "-r", "router1"])
    mock_delete_node.assert_called_once_with("router1", is_folder=False)

def test_node_list(mock_list_nodes, app)

Expand source code

@patch("connpy.services.node_service.NodeService.list_nodes")
def test_node_list(mock_list_nodes, app):
    """Test 'list nodes' invokes node service."""
    mock_list_nodes.return_value = ["router1", "server1"]
    app.start(["list", "nodes"])
    # Should be called during init and during the list command
    assert mock_list_nodes.call_count >= 2

Test 'list nodes' invokes node service.

def test_node_mod(mock_q_nodes, mock_q_edit, mock_update_node, mock_get_details, mock_list_nodes, app)

Expand source code

@patch("connpy.services.node_service.NodeService.list_nodes")
@patch("connpy.services.node_service.NodeService.get_node_details")
@patch("connpy.services.node_service.NodeService.update_node")
@patch("connpy.cli.forms.Forms.questions_edit")
@patch("connpy.cli.forms.Forms.questions_nodes")
def test_node_mod(mock_q_nodes, mock_q_edit, mock_update_node, mock_get_details, mock_list_nodes, app):
    mock_list_nodes.return_value = ["router1"]
    mock_get_details.return_value = {"host": "1.1.1.1", "port": 22}
    mock_q_edit.return_value = {"host": True}
    mock_q_nodes.return_value = {"host": "2.2.2.2", "port": 22}

    app.start(["node", "-e", "router1"])
    mock_update_node.assert_called_once()

def test_node_show(mock_data, app)

Expand source code

@patch("connpy.printer.data")
def test_node_show(mock_data, app):
    app.nodes_list = ["router1"]
    app.config.getitem = MagicMock(return_value={"host": "1.1.1.1"})
    app.start(["node", "-s", "router1"])
    mock_data.assert_called()

def test_plugin_add(mock_verify, mock_copy, mock_exists, app)

Expand source code

@patch("os.path.exists")
@patch("shutil.copy2")
@patch("connpy.plugins.Plugins.verify_script")
def test_plugin_add(mock_verify, mock_copy, mock_exists, app):
    def mock_exists_side_effect(path):
        if "testplug.py" in path: return False
        if "testplug.py.bkp" in path: return False
        if "file.py" in path: return True
        return True
    mock_exists.side_effect = mock_exists_side_effect
    mock_verify.return_value = None
    app.commands = []
    app.start(["plugin", "--add", "testplug", "file.py"])
    mock_copy.assert_called()

def test_plugin_delete(mock_delete, app)

Expand source code

@patch("connpy.services.plugin_service.PluginService.delete_plugin")
def test_plugin_delete(mock_delete, app):
    app.start(["plugin", "--del", "testplug"])
    mock_delete.assert_called_once_with("testplug")

def test_plugin_disable(mock_disable, app)

Expand source code

@patch("connpy.services.plugin_service.PluginService.disable_plugin")
def test_plugin_disable(mock_disable, app):
    app.start(["plugin", "--disable", "testplug"])
    mock_disable.assert_called_once_with("testplug")

def test_plugin_enable(mock_enable, app)

Expand source code

@patch("connpy.services.plugin_service.PluginService.enable_plugin")
def test_plugin_enable(mock_enable, app):
    app.start(["plugin", "--enable", "testplug"])
    mock_enable.assert_called_once_with("testplug")

def test_plugin_list(mock_list_plugins, app)

Expand source code

@patch("connpy.services.plugin_service.PluginService.list_plugins")
def test_plugin_list(mock_list_plugins, app):
    mock_list_plugins.return_value = {"testplug": {"enabled": True}}
    app.start(["plugin", "--list"])
    mock_list_plugins.assert_called_once()

def test_profile_add(mock_q_profiles, mock_add_profile, mock_list_profiles, app)

Expand source code

@patch("connpy.services.profile_service.ProfileService.list_profiles")
@patch("connpy.services.profile_service.ProfileService.add_profile")
@patch("connpy.cli.forms.Forms.questions_profiles")
def test_profile_add(mock_q_profiles, mock_add_profile, mock_list_profiles, app):
    mock_list_profiles.return_value = ["default"]
    mock_q_profiles.return_value = {"host": "test"}
    app.start(["profile", "-a", "new_profile"])
    mock_add_profile.assert_called_once_with("new_profile", {"host": "test"})

def test_profile_del(mock_prompt, mock_delete_profile, mock_get_profile, app)

Expand source code

@patch("connpy.services.profile_service.ProfileService.get_profile")
@patch("connpy.services.profile_service.ProfileService.delete_profile")
@patch("inquirer.prompt")
def test_profile_del(mock_prompt, mock_delete_profile, mock_get_profile, app):
    mock_get_profile.return_value = {"host": "test"}
    mock_prompt.return_value = {"delete": True}
    app.start(["profile", "-r", "test_profile"])
    mock_delete_profile.assert_called_once_with("test_profile")

def test_profile_list(mock_print, mock_list_profiles, app)

Expand source code

@patch("connpy.services.profile_service.ProfileService.list_profiles")
@patch("connpy.connapp.printer.console.print")
def test_profile_list(mock_print, mock_list_profiles, app):
    """Test 'profile list' invokes profile service correctly."""
    mock_list_profiles.return_value = ["default", "office-user"]
    app.start(["list", "profiles"])
    assert mock_list_profiles.call_count >= 2

Test 'profile list' invokes profile service correctly.

def test_profile_mod(mock_q_profiles, mock_q_edit, mock_update_profile, mock_get_profile, app)

Expand source code

@patch("connpy.services.profile_service.ProfileService.get_profile")
@patch("connpy.services.profile_service.ProfileService.update_profile")
@patch("connpy.cli.forms.Forms.questions_edit")
@patch("connpy.cli.forms.Forms.questions_profiles")
def test_profile_mod(mock_q_profiles, mock_q_edit, mock_update_profile, mock_get_profile, app):
    mock_get_profile.return_value = {"host": "test", "port": 22}
    mock_q_edit.return_value = {"host": True}
    mock_q_profiles.return_value = {"id": "test_profile", "host": "new_host", "port": 22}
    app.start(["profile", "-e", "test_profile"])
    mock_update_profile.assert_called_once_with("test_profile", {"id": "test_profile", "host": "new_host", "port": 22})

def test_profile_show(mock_data, mock_get_profile, app)

Expand source code

@patch("connpy.services.profile_service.ProfileService.get_profile")
@patch("connpy.printer.data")
def test_profile_show(mock_data, mock_get_profile, app):
    mock_get_profile.return_value = {"host": "test"}
    app.start(["profile", "-s", "test_profile"])
    mock_data.assert_called()

def test_run(mock_run_commands, app)

Expand source code

@patch("connpy.services.execution_service.ExecutionService.run_commands")
def test_run(mock_run_commands, app):
    app.start(["run", "node1", "command1", "command2"])
    mock_run_commands.assert_called_once()
    assert mock_run_commands.call_args[1]["nodes_filter"] == "node1"
    assert mock_run_commands.call_args[1]["commands"] == ["command1 command2"]

def test_type_node_reserved_word(app)

Expand source code

def test_type_node_reserved_word(app):
    app.commands = ["bulk", "ai", "run"]
    with patch("sys.argv", ["connpy", "node", "-a", "bulk"]):
        with pytest.raises(SystemExit) as exc:
            app._type_node("bulk")
        assert exc.value.code == 2
    
    # In move/copy it also raises because destination cannot be reserved
    with patch("sys.argv", ["connpy", "mv", "test1", "bulk"]):
        with pytest.raises(SystemExit) as exc:
            app._type_node("bulk")
        assert exc.value.code == 2