add tests

qiskit_ibm_runtime/base_primitive.py

@@ -75,15 +75,6 @@ def __init__(
         self._service: QiskitRuntimeService = None
         self._backend: Optional[IBMBackend] = None
 
-        if options is None:
-            self._options = asdict(Options())
-        elif isinstance(options, Options):
-            self._options = asdict(copy.deepcopy(options))
-        else:
-            options_copy = copy.deepcopy(options)
-            default_options = asdict(Options())
-            self._options = Options._merge_options(default_options, options_copy)
-
         if isinstance(session, Session):
             self._session = session
             self._service = self._session.service
@@ -148,6 +139,21 @@ def __init__(
                 raise ValueError(
                     "A backend or session must be specified when not using ibm_cloud channel."
                 )
+        self._simulator_backend = (
+            self._backend.configuration().simulator if self._backend else False
+        )
+
+        if options is None:
+            self._options = asdict(Options())
+        elif isinstance(options, Options):
+            self._options = asdict(copy.deepcopy(options))
+        else:
+            options_copy = copy.deepcopy(options)
+            default_options = asdict(Options())
+            self._options = Options._merge_options_with_defaults(
+                default_options, options_copy, is_simulator=self._simulator_backend
+            )
+
         # self._first_run = True
         # self._circuits_map = {}
         # if self.circuits:
@@ -169,8 +175,9 @@ def _run_primitive(self, primitive_inputs: Dict, user_kwargs: Dict) -> RuntimeJo
         Returns:
             Submitted job.
         """
-        is_simulator = self._backend.configuration().simulator if self._backend else False
-        combined = Options._finalize_options(self._options, user_kwargs, is_simulator)
+        combined = Options._merge_options_with_defaults(
+            self._options, user_kwargs, self._simulator_backend
+        )
         self._validate_options(combined)
 
         primitive_inputs.update(Options._get_program_inputs(combined))
@@ -228,7 +235,9 @@ def set_options(self, **fields: Any) -> None:
         Args:
             **fields: The fields to update the options
         """
-        self._options = Options._merge_options(self._options, fields)
+        self._options = Options._merge_options_with_defaults(
+            self._options, fields, self._simulator_backend
+        )
 
     @abstractmethod
     def _validate_options(self, options: dict) -> None:

qiskit_ibm_runtime/options/options.py

@@ -27,9 +27,8 @@
 from .resilience_options import ResilienceOptions, _ZneOptions, _PecOptions
 from .twirling_options import TwirlingOptions
 from ..runtime_options import RuntimeOptions
-from .resilience_level_defaults import _default_resilience_options
 
-DDSequenceType = Literal["XX", "XpXm", "XY4"]
+DDSequenceType = Literal[None, "XX", "XpXm", "XY4"]
 
 
 @_flexible
@@ -106,7 +105,7 @@ class Options:
     optimization_level: Optional[int] = None
     resilience_level: Optional[int] = None
     max_execution_time: Optional[int] = None
-    dynamical_decoupling: Optional[DDSequenceType] = "XX"
+    dynamical_decoupling: Optional[DDSequenceType] = None
     transpilation: Union[TranspilationOptions, Dict] = field(default_factory=TranspilationOptions)
     resilience: Union[ResilienceOptions, Dict] = field(default_factory=ResilienceOptions)
     execution: Union[ExecutionOptions, Dict] = field(default_factory=ExecutionOptions)
@@ -207,34 +206,6 @@ def validate_options(options: dict) -> None:
         ExecutionOptions.validate_execution_options(options.get("execution"))
         SimulatorOptions.validate_simulator_options(options.get("simulator"))
 
-    @staticmethod
-    def _remove_none_values(options: dict) -> dict:
-        """Remove `None` values from the options dictionary."""
-        new_options = {}
-        for key, value in options.items():
-            if value is not None:
-                if isinstance(value, dict):
-                    new_suboptions = {}
-                    for subkey, subvalue in value.items():
-                        if subvalue is not None:
-                            new_suboptions[subkey] = subvalue
-                    new_options[key] = new_suboptions
-                else:
-                    new_options[key] = value
-
-        return new_options
-
-    @staticmethod
-    def _set_default_resilience_options(options: dict) -> dict:
-        """Set default resilience options for resilience level 2."""
-        if options["resilience_level"] == 2:
-            if not options["resilience"]["noise_factors"]:
-                options["resilience"]["noise_factors"] = (1, 3, 5)
-            if not options["resilience"]["extrapolator"]:
-                options["resilience"]["extrapolator"] = "LinearExtrapolator"
-
-        return options
-
     @staticmethod
     def _get_runtime_options(options: dict) -> dict:
         """Extract runtime options.
@@ -301,8 +272,12 @@ def _update_options(old: dict, new: dict, matched: Optional[dict] = None) -> Non
         return combined
 
     @classmethod
-    def _finalize_options(cls, primitive_options: dict, overwrite_options: Optional[dict] = None, is_simulator: bool = False):
-
+    def _merge_options_with_defaults(
+        cls,
+        primitive_options: dict,
+        overwrite_options: Optional[dict] = None,
+        is_simulator: bool = False,
+    ):
         def _get_merged_value(name, first: dict = None, second: dict = None):
             first = first or overwrite_options
             second = second or primitive_options
@@ -311,13 +286,27 @@ def _get_merged_value(name, first: dict = None, second: dict = None):
         # 1. Determine optimization and resilience levels
         optimization_level = _get_merged_value("optimization_level")
         resilience_level = _get_merged_value("resilience_level")
-        noise_model = _get_merged_value("noise_model", first=overwrite_options.get("simulator", {}), second=primitive_options.get("simulator", {}))
+        noise_model = _get_merged_value(
+            "noise_model",
+            first=overwrite_options.get("simulator", {}),
+            second=primitive_options.get("simulator", {}),
+        )
         if optimization_level is None:
-            optimization_level = cls._DEFAULT_NOISELESS_OPTIMIZATION_LEVEL if (is_simulator and noise_model is None) else cls._DEFAULT_OPTIMIZATION_LEVEL
+            optimization_level = (
+                cls._DEFAULT_NOISELESS_OPTIMIZATION_LEVEL
+                if (is_simulator and noise_model is None)
+                else cls._DEFAULT_OPTIMIZATION_LEVEL
+            )
         if resilience_level is None:
-            resilience_level = cls._DEFAULT_NOISELESS_RESILIENCE_LEVEL if (is_simulator and not noise_model is None) else cls._DEFAULT_RESILIENCE_LEVEL
+            resilience_level = (
+                cls._DEFAULT_NOISELESS_RESILIENCE_LEVEL
+                if (is_simulator and noise_model is None)
+                else cls._DEFAULT_RESILIENCE_LEVEL
+            )
 
         # 2. Determine the default resilience options
+        if resilience_level not in _DEFAULT_RESILIENCE_LEVEL_OPTIONS.keys():
+            raise ValueError(f"resilience_level {resilience_level} is not a valid value.")
         default_options = asdict(_DEFAULT_RESILIENCE_LEVEL_OPTIONS[resilience_level])
         default_options["optimization_level"] = optimization_level
 
@@ -333,38 +322,59 @@ def _get_merged_value(name, first: dict = None, second: dict = None):
         return final_options
 
 
-
 @dataclass(frozen=True)
 class _ResilienceLevel0Options:
     resilience_level: int = 0
-    resilience: ResilienceOptions = field(default=ResilienceOptions(measure_noise_mitigation=False, zne_mitigation=False, pec_mitigation=False))
+    resilience: ResilienceOptions = field(
+        default=ResilienceOptions(
+            measure_noise_mitigation=False, zne_mitigation=False, pec_mitigation=False
+        )
+    )
     twirling: TwirlingOptions = field(default=TwirlingOptions(gates=False, measure=False))
 
 
 @dataclass(frozen=True)
 class _ResilienceLevel1Options:
     resilience_level: int = 1
-    resilience: ResilienceOptions = field(default=ResilienceOptions(measure_noise_mitigation=True, zne_mitigation=False, pec_mitigation=False))
-    twirling: TwirlingOptions = field(default=TwirlingOptions(gates=True, measure=True, strategy="active-accum"))
+    resilience: ResilienceOptions = field(
+        default=ResilienceOptions(
+            measure_noise_mitigation=True, zne_mitigation=False, pec_mitigation=False
+        )
+    )
+    twirling: TwirlingOptions = field(
+        default=TwirlingOptions(gates=True, measure=True, strategy="active-accum")
+    )
 
 
 @dataclass(frozen=True)
 class _ResilienceLevel2Options:
     resilience_level: int = 2
-    resilience: ResilienceOptions = field(default=ResilienceOptions(measure_noise_mitigation=True, pec_mitigation=False, **asdict(_ZneOptions())))
-    twirling: TwirlingOptions = field(default=TwirlingOptions(gates=True, measure=True, strategy="active-accum"))
+    resilience: ResilienceOptions = field(
+        default=ResilienceOptions(
+            measure_noise_mitigation=True, pec_mitigation=False, **asdict(_ZneOptions())
+        )
+    )
+    twirling: TwirlingOptions = field(
+        default=TwirlingOptions(gates=True, measure=True, strategy="active-accum")
+    )
 
 
 @dataclass(frozen=True)
 class _ResilienceLevel3Options:
     resilience_level: int = 3
-    resilience: ResilienceOptions = field(default=ResilienceOptions(measure_noise_mitigation=True, zne_mitigation=False, **asdict(_PecOptions())))
-    twirling: TwirlingOptions = field(default=TwirlingOptions(gates=True, measure=True, strategy="active"))
+    resilience: ResilienceOptions = field(
+        default=ResilienceOptions(
+            measure_noise_mitigation=True, zne_mitigation=False, **asdict(_PecOptions())
+        )
+    )
+    twirling: TwirlingOptions = field(
+        default=TwirlingOptions(gates=True, measure=True, strategy="active")
+    )
 
 
 _DEFAULT_RESILIENCE_LEVEL_OPTIONS = {
     0: _ResilienceLevel0Options(),
     1: _ResilienceLevel1Options(),
     2: _ResilienceLevel2Options(),
-    3: _ResilienceLevel3Options()
+    3: _ResilienceLevel3Options(),
 }

qiskit_ibm_runtime/sampler.py

@@ -166,13 +166,11 @@ def _validate_options(self, options: dict) -> None:
             qctrl_validate(options)
             return
 
-        if options.get("resilience_level") and not options.get("resilience_level") in [
-            0,
-            1,
-        ]:
+        valid_levels = list(range(Options._MAX_RESILIENCE_LEVEL_SAMPLER + 1))
+        if options.get("resilience_level") and not options.get("resilience_level") in valid_levels:
             raise ValueError(
-                f"resilience_level can only take the values "
-                f"{list(range(Options._MAX_RESILIENCE_LEVEL_SAMPLER + 1))} in Sampler"
+                f"resilience_level {options.get('resilience_level')} is not a valid value."
+                f"It can only take the values {valid_levels} in Sampler."
             )
         Options.validate_options(options)
 

test/unit/test_ibm_primitives.py

@@ -34,6 +34,7 @@
 )
 from qiskit_ibm_runtime.ibm_backend import IBMBackend
 import qiskit_ibm_runtime.session as session_pkg
+from qiskit_ibm_runtime.options.utils import _remove_dict_none_values
 
 from ..ibm_test_case import IBMTestCase
 from ..utils import (
@@ -81,9 +82,7 @@ def test_dict_options(self):
             for options in options_vars:
                 with self.subTest(primitive=cls, options=options):
                     inst = cls(session=MagicMock(spec=MockSession), options=options)
-                    expected = asdict(Options())
-                    self._update_dict(expected, copy.deepcopy(options))
-                    self.assertDictEqual(expected, inst.options.__dict__)
+                    self.assertTrue(dict_paritally_equal(inst.options.__dict__, options))
 
     def test_backend_in_options(self):
         """Test specifying backend in options."""
@@ -533,10 +532,6 @@ def test_accept_level_1_options(self):
                     # Make sure the values are equal.
                     inst1_options = inst1.options.__dict__
                     expected_dict = inst2.options.__dict__
-                    self.assertTrue(
-                        dict_paritally_equal(inst1_options, expected_dict),
-                        f"inst_options={inst1_options}, options={opts}",
-                    )
                     # Make sure the structure didn't change.
                     self.assertTrue(
                         dict_keys_equal(inst1_options, expected_dict),

test/unit/test_options.py

@@ -207,7 +207,7 @@ def test_unsupported_options(self):
                 "noise_factors": (0, 2, 4),
                 "extrapolator": "LinearExtrapolator",
             },
-            "twirling": {}
+            "twirling": {},
         }
         Options.validate_options(options)
         for opt in ["simulator", "transpilation", "execution"]:
@@ -315,21 +315,60 @@ def test_qctrl_overrides(self):
                 _warn_and_clean_options(option)
                 self.assertEqual(expected_, option)
 
-    def test_final_options_res0(self):
-        """Test final options with resilience 0."""
+    def test_merge_with_defaults_overwrite(self):
+        """Test merge_with_defaults with different overwrite."""
+        expected = {"twirling": {"measure": True}}
         all_options = [
-            ({"twirling": {"measure": True}}, {}, {"twirling": {"measure": True}}),
-            ({}, {"twirling": {"measure": True}}, {"twirling": {"measure": True}}),
-            ({"twirling": {"measure": True}}, {"twirling": {"measure": False}}, {"twirling": {"measure": False}})
+            ({"twirling": {"measure": True}}, {}),
+            ({}, {"twirling": {"measure": True}}),
+            ({"twirling": {"measure": False}}, {"twirling": {"measure": True}}),
         ]
 
-        for old, new, expected in all_options:
+        for old, new in all_options:
             with self.subTest(old=old, new=new):
                 old["resilience_level"] = 0
-                final = Options._finalize_options(old, new)
+                final = Options._merge_options_with_defaults(old, new)
                 self.assertTrue(dict_paritally_equal(final, expected))
                 self.assertEqual(final["resilience_level"], 0)
                 res_dict = final["resilience"]
                 self.assertFalse(res_dict["measure_noise_mitigation"])
                 self.assertFalse(res_dict["zne_mitigation"])
                 self.assertFalse(res_dict["pec_mitigation"])
+
+    def test_merge_with_defaults_different_level(self):
+        """Test merge_with_defaults with different resilience level."""
+
+        old = {"resilience_level": 0}
+        new = {"resilience_level": 3, "measure_noise_mitigation": False}
+        final = Options._merge_options_with_defaults(old, new)
+        self.assertEqual(final["resilience_level"], 3)
+        res_dict = final["resilience"]
+        self.assertFalse(res_dict["measure_noise_mitigation"])
+        self.assertFalse(res_dict["zne_mitigation"])
+        self.assertTrue(res_dict["pec_mitigation"])
+
+    def test_merge_with_defaults_noiseless_simulator(self):
+        """Test merge_with_defaults with noiseless simulator."""
+
+        new = {"measure_noise_mitigation": True}
+        final = Options._merge_options_with_defaults({}, new, is_simulator=True)
+        self.assertEqual(final["resilience_level"], 0)
+        self.assertEqual(final["optimization_level"], 1)
+        res_dict = final["resilience"]
+        self.assertTrue(res_dict["measure_noise_mitigation"])
+        self.assertFalse(res_dict["zne_mitigation"])
+        self.assertFalse(res_dict["pec_mitigation"])
+
+    def test_merge_with_defaults_noisy_simulator(self):
+        """Test merge_with_defaults with noisy simulator."""
+
+        new = {"measure_noise_mitigation": False}
+        final = Options._merge_options_with_defaults(
+            {"simulator": {"noise_model": "foo"}}, new, is_simulator=True
+        )
+        self.assertEqual(final["resilience_level"], 1)
+        self.assertEqual(final["optimization_level"], 3)
+        res_dict = final["resilience"]
+        self.assertFalse(res_dict["measure_noise_mitigation"])
+        self.assertFalse(res_dict["zne_mitigation"])
+        self.assertFalse(res_dict["pec_mitigation"])