[Test] In test_dcv_configuration surface dcv-connect errors in assertion error message. Also added a locking mechanism to serialize the modifications on known_hosts files and avoid interference between concurrent test execution.

Author: gmarciani

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

@@ -9,6 +9,8 @@
 # or in the "LICENSE.txt" file accompanying this file.
 # This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, express or implied.
 # See the License for the specific language governing permissions and limitations under the License.
+import contextlib
+import fcntl
 import json
 import logging
 import os as operating_system
@@ -241,14 +243,26 @@ def _check_error_cases(remote_command_executor, dcv_authenticator_port):
     logging.info(f"Completed checks for authentication failure on {remote_command_executor.target}")
 
 
+@contextlib.contextmanager
+def _temporary_known_host(hostname, host_keys_file, env):
+    """Add SSH host keys for hostname, yield, then remove them. Serialized via file lock across processes."""
+    lock_file = host_keys_file + ".lock"
+    with open(lock_file, "w") as lf:
+        fcntl.flock(lf, fcntl.LOCK_EX)
+        try:
+            add_keys_to_known_hosts(hostname, host_keys_file)
+            yield
+        finally:
+            remove_keys_from_known_hosts(hostname, host_keys_file, env=env)
+
+
 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")
     host_keys_path = Path(host_keys_file)
@@ -260,18 +274,18 @@ def _test_show_url(cluster, region, dcv_port, access_from, use_login_node=False)
     except Exception as e:
         logging.warning(f"Failed to prepare known_hosts file {host_keys_file}: {e}")
 
-    add_keys_to_known_hosts(node_ip, host_keys_file)
-
     dcv_connect_args = ["pcluster", "dcv-connect", "--cluster-name", cluster.name, "--show-url"]
 
     if use_login_node:
         dcv_connect_args.extend(["--login-node-ip", node_ip])
 
-    try:
-        result = run_pcluster_command(dcv_connect_args, env=env)
-    finally:
-        # remove ssh key from jenkins user known hosts file
-        remove_keys_from_known_hosts(node_ip, host_keys_file, env=env)
+    with _temporary_known_host(node_ip, host_keys_file, env):
+        try:
+            result = run_pcluster_command(dcv_connect_args, env=env)
+        except subprocess.CalledProcessError as e:
+            raise AssertionError(
+                f"Command {e.cmd} failed (exit {e.returncode}).\nstderr: {e.stderr}\nstdout: {e.stdout}"
+            ) from e
 
     assert_that(result.stdout).matches(
         r"Please use the following one-time URL in your browser within 30 seconds:\n"