diff --git a/requirements-testing.txt b/requirements-testing.txt
index d2e09051..26b875de 100644
--- a/requirements-testing.txt
+++ b/requirements-testing.txt
@@ -1,7 +1,7 @@
 backoff == 2.2.*
 coverage[toml] ~= 7.6
 coverage-conditional-plugin == 0.9.*
-deadline-cloud-test-fixtures == 0.16.*
+deadline-cloud-test-fixtures == 0.17.*
 flaky == 3.8.*
 pytest ~= 8.3
 pytest-cov == 5.0.*
diff --git a/test/e2e/test_job_submissions.py b/test/e2e/test_job_submissions.py
index 15d36fc9..b4f7cba1 100644
--- a/test/e2e/test_job_submissions.py
+++ b/test/e2e/test_job_submissions.py
@@ -65,7 +65,7 @@ def test_success(
         os.environ["OPERATING_SYSTEM"] == "windows",
         reason="Linux specific queue crendentials test",
     )
-    def test_queue_credentials_file_is_secure(
+    def test_queue_credentials_file_is_secure_from_other_users(
         self,
         deadline_resources,
         session_worker: EC2InstanceWorker,
@@ -73,7 +73,7 @@ def test_queue_credentials_file_is_secure(
         generic_non_queue_job_user: PosixSessionUser,
         deadline_client: DeadlineClient,
     ) -> None:
-        # Test to verify that the queue credentials can never be accessed by a different queue or user
+        # Test to verify that the queue credentials can never be accessed by a different user on the same machine
 
         job = submit_custom_job(
             "Test Sleep",
@@ -126,6 +126,61 @@ def sessions_exist(current_job: Job) -> bool:
 
             assert check_other_user_cannot_access_credentials_result.exit_code != 0
 
+        finally:
+            deadline_client.update_job(
+                farmId=job.farm.id,
+                queueId=job.queue.id,
+                jobId=job.id,
+                targetTaskRunStatus="CANCELED",
+            )
+            job.wait_until_complete(client=deadline_client)
+
+        return
+
+    @pytest.mark.skipif(
+        os.environ["OPERATING_SYSTEM"] == "windows",
+        reason="Linux specific queue crendentials test",
+    )
+    def test_queue_credentials_file_is_secure_from_other_queues(
+        self,
+        deadline_resources,
+        session_worker: EC2InstanceWorker,
+        deadline_client: DeadlineClient,
+    ) -> None:
+        # Test to verify that the queue credentials can never be accessed by a different queue's job user
+
+        job = submit_sleep_job(
+            "Test Sleep",
+            deadline_client,
+            deadline_resources.farm,
+            deadline_resources.queue_a,
+        )
+        try:
+
+            @backoff.on_predicate(
+                wait_gen=backoff.constant,
+                max_time=120,
+                interval=10,
+            )
+            def is_job_started(current_job: Job) -> bool:
+                current_job.refresh_job_info(client=deadline_client)
+                LOG.info(f"Waiting for job {current_job.id} to be created")
+                return current_job.lifecycle_status != "CREATE_IN_PROGRESS"
+
+            assert is_job_started(job)
+
+            @backoff.on_predicate(backoff.constant, interval=5, max_time=60)
+            def sessions_exist(current_job: Job) -> bool:
+                sessions: list[dict[str, Any]] = deadline_client.list_sessions(
+                    farmId=current_job.farm.id, queueId=current_job.queue.id, jobId=current_job.id
+                ).get("sessions")
+
+                return len(sessions) > 0
+
+            assert sessions_exist(job)
+
+            queue_credentials_directory = f"/var/lib/deadline/queues/{job.queue.id}"
+
             # Verify that another queue's user cannot access the credentials file through a job
             second_queue_job = submit_custom_job(
                 "Test Getting Primary Queue Credentials File",
diff --git a/test/e2e/test_worker_config.py b/test/e2e/test_worker_config.py
index 45d6c225..866bc1a5 100644
--- a/test/e2e/test_worker_config.py
+++ b/test/e2e/test_worker_config.py
@@ -5,6 +5,7 @@
 """
 import logging
 import os
+from time import sleep
 from typing import Any, Callable, Optional
 import backoff
 import boto3
@@ -21,6 +22,64 @@
 @pytest.mark.parametrize("operating_system", [os.environ["OPERATING_SYSTEM"]], indirect=True)
 class TestWorkerConfiguration:
 
+    def test_worker_requires_no_instance_profile(
+        self,
+        deadline_resources,
+        deadline_client: DeadlineClient,
+        worker_config: DeadlineWorkerConfiguration,
+        function_worker_factory: Callable[[DeadlineWorkerConfiguration], EC2InstanceWorker],
+    ) -> None:
+
+        # Create a EC2 worker with disallow-instance-profiles option for the worker agent
+        # Note that the EC2 instance is created with an instance profile, so no job will ever be picked up by this worker
+        function_worker_factory(
+            dataclasses.replace(
+                worker_config,
+                disallow_instance_profile="True",
+                fleet=deadline_resources.scaling_fleet,
+            )
+        )
+
+        # Check that the worker agent will eventually shut down
+        job = submit_sleep_job(
+            "Test Job with worker instance profiles disallowed",
+            deadline_client,
+            deadline_resources.farm,
+            deadline_resources.queue_a,
+        )
+        try:
+            # Wait until the job is finished creation
+
+            @backoff.on_exception(
+                exception=Exception,
+                wait_gen=backoff.constant,
+                max_time=120,
+                interval=10,
+            )
+            def check_job_created() -> None:
+                job.refresh_job_info(client=deadline_client)
+                assert job.lifecycle_status != "CREATE_IN_PROGRESS"
+
+            check_job_created()
+
+            # Sleep 30 seconds to allow the worker to pick up the job, the worker will not pick up the job due to the instance profile
+            sleep(30)
+
+            def check_job_not_picked_up() -> None:
+                # Check that the job is never picked up, since the worker will not pick up any jobs due to the instance profile
+                job.refresh_job_info(client=deadline_client)
+                assert job.task_run_status in ["PENDING", "READY"]
+
+            check_job_not_picked_up()
+
+        finally:
+            deadline_client.update_job(
+                farmId=job.farm.id,
+                queueId=job.queue.id,
+                jobId=job.id,
+                targetTaskRunStatus="CANCELED",
+            )
+
     def test_worker_local_session_logs_can_be_turned_off(
         self,
         deadline_resources,