Merge pull request #164 from sadsfae/fixos

fix: validate OS selection in schedule/mod-cloud

Author: sadsfae

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

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)

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(

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

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

tests/test_commands_unified_schedule.py

@@ -97,13 +97,37 @@ 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"')
 
         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

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