diff --git a/src/quads_client/commands/cloud.py b/src/quads_client/commands/cloud.py index 8dd019c..9b2bac3 100644 --- a/src/quads_client/commands/cloud.py +++ b/src/quads_client/commands/cloud.py @@ -3,6 +3,7 @@ from tabulate import tabulate from quads_client.error_handler import require_connection +from quads_client.utils import resolve_os class CloudCommands: @@ -342,7 +343,11 @@ def cmd_mod_cloud(self, args): return i += 2 elif parts[i] == "os" and i + 1 < len(parts): - updates["ostype"] = parts[i + 1] + resolved_os, os_error = resolve_os(self.shell.connection.api, parts[i + 1]) + if os_error: + self.shell.perror(os_error) + return + updates["ostype"] = resolved_os i += 2 elif parts[i] == "wipe": updates["wipe"] = True diff --git a/src/quads_client/commands/schedule.py b/src/quads_client/commands/schedule.py index 9df9c43..e051b33 100644 --- a/src/quads_client/commands/schedule.py +++ b/src/quads_client/commands/schedule.py @@ -3,7 +3,7 @@ from quads_client.arg_parser import parse_extend_args, parse_schedule_admin_args, parse_shrink_args from quads_client.error_handler import handle_api_error, require_admin, require_connection -from quads_client.utils import format_schedule_datetime, parse_api_datetime +from quads_client.utils import format_schedule_datetime, parse_api_datetime, resolve_os def parse_flexible_datetime(date_str): @@ -261,7 +261,11 @@ def cmd_schedule_admin(self, args): if parsed["qinq"] is not None: batch_data["qinq"] = parsed["qinq"] if parsed.get("os"): - batch_data["ostype"] = parsed["os"] + resolved_os, os_error = resolve_os(self.shell.connection.api, parsed["os"]) + if os_error: + self.shell.perror(os_error) + return + batch_data["ostype"] = resolved_os try: result = self.shell.connection.api.create_schedules_batch(batch_data) diff --git a/src/quads_client/commands/user.py b/src/quads_client/commands/user.py index e14b3ce..41d8765 100644 --- a/src/quads_client/commands/user.py +++ b/src/quads_client/commands/user.py @@ -10,6 +10,7 @@ extract_cloud_number, extract_hostname, get_username_short, + resolve_os, ) @@ -729,7 +730,11 @@ def cmd_schedule(self, args): if parsed["vlan"]: assignment_data["vlan"] = parsed["vlan"] if parsed["os"]: - assignment_data["ostype"] = parsed["os"] + resolved_os, os_error = resolve_os(self.shell.connection.api, parsed["os"]) + if os_error: + self.shell.perror(os_error) + return + assignment_data["ostype"] = resolved_os # Step 1: Create self-assignment (SSM endpoint auto-assigns cloud) assignment = auto_refresh_on_auth_error( diff --git a/src/quads_client/utils.py b/src/quads_client/utils.py index 70d6109..44fe370 100644 --- a/src/quads_client/utils.py +++ b/src/quads_client/utils.py @@ -271,3 +271,33 @@ def validate_cloud_exists(api, cloud_name: str) -> bool: """ clouds = api.filter_clouds({"name": cloud_name}) return bool(clouds) + + +def resolve_os(api, os_input: str): + """ + Validate and resolve an OS selection against the server's OS list. + + Matches by Title (case-insensitive) or by numeric Id. + Returns (canonical_title, None) on success, (None, error_string) on failure. + """ + os_list = api.get_os_list() + if not os_list: + return None, f"OS '{os_input}' not found" + + os_lower = os_input.lower() + for entry in os_list: + title = entry.get("Title", "") + if title.lower() == os_lower: + return title, None + + if os_input.isdigit(): + os_id = int(os_input) + for entry in os_list: + if entry.get("Id") == os_id: + return entry.get("Title"), None + + available = ", ".join(entry.get("Title", "") for entry in os_list if entry.get("Title")) + error = f"OS '{os_input}' not found" + if available: + error += f"\nAvailable: {available}" + return None, error diff --git a/tests/test_commands_cloud.py b/tests/test_commands_cloud.py index 933919d..bd78be3 100644 --- a/tests/test_commands_cloud.py +++ b/tests/test_commands_cloud.py @@ -233,8 +233,28 @@ def test_mod_cloud_with_os(mock_shell): mock_shell.connection.is_connected = True mock_shell.connection.api.get_active_cloud_assignment.return_value = {"id": 42} mock_shell.connection.api.update_assignment.return_value = {"status": "success"} + mock_shell.connection.api.get_os_list.return_value = [ + {"Id": 1, "Title": "RHEL 9.4", "Release Name": "", "Family": "Redhat"}, + ] cloud_cmd = CloudCommands(mock_shell) cloud_cmd.cmd_mod_cloud('cloud05 os "RHEL 9.4"') mock_shell.connection.api.update_assignment.assert_called_once_with(42, {"ostype": "RHEL 9.4"}) + + +def test_mod_cloud_with_invalid_os(mock_shell): + """Test mod-cloud rejects invalid OS and does not call API""" + mock_shell.connection.is_connected = True + mock_shell.connection.api.get_active_cloud_assignment.return_value = {"id": 42} + mock_shell.connection.api.get_os_list.return_value = [ + {"Id": 1, "Title": "RHEL 9.4", "Release Name": "", "Family": "Redhat"}, + ] + + cloud_cmd = CloudCommands(mock_shell) + cloud_cmd.cmd_mod_cloud('cloud05 os "Windows 11"') + + mock_shell.connection.api.update_assignment.assert_not_called() + mock_shell.perror.assert_called() + error_msg = mock_shell.perror.call_args[0][0] + assert "not found" in error_msg diff --git a/tests/test_commands_unified_schedule.py b/tests/test_commands_unified_schedule.py index 6078015..1346dbd 100644 --- a/tests/test_commands_unified_schedule.py +++ b/tests/test_commands_unified_schedule.py @@ -97,6 +97,9 @@ def test_schedule_ssm_with_os(self, mock_shell): "owner": "alice", } mock_shell.connection.api.create_schedule.return_value = {"id": 1} + mock_shell.connection.api.get_os_list.return_value = [ + {"Id": 1, "Title": "RHEL 9.4", "Release Name": "", "Family": "Redhat"}, + ] user_cmd = UserCommands(mock_shell) user_cmd.cmd_schedule('1 description "Test" os "RHEL 9.4"') @@ -104,6 +107,27 @@ def test_schedule_ssm_with_os(self, mock_shell): call_args = mock_shell.connection.api.create_self_assignment.call_args[0][0] assert call_args["ostype"] == "RHEL 9.4" + def test_schedule_ssm_with_invalid_os(self, mock_shell): + """Test SSM schedule rejects invalid OS and does not call API""" + mock_shell.connection.is_connected = True + mock_shell.connection.is_authenticated = True + mock_shell.connection.is_admin = False + mock_shell.connection.username = "alice@example.com" + mock_shell.connection.api.filter_available.return_value = [ + {"name": "host01.example.com"}, + ] + mock_shell.connection.api.get_os_list.return_value = [ + {"Id": 1, "Title": "RHEL 9.4", "Release Name": "", "Family": "Redhat"}, + ] + + user_cmd = UserCommands(mock_shell) + user_cmd.cmd_schedule('1 description "Test" os "--help"') + + mock_shell.connection.api.create_self_assignment.assert_not_called() + mock_shell.perror.assert_called() + error_msg = mock_shell.perror.call_args[0][0] + assert "not found" in error_msg + def test_schedule_ssm_insufficient_hosts(self, mock_shell): """Test SSM schedule with insufficient available hosts""" mock_shell.connection.is_connected = True @@ -703,6 +727,9 @@ def test_schedule_admin_batch_with_os(self, mock_shell): "schedules_created": 1, "hostnames": ["host01.example.com"], } + mock_shell.connection.api.get_os_list.return_value = [ + {"Id": 1, "Title": "RHEL 9.4", "Release Name": "", "Family": "Redhat"}, + ] schedule_cmd = ScheduleCommands(mock_shell) cmd = ( @@ -713,3 +740,25 @@ def test_schedule_admin_batch_with_os(self, mock_shell): batch_data = mock_shell.connection.api.create_schedules_batch.call_args[0][0] assert batch_data["ostype"] == "RHEL 9.4" + + def test_schedule_admin_batch_with_invalid_os(self, mock_shell): + """Test batch schedule rejects invalid OS and does not call API""" + mock_shell.connection.is_connected = True + mock_shell.connection.is_authenticated = True + mock_shell.connection.is_admin = True + mock_shell.connection.api.filter_clouds.return_value = [{"name": "cloud02"}] + mock_shell.connection.api.get_os_list.return_value = [ + {"Id": 1, "Title": "RHEL 9.4", "Release Name": "", "Family": "Redhat"}, + ] + + schedule_cmd = ScheduleCommands(mock_shell) + cmd = ( + 'cloud02 host01 "2026-05-11 22:00" "2026-06-11 22:00" ' + 'description "Test" cloud-owner jdoe cloud-ticket 123 os "bogus"' + ) + schedule_cmd.cmd_schedule_admin(cmd) + + mock_shell.connection.api.create_schedules_batch.assert_not_called() + mock_shell.perror.assert_called() + error_msg = mock_shell.perror.call_args[0][0] + assert "not found" in error_msg diff --git a/tests/test_utils.py b/tests/test_utils.py index e170a1c..d3cf248 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -12,6 +12,7 @@ get_ssl_indicator, get_ssl_status_text, get_username_short, + resolve_os, validate_cloud_exists, ) @@ -485,3 +486,75 @@ def test_validate_cloud_exists_false(): assert result is False mock_api.filter_clouds.assert_called_once_with({"name": "cloud99"}) + + +OS_LIST = [ + {"Id": 1, "Title": "Rocky Linux 9.8", "Release Name": "", "Family": "Redhat"}, + {"Id": 2, "Title": "Fedora 42", "Release Name": "", "Family": "Redhat"}, + {"Id": 3, "Title": "RHEL 10.0", "Release Name": "", "Family": "Redhat"}, +] + + +def test_resolve_os_exact_title(): + mock_api = MagicMock() + mock_api.get_os_list.return_value = OS_LIST + title, error = resolve_os(mock_api, "RHEL 10.0") + assert title == "RHEL 10.0" + assert error is None + + +def test_resolve_os_case_insensitive(): + mock_api = MagicMock() + mock_api.get_os_list.return_value = OS_LIST + title, error = resolve_os(mock_api, "rhel 10.0") + assert title == "RHEL 10.0" + assert error is None + title, error = resolve_os(mock_api, "fedora 42") + assert title == "Fedora 42" + assert error is None + + +def test_resolve_os_by_id(): + mock_api = MagicMock() + mock_api.get_os_list.return_value = OS_LIST + title, error = resolve_os(mock_api, "2") + assert title == "Fedora 42" + assert error is None + title, error = resolve_os(mock_api, "3") + assert title == "RHEL 10.0" + assert error is None + + +def test_resolve_os_no_match(): + mock_api = MagicMock() + mock_api.get_os_list.return_value = OS_LIST + title, error = resolve_os(mock_api, "--help") + assert title is None + assert "not found" in error + assert "Available:" in error + + +def test_resolve_os_no_match_shows_available(): + mock_api = MagicMock() + mock_api.get_os_list.return_value = OS_LIST + title, error = resolve_os(mock_api, "Windows 11") + assert title is None + assert "Rocky Linux 9.8" in error + assert "Fedora 42" in error + assert "RHEL 10.0" in error + + +def test_resolve_os_empty_list(): + mock_api = MagicMock() + mock_api.get_os_list.return_value = [] + title, error = resolve_os(mock_api, "RHEL 10.0") + assert title is None + assert "not found" in error + + +def test_resolve_os_none_list(): + mock_api = MagicMock() + mock_api.get_os_list.return_value = None + title, error = resolve_os(mock_api, "RHEL 10.0") + assert title is None + assert "not found" in error