From 81fe57f06c23e3b8ad38815d19d8ff6f6477cae0 Mon Sep 17 00:00:00 2001 From: "Diego M. Rodriguez" Date: Thu, 30 Jul 2026 19:54:04 +0200 Subject: [PATCH 1/6] Add support for jobs in registry --- test/registries.py | 127 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 124 insertions(+), 3 deletions(-) diff --git a/test/registries.py b/test/registries.py index 3e67d8e47..ff816b1e7 100644 --- a/test/registries.py +++ b/test/registries.py @@ -19,6 +19,7 @@ from collections import defaultdict from dataclasses import dataclass, field from typing import TYPE_CHECKING, Literal, TypeAlias +from urllib.parse import parse_qsl, urlparse from responses import GET, POST, CallbackResponse, Response from responses.registries import FirstMatchRegistry @@ -116,6 +117,25 @@ def from_( ) +@dataclass +class Job: + """Registry representation of a job.""" + + id: str + """Job id.""" + + backend_name: str + """Backend name.""" + + program: Literal["sampler", "estimator"] = "sampler" + + status: Literal["queued", "running", "completed", "cancelled", "failed"] = "completed" + """Job status.""" + + tags: list[str] = field(default_factory=list) + """List of job tags.""" + + class BaseRegistry(FirstMatchRegistry): """Registry that dynamically serves IBM Quantum Compute responses. @@ -141,13 +161,17 @@ class BaseRegistry(FirstMatchRegistry): """Instances in this registry, keyed by instance name.""" backends: dict[str, dict[str, Backend]] - """Backends in this registry, keyed by instance name.""" + """Backends in this registry, keyed by instance name and backend name.""" + + jobs: dict[str, dict[str, Job]] + """Jobs in this registry, keyed by instance name and job id.""" def __init__(self) -> None: super().__init__() self.instances = {} self.backends = defaultdict(dict) + self.jobs = defaultdict(dict) # Add callbacks for IBM Global Search and Global Catalog. self.add( @@ -215,7 +239,21 @@ def __init__(self) -> None: CallbackResponse( method=POST, url="https://my-region.quantum.cloud.ibm.com/api/v1/jobs", - callback=self.callback_jobs, + callback=self.callback_jobs_post, + ), + ) + self.add( + CallbackResponse( + method=GET, + url="https://my-region.quantum.cloud.ibm.com/api/v1/jobs", + callback=self.callback_jobs_get, + ), + ) + self.add( + CallbackResponse( + method=GET, + url=re.compile(r"https://my-region.quantum.cloud.ibm.com/api/v1/jobs/\w"), + callback=self.callback_jobs_id, ), ) @@ -232,6 +270,10 @@ def add_backend(self, backend: Backend, instance: str | None = None) -> None: for name in instances: self.backends[name][backend.name] = backend + def add_job(self, job: Job, instance: str) -> None: + """Add a new job to the registry.""" + self.jobs[instance][job.id] = job + def callback_global_search(self, _: PreparedRequest) -> CallbackResult: """Callback for the IBM Cloud Global Search API. @@ -366,7 +408,7 @@ def callback_backends_status(self, request: PreparedRequest) -> CallbackResult: } return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) - def callback_jobs(self, request: PreparedRequest) -> CallbackResult: + def callback_jobs_post(self, request: PreparedRequest) -> CallbackResult: """Callback for the IBM Quantum Compute API ``/jobs`` endpoint. Dynamically return a job, based on the contents of `self.backends`. @@ -390,6 +432,85 @@ def callback_jobs(self, request: PreparedRequest) -> CallbackResult: } return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) + def callback_jobs_get(self, request: PreparedRequest) -> CallbackResult: + """Callback for the IBM Quantum Compute API ``/jobs`` endpoint. + + Dynamically return a list of job, based on the contents of `self.jobs`. + + References: + https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/jobs + """ + # Validate the instance CRN and backend name. + instance_crn = request.headers.get("Service-CRN") + instance = next( + instance for instance in self.instances.values() if instance.crn == instance_crn + ) + if instance.name not in self.backends: + return (404, {"Content-Type": "application/json"}, "{}") + + # Retrieve query parameters. + query_dict: dict = dict(parse_qsl(urlparse(request.url).query)) + limit = int(query_dict.get("limit", 0)) + offset = int(query_dict.get("offset", 0)) + program = query_dict.get("program", None) + pending = query_dict.get("pending", None) + + # Build filter based on query parameters. + filter_: dict[str, tuple] = {} + if program: + filter_["program"] = (program,) + if pending == "true": + filter_["status"] = ("queued", "running") + elif pending == "false": + filter_["status"] = ("completed", "cancelled", "failed") + + jobs = [ + { + "id": job.id, + "backend": job.backend_name, + "status": job.status.capitalize(), + "program": {"id": job.program}, + } + for job in self.jobs[instance.name].values() + if all(getattr(job, key) in value for key, value in filter_.items()) + ] + + response_body = { + "jobs": jobs[offset : offset + limit], + "count": len(jobs), + "limit": limit, + "offset": offset, + } + return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) + + def callback_jobs_id(self, request: PreparedRequest) -> CallbackResult: + """Callback for the IBM Quantum Compute API ``/jobs/{}`` endpoint. + + Dynamically return a job, based on the contents of `self.jobs`. + + References: + https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/jobs + """ + # Validate the instance CRN and backend name. + instance_crn = request.headers.get("Service-CRN") + instance = next( + instance for instance in self.instances.values() if instance.crn == instance_crn + ) + job_id = request.path_url.split("/")[-1].split("?")[0] + + if instance.name not in self.backends or job_id not in self.jobs[instance.name]: + return (404, {"Content-Type": "application/json"}, "{}") + + job = self.jobs[instance.name][job_id] + + response_body = { + "id": job.id, + "backend": job.backend_name, + "status": job.status.capitalize(), + "program": {"id": job.program}, + } + return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) + class DefaultRegistry(BaseRegistry): """Registry with two instances, with one common backend and two unique backends. From ab9fcdc6c66bf0e03bc4d694410a43cf4fdec940 Mon Sep 17 00:00:00 2001 From: "Diego M. Rodriguez" Date: Thu, 30 Jul 2026 19:54:29 +0200 Subject: [PATCH 2/6] Update test_job_retrieval --- test/unit/test_job_retrieval.py | 222 +++++++++++++++++--------------- 1 file changed, 117 insertions(+), 105 deletions(-) diff --git a/test/unit/test_job_retrieval.py b/test/unit/test_job_retrieval.py index 728d342cf..2c7e0c388 100644 --- a/test/unit/test_job_retrieval.py +++ b/test/unit/test_job_retrieval.py @@ -14,9 +14,12 @@ from datetime import datetime, timedelta, timezone -from ..decorators import run_cloud_fake +from qiskit_ibm_runtime.qiskit_runtime_service import QiskitRuntimeService + +from ..decorators import mock_responses from ..ibm_test_case import IBMTestCase from ..program import run_program +from ..registries import Job from ..utils import mock_wait_for_final_state from .mock.fake_runtime_service import FakeRuntimeService @@ -31,164 +34,150 @@ def setUp(self): channel="ibm_quantum_platform", token="my_token" ) - @run_cloud_fake - def test_retrieve_job(self, service): + @mock_responses + def test_retrieve_job(self, registry): """Test retrieving a job.""" - program_id = "sampler" - params = {"param1": "foo"} - job = run_program(service=service, program_id=program_id, inputs=params) - rjob = service.job(job.job_id()) - self.assertEqual(job.job_id(), rjob.job_id()) - self.assertEqual(program_id, rjob.primitive_id) + registry.add_job(Job("1", "common_backend", program="sampler"), "a") + + service = QiskitRuntimeService(token="my_token") + job = service.job("1") + self.assertEqual("1", job.job_id()) + self.assertEqual("sampler", job.primitive_id) - @run_cloud_fake - def test_jobs_no_limit(self, service): + @mock_responses + def test_jobs_no_limit(self, registry): """Test retrieving jobs without limit.""" - program_id = "sampler" + for i in range(25): + registry.add_job(Job(str(i), "common_backend"), "a") - jobs = [] - for _ in range(25): - jobs.append(run_program(service, program_id)) - rjobs = service.jobs(limit=None) - self.assertEqual(25, len(rjobs)) + service = QiskitRuntimeService(token="my_token", instance="a") + jobs = service.jobs(limit=None) + self.assertEqual(25, len(jobs)) - @run_cloud_fake - def test_jobs_limit(self, service): + @mock_responses + def test_jobs_limit(self, registry): """Test retrieving jobs with limit.""" - program_id = "sampler" - - jobs = [] - job_count = 25 - for _ in range(job_count): - jobs.append(run_program(service, program_id)) + for i in range(25): + registry.add_job(Job(str(i), "common_backend"), "a") + service = QiskitRuntimeService(token="my_token", instance="a") limits = [21, 30] for limit in limits: with self.subTest(limit=limit): - rjobs = service.jobs(limit=limit) - self.assertEqual(min(limit, job_count), len(rjobs)) + jobs = service.jobs(limit=limit) + self.assertEqual(min(limit, 25), len(jobs)) - @run_cloud_fake - def test_jobs_skip(self, service): + @mock_responses + def test_jobs_skip(self, registry): """Test retrieving jobs with skip.""" - program_id = "sampler" + for i in range(5): + registry.add_job(Job(str(i), "common_backend"), "a") - jobs = [] - for _ in range(5): - jobs.append(run_program(service, program_id)) - rjobs = service.jobs(skip=4) - self.assertEqual(1, len(rjobs)) + service = QiskitRuntimeService(token="my_token", instance="a") + jobs = service.jobs(skip=4) + self.assertEqual(1, len(jobs)) - @run_cloud_fake - def test_backend_instance_warnings(self, service): + @mock_responses + def test_backend_instance_warnings(self, registry): """Test backend instance warnings do not appear.""" - program_id = "sampler" - params = {"param1": "foo"} - job = run_program(service=service, program_id=program_id, inputs=params) + registry.add_job(Job("1", "common_backend", program="sampler"), "a") + + service = QiskitRuntimeService(token="my_token", instance="a") with self.assertNoLogs("qiskit_ibm_runtime", level="WARNING"): service.jobs() with self.assertNoLogs("qiskit_ibm_runtime", level="WARNING"): - service.job(job.job_id()) + service.job("1") - def test_jobs_skip_limit(self): + @mock_responses + def test_jobs_skip_limit(self, registry): """Test retrieving jobs with skip and limit.""" - service = self._ibm_quantum_service - program_id = "sampler" + for i in range(10): + registry.add_job(Job(str(i), "common_backend"), "a") - jobs = [] - for _ in range(10): - jobs.append(run_program(service, program_id)) - rjobs = service.jobs(skip=4, limit=2) - self.assertEqual(2, len(rjobs)) + service = QiskitRuntimeService(token="my_token", instance="a") + jobs = service.jobs(skip=4, limit=2) + self.assertEqual(2, len(jobs)) - @run_cloud_fake - def test_jobs_pending(self, service): + @mock_responses + def test_jobs_pending(self, registry): """Test retrieving pending jobs (QUEUED, RUNNING).""" - program_id = "sampler" + pending_jobs_count, _ = self.populate_jobs_with_all_statuses(registry) - _, pending_jobs_count, _ = self._populate_jobs_with_all_statuses( - service, program_id=program_id - ) - rjobs = service.jobs(pending=True) - self.assertEqual(pending_jobs_count, len(rjobs)) + service = QiskitRuntimeService(token="my_token", instance="a") + jobs = service.jobs(pending=True) + self.assertEqual(pending_jobs_count, len(jobs)) - def test_jobs_limit_pending(self): + @mock_responses + def test_jobs_limit_pending(self, registry): """Test retrieving pending jobs (QUEUED, RUNNING) with limit.""" - service = self._ibm_quantum_service - program_id = "sampler" + self.populate_jobs_with_all_statuses(registry) - self._populate_jobs_with_all_statuses(service, program_id=program_id) + service = QiskitRuntimeService(token="my_token", instance="a") limit = 4 - rjobs = service.jobs(limit=limit, pending=True) - self.assertEqual(limit, len(rjobs)) + jobs = service.jobs(limit=limit, pending=True) + self.assertEqual(limit, len(jobs)) - def test_jobs_skip_pending(self): + @mock_responses + def test_jobs_skip_pending(self, registry): """Test retrieving pending jobs (QUEUED, RUNNING) with skip.""" - service = self._ibm_quantum_service - program_id = "sampler" + pending_jobs_count, _ = self.populate_jobs_with_all_statuses(registry) - _, pending_jobs_count, _ = self._populate_jobs_with_all_statuses( - service, program_id=program_id - ) + service = QiskitRuntimeService(token="my_token", instance="a") skip = 4 - rjobs = service.jobs(skip=skip, pending=True) - self.assertEqual(pending_jobs_count - skip, len(rjobs)) + jobs = service.jobs(skip=skip, pending=True) + self.assertEqual(pending_jobs_count - skip, len(jobs)) - def test_jobs_limit_skip_pending(self): + @mock_responses + def test_jobs_limit_skip_pending(self, registry): """Test retrieving pending jobs (QUEUED, RUNNING) with limit and skip.""" - service = self._ibm_quantum_service - program_id = "sampler" + self.populate_jobs_with_all_statuses(registry) - self._populate_jobs_with_all_statuses(service, program_id=program_id) + service = QiskitRuntimeService(token="my_token", instance="a") limit = 2 skip = 3 - rjobs = service.jobs(limit=limit, skip=skip, pending=True) - self.assertEqual(limit, len(rjobs)) + jobs = service.jobs(limit=limit, skip=skip, pending=True) + self.assertEqual(limit, len(jobs)) - def test_jobs_returned(self): + @mock_responses + def test_jobs_returned(self, registry): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED).""" - service = self._ibm_quantum_service - program_id = "sampler" + _, returned_jobs_count = self.populate_jobs_with_all_statuses(registry) - _, _, returned_jobs_count = self._populate_jobs_with_all_statuses( - service, program_id=program_id - ) - rjobs = service.jobs(pending=False) - self.assertEqual(returned_jobs_count, len(rjobs)) + service = QiskitRuntimeService(token="my_token", instance="a") + jobs = service.jobs(pending=False) + self.assertEqual(returned_jobs_count, len(jobs)) - def test_jobs_limit_returned(self): + @mock_responses + def test_jobs_limit_returned(self, registry): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED) with limit.""" - service = self._ibm_quantum_service - program_id = "sampler" + self.populate_jobs_with_all_statuses(registry) - self._populate_jobs_with_all_statuses(service, program_id=program_id) + service = QiskitRuntimeService(token="my_token", instance="a") limit = 6 - rjobs = service.jobs(limit=limit, pending=False) - self.assertEqual(limit, len(rjobs)) + jobs = service.jobs(limit=limit, pending=False) + self.assertEqual(limit, len(jobs)) - def test_jobs_skip_returned(self): + @mock_responses + def test_jobs_skip_returned(self, registry): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED) with skip.""" - service = self._ibm_quantum_service - program_id = "sampler" + _, returned_jobs_count = self.populate_jobs_with_all_statuses(registry) - _, _, returned_jobs_count = self._populate_jobs_with_all_statuses( - service, program_id=program_id - ) + service = QiskitRuntimeService(token="my_token", instance="a") skip = 4 - rjobs = service.jobs(skip=skip, pending=False) - self.assertEqual(returned_jobs_count - skip, len(rjobs)) + jobs = service.jobs(skip=skip, pending=False) + self.assertEqual(returned_jobs_count - skip, len(jobs)) - def test_jobs_limit_skip_returned(self): + @mock_responses + def test_jobs_limit_skip_returned(self, registry): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED) with limit and skip.""" - service = self._ibm_quantum_service - program_id = "sampler" + self.populate_jobs_with_all_statuses(registry) - self._populate_jobs_with_all_statuses(service, program_id=program_id) + service = QiskitRuntimeService(token="my_token", instance="a") limit = 6 skip = 2 - rjobs = service.jobs(limit=limit, skip=skip, pending=False) - self.assertEqual(limit, len(rjobs)) + jobs = service.jobs(limit=limit, skip=skip, pending=False) + self.assertEqual(limit, len(jobs)) def test_jobs_filter_by_job_tags(self): """Test retrieving jobs by job tags.""" @@ -277,6 +266,29 @@ def test_different_instance(self): rjob = service.job(job.job_id()) self.assertIsNotNone(rjob.backend()) + def populate_jobs_with_all_statuses(self, registry, program_id="sampler"): + """Populate the database with jobs of all statuses.""" + pending_jobs_count = 0 + returned_jobs_count = 0 + status_count = { + "running": 3, + "completed": 4, + "queued": 2, + "failed": 3, + "cancelled": 2, + } + + job_id = 1 + for stat, count in status_count.items(): + for _ in range(count): + registry.add_job(Job(str(job_id), "common_backend", program_id, stat), "a") + if stat in ("running", "queued"): + pending_jobs_count += 1 + else: + returned_jobs_count += 1 + job_id += 1 + return pending_jobs_count, returned_jobs_count + def _populate_jobs_with_all_statuses(self, service, program_id): """Populate the database with jobs of all statuses.""" jobs = [] From c8fda40637e22161034588ec24344957c92ce9eb Mon Sep 17 00:00:00 2001 From: "Diego M. Rodriguez" Date: Tue, 4 Aug 2026 18:47:48 +0200 Subject: [PATCH 3/6] Simplify, avoiding jobs endpoint --- test/registries.py | 72 +++-------- test/unit/test_job_retrieval.py | 222 +++++++++++++++----------------- 2 files changed, 124 insertions(+), 170 deletions(-) diff --git a/test/registries.py b/test/registries.py index ff816b1e7..224aaab18 100644 --- a/test/registries.py +++ b/test/registries.py @@ -132,9 +132,6 @@ class Job: status: Literal["queued", "running", "completed", "cancelled", "failed"] = "completed" """Job status.""" - tags: list[str] = field(default_factory=list) - """List of job tags.""" - class BaseRegistry(FirstMatchRegistry): """Registry that dynamically serves IBM Quantum Compute responses. @@ -225,7 +222,7 @@ def __init__(self) -> None: ) ) - # Add responses for the IBM Quantum Compute `/instances` endpoint. + # Add responses for the IBM Quantum Compute `/instances` endpoints. self.add( Response( method=GET, @@ -323,10 +320,7 @@ def callback_backends(self, request: PreparedRequest) -> CallbackResult: https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/backends """ # Validate the instance CRN. - instance_crn = request.headers.get("Service-CRN") - instance = next( - instance for instance in self.instances.values() if instance.crn == instance_crn - ) + instance = self.get_crn_from_request(request) if instance.name not in self.backends: return (404, {"Content-Type": "application/json"}, "{}") @@ -351,10 +345,7 @@ def callback_backends_configuration(self, request: PreparedRequest) -> CallbackR https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/backends """ # Validate the instance CRN and backend name. - instance_crn = request.headers.get("Service-CRN") - instance = next( - instance for instance in self.instances.values() if instance.crn == instance_crn - ) + instance = self.get_crn_from_request(request) backend_name = request.path_url.split("/")[4] if instance.name not in self.backends or backend_name not in self.backends[instance.name]: return (404, {"Content-Type": "application/json"}, "{}") @@ -371,10 +362,7 @@ def callback_backends_properties(self, request: PreparedRequest) -> CallbackResu https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/backends """ # Validate the instance CRN and backend name. - instance_crn = request.headers.get("Service-CRN") - instance = next( - instance for instance in self.instances.values() if instance.crn == instance_crn - ) + instance = self.get_crn_from_request(request) backend_name = request.path_url.split("/")[4] if instance.name not in self.backends or backend_name not in self.backends[instance.name]: return (404, {"Content-Type": "application/json"}, "{}") @@ -391,10 +379,7 @@ def callback_backends_status(self, request: PreparedRequest) -> CallbackResult: https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/backends """ # Validate the instance CRN and backend name. - instance_crn = request.headers.get("Service-CRN") - instance = next( - instance for instance in self.instances.values() if instance.crn == instance_crn - ) + instance = self.get_crn_from_request(request) backend_name = request.path_url.split("/")[4] if instance.name not in self.backends or backend_name not in self.backends[instance.name]: return (404, {"Content-Type": "application/json"}, "{}") @@ -417,10 +402,7 @@ def callback_jobs_post(self, request: PreparedRequest) -> CallbackResult: https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/jobs """ # Validate the instance CRN and backend name. - instance_crn = request.headers.get("Service-CRN") - instance = next( - instance for instance in self.instances.values() if instance.crn == instance_crn - ) + instance = self.get_crn_from_request(request) backend_name = json.loads(str(request.body))["backend"] if instance.name not in self.backends or backend_name not in self.backends[instance.name]: @@ -440,30 +422,11 @@ def callback_jobs_get(self, request: PreparedRequest) -> CallbackResult: References: https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/jobs """ - # Validate the instance CRN and backend name. - instance_crn = request.headers.get("Service-CRN") - instance = next( - instance for instance in self.instances.values() if instance.crn == instance_crn - ) + # Validate the instance CRN. + instance = self.get_crn_from_request(request) if instance.name not in self.backends: return (404, {"Content-Type": "application/json"}, "{}") - # Retrieve query parameters. - query_dict: dict = dict(parse_qsl(urlparse(request.url).query)) - limit = int(query_dict.get("limit", 0)) - offset = int(query_dict.get("offset", 0)) - program = query_dict.get("program", None) - pending = query_dict.get("pending", None) - - # Build filter based on query parameters. - filter_: dict[str, tuple] = {} - if program: - filter_["program"] = (program,) - if pending == "true": - filter_["status"] = ("queued", "running") - elif pending == "false": - filter_["status"] = ("completed", "cancelled", "failed") - jobs = [ { "id": job.id, @@ -472,14 +435,13 @@ def callback_jobs_get(self, request: PreparedRequest) -> CallbackResult: "program": {"id": job.program}, } for job in self.jobs[instance.name].values() - if all(getattr(job, key) in value for key, value in filter_.items()) ] response_body = { - "jobs": jobs[offset : offset + limit], + "jobs": jobs, "count": len(jobs), - "limit": limit, - "offset": offset, + "limit": 20, + "offset": 0, } return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) @@ -492,10 +454,7 @@ def callback_jobs_id(self, request: PreparedRequest) -> CallbackResult: https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/jobs """ # Validate the instance CRN and backend name. - instance_crn = request.headers.get("Service-CRN") - instance = next( - instance for instance in self.instances.values() if instance.crn == instance_crn - ) + instance = self.get_crn_from_request(request) job_id = request.path_url.split("/")[-1].split("?")[0] if instance.name not in self.backends or job_id not in self.jobs[instance.name]: @@ -510,6 +469,13 @@ def callback_jobs_id(self, request: PreparedRequest) -> CallbackResult: "program": {"id": job.program}, } return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) + + def get_crn_from_request(self, request: PreparedRequest) -> Instance: + """Retrieve the instance CRN from the request headers.""" + instance_crn = request.headers.get("Service-CRN") + return next( + instance for instance in self.instances.values() if instance.crn == instance_crn + ) class DefaultRegistry(BaseRegistry): diff --git a/test/unit/test_job_retrieval.py b/test/unit/test_job_retrieval.py index 2c7e0c388..728d342cf 100644 --- a/test/unit/test_job_retrieval.py +++ b/test/unit/test_job_retrieval.py @@ -14,12 +14,9 @@ from datetime import datetime, timedelta, timezone -from qiskit_ibm_runtime.qiskit_runtime_service import QiskitRuntimeService - -from ..decorators import mock_responses +from ..decorators import run_cloud_fake from ..ibm_test_case import IBMTestCase from ..program import run_program -from ..registries import Job from ..utils import mock_wait_for_final_state from .mock.fake_runtime_service import FakeRuntimeService @@ -34,150 +31,164 @@ def setUp(self): channel="ibm_quantum_platform", token="my_token" ) - @mock_responses - def test_retrieve_job(self, registry): + @run_cloud_fake + def test_retrieve_job(self, service): """Test retrieving a job.""" - registry.add_job(Job("1", "common_backend", program="sampler"), "a") - - service = QiskitRuntimeService(token="my_token") - job = service.job("1") - self.assertEqual("1", job.job_id()) - self.assertEqual("sampler", job.primitive_id) + program_id = "sampler" + params = {"param1": "foo"} + job = run_program(service=service, program_id=program_id, inputs=params) + rjob = service.job(job.job_id()) + self.assertEqual(job.job_id(), rjob.job_id()) + self.assertEqual(program_id, rjob.primitive_id) - @mock_responses - def test_jobs_no_limit(self, registry): + @run_cloud_fake + def test_jobs_no_limit(self, service): """Test retrieving jobs without limit.""" - for i in range(25): - registry.add_job(Job(str(i), "common_backend"), "a") + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") - jobs = service.jobs(limit=None) - self.assertEqual(25, len(jobs)) + jobs = [] + for _ in range(25): + jobs.append(run_program(service, program_id)) + rjobs = service.jobs(limit=None) + self.assertEqual(25, len(rjobs)) - @mock_responses - def test_jobs_limit(self, registry): + @run_cloud_fake + def test_jobs_limit(self, service): """Test retrieving jobs with limit.""" - for i in range(25): - registry.add_job(Job(str(i), "common_backend"), "a") + program_id = "sampler" + + jobs = [] + job_count = 25 + for _ in range(job_count): + jobs.append(run_program(service, program_id)) - service = QiskitRuntimeService(token="my_token", instance="a") limits = [21, 30] for limit in limits: with self.subTest(limit=limit): - jobs = service.jobs(limit=limit) - self.assertEqual(min(limit, 25), len(jobs)) + rjobs = service.jobs(limit=limit) + self.assertEqual(min(limit, job_count), len(rjobs)) - @mock_responses - def test_jobs_skip(self, registry): + @run_cloud_fake + def test_jobs_skip(self, service): """Test retrieving jobs with skip.""" - for i in range(5): - registry.add_job(Job(str(i), "common_backend"), "a") + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") - jobs = service.jobs(skip=4) - self.assertEqual(1, len(jobs)) + jobs = [] + for _ in range(5): + jobs.append(run_program(service, program_id)) + rjobs = service.jobs(skip=4) + self.assertEqual(1, len(rjobs)) - @mock_responses - def test_backend_instance_warnings(self, registry): + @run_cloud_fake + def test_backend_instance_warnings(self, service): """Test backend instance warnings do not appear.""" - registry.add_job(Job("1", "common_backend", program="sampler"), "a") - - service = QiskitRuntimeService(token="my_token", instance="a") + program_id = "sampler" + params = {"param1": "foo"} + job = run_program(service=service, program_id=program_id, inputs=params) with self.assertNoLogs("qiskit_ibm_runtime", level="WARNING"): service.jobs() with self.assertNoLogs("qiskit_ibm_runtime", level="WARNING"): - service.job("1") + service.job(job.job_id()) - @mock_responses - def test_jobs_skip_limit(self, registry): + def test_jobs_skip_limit(self): """Test retrieving jobs with skip and limit.""" - for i in range(10): - registry.add_job(Job(str(i), "common_backend"), "a") + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") - jobs = service.jobs(skip=4, limit=2) - self.assertEqual(2, len(jobs)) + jobs = [] + for _ in range(10): + jobs.append(run_program(service, program_id)) + rjobs = service.jobs(skip=4, limit=2) + self.assertEqual(2, len(rjobs)) - @mock_responses - def test_jobs_pending(self, registry): + @run_cloud_fake + def test_jobs_pending(self, service): """Test retrieving pending jobs (QUEUED, RUNNING).""" - pending_jobs_count, _ = self.populate_jobs_with_all_statuses(registry) + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") - jobs = service.jobs(pending=True) - self.assertEqual(pending_jobs_count, len(jobs)) + _, pending_jobs_count, _ = self._populate_jobs_with_all_statuses( + service, program_id=program_id + ) + rjobs = service.jobs(pending=True) + self.assertEqual(pending_jobs_count, len(rjobs)) - @mock_responses - def test_jobs_limit_pending(self, registry): + def test_jobs_limit_pending(self): """Test retrieving pending jobs (QUEUED, RUNNING) with limit.""" - self.populate_jobs_with_all_statuses(registry) + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") + self._populate_jobs_with_all_statuses(service, program_id=program_id) limit = 4 - jobs = service.jobs(limit=limit, pending=True) - self.assertEqual(limit, len(jobs)) + rjobs = service.jobs(limit=limit, pending=True) + self.assertEqual(limit, len(rjobs)) - @mock_responses - def test_jobs_skip_pending(self, registry): + def test_jobs_skip_pending(self): """Test retrieving pending jobs (QUEUED, RUNNING) with skip.""" - pending_jobs_count, _ = self.populate_jobs_with_all_statuses(registry) + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") + _, pending_jobs_count, _ = self._populate_jobs_with_all_statuses( + service, program_id=program_id + ) skip = 4 - jobs = service.jobs(skip=skip, pending=True) - self.assertEqual(pending_jobs_count - skip, len(jobs)) + rjobs = service.jobs(skip=skip, pending=True) + self.assertEqual(pending_jobs_count - skip, len(rjobs)) - @mock_responses - def test_jobs_limit_skip_pending(self, registry): + def test_jobs_limit_skip_pending(self): """Test retrieving pending jobs (QUEUED, RUNNING) with limit and skip.""" - self.populate_jobs_with_all_statuses(registry) + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") + self._populate_jobs_with_all_statuses(service, program_id=program_id) limit = 2 skip = 3 - jobs = service.jobs(limit=limit, skip=skip, pending=True) - self.assertEqual(limit, len(jobs)) + rjobs = service.jobs(limit=limit, skip=skip, pending=True) + self.assertEqual(limit, len(rjobs)) - @mock_responses - def test_jobs_returned(self, registry): + def test_jobs_returned(self): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED).""" - _, returned_jobs_count = self.populate_jobs_with_all_statuses(registry) + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") - jobs = service.jobs(pending=False) - self.assertEqual(returned_jobs_count, len(jobs)) + _, _, returned_jobs_count = self._populate_jobs_with_all_statuses( + service, program_id=program_id + ) + rjobs = service.jobs(pending=False) + self.assertEqual(returned_jobs_count, len(rjobs)) - @mock_responses - def test_jobs_limit_returned(self, registry): + def test_jobs_limit_returned(self): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED) with limit.""" - self.populate_jobs_with_all_statuses(registry) + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") + self._populate_jobs_with_all_statuses(service, program_id=program_id) limit = 6 - jobs = service.jobs(limit=limit, pending=False) - self.assertEqual(limit, len(jobs)) + rjobs = service.jobs(limit=limit, pending=False) + self.assertEqual(limit, len(rjobs)) - @mock_responses - def test_jobs_skip_returned(self, registry): + def test_jobs_skip_returned(self): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED) with skip.""" - _, returned_jobs_count = self.populate_jobs_with_all_statuses(registry) + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") + _, _, returned_jobs_count = self._populate_jobs_with_all_statuses( + service, program_id=program_id + ) skip = 4 - jobs = service.jobs(skip=skip, pending=False) - self.assertEqual(returned_jobs_count - skip, len(jobs)) + rjobs = service.jobs(skip=skip, pending=False) + self.assertEqual(returned_jobs_count - skip, len(rjobs)) - @mock_responses - def test_jobs_limit_skip_returned(self, registry): + def test_jobs_limit_skip_returned(self): """Test retrieving returned jobs (COMPLETED, FAILED, CANCELLED) with limit and skip.""" - self.populate_jobs_with_all_statuses(registry) + service = self._ibm_quantum_service + program_id = "sampler" - service = QiskitRuntimeService(token="my_token", instance="a") + self._populate_jobs_with_all_statuses(service, program_id=program_id) limit = 6 skip = 2 - jobs = service.jobs(limit=limit, skip=skip, pending=False) - self.assertEqual(limit, len(jobs)) + rjobs = service.jobs(limit=limit, skip=skip, pending=False) + self.assertEqual(limit, len(rjobs)) def test_jobs_filter_by_job_tags(self): """Test retrieving jobs by job tags.""" @@ -266,29 +277,6 @@ def test_different_instance(self): rjob = service.job(job.job_id()) self.assertIsNotNone(rjob.backend()) - def populate_jobs_with_all_statuses(self, registry, program_id="sampler"): - """Populate the database with jobs of all statuses.""" - pending_jobs_count = 0 - returned_jobs_count = 0 - status_count = { - "running": 3, - "completed": 4, - "queued": 2, - "failed": 3, - "cancelled": 2, - } - - job_id = 1 - for stat, count in status_count.items(): - for _ in range(count): - registry.add_job(Job(str(job_id), "common_backend", program_id, stat), "a") - if stat in ("running", "queued"): - pending_jobs_count += 1 - else: - returned_jobs_count += 1 - job_id += 1 - return pending_jobs_count, returned_jobs_count - def _populate_jobs_with_all_statuses(self, service, program_id): """Populate the database with jobs of all statuses.""" jobs = [] From 1ec13b9a6273c2e608d54129c79578147b0993b2 Mon Sep 17 00:00:00 2001 From: "Diego M. Rodriguez" Date: Fri, 7 Aug 2026 12:18:00 +0200 Subject: [PATCH 4/6] Add tool for storing jobs Co-Authored-By: Claude --- test/registries.py | 3 +- tools/store_job.py | 87 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) create mode 100755 tools/store_job.py diff --git a/test/registries.py b/test/registries.py index 224aaab18..e41b595eb 100644 --- a/test/registries.py +++ b/test/registries.py @@ -19,7 +19,6 @@ from collections import defaultdict from dataclasses import dataclass, field from typing import TYPE_CHECKING, Literal, TypeAlias -from urllib.parse import parse_qsl, urlparse from responses import GET, POST, CallbackResponse, Response from responses.registries import FirstMatchRegistry @@ -469,7 +468,7 @@ def callback_jobs_id(self, request: PreparedRequest) -> CallbackResult: "program": {"id": job.program}, } return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) - + def get_crn_from_request(self, request: PreparedRequest) -> Instance: """Retrieve the instance CRN from the request headers.""" instance_crn = request.headers.get("Service-CRN") diff --git a/tools/store_job.py b/tools/store_job.py new file mode 100755 index 000000000..9245836b2 --- /dev/null +++ b/tools/store_job.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +# This code is part of Qiskit. +# +# (C) Copyright IBM 2024-2026. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Retrieve and store a job for using with backwards-compatibility tests. + +Retrieve a job from an instance, and store the json responsed for its details and results in the +tests directory used for backwards-compatibility tests. +""" + +import argparse +import json +from pathlib import Path + +from responses import PassthroughResponse, RequestsMock + +from qiskit_ibm_runtime import QiskitRuntimeService + + +def main(instance: str, job_id: str, cloud_url: str, dir_: Path) -> None: + """Retrieve and store a job for using with backwards-compatibility tests. + + Args: + instance: CRN of the instance. + job_id: ID of the job to retrieve and store. + cloud_url: Cloud URL for the endpoints. + dir_: Path where to write the responses to. + """ + job_url = f"{cloud_url}/api/v1/jobs/{job_id}" + results_url = f"{cloud_url}/api/v1/jobs/{job_id}/results" + + service = QiskitRuntimeService(instance=instance) + # Force additional queries to take place outside mocking responses. + service.backends() + + # Intercept HTTP responses for the jobs endpoints. + with RequestsMock() as responses: + job_details_endpoint = responses.add(PassthroughResponse(method="GET", url=job_url)) + job_results_endpoint = responses.add(PassthroughResponse(method="GET", url=results_url)) + job = service.job(job_id) + job.result() + + job_details = job_details_endpoint.calls[0].response.text + job_results = job_results_endpoint.calls[0].response.text + + # Anonymize some data. + job_details = json.dumps( + {**json.loads(job_details), "user_id": "unknown"}, separators=(",", ":") + ) + + # Write the responses to files. + (dir_ / f"{job_id}_details.json").write_text(job_details, encoding="utf-8") + (dir_ / f"{job_id}_results.json").write_text(job_results, encoding="utf-8") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--instance", type=str, help="Instance CRN.") + parser.add_argument("--job-id", type=str, help="Job ID.") + parser.add_argument( + "--cloud-url", + type=str, + default="https://quantum.cloud.ibm.com", + help="Cloud URL for the endpoints.", + ) + parser.add_argument( + "--dir", + type=Path, + default=Path("test/unit/backwards_compatibility/resources"), + help="Path where to write the responses.", + ) + args = parser.parse_args() + main( + instance=args.instance, + job_id=args.job_id, + cloud_url=args.cloud_url, + dir_=args.dir, + ) From 4cb7f90b5d6ec74b1e109010f4ff51979446844e Mon Sep 17 00:00:00 2001 From: "Diego M. Rodriguez" Date: Fri, 7 Aug 2026 14:06:43 +0200 Subject: [PATCH 5/6] Add support for job raw data --- test/registries.py | 57 ++++++++++++++++--- test/unit/backwards_compatibility/__init__.py | 19 +++++++ .../d9qp8cvpemts73crjmsg_details.json | 1 + .../d9qp8cvpemts73crjmsg_results.json | 1 + .../backwards_compatibility/test_load_jobs.py | 53 +++++++++++++++++ 5 files changed, 123 insertions(+), 8 deletions(-) create mode 100644 test/unit/backwards_compatibility/__init__.py create mode 100644 test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_details.json create mode 100644 test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_results.json create mode 100644 test/unit/backwards_compatibility/test_load_jobs.py diff --git a/test/registries.py b/test/registries.py index e41b595eb..531c3eeb8 100644 --- a/test/registries.py +++ b/test/registries.py @@ -131,6 +131,12 @@ class Job: status: Literal["queued", "running", "completed", "cancelled", "failed"] = "completed" """Job status.""" + raw_details: str | None = None + """Response for the job details.""" + + raw_results: str | None = None + """Response for the job results.""" + class BaseRegistry(FirstMatchRegistry): """Registry that dynamically serves IBM Quantum Compute responses. @@ -248,10 +254,17 @@ def __init__(self) -> None: self.add( CallbackResponse( method=GET, - url=re.compile(r"https://my-region.quantum.cloud.ibm.com/api/v1/jobs/\w"), + url=re.compile(r"https://my-region.quantum.cloud.ibm.com/api/v1/jobs/\w+"), callback=self.callback_jobs_id, ), ) + self.add( + CallbackResponse( + method=GET, + url=re.compile(r"https://my-region.quantum.cloud.ibm.com/api/v1/jobs/\w+/results"), + callback=self.callback_jobs_results, + ), + ) def add_instance(self, instance: Instance) -> None: """Add a new instance to the registry.""" @@ -461,13 +474,41 @@ def callback_jobs_id(self, request: PreparedRequest) -> CallbackResult: job = self.jobs[instance.name][job_id] - response_body = { - "id": job.id, - "backend": job.backend_name, - "status": job.status.capitalize(), - "program": {"id": job.program}, - } - return (200, {"Content-Type": "application/json"}, json.dumps(response_body)) + if job.raw_details: + response_body = job.raw_details + else: + response_body = json.dumps( + { + "id": job.id, + "backend": job.backend_name, + "status": job.status.capitalize(), + "program": {"id": job.program}, + } + ) + return (200, {"Content-Type": "application/json"}, response_body) + + def callback_jobs_results(self, request: PreparedRequest) -> CallbackResult: + """Callback for the IBM Quantum Compute API ``/jobs/{}/results`` endpoint. + + Dynamically return a job results, based on the contents of `self.jobs`. + + References: + https://quantum.cloud.ibm.com/docs/en/api/qiskit-runtime-rest/tags/jobs + """ + # Validate the instance CRN and backend name. + instance = self.get_crn_from_request(request) + job_id = request.path_url.split("/")[-2].split("?")[0] + + if instance.name not in self.backends or job_id not in self.jobs[instance.name]: + return (404, {"Content-Type": "application/json"}, "{}") + + job = self.jobs[instance.name][job_id] + + if job.raw_results: + response_body = job.raw_results + else: + response_body = json.dumps({}) + return (200, {"Content-Type": "application/json"}, response_body) def get_crn_from_request(self, request: PreparedRequest) -> Instance: """Retrieve the instance CRN from the request headers.""" diff --git a/test/unit/backwards_compatibility/__init__.py b/test/unit/backwards_compatibility/__init__.py new file mode 100644 index 000000000..b70291d84 --- /dev/null +++ b/test/unit/backwards_compatibility/__init__.py @@ -0,0 +1,19 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2026. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests for backwards compatibility. + +Stored jobs: +* `d9qp8cvpemts73crjmsg` - circuit from + `test/integration/test_executor_sampler/TestSampler::test_sampler_with_parametric_circuits` + with no twirling, against `ibm_fez`, with version `0.49`. +""" diff --git a/test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_details.json b/test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_details.json new file mode 100644 index 000000000..01d10046f --- /dev/null +++ b/test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_details.json @@ -0,0 +1 @@ +{"backend":"ibm_fez","created":"2026-08-07T08:15:47.460045Z","id":"d9qp8cvpemts73crjmsg","cost":600,"program":{"id":"executor"},"state":{"status":"Completed"},"status":"Completed","user_id":"unknown"} \ No newline at end of file diff --git a/test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_results.json b/test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_results.json new file mode 100644 index 000000000..6eeffc16b --- /dev/null +++ b/test/unit/backwards_compatibility/resources/d9qp8cvpemts73crjmsg_results.json @@ -0,0 +1 @@ +{"schema_version": "v2.0", "data": [{"results": {"meas": {"data": "eJxNWM+vW0cZPTN37IxdF8aO27rVK5p3nxPdvAbkhgg9IYTGjgGnBORUXWTBwpQgVYJFKhUJsRo7TuU8QuVERSqoi5uooLTqIuKHxNIJpUqrCj0QC5ZvQSX+DM4396XFeT/8rme++X6c73xn0v/Yha+4IbCNruk5BQR+mYZxEa6rXA2a/4bh4tRMjYprYAq43q3wJlpcaFHnFrT5DFjDKBXawTSBHs0ohJpum04NtP+MLHEGceyU67ZbiFyB4Fww2AzNO/wrWmPqLgS4MHFhEGFbKF8MYjoMG7UPYLl+qeZQASbHEzc7m+zbseTHyGCCmpjOAjOj6PjT/qUaF43R4ZbA47FuGXQj1Db2x9C2GYy1MTdQNur9BtQatXqPB2i9v+RBRTQ6FvnQGHumYw22H+7t6jseNve4Hpmkj+Bdy3s/mxe4X/rcLOmlX11snuNh0ekF4xsyXzOsljk2A+CP7zdPPG691ejUrzE5BbxxU3un7f11rT2fxPXDvdN6tJxvrYz5bQkbIvJ/BwU7GaLYcM0y2hznjGWuPXKoxerNuJkVzyzGuT3Z48la/401As55GI3MlptId3Auvqrygqdc9u924KWmtNYHGpqGfG8EvGfxQCoCHgr46MWMt1zJOvCHFuu0j/RiIe/BWjA18pIayIb7wALJCk0XmDieVaYV/NT00xqoWyZZKYIs8/yxqYuNKNgKlX3xsTJrWI3PHlt5F5WpThtUD51NduYe2unkD2EQB4824DrA4hOsemMMmrTgLH+hZKiRPt5nIF1ajYTyoHUUkueyFVruutoliOT1IvP4C8mw7JUiQ6saT9jGqQLoaO8R/xCjN5Iq14zeEsGY2zb/fF0ZO/CS2EN/P1qn2BeV/3RFuzvSTBkboKjtCIzuSfPUWwEjwXYN5vKm5UuTwjHBrR+lgxtO8Ndp7nXiN/txkroB2rh7Ytr8kn+PHJr74vCpJjpR35XCpddKzlGRyb7HnF2xkvN7mqdEgQcEOlqXBKr16yg5UVdSaqznVmVOi40dVJWw8sUUNCwT6yKzQZgOto4chfkq7IItJyfrQh0mNC3QI0R0wlYrAadgbm1orXYHNIVMV1Db8zYVfOUSmmmj5IOIzS6ylFWbzsDAH73X6Rd71Sd3BUSWHjG0RSy3d2VVmFWoJRZjtHFB7ls8CAX5QLZvYyZQjxg3sNdiLZ0nsbAC0hcv00/gpORJp1IoHsTnmrFkkqyCa1OK+8yt2ZZTZNUq5QjRBLsntkmxGu9jX9x4N4Z7wnOwTfGZjTPiLuOISx47x4qJ8Y20n/WI544qiAvJ3QX7lc0Yf5OaZl5VxPt+2LoV/De7igEqy0++BT9j1EZbATv6P5S8kTCF+Kt+s8nxXmpQBR2uiTF/QThXamHTY2+NbqUucSn8qlOxBP1lNmcwD/l8JJ+xx4lLZVwFA35Vjcnn287dI1SZA9KZJQ9YLvrgyJasflx5q8ze0hzBXedyZF1wKGcKc6mGAIQMgrN8s0KheuhsEj3oim3YTuzZ20c2b7Ne3GbflhStSKHpzxRERZ98c6PazYZyA8mGtiq5btNPNhqfVM2TXlWjsKeVS66ohBGiCiFLCzopuU7Fow0Im5jRhoagJCotKcZCwh44IR5+EGU1of2C2JY0s4MSMbnKS76V1LeMl2Qw503JQ036i0A1Qm7eNchUBKizkfhvky22rq4f0PC2UFWcY2yQAvwXtJyyjLgGPCRueIieuYT2jVWV4+5mO6MXdelLhL0OD9EESCMFZ1+pCiSVjBydLIzn5HGZw+g2UpzKC5MPKzafQlAeUk6z1ifbibeWjL0aJzCXSg4QNq3GjkbpGr+GejscxpCynQpxRQIeWOdkIlwdYLlQzjgjw56r4gIDveUPFZZKcnYjfDZZajdmQYzL4Wb+woVaOJumcYKy7cPuMtFkErM9qqjJaVQZqDF2shEzHNyUoNEtUwq34GJdzhC4EevjHa5Q7hKUov8ttvG0vV4Exw6aEeaz2X6oGiDc/E6UqWfxk5whPeBZfair7E5hDLjNhOTqRV3Z86jzn/l+8Pi/V48i6mn8WOj8/CyIB5RcMgFzdGR+ct4FSrN+Q+aleyJ1nkRSQ03JG4Ohdh+Ohc59gtbQdiaCNAG4wWqWRFoXujRCHlAkOB4ThIxnyVJPdnkVL/gK3ArDplVdYmJq1pQqzGLKiuBywmn0J8Q62WAgjKHtzkKmDL97oQvfZtuEn9ZE1oQL3eK4rbk+m5cHcqT4jKi9aet4Ti1Jrk3EeVA/17YffM8PRpbny/y4Symw0rmxXunqSbwfOp6ns7Oagh4qK5PaNY0M9pXRWUv3SySU+0X4PepL+UjGIKezF3aPRUS/mzBAeCfgllh/MWuyUDgufapJ13GZW6tW22imoGWd9iVa92S9ZnSnCbOp9pE0NtGVkoFnUe6yc9lt3GRmTDbsY/2K6PmZGew3v/Gy0PEn8mOXYWXBmqnVDyKpQ52lwu2PAi5SzMwO4+SgeJxa5PT9K6GtV2zB4ycwXamDjX444LTNf+f74+4/WTi1iPH5iGPwl9T87/34NfYzY4B649XXWNYDM1Qu78xPvpZp60wZOsP3FKZtxvpB6+sxLGw2f/Pju8GqZ9yxUfS9H2HatTUBxpPHpnOCoiBN9XCXMaoF5RArvlKZnmDTJHb+ymn0D2OH5zlH37oZ9Ydcb8fE/hP60MdlCN8dhwPoOo7/jAy/tJNWWz250Wqz53y/598oLth3xl6KpDs2FqsY7ytc8Q8m+8XBMXLXws1UHLQ2BWGcX3F7UC1qHP3OyBeX2YaZ3fhVKLZiPidgRnlsTjZefcqkjy0++oEg/EHzKdXQuoNsJIO6zK+yPKV/+QWW57+YTYN3585ktTqGG42RaIwmsb/+0orQmrBtphQWC8vu1te9m/kwbpS6s2O+t/3ng1B8AaGxi4Wqn7qVsUnGPecHr+M/iAMf6p+qqFdxcvYhMZMXfxlkd0q/eaW4LF6BhBhaaXr2LOXELSXUvTG2O4Q+g1A/QUfGApSh3Jo6XsZqj7P/GC86QnmqZTiotngXQlto7Tk+u8prTz4mRtv58xUzefsqSeG2npI1FVrdtV+Naa+bybyM87PSvX6Nu5TO8lqBQGsV4S0zwFyVHDjiKtZBZEgskypQhohQmKy7R2MZB2kSkCGCkIrcESlU3cMoXTGwOKbRCtKxY94ZD9K853cm0rUnF05Rc8J1DX4MVLqfetE1z2hsMukuI3tCEpcyjCsq123pYpOzO+e8f5HnhrKAs3uEfHsmjc1XS6VhRpyL58JglEwBNWEDgZQQdQ/+NhKzJprMRPknjaBEDHpl6dOmCtUlVSUiNODbzE7iW4mNomBJi18WPZsYlEwVTfRJv+Z04WmZ9zu2OLoHPVJ9nDAj/aw+lJRqo0NSVmIgow+1QVJQ9gbmvbSYfpwSqpbrSJpRaETRKCX3JT0ByVUwMu0obkKid7TsoBJervVo4OijWMz5GGM5dI+c+fwl6nmt5NIN04gpbEb5+tlmV8Qwb1s+byrKg5FJ89j75z+/8lE2+hxJBCSBJc6JtknXL6WDauZYbIVU2vgQ1QzHCS/MhTA/ssL7OFsU52E5hmJndYqDMz+6trrokhLCS1byrU7isY5MLh6k6SDZKUqxFzao43y8neQuLmA+oJSpAtWxki0Og0ZYxUfIc9RFnFBViuxu5RrjHZSBhK6UwpH6hZwjZq8s0i/jW3I3KUU0RpGsdiEmDTttkMkkB6nzuC6zI10sAOuEkspAN1BsPDIniPbJNuGvw1oc4k24FhsJfrxsplv8TDRpXoGowTu/3BcVRmqFsutCLeAMi55RTnoXUzJD6skdSBPA8QSqonoKMIiGYHYtp/1wjtNMX/mrt3j3c6IcXa0Kf6RfkY4z7EpfQcWLku5yEW99mGtqLTbkOS0w8W2RgJFNiQ4hFotslVdXxONMixHV7QjQk7q6LUpiqKNTSWw9OdfD0U1S/ieg1KPRU9IcDGfAxdzN24VL2IEtnZYtS3ONOBH4CSu4/NK2v7rPiXyIfsZ8VPdDx1sFCxmXJJxMyriQc27J/1gkNct5zgvDsxABMei3hSoWQ1Iiezaqy3Rwc4N+ZDnHCAyH27Sp/UZohhx3siBHpkvu/wDgbWoC", "shape": [5, 4, 1000, 2], "dtype": "bool"}}, "metadata": {"scheduler_timing": null, "stretch_values": null}}, {"results": {"meas": {"data": "eJw9UkuOE0EMfeWuZNxRg4poNNLsPIFFC80iR6hBg5RlJC6A4AIcwcmwCBFCOUIzaw7RS47CkiPw3IkoRYrbn+fnZ+MR6BqIfgKgmhZAi4T3KLVPmIWvFnEUWHVhjivewIVlqEzke7DiEUgVkgSJsQ/JykeHZni+Yb01mnBLsOTYdiMIx0zCORK/aCUUo5HaDPMK2IgdkKcukTjLcranN7Z4YNBQU2B2yZgpukZOPme881Jr2mJCsPgZGuRTlWv4+lpj1oK+rpCtft/cKyyRSF41wBcOU22Ow4BxJy7z3BHbeoxwhxR51ZiwC2F16ccFzl1E9edAV/jf0cNxKvR2J98G7JWYz9VovSWZUOsPSAi/Y6C1LynIUXTAC6MyHdpfMm//YlkGPBGusKLYgmLNuxRiGNU76o9gwPqbupyRUU65D/AtDqqfc96Wl9zdo9md1nUs8ApuSlL32JMBTeIIyjyWeJhkik3I9KfR5hkpFrBmY+xjmc1Z/SZDZRSNxbSQ1xe/MI+ZfY2SkMWOZsHQz9h2RZsS1akFO36FTlpwwvNyLwP9f7ySBYR3wBvZbIDVxZ+I69oN0EXwDs2nCwwAJ+WnqXiPu4mQDc2IE4YT+n/chlGQ", "shape": [3, 1000, 2], "dtype": "bool"}}, "metadata": {"scheduler_timing": null, "stretch_values": null}}], "metadata": {"chunk_timing": [{"start": "2026-08-07T08:43:35.622741", "stop": "2026-08-07T08:43:42.687708", "parts": [{"idx_item": 0, "size": 20, "permutation": [0, 1], "element_range": [0, 20, 1]}, {"idx_item": 1, "size": 3, "permutation": [0], "element_range": [0, 3, 1]}]}]}, "passthrough_data": {"post_processor": {"circuits_metadata": [{"list": [1, 2, 3]}, {"tuple": [1, 2, 3]}], "meas_type": "classified", "shots": 1000, "twirling": false, "version": "v0.1"}}, "semantic_role": "sampler_v2"} \ No newline at end of file diff --git a/test/unit/backwards_compatibility/test_load_jobs.py b/test/unit/backwards_compatibility/test_load_jobs.py new file mode 100644 index 000000000..a07ecb269 --- /dev/null +++ b/test/unit/backwards_compatibility/test_load_jobs.py @@ -0,0 +1,53 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2026. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests for backwards compatibility when loading jobs from earlier versions.""" + +from pathlib import Path + +from qiskit.primitives import PrimitiveResult + +from qiskit_ibm_runtime.fake_provider import FakeFez +from qiskit_ibm_runtime.qiskit_runtime_service import QiskitRuntimeService + +from ...decorators import mock_responses +from ...ibm_test_case import IBMTestCase +from ...registries import Backend, Job, OneInstanceNoBackendsRegistry + + +class SamplerTestCase(IBMTestCase): + """Test for loading Sampler jobs from earlier versions.""" + + @mock_responses(OneInstanceNoBackendsRegistry) + def test_wrapper_sampler_jobs(self, registry: OneInstanceNoBackendsRegistry) -> None: + """Test stored WrapperSampler jobs.""" + job_id = "d9qp8cvpemts73crjmsg" + + resources_path = Path(__file__).resolve().parent / "resources" + job_details = (resources_path / f"{job_id}_details.json").read_text(encoding="utf-8") + job_results = (resources_path / f"{job_id}_results.json").read_text(encoding="utf-8") + + # Prepare the contents of the registry. + registry.add_backend(Backend.from_(FakeFez)) + registry.add_job( + Job(job_id, "ibm_fez", raw_details=job_details, raw_results=job_results), "a" + ) + + service = QiskitRuntimeService(token="my_token") + job = service.job(job_id) + result = job.result() + + # Job should be an executor (wrapped sampler) job. + self.assertEqual(job.primitive_id, "executor") + # Result should be loaded correctly + self.assertIsInstance(result, PrimitiveResult) + self.assertIsInstance(result.metadata, dict) From 6707b1fed550c0c1b356e31d561c3c7675e69cc0 Mon Sep 17 00:00:00 2001 From: "Diego M. Rodriguez" Date: Fri, 7 Aug 2026 14:08:59 +0200 Subject: [PATCH 6/6] Fix docstring --- test/registries.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/registries.py b/test/registries.py index 531c3eeb8..19d8fa5dc 100644 --- a/test/registries.py +++ b/test/registries.py @@ -511,7 +511,7 @@ def callback_jobs_results(self, request: PreparedRequest) -> CallbackResult: return (200, {"Content-Type": "application/json"}, response_body) def get_crn_from_request(self, request: PreparedRequest) -> Instance: - """Retrieve the instance CRN from the request headers.""" + """Retrieve the `Instance` from the request headers.""" instance_crn = request.headers.get("Service-CRN") return next( instance for instance in self.instances.values() if instance.crn == instance_crn