Fix failing eagle integration tests (#1980)

* (WIP) Test integration tests in fork

* Update test_cancel_job_queued

* Update tests

---------

Co-authored-by: ptristan3 <[email protected]>

test/integration/test_estimator_v2.py

@@ -50,7 +50,7 @@ def test_estimator_v2_session(self, service):
         theta2 = [0, 1, 1, 2, 3, 5, 8, 13]
         theta3 = [1, 2, 3, 4, 5, 6]
 
-        with Session(service, self.dependencies.qpu) as session:
+        with Session(service, self._backend) as session:
             estimator = EstimatorV2(mode=session)
 
             job = estimator.run([(psi1, H1, [theta1])])

test/integration/test_ibm_job.py

@@ -122,7 +122,18 @@ def test_retrieve_completed_jobs(self):
             backend_name=self.sim_backend.name, limit=3, pending=False
         )
         for job in completed_job_list:
-            self.assertTrue(job.status() in ["DONE", "CANCELLED", "ERROR"])
+            self.assertTrue(
+                job.status()
+                # Update when RuntimeJob is removed in favor of RuntimeJobV2
+                in [
+                    "DONE",
+                    "CANCELLED",
+                    "ERROR",
+                    JobStatus.DONE,
+                    JobStatus.CANCELLED,
+                    JobStatus.ERROR,
+                ]
+            )
 
     def test_retrieve_pending_jobs(self):
         """Test retrieving jobs with the pending filter."""
