add test & decorator

test/decorators.py

@@ -49,6 +49,20 @@ def _wrapper(self, service):
     return _wrapper
 
 
+def cloud_only(func):
+    """Decorator that runs a test using only ibm_cloud services."""
+
+    @wraps(func)
+    def _wrapper(self, service):
+        if service._channel != "ibm_cloud":
+            raise SkipTest(
+                f"Skipping integration test. {self} does not support channel type {service._channel}"
+            )
+        func(self, service)
+
+    return _wrapper
+
+
 def run_quantum_and_cloud_fake(func):
     """Decorator that runs a test using both quantum and cloud fake services."""
 

test/integration/test_qctrl.py

@@ -12,13 +12,15 @@
 
 """Tests for job functions using real runtime service."""
 
-
+import time
 from qiskit.test.reference_circuits import ReferenceCircuits
+from qiskit.providers.jobstatus import JobStatus
 
-from qiskit_ibm_runtime import Sampler, Session
+from qiskit_ibm_runtime import Sampler, Session, Options
 
 from ..ibm_test_case import IBMIntegrationJobTestCase
-from ..decorators import run_integration_test
+from ..decorators import run_integration_test, cloud_only
+from ..utils import cancel_job_safe
 
 
 class TestQCTRL(IBMIntegrationJobTestCase):
@@ -28,13 +30,35 @@ def setUp(self) -> None:
         super().setUp()
         self.bell = ReferenceCircuits.bell()
         self.backend = "ibmq_qasm_simulator"
+        # does q-ctrl work with simulator?
+        # self.dependencies.service.least_busy(simulator=False)
 
     @run_integration_test
+    @cloud_only
     def test_qctrl(self, service):
-        """Test qctrl."""
+        """Test simple bell circuit."""
         service._channel_strategy = "q-ctrl"
         with Session(service, self.backend) as session:
-            sampler = Sampler(session=session)
+            options = Options(resilience_level=1)
+            sampler = Sampler(session=session, options=options)
 
             result = sampler.run(circuits=self.bell).result()
             self.assertTrue(result)
+
+    @run_integration_test
+    @cloud_only
+    def test_cancel_qctrl_job(self, service):
+        """Test canceling qctrl job."""
+        service._channel_strategy = "q-ctrl"
+
+        with Session(service, self.backend) as session:
+            options = Options(resilience_level=1)
+            sampler = Sampler(session=session, options=options)
+
+            job = sampler.run([self.bell] * 10)
+
+        rjob = service.job(job.job_id())
+        if not cancel_job_safe(rjob, self.log):
+            return
+        time.sleep(5)
+        self.assertEqual(rjob.status(), JobStatus.CANCELLED)