diff --git a/qiskit_ibm_runtime/__init__.py b/qiskit_ibm_runtime/__init__.py
index 2d0884242..11975505e 100644
--- a/qiskit_ibm_runtime/__init__.py
+++ b/qiskit_ibm_runtime/__init__.py
@@ -98,7 +98,7 @@
     from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager
 
     from qiskit_ibm_runtime import Session
-    from qiskit_ibm_runtime import Sampler
+    from qiskit_ibm_runtime import SamplerV2 as Sampler
     from qiskit_ibm_runtime.fake_provider import FakeManilaV2
 
     # Bell Circuit
diff --git a/qiskit_ibm_runtime/base_primitive.py b/qiskit_ibm_runtime/base_primitive.py
index 33f1b36a8..bfa4b739e 100644
--- a/qiskit_ibm_runtime/base_primitive.py
+++ b/qiskit_ibm_runtime/base_primitive.py
@@ -35,7 +35,7 @@
 from .ibm_backend import IBMBackend
 from .utils.default_session import get_cm_session
 from .utils.deprecation import issue_deprecation_msg
-from .utils.utils import validate_isa_circuits
+from .utils.utils import validate_isa_circuits, is_simulator
 from .constants import DEFAULT_DECODERS
 from .qiskit_runtime_service import QiskitRuntimeService
 from .fake_provider.local_service import QiskitRuntimeLocalService
@@ -56,15 +56,15 @@ class BasePrimitiveV2(ABC, Generic[OptionsT]):
 
     def __init__(
         self,
-        backend: Optional[Union[str, IBMBackend]] = None,
-        session: Optional[Union[Session, str, IBMBackend]] = None,
+        backend: Optional[Union[str, BackendV1, BackendV2]] = None,
+        session: Optional[Session] = None,
         options: Optional[Union[Dict, OptionsT]] = None,
     ):
         """Initializes the primitive.
 
         Args:
 
-            backend: Backend to run the primitive. This can be a backend name or an :class:`IBMBackend`
+            backend: Backend to run the primitive. This can be a backend name or a ``Backend``
                 instance. If a name is specified, the default account (e.g. ``QiskitRuntimeService()``)
                 is used.
 
@@ -75,31 +75,33 @@ def __init__(
                 :class:`qiskit_ibm_runtime.Session` context manager, then the session is used.
                 Otherwise if IBM Cloud channel is used, a default backend is selected.
 
-            options: Primitive options, see :class:`Options` for detailed description.
-                The ``backend`` keyword is still supported but is deprecated.
+            options: Primitive options, see :class:`qiskit_ibm_runtime.options.EstimatorOptions`
+                and :class:`qiskit_ibm_runtime.options.SamplerOptions` for detailed description
+                on estimator and sampler options, respectively.
 
         Raises:
             ValueError: Invalid arguments are given.
         """
         self._session: Optional[Session] = None
-        self._service: QiskitRuntimeService = None
-        self._backend: Optional[IBMBackend] = None
+        self._service: QiskitRuntimeService | QiskitRuntimeLocalService = None
+        self._backend: Optional[BackendV1 | BackendV2] = None
 
         self._set_options(options)
 
         if isinstance(session, Session):
             self._session = session
             self._service = self._session.service
-            self._backend = self._service.backend(
-                name=self._session.backend(), instance=self._session._instance
-            )
+            self._backend = self._session._backend
             return
-        elif session is not None:
+        elif session is not None:  # type: ignore[unreachable]
             raise ValueError("session must be of type Session or None")
 
-        if isinstance(backend, IBMBackend):
+        if isinstance(backend, IBMBackend):  # type: ignore[unreachable]
             self._service = backend.service
             self._backend = backend