@@ -162,7 +173,16 @@ def test_retrieve_jobs_status(self):
 
         for job in backend_jobs:
             self.assertTrue(
-                job.status() in ["DONE", "CANCELLED", "ERROR"],
+                job.status()
+                # Update when RuntimeJob is removed in favor of RuntimeJobV2
+                in [
+                    "DONE",
+                    "CANCELLED",
+                    "ERROR",
+                    JobStatus.DONE,
+                    JobStatus.CANCELLED,
+                    JobStatus.ERROR,
+                ],
                 "Job {} has status {} when it should be DONE, CANCELLED, or ERROR".format(
                     job.job_id(), job.status()
                 ),

test/integration/test_job.py

@@ -68,9 +68,10 @@ def test_run_program_log_level(self, service):
     @production_only
     def test_cancel_job_queued(self, service):
         """Test canceling a queued job."""
-        real_device_name = get_real_device(service)
-        _ = self._run_program(service, circuits=[(bell(),)] * 10, backend=real_device_name)
-        job = self._run_program(service, circuits=[(bell(),)] * 2, backend=real_device_name)
+        _ = self._run_program(
+            service,
+        )
+        job = self._run_program(service)
         wait_for_status(job, "QUEUED")
         if not cancel_job_safe(job, self.log):
             return

test/integration/test_noise_learner.py

@@ -13,6 +13,7 @@
 """Integration tests for NoiseLearner."""
 
 from copy import deepcopy
+from unittest import SkipTest
 
 from qiskit.circuit import QuantumCircuit
 
@@ -55,6 +56,8 @@ def setUp(self) -> None:
     @run_integration_test
     def test_with_default_options(self, service):  # pylint: disable=unused-argument
         """Test noise learner with default options."""
+        if self._backend.simulator:
+            raise SkipTest("Simulator does not have coupling map")
         options = NoiseLearnerOptions()
         learner = NoiseLearner(mode=self._backend, options=options)
 
@@ -65,6 +68,8 @@ def test_with_default_options(self, service):  # pylint: disable=unused-argument
     @run_integration_test
     def test_with_non_default_options(self, service):  # pylint: disable=unused-argument
         """Test noise learner with non-default options."""
+        if self._backend.simulator:
+            raise SkipTest("Simulator does not have coupling map")
         options = NoiseLearnerOptions()
         options.max_layers_to_learn = 1
         options.layer_pair_depths = [0, 1]
@@ -80,6 +85,8 @@ def test_with_non_default_options(self, service):  # pylint: disable=unused-argu
     @run_integration_test
     def test_with_no_layers(self, service):  # pylint: disable=unused-argument
         """Test noise learner when `max_layers_to_learn` is `0`."""
+        if self._backend.simulator:
+            raise SkipTest("Skip if simulator backend is used.")
         options = NoiseLearnerOptions()
         options.max_layers_to_learn = 0
         learner = NoiseLearner(mode=self._backend, options=options)
@@ -95,6 +102,8 @@ def test_with_no_layers(self, service):  # pylint: disable=unused-argument
     @run_integration_test
     def test_learner_plus_estimator(self, service):  # pylint: disable=unused-argument
         """Test feeding noise learner data to estimator."""
+        if self._backend.simulator:
+            raise SkipTest("Simulator does not have coupling map")
 
         options = EstimatorOptions()
         options.resilience.zne_mitigation = True  # pylint: disable=assigning-non-slot

test/integration/test_retrieve_job.py

@@ -67,7 +67,6 @@ def test_retrieve_all_jobs(self, service):
         for rjob in rjobs:
             if rjob.job_id() == job.job_id():
                 self.assertEqual(job.primitive_id, rjob.primitive_id)
-                self.assertEqual(job.status(), rjob.status())
                 found = True
                 break
         self.assertTrue(found, f"Job {job.job_id()} not returned.")
@@ -115,7 +114,6 @@ def test_retrieve_returned_jobs(self, service):
         for rjob in rjobs:
             if rjob.job_id() == job.job_id():
                 self.assertEqual(job.primitive_id, rjob.primitive_id)
-                self.assertEqual(job.status(), rjob.status())
                 found = True
                 break
         self.assertTrue(found, f"Returned job {job.job_id()} not retrieved.")

test/integration/test_sampler_v2.py

@@ -81,7 +81,7 @@ def setUp(self):
     def test_sampler_run(self, service):
         """Test Sampler.run()."""
 
-        with Session(service, self.dependencies.qpu) as session:
+        with Session(service, self._backend) as session:
             _, _, target = self._cases[1]
             with self.subTest("single"):
                 sampler = Sampler(mode=session, options=self._options)
@@ -162,7 +162,7 @@ def test_run_2qubit(self):
     def test_run_single_circuit(self, service):
         """Test for single circuit case."""
         pm = generate_preset_pass_manager(optimization_level=1, target=self._backend.target)
-        with Session(service, self.dependencies.qpu) as session:
+        with Session(service, self._backend) as session:
             sampler = Sampler(mode=session, options=self._options)
 
             with self.subTest("No parameter"):
@@ -253,7 +253,7 @@ def test_run_empty_parameter(self, service):
     @run_integration_test
     def test_run_numpy_params(self, service):
         """Test for numpy array as parameter values"""
-        with Session(service, self.dependencies.qpu) as session:
+        with Session(service, self._backend) as session:
             qc = RealAmplitudes(num_qubits=2, reps=2)
             qc.measure_all()
             qc = transpile(circuits=qc, backend=self._backend)
@@ -276,7 +276,7 @@ def test_run_numpy_params(self, service):
     @run_integration_test
     def test_run_with_shots_option(self, service):
         """test with shots option."""
-        with Session(service, self.dependencies.qpu) as session:
+        with Session(service, self._backend) as session:
             _, _, _ = self._cases[1]
             shots = 100
             with self.subTest("init option"):
@@ -363,7 +363,7 @@ def test_circuit_with_unitary(self, service):
         """Test for circuit with unitary gate."""
         pm = generate_preset_pass_manager(optimization_level=1, target=self._backend.target)
 
-        with Session(service, self.dependencies.qpu) as session:
+        with Session(service, self._backend) as session:
             with self.subTest("identity"):
                 gate = UnitaryGate(np.eye(2))
 
@@ -398,7 +398,7 @@ def test_metadata(self):
     @run_integration_test
     def test_circuit_with_multiple_cregs(self, service):
         """Test for circuit with multiple classical registers."""
-        with Session(service, self.dependencies.qpu) as session:
+        with Session(service, self._backend) as session:
             cases = []
             pm = generate_preset_pass_manager(optimization_level=1, target=self._backend.target)