[IntegTest][Release-3.14] Fix test_dcv in ADC regions by ensuring known_hosts path exists to avoid cat command failure

tests/integration-tests/tests/dcv/test_dcv.py

@@ -12,7 +12,9 @@
 import logging
 import os as operating_system
 import re
+import stat
 import subprocess
+from pathlib import Path
 
 import pytest
 import requests
@@ -157,23 +159,37 @@ def _check_error_cases(remote_command_executor, dcv_authenticator_port):
     )
 
 
-def _test_show_url(cluster, region, dcv_port, access_from, use_login_node=False):
+def _test_show_url(cluster, region, dcv_port, access_from, use_login_node=False):  # noqa: C901
     """Test dcv-connect with --show-url."""
     env = operating_system.environ.copy()
     env["AWS_DEFAULT_REGION"] = region
 
     node_ip = cluster.get_login_node_public_ip() if use_login_node else cluster.head_node_ip
 
-    # add ssh key to jenkins user known hosts file to avoid ssh keychecking prompt
+    # Ensure known_hosts path exists to avoid `cat` command returning non-zero exit when testing in ADC region.
     host_keys_file = operating_system.path.expanduser("~/.ssh/known_hosts")
-    logging.info(f"Add ip address {node_ip} to known hosts file {host_keys_file}")
+    host_keys_path = Path(host_keys_file)
+    try:
+        host_keys_path.parent.mkdir(parents=True, exist_ok=True)
+        if not host_keys_path.exists():
+            host_keys_path.touch()
+            host_keys_path.chmod(stat.S_IRUSR | stat.S_IWUSR)  # 0600
+    except Exception as e:
+        logging.warning(f"Failed to prepare known_hosts file {host_keys_file}: {e}")
 
-    result = subprocess.check_output("cat {0}".format(host_keys_file), shell=True)
-    logging.info(f"Original content of known hosts file {host_keys_file}: {result}")
+    try:
+        before_content = subprocess.check_output(f"cat {host_keys_file}", shell=True)
+    except subprocess.CalledProcessError:
+        before_content = b""
+    logging.info(f"Original content of known hosts file {host_keys_file}: {before_content}")
 
     add_keys_to_known_hosts(node_ip, host_keys_file)
-    result = subprocess.check_output("cat {0}".format(host_keys_file), shell=True)
-    logging.info(f"New content of known hosts file {host_keys_file}: {result}")
+
+    try:
+        after_content = subprocess.check_output(f"cat {host_keys_file}", shell=True)
+    except subprocess.CalledProcessError:
+        after_content = b""
+    logging.info(f"New content of known hosts file {host_keys_file}: {after_content}")
 
     dcv_connect_args = ["pcluster", "dcv-connect", "--cluster-name", cluster.name, "--show-url"]