+        elif isinstance(backend, (BackendV1, BackendV2)):
+            self._service = QiskitRuntimeLocalService()
+            self._backend = backend
         elif isinstance(backend, str):
             self._service = (
                 QiskitRuntimeService()
@@ -123,6 +125,12 @@ def __init__(
                 raise ValueError(
                     "A backend or session must be specified when not using ibm_cloud channel."
                 )
+            issue_deprecation_msg(
+                "Not providing a backend is deprecated",
+                "0.22.0",
+                "Passing in a backend will be required, please provide a backend.",
+                3,
+            )
 
     def _run(self, pubs: Union[list[EstimatorPub], list[SamplerPub]]) -> RuntimeJobV2:
         """Run the primitive.
@@ -142,13 +150,11 @@ def _run(self, pubs: Union[list[EstimatorPub], list[SamplerPub]]) -> RuntimeJobV
 
         if self._backend:
             for pub in pubs:
-                if (
-                    getattr(self._backend, "target", None)
-                    and not self._backend.configuration().simulator
-                ):
+                if getattr(self._backend, "target", None) and not is_simulator(self._backend):
                     validate_isa_circuits([pub.circuit], self._backend.target)
 
-                self._backend.check_faulty(pub.circuit)
+                if isinstance(self._backend, IBMBackend):
+                    self._backend.check_faulty(pub.circuit)
 
         logger.info("Submitting job using options %s", primitive_options)
 
@@ -162,16 +168,23 @@ def _run(self, pubs: Union[list[EstimatorPub], list[SamplerPub]]) -> RuntimeJobV
             )
 
         if self._backend:
-            runtime_options["backend"] = self._backend.name
-            if "instance" not in runtime_options:
+            runtime_options["backend"] = self._backend
+            if "instance" not in runtime_options and isinstance(self._backend, IBMBackend):
                 runtime_options["instance"] = self._backend._instance
 
+        if isinstance(self._service, QiskitRuntimeService):
+            return self._service.run(
+                program_id=self._program_id(),
+                options=runtime_options,
+                inputs=primitive_inputs,
+                callback=options_dict.get("environment", {}).get("callback", None),
+                result_decoder=DEFAULT_DECODERS.get(self._program_id()),
+            )
+
         return self._service.run(
-            program_id=self._program_id(),
+            program_id=self._program_id(),  # type: ignore[arg-type]
             options=runtime_options,
             inputs=primitive_inputs,
-            callback=options_dict.get("environment", {}).get("callback", None),
-            result_decoder=DEFAULT_DECODERS.get(self._program_id()),
         )
 
     @property
diff --git a/qiskit_ibm_runtime/estimator.py b/qiskit_ibm_runtime/estimator.py
index 554edbc11..dc17fef31 100644
--- a/qiskit_ibm_runtime/estimator.py
+++ b/qiskit_ibm_runtime/estimator.py
@@ -121,8 +121,7 @@ def __init__(
                 :class:`qiskit_ibm_runtime.Session` context manager, then the session is used.
                 Otherwise if IBM Cloud channel is used, a default backend is selected.
 
-            options: Primitive options, see :class:`Options` for detailed description.
-                The ``backend`` keyword is still supported but is deprecated.
+            options: Estimator options, see :class:`EstimatorOptions` for detailed description.
 
         Raises:
             NotImplementedError: If "q-ctrl" channel strategy is used.
diff --git a/qiskit_ibm_runtime/fake_provider/local_service.py b/qiskit_ibm_runtime/fake_provider/local_service.py
index cb6570f35..b59fcf466 100644
--- a/qiskit_ibm_runtime/fake_provider/local_service.py
+++ b/qiskit_ibm_runtime/fake_provider/local_service.py
@@ -17,6 +17,8 @@
 import logging
 import copy
 from typing import Dict, Union, Literal
+import warnings
+from dataclasses import asdict
 
 from qiskit.utils import optionals
 from qiskit.providers.backend import BackendV1, BackendV2
@@ -25,6 +27,7 @@
 
 from ..runtime_options import RuntimeOptions
 from ..ibm_backend import IBMBackend
+from ..qiskit.primitives import BackendEstimatorV2, BackendSamplerV2  # type: ignore[attr-defined]
 
 logger = logging.getLogger(__name__)
 
@@ -66,19 +69,20 @@ def run(
             ValueError: If input is invalid.
             NotImplementedError: If using V2 primitives.
         """
-        # qrt_options: RuntimeOptions = options
         if isinstance(options, Dict):
-            qrt_options = RuntimeOptions(**options)
+            qrt_options = copy.deepcopy(options)
         else:
-            qrt_options = options
+            qrt_options = asdict(options)
+
+        backend = qrt_options.pop("backend", None)
 
         if program_id not in ["sampler", "estimator"]:
             raise ValueError("Only sampler and estimator are supported in local testing mode.")
-        if isinstance(qrt_options.backend, IBMBackend):
+        if isinstance(backend, IBMBackend):
             raise ValueError(
                 "Local testing mode is not supported when a cloud-based backend is used."
             )
-        if isinstance(qrt_options.backend, str):
+        if isinstance(backend, str):
             raise ValueError(
                 "Passing a backend name is not supported in local testing mode. "
                 "Please pass a backend instance."
@@ -94,25 +98,30 @@ def run(
             if program_id == "estimator":
                 primitive_inputs["observables"] = inputs.pop("observables")
             inputs.pop("parameters", None)
+
+            if optionals.HAS_AER:
+                # pylint: disable=import-outside-toplevel
+                from qiskit_aer.backends.aerbackend import AerBackend
+
+                if isinstance(backend, AerBackend):
+                    return self._run_aer_primitive_v1(
+                        primitive=program_id, options=inputs, inputs=primitive_inputs
+                    )
+
+            return self._run_backend_primitive_v1(
+                backend=backend,
+                primitive=program_id,
+                options=inputs,
+                inputs=primitive_inputs,
+            )
         else:
             primitive_inputs = {"pubs": inputs.pop("pubs")}
-            raise NotImplementedError("V2 primitives are not supported in local mode.")
-
-        if optionals.HAS_AER:
-            # pylint: disable=import-outside-toplevel
-            from qiskit_aer.backends.aerbackend import AerBackend
-
-            if isinstance(qrt_options.backend, AerBackend):
-                return self._run_aer_primitive_v1(
-                    primitive=program_id, options=inputs, inputs=primitive_inputs
-                )
-
-        return self._run_backend_primitive_v1(
-            backend=qrt_options.backend,
-            primitive=program_id,
-            options=inputs,
-            inputs=primitive_inputs,
-        )
+            return self._run_backend_primitive_v2(
+                backend=backend,
+                primitive=program_id,
+                options=inputs.get("options", {}),
+                inputs=primitive_inputs,
+            )
 
     def _run_aer_primitive_v1(
         self, primitive: Literal["sampler", "estimator"], options: dict, inputs: dict
@@ -197,3 +206,40 @@ def _run_backend_primitive_v1(
 
         primitive_inst.set_transpile_options(**transpilation_options)
         return primitive_inst.run(**inputs, **run_options)
+
+    def _run_backend_primitive_v2(
+        self,
+        backend: BackendV1 | BackendV2,
+        primitive: Literal["sampler", "estimator"],
+        options: dict,
+        inputs: dict,
+    ) -> PrimitiveJob:
+        """Run V2 backend primitive.
+
+        Args:
+            backend: The backend to run the primitive on.
+            primitive: Name of the primitive.
+            options: Primitive options to use.
+            inputs: Primitive inputs.
+
+        Returns:
+            The job object of the result of the primitive.
+        """
+        options_copy = copy.deepcopy(options)
+
+        prim_options = {}
+        if seed_simulator := options_copy.pop("simulator", {}).pop("seed_simulator", None):
+            prim_options["seed_simulator"] = seed_simulator
+        if primitive == "sampler":
+            if default_shots := options_copy.pop("default_shots", None):
+                prim_options["default_shots"] = default_shots
+            primitive_inst = BackendSamplerV2(backend=backend, options=prim_options)
+        else:
+            if default_precision := options_copy.pop("default_precision", None):
+                prim_options["default_precision"] = default_precision
+            primitive_inst = BackendEstimatorV2(backend=backend, options=prim_options)
+
+        if options_copy:
+            warnings.warn(f"Options {options_copy} have no effect in local testing mode.")
+
+        return primitive_inst.run(**inputs)
diff --git a/qiskit_ibm_runtime/options/utils.py b/qiskit_ibm_runtime/options/utils.py
index 1cff571b6..3bedff333 100644
--- a/qiskit_ibm_runtime/options/utils.py
+++ b/qiskit_ibm_runtime/options/utils.py
@@ -25,6 +25,7 @@
 
 from qiskit.providers.backend import Backend
 
+from ..utils.utils import is_simulator
 
 if TYPE_CHECKING:
     from ..options.options import BaseOptions
@@ -47,9 +48,7 @@ def set_default_error_levels(
     Returns:
         options with correct error level defaults.
     """
-    is_sim = False
-    if hasattr(backend, "configuration"):
-        is_sim = getattr(backend.configuration(), "simulator", False)
+    is_sim = is_simulator(backend)
 
     if options.get("optimization_level") is None:
         if is_sim and not options.get("simulator", {}).get("noise_model"):
diff --git a/qiskit_ibm_runtime/sampler.py b/qiskit_ibm_runtime/sampler.py
index 2277a0303..f11c09869 100644
--- a/qiskit_ibm_runtime/sampler.py
+++ b/qiskit_ibm_runtime/sampler.py
@@ -80,8 +80,7 @@ def __init__(
                 :class:`qiskit_ibm_runtime.Session` context manager, then the session is used.
                 Otherwise if IBM Cloud channel is used, a default backend is selected.
 
-            options: Primitive options, see :class:`Options` for detailed description.
-                The ``backend`` keyword is still supported but is deprecated.
+            options: Sampler options, see :class:`SamplerOptions` for detailed description.
 
         Raises:
             NotImplementedError: If "q-ctrl" channel strategy is used.
diff --git a/qiskit_ibm_runtime/utils/utils.py b/qiskit_ibm_runtime/utils/utils.py
index 7f60c6970..7eb39bf26 100644
--- a/qiskit_ibm_runtime/utils/utils.py
+++ b/qiskit_ibm_runtime/utils/utils.py
@@ -11,12 +11,13 @@
 # that they have been altered from the originals.
 
 """General utility functions."""
+
+from __future__ import annotations
 import copy
 import keyword
 import logging
 import os
 import re
-import hashlib
 from queue import Queue
 from threading import Condition
 from typing import List, Optional, Any, Dict, Union, Tuple, Sequence
@@ -29,9 +30,24 @@
 from ibm_platform_services import ResourceControllerV2  # pylint: disable=import-error
 from qiskit.circuit import QuantumCircuit
 from qiskit.transpiler import Target
+from qiskit.providers.backend import BackendV1, BackendV2
 from qiskit_ibm_runtime.exceptions import IBMInputValueError
 
 
+def is_simulator(backend: BackendV1 | BackendV2) -> bool:
+    """Return true if the backend is a simulator.
+
+    Args:
+        backend: Backend to check.
+
+    Returns:
+        True if backend is a simulator.
+    """
+    if hasattr(backend, "configuration"):
+        return getattr(backend.configuration(), "simulator", False)
+    return getattr(backend, "simulator", False)
+
+
 def is_isa_circuit(circuit: QuantumCircuit, target: Target) -> str:
     """Checks if the circuit is an ISA circuit, meaning that it has a layout and that it
     only uses instructions that exist in the target.
@@ -310,13 +326,6 @@ def _filter_value(data: Dict[str, Any], filter_keys: List[Union[str, Tuple[str,
                 _filter_value(value, filter_keys)
 
 
-def _hash(hash_str: str) -> str:
-    """Hashes and returns a digest.
-    blake2s is supposedly faster than SHAs.
-    """
-    return hashlib.blake2s(hash_str.encode()).hexdigest()
-
-
 class RefreshQueue(Queue):
     """A queue that replaces the oldest item with the new item being added when full.
 
diff --git a/test/unit/test_estimator.py b/test/unit/test_estimator.py
index f420e6641..5d4c60a15 100644
--- a/test/unit/test_estimator.py
+++ b/test/unit/test_estimator.py
@@ -12,8 +12,6 @@
 
 """Tests for estimator class."""
 
-from unittest.mock import MagicMock
-
 import numpy as np
 from ddt import data, ddt
 
@@ -28,7 +26,6 @@
 from ..ibm_test_case import IBMTestCase
 from ..utils import (
     get_mocked_backend,
-    MockSession,
     dict_paritally_equal,
     transpile_pubs,
     get_primitive_inputs,
@@ -115,11 +112,10 @@ def test_unsupported_values_for_estimator_options(self):
 
     def test_pec_simulator(self):
         """Test error is raised when using pec on simulator without coupling map."""
+        backend = get_mocked_backend()
+        backend.configuration().simulator = True
 
-        session = MagicMock(spec=MockSession)
-        session.service.backend().configuration().simulator = True
-
-        inst = EstimatorV2(session=session, options={"resilience": {"pec_mitigation": True}})
+        inst = EstimatorV2(backend=backend, options={"resilience": {"pec_mitigation": True}})
         with self.assertRaises(ValueError) as exc:
             inst.run(**get_primitive_inputs(inst))
         self.assertIn("coupling map", str(exc.exception))
diff --git a/test/unit/test_ibm_primitives_v2.py b/test/unit/test_ibm_primitives_v2.py
index 4286338a9..a522c8eae 100644
--- a/test/unit/test_ibm_primitives_v2.py
+++ b/test/unit/test_ibm_primitives_v2.py
@@ -47,6 +47,7 @@
     get_primitive_inputs,
     get_mocked_backend,
     bell,
+    get_mocked_session,
 )
 
 
@@ -75,8 +76,9 @@ def test_dict_options(self, primitive):
             },
             {"default_shots": 1000},
         ]
+        backend = get_mocked_backend()
         for options in options_vars:
-            inst = primitive(session=MagicMock(spec=MockSession), options=options)
+            inst = primitive(backend=backend, options=options)
             self.assertTrue(dict_paritally_equal(asdict(inst.options), options))
 
     @combine(
@@ -121,9 +123,10 @@ def test_image(self, primitive, opts):
     @data(EstimatorV2, SamplerV2)
     def test_options_copied(self, primitive):
         """Test modifying original options does not affect primitives."""
+        backend = get_mocked_backend()
         options = primitive._options_class()
         options.max_execution_time = 100
-        inst = primitive(session=MagicMock(spec=MockSession), options=options)
+        inst = primitive(backend=backend, options=options)
         options.max_execution_time = 200
         self.assertEqual(inst.options.max_execution_time, 100)
 
@@ -131,24 +134,24 @@ def test_options_copied(self, primitive):
     def test_init_with_backend_str(self, primitive):
         """Test initializing a primitive with a backend name."""
         backend_name = "ibm_gotham"
+        mock_backend = get_mocked_backend(name=backend_name)
+        mock_service_inst = mock_backend.service
 
-        with patch("qiskit_ibm_runtime.base_primitive.QiskitRuntimeService") as mock_service:
-            mock_service.reset_mock()
-            mock_service_inst = MagicMock()
-            mock_service.return_value = mock_service_inst
-            mock_backend = MagicMock()
-            mock_backend.name = backend_name
-            mock_backend.target = None
-            mock_service.global_service = None
-            mock_service_inst.backend.return_value = mock_backend
+        class MockQRTService:
+            """Mock class used to create a new QiskitRuntimeService."""
+
+            global_service = None
 
+            def __new__(cls, *args, **kwargs):  # pylint: disable=unused-argument
+                return mock_service_inst
+
+        with patch("qiskit_ibm_runtime.base_primitive.QiskitRuntimeService", new=MockQRTService):
             inst = primitive(backend=backend_name)
-            mock_service.assert_called_once()
             self.assertIsNone(inst.session)
             inst.run(**get_primitive_inputs(inst))
             mock_service_inst.run.assert_called_once()
             runtime_options = mock_service_inst.run.call_args.kwargs["options"]
-            self.assertEqual(runtime_options["backend"], backend_name)
+            self.assertEqual(runtime_options["backend"], mock_backend)
 
     @data(EstimatorV2, SamplerV2)
     def test_init_with_session_backend_str(self, primitive):
@@ -173,7 +176,7 @@ def test_init_with_backend_instance(self, primitive):
         inst.run(**get_primitive_inputs(inst))
         service.run.assert_called_once()
         runtime_options = service.run.call_args.kwargs["options"]
-        self.assertEqual(runtime_options["backend"], backend.name)
+        self.assertEqual(runtime_options["backend"], backend)
 
         with self.assertRaises(ValueError) as exc:
             inst = primitive(session=backend)
@@ -183,12 +186,11 @@ def test_init_with_backend_instance(self, primitive):
     @data(EstimatorV2, SamplerV2)
     def test_init_with_backend_session(self, primitive):
         """Test initializing a primitive with both backend and session."""
-        session = MagicMock(spec=MockSession)
         backend_name = "ibm_gotham"
+        session = get_mocked_session(get_mocked_backend(backend_name))
 
         session.reset_mock()
         inst = primitive(session=session, backend=backend_name)
-        inst._backend.target = None
         self.assertIsNotNone(inst.session)
         inst.run(**get_primitive_inputs(inst))
         session.run.assert_called_once()
@@ -239,7 +241,7 @@ def test_default_session_cm_new_backend(self, primitive):
             inst.run(**get_primitive_inputs(inst))
             service.run.assert_called_once()
             runtime_options = service.run.call_args.kwargs["options"]
-            self.assertEqual(runtime_options["backend"], backend.name)
+            self.assertEqual(runtime_options["backend"], backend)
 
     @data(EstimatorV2, SamplerV2)
     def test_no_session(self, primitive):
@@ -438,11 +440,10 @@ def test_run_same_session(self):
         """Test multiple runs within a session."""
         num_runs = 5
         primitives = [EstimatorV2, SamplerV2]
-        session = MagicMock(spec=MockSession)
+        session = get_mocked_session()
         for idx in range(num_runs):
             cls = primitives[idx % len(primitives)]
             inst = cls(session=session)
-            inst._backend.target = None
             inst.run(**get_primitive_inputs(inst))
         self.assertEqual(session.run.call_count, num_runs)
 
@@ -457,9 +458,9 @@ def test_set_options(self, primitive, new_opts):
         """Test set options."""
         opt_cls = primitive._options_class
         options = opt_cls(default_shots=100)
+        backend = get_mocked_backend()
 
-        session = MagicMock(spec=MockSession)
-        inst = primitive(session=session, options=options)
+        inst = primitive(backend=backend, options=options)
         inst.options.update(**new_opts)
         # Make sure the values are equal.
         inst_options = asdict(inst.options)
@@ -489,12 +490,12 @@ def test_accept_level_1_options(self, primitive):
         expected_list[1].default_shots = 1024
         expected_list[2].simulator.seed_simulator = 123
         expected_list[3].environment.log_level = "ERROR"
+        backend = get_mocked_backend()
 
-        session = MagicMock(spec=MockSession)
         for opts, expected in zip(options_dicts, expected_list):
             with self.subTest(options=opts):
-                inst1 = primitive(session=session, options=opts)
-                inst2 = primitive(session=session, options=expected)
+                inst1 = primitive(backend=backend, options=opts)
+                inst2 = primitive(backend=backend, options=expected)
                 self.assertEqual(inst1.options, inst2.options)
 
     @data(EstimatorV2, SamplerV2)
diff --git a/test/unit/test_local_mode.py b/test/unit/test_local_mode.py
index 7ba496f2b..0d5203b07 100644
--- a/test/unit/test_local_mode.py
+++ b/test/unit/test_local_mode.py
@@ -12,10 +12,13 @@
 
 """Tests for local mode."""
 
+import warnings
+
 from ddt import data, ddt
 
 from qiskit_aer import AerSimulator
-from qiskit.primitives import EstimatorResult, SamplerResult
+from qiskit.primitives import EstimatorResult, SamplerResult, PrimitiveResult, PubResult
+from qiskit.primitives.containers.data_bin import DataBin
 
 from qiskit_ibm_runtime.fake_provider import FakeManila, FakeManilaV2
 from qiskit_ibm_runtime import (
@@ -24,6 +27,8 @@
     Options,
     Session,
     Batch,
+    SamplerV2,
+    EstimatorV2,
 )
 
 from ..ibm_test_case import IBMTestCase
@@ -34,8 +39,8 @@
 
 
 @ddt
-class TestLocalMode(IBMTestCase):
-    """Class for testing the Sampler and Estimator classes."""
+class TestLocalModeV1(IBMTestCase):
+    """Class for testing local mode for v1 primitives."""
 
     @combine(backend=[FakeManila(), FakeManilaV2(), AerSimulator()], num_sets=[1, 3])
     def test_v1_sampler(self, backend, num_sets):
@@ -119,6 +124,104 @@ def test_estimator_v1_session(self, session_cls, backend):
             self.assertEqual(len(result.values), 1)
             self.assertEqual(len(result.metadata), 1)
 
+
+@ddt
+class TestLocalModeV2(IBMTestCase):
+    """Class for testing local mode for V2 primitives."""
+
+    @combine(backend=[FakeManila(), FakeManilaV2(), AerSimulator()], num_sets=[1, 3])
+    def test_v2_sampler(self, backend, num_sets):
+        """Test V2 Sampler on a local backend."""
+        inst = SamplerV2(backend=backend)
+        job = inst.run(**get_primitive_inputs(inst, backend=backend, num_sets=num_sets))
+        result = job.result()
+        self.assertIsInstance(result, PrimitiveResult)
+        self.assertEqual(len(result), num_sets)
+        for pub_result in result:
+            self.assertIsInstance(pub_result, PubResult)
+            self.assertIsInstance(pub_result.data, DataBin)
+            self.assertIsInstance(pub_result.metadata, dict)
+
+    @combine(backend=[FakeManila(), FakeManilaV2(), AerSimulator()], num_sets=[1, 3])
+    def test_v2_estimator(self, backend, num_sets):
+        """Test V2 Estimator on a local backend."""
+        inst = EstimatorV2(backend=backend)
+        job = inst.run(**get_primitive_inputs(inst, backend=backend, num_sets=num_sets))
+        result = job.result()
+        self.assertIsInstance(result, PrimitiveResult)
+        self.assertEqual(len(result), num_sets)
+        for pub_result in result:
+            self.assertIsInstance(pub_result, PubResult)
+            self.assertIsInstance(pub_result.data, DataBin)
+            self.assertIsInstance(pub_result.metadata, dict)
+
+    @data(FakeManila(), FakeManilaV2(), AerSimulator.from_backend(FakeManila()))
+    def test_v2_sampler_with_accepted_options(self, backend):
+        """Test V2 sampler with accepted options."""
+        options = {"default_shots": 10, "simulator": {"seed_simulator": 42}}
+        inst = SamplerV2(backend=backend, options=options)
+        job = inst.run(**get_primitive_inputs(inst, backend=backend))
+        pub_result = job.result()[0]
+        self.assertEqual(pub_result.data.meas.num_shots, 10)
+        self.assertDictEqual(
+            pub_result.data.meas.get_counts(), {"00010": 1, "00011": 2, "00000": 7}
+        )
+
+    @data(FakeManila(), FakeManilaV2(), AerSimulator.from_backend(FakeManila()))
+    def test_v2_estimator_with_accepted_options(self, backend):
+        """Test V1 estimator with accepted options."""
+        options = {"default_precision": 0.03125, "simulator": {"seed_simulator": 42}}
+        inst = EstimatorV2(backend=backend, options=options)
+        job = inst.run(**get_primitive_inputs(inst, backend=backend))
+        pub_result = job.result()[0]
+        self.assertDictEqual(pub_result.metadata, {"target_precision": 0.03125})
+        self.assertEqual(pub_result.data.evs[0], 0.197265625)
+
+    @combine(
+        primitive=[SamplerV2, EstimatorV2], backend=[FakeManila(), FakeManilaV2(), AerSimulator()]
+    )
+    def test_primitve_v2_with_not_accepted_options(self, primitive, backend):
+        """Test V1 primitive with accepted options."""
+        options = {
+            "max_execution_time": 200,
+            "dynamical_decoupling": {"enable": True},
+            "simulator": {"seed_simulator": 42},
+        }
+        inst = primitive(backend=backend, options=options)
+        with warnings.catch_warnings(record=True) as warns:
+            job = inst.run(**get_primitive_inputs(inst, backend=backend))
+            _ = job.result()
+            self.assertEqual(len(warns), 1)
+            self.assertIn("dynamical_decoupling", str(warns[0].message))
+
+    @combine(session_cls=[Session, Batch], backend=[FakeManila(), FakeManilaV2(), AerSimulator()])
+    def test_sampler_v2_session(self, session_cls, backend):
+        """Testing running v2 sampler inside session."""
+        with session_cls(backend=backend) as session:
+            inst = SamplerV2(session=session)
+            job = inst.run(**get_primitive_inputs(inst, backend=backend))
+            result = job.result()
+            self.assertIsInstance(result, PrimitiveResult)
+            self.assertEqual(len(result), 1)
+            for pub_result in result:
+                self.assertIsInstance(pub_result, PubResult)
+                self.assertIsInstance(pub_result.data, DataBin)
+                self.assertIsInstance(pub_result.metadata, dict)
+
+    @combine(session_cls=[Session, Batch], backend=[FakeManila(), FakeManilaV2(), AerSimulator()])
+    def test_estimator_v2_session(self, session_cls, backend):
+        """Testing running v2 estimator inside session."""
+        with session_cls(backend=backend) as session:
+            inst = EstimatorV2(session=session)
+            job = inst.run(**get_primitive_inputs(inst, backend=backend))
+            result = job.result()
+            self.assertIsInstance(result, PrimitiveResult)
+            self.assertEqual(len(result), 1)
+            for pub_result in result:
+                self.assertIsInstance(pub_result, PubResult)
+                self.assertIsInstance(pub_result.data, DataBin)
+                self.assertIsInstance(pub_result.metadata, dict)
+
     @data(FakeManila(), FakeManilaV2(), AerSimulator())
     def test_non_primitve(self, backend):
         """Test calling non-primitive in local mode."""
diff --git a/test/utils.py b/test/utils.py
index 34427d674..94e8d5b5a 100644
--- a/test/utils.py
+++ b/test/utils.py
@@ -18,7 +18,7 @@
 import itertools
 import unittest
 from unittest import mock
-from typing import Dict, Optional
+from typing import Dict, Optional, Any
 from datetime import datetime
 from ddt import data, unpack
 
@@ -321,6 +321,17 @@ def get_mocked_backend(
     return mock_backend
 
 
+def get_mocked_session(backend: Any = None) -> mock.MagicMock:
+    """Return a mocked session object."""
+    session = mock.MagicMock(spec=Session)
+    session._instance = None
+    session._backend = backend or get_mocked_backend()
+    session._service = getattr(backend, "service", None) or mock.MagicMock(
+        spec=QiskitRuntimeService
+    )
+    return session
+
+
 def submit_and_cancel(backend: IBMBackend, logger: logging.Logger) -> RuntimeJob:
     """Submit and cancel a job.