diff --git a/.github/workflows/q-ctrl-tests.yml b/.github/workflows/q-ctrl-tests.yml
new file mode 100644
index 000000000..bec27ac69
--- /dev/null
+++ b/.github/workflows/q-ctrl-tests.yml
@@ -0,0 +1,53 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2023.
+#
+# 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.
+
+name: Q-CTRL Tests
+on:
+  push:
+      tags:
+        - "*"
+  workflow_dispatch:
+jobs:
+  integration-tests:
+    name: Run integration tests - ${{ matrix.environment }}
+    runs-on: ${{ matrix.os }}
+    strategy:
+      # avoid cancellation of in-progress jobs if any matrix job fails
+      fail-fast: false
+      matrix:
+        python-version: [ 3.9 ]
+        os: [ "ubuntu-latest" ]
+        environment: [ "ibm-cloud-staging" ]
+    environment: ${{ matrix.environment }}
+    env:
+      QISKIT_IBM_TOKEN: ${{ secrets.QISKIT_IBM_TOKEN_QCTRL }}
+      QISKIT_IBM_URL: ${{ secrets.QISKIT_IBM_URL }}
+      QISKIT_IBM_INSTANCE: ${{ secrets.QISKIT_IBM_INSTANCE_QCTRL }}
+      CHANNEL_STRATEGY: q-ctrl
+      LOG_LEVEL: DEBUG
+      STREAM_LOG: True
+      QISKIT_IN_PARALLEL: True
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e .
+          pip install -c constraints.txt -r requirements-dev.txt
+      - name: Run q-ctrl tests
+        run: python -m unittest test/qctrl/test_qctrl.py
diff --git a/test/decorators.py b/test/decorators.py
index 007b7944e..dfef9f8e0 100644
--- a/test/decorators.py
+++ b/test/decorators.py
@@ -69,13 +69,14 @@ def _wrapper(self, *args, **kwargs):
 
 
 def _get_integration_test_config():
-    token, url, instance = (
+    token, url, instance, channel_strategy = (
         os.getenv("QISKIT_IBM_TOKEN"),
         os.getenv("QISKIT_IBM_URL"),
         os.getenv("QISKIT_IBM_INSTANCE"),
+        os.getenv("CHANNEL_STRATEGY"),
     )
     channel: Any = "ibm_quantum" if url.find("quantum-computing.ibm.com") >= 0 else "ibm_cloud"
-    return channel, token, url, instance
+    return channel, token, url, instance, channel_strategy
 
 
 def run_integration_test(func):
@@ -115,7 +116,7 @@ def _wrapper(self, *args, **kwargs):
                 ["ibm_cloud", "ibm_quantum"] if supported_channel is None else supported_channel
             )
 
-            channel, token, url, instance = _get_integration_test_config()
+            channel, token, url, instance, channel_strategy = _get_integration_test_config()
             if not all([channel, token, url]):
                 raise Exception("Configuration Issue")  # pylint: disable=broad-exception-raised
 
@@ -131,6 +132,7 @@ def _wrapper(self, *args, **kwargs):
                     channel=channel,
                     token=token,
                     url=url,
+                    channel_strategy=channel_strategy,
                 )
             dependencies = IntegrationTestDependencies(
                 channel=channel,
@@ -138,6 +140,7 @@ def _wrapper(self, *args, **kwargs):
                 url=url,
                 instance=instance,
                 service=service,
+                channel_strategy=channel_strategy,
             )
             kwargs["dependencies"] = dependencies
             func(self, *args, **kwargs)
@@ -156,6 +159,7 @@ class IntegrationTestDependencies:
     token: str
     channel: str
     url: str
+    channel_strategy: str
 
 
 def integration_test_setup_with_backend(
diff --git a/test/integration/test_account.py b/test/integration/test_account.py
index ffe8929e2..9b0b86364 100644
--- a/test/integration/test_account.py
+++ b/test/integration/test_account.py
@@ -24,6 +24,7 @@
     get_resource_controller_api_url,
     get_iam_api_url,
 )
+from qiskit_ibm_runtime.exceptions import IBMNotAuthorizedError
 from ..ibm_test_case import IBMIntegrationTestCase
 from ..decorators import IntegrationTestDependencies
 
@@ -51,6 +52,28 @@ def _skip_on_ibm_quantum(self):
         if self.dependencies.channel == "ibm_quantum":
             self.skipTest("Not supported on ibm_quantum")
 
+    def test_channel_strategy(self):
+        """Test passing in a channel strategy."""
+        self._skip_on_ibm_quantum()
+        # test when channel strategy not supported by instance
+        with self.assertRaises(IBMNotAuthorizedError):
+            QiskitRuntimeService(
+                channel="ibm_cloud",
+                url=self.dependencies.url,
+                token=self.dependencies.token,
+                instance=self.dependencies.instance,
+                channel_strategy="q-ctrl",
+            )
+        # test passing in default
+        service = QiskitRuntimeService(
+            channel="ibm_cloud",
+            url=self.dependencies.url,
+            token=self.dependencies.token,
+            instance=self.dependencies.instance,
+            channel_strategy="default",
+        )
+        self.assertTrue(service)
+
     def test_resolve_crn_for_valid_service_instance_name(self):
         """Verify if CRN is transparently resolved based for an existing service instance name."""
         self._skip_on_ibm_quantum()
diff --git a/test/qctrl/test_qctrl.py b/test/qctrl/test_qctrl.py
new file mode 100644
index 000000000..e22b51185
--- /dev/null
+++ b/test/qctrl/test_qctrl.py
@@ -0,0 +1,394 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2023.
+#
+# 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 job functions using real runtime service."""
+
+import time
+
+from qiskit import QuantumCircuit
+from qiskit.quantum_info import Statevector, hellinger_fidelity
+from qiskit.test.reference_circuits import ReferenceCircuits
+from qiskit.providers.jobstatus import JobStatus
+from qiskit.quantum_info import SparsePauliOp
+
+from qiskit_ibm_runtime import Sampler, Session, Options, Estimator, QiskitRuntimeService
+from qiskit_ibm_runtime.exceptions import IBMNotAuthorizedError
+
+from ..ibm_test_case import IBMIntegrationTestCase
+from ..decorators import run_integration_test
+from ..utils import cancel_job_safe
+
+FIDELITY_THRESHOLD = 0.9
+DIFFERENCE_THRESHOLD = 0.1
+
+
+class TestQCTRL(IBMIntegrationTestCase):
+    """Integration tests for QCTRL integration."""
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.bell = ReferenceCircuits.bell()
+        self.backend = "alt_canberra"
+
+    def test_channel_strategy_parameter(self):
+        """Test passing in channel strategy parameter for a q-ctrl instance."""
+        service = QiskitRuntimeService(
+            channel="ibm_cloud",
+            url=self.dependencies.url,
+            token=self.dependencies.token,
+            instance=self.dependencies.instance,
+            channel_strategy="q-ctrl",
+        )
+        self.assertTrue(service)
+
+    def test_invalid_channel_strategy_parameter(self):
+        """Test passing in invalid channel strategy parameter for a q-ctrl instance."""
+        with self.assertRaises(IBMNotAuthorizedError):
+            QiskitRuntimeService(
+                channel="ibm_cloud",
+                url=self.dependencies.url,
+                token=self.dependencies.token,
+                instance=self.dependencies.instance,
+                channel_strategy=None,
+            )
+
+    @run_integration_test
+    def test_cancel_qctrl_job(self, service):
+        """Test canceling qctrl job."""
+        with Session(service, self.backend) as session:
+            options = Options(resilience_level=1)
+            sampler = Sampler(session=session, options=options)
+
+            job = sampler.run([self.bell] * 10)
+
+        rjob = service.job(job.job_id())
+        if not cancel_job_safe(rjob, self.log):
+            return
+        self.assertEqual(rjob.status(), JobStatus.CANCELLED)
+
+    @run_integration_test
+    def test_sampler_qctrl_bell(self, service):
+        """Test qctrl bell state"""
+        # Set shots for experiment
+        shots = 1000
+
+        # Create Bell test circuit
+        bell_circuit = QuantumCircuit(2)
+        bell_circuit.h(0)
+        bell_circuit.cx(0, 1)
+
+        # Add measurements for the sampler
+        bell_circuit_sampler = bell_circuit.copy()
+        bell_circuit_sampler.measure_active()
+
+        # Execute circuit in a session with sampler
+        with Session(service, backend=self.backend):
+            options = Options(resilience_level=1)
+            sampler = Sampler(options=options)
+
+            result = sampler.run(bell_circuit_sampler, shots=shots).result()
+            results_dict = {
+                "{0:02b}".format(key): value for key, value in result.quasi_dists[0].items()
+            }  # convert keys to bitstrings
+
+        ideal_result = {
+            key: val / shots for key, val in Statevector(bell_circuit).probabilities_dict().items()
+        }
+        fidelity = hellinger_fidelity(results_dict, ideal_result)
+
+        self.assertGreater(fidelity, FIDELITY_THRESHOLD)
+
+    @run_integration_test
+    def test_sampler_qctrl_ghz(self, service):
+        """Test qctrl small GHZ"""
+        shots = 1000
+        num_qubits = 5
+        ghz_circuit = QuantumCircuit(num_qubits)
+        ghz_circuit.h(0)
+        for i in range(num_qubits - 1):
+            ghz_circuit.cx(i, i + 1)
+
+        # Add measurements for the sampler
+        ghz_circuit_sampler = ghz_circuit.copy()
+        ghz_circuit_sampler.measure_active()
+
+        # Execute circuit in a session with sampler
+        with Session(service, backend=self.backend):
+            options = Options(resilience_level=1)
+            sampler = Sampler(options=options)
+
+            result = sampler.run(ghz_circuit_sampler, shots=shots).result()
+            results_dict = {
+                f"{{0:0{num_qubits}b}}".format(key): value
+                for key, value in result.quasi_dists[0].items()
+            }  # convert keys to bitstrings
+
+        ideal_result = {
+            key: val / shots for key, val in Statevector(ghz_circuit).probabilities_dict().items()
+        }
+        fidelity = hellinger_fidelity(results_dict, ideal_result)
+        self.assertGreater(fidelity, FIDELITY_THRESHOLD)
+
+    @run_integration_test
+    def test_sampler_qctrl_superposition(self, service):
+        """Test qctrl small superposition"""
+
+        shots = 1000
+        num_qubits = 5
+        superposition_circuit = QuantumCircuit(num_qubits)
+        superposition_circuit.h(range(num_qubits))
+
+        # Add measurements for the sampler
+        superposition_circuit_sampler = superposition_circuit.copy()
+        superposition_circuit_sampler.measure_active()
+
+        # Execute circuit in a session with sampler
+        with Session(service, backend=self.backend):
+            options = Options(resilience_level=1)
+            sampler = Sampler(options=options)
+
+            result = sampler.run(superposition_circuit_sampler, shots=shots).result()
+            results_dict = {
+                f"{{0:0{num_qubits}b}}".format(key): value
+                for key, value in result.quasi_dists[0].items()
+            }  # convert keys to bitstrings
+
+        ideal_result = {
+            key: val / shots
+            for key, val in Statevector(superposition_circuit).probabilities_dict().items()
+        }
+        fidelity = hellinger_fidelity(results_dict, ideal_result)
+        self.assertGreater(fidelity, FIDELITY_THRESHOLD)
+
+    @run_integration_test
+    def test_sampler_qctrl_computational_states(self, service):
+        """Test qctrl computational states"""
+        shots = 1000
+        num_qubits = 3
+        computational_states_circuits = []
+        for idx in range(2**num_qubits):
+            circuit = QuantumCircuit(num_qubits)
+            bitstring = f"{{0:0{num_qubits}b}}".format(idx)
+            for bit_pos, bit in enumerate(
+                bitstring[::-1]
+            ):  # convert to little-endian (qiskit convention)
+                if bit == "1":
+                    circuit.x(bit_pos)
+            computational_states_circuits.append(circuit)
+
+        # Add measurements for the sampler
+        computational_states_sampler_circuits = []
+        for circuit in computational_states_circuits:
+            circuit_sampler = circuit.copy()
+            circuit_sampler.measure_all()
+            computational_states_sampler_circuits.append(circuit_sampler)
+
+        # Execute circuit in a session with sampler
+        with Session(service, backend=self.backend):
+            options = Options(resilience_level=1)
+            sampler = Sampler(options=options)
+
+            result = sampler.run(computational_states_sampler_circuits, shots=shots).result()
+            results_dict_list = [
+                {f"{{0:0{num_qubits}b}}".format(key): value for key, value in quasis.items()}
+                for quasis in result.quasi_dists
+            ]  # convert keys to bitstrings
+
+        ideal_results_list = [
+            {key: val / shots for key, val in Statevector(circuit).probabilities_dict().items()}
+            for circuit in computational_states_circuits
+        ]
+        fidelities = [
+            hellinger_fidelity(results_dict, ideal_result)
+            for results_dict, ideal_result in zip(results_dict_list, ideal_results_list)
+        ]
+
+        for fidelity in fidelities:
+            self.assertGreater(fidelity, FIDELITY_THRESHOLD)
+
+    @run_integration_test
+    def test_estimator_qctrl_bell(self, service):
+        """Test estimator qctrl bell state"""
+        # Set shots for experiment
+        shots = 1000
+
+        # Create Bell test circuit
+        bell_circuit = QuantumCircuit(2)
+        bell_circuit.h(0)
+        bell_circuit.cx(0, 1)
+
+        # Measure some observables in the estimator
+        observables = [SparsePauliOp("ZZ"), SparsePauliOp("IZ"), SparsePauliOp("ZI")]
+
+        # Execute circuit in a session with estimator
+        with Session(service, backend=self.backend):
+            estimator = Estimator()
+
+            result = estimator.run(
+                [bell_circuit] * len(observables), observables=observables, shots=shots
+            ).result()
+
+        ideal_result = [
+            Statevector(bell_circuit).expectation_value(observable).real
+            for observable in observables
+        ]
+        absolute_difference = [
+            abs(obs_theory - obs_exp) for obs_theory, obs_exp in zip(ideal_result, result.values)
+        ]
+        # absolute_difference_dict = {
+        #     obs.paulis[0].to_label(): diff for obs, diff in zip(observables, absolute_difference)
+        # }
+
+        for diff in absolute_difference:
+            self.assertLess(diff, DIFFERENCE_THRESHOLD)
+
+    @run_integration_test
+    def test_estimator_qctrl_ghz(self, service):
+        """Test estimator qctrl GHZ state"""
+        shots = 1000
+        num_qubits = 5
+        ghz_circuit = QuantumCircuit(num_qubits)
+        ghz_circuit.h(0)
+        for i in range(num_qubits - 1):
+            ghz_circuit.cx(i, i + 1)
+
+        # Measure some observables in the estimator
+        observables = [
+            SparsePauliOp("Z" * num_qubits),
+            SparsePauliOp("I" * (num_qubits - 1) + "Z"),
+            SparsePauliOp("Z" + "I" * (num_qubits - 1)),
+        ]
+
+        # Execute circuit in a session with estimator
+        with Session(service, backend=self.backend):
+            estimator = Estimator()
+
+            result = estimator.run(
+                [ghz_circuit] * len(observables), observables=observables, shots=shots
+            ).result()
+
+        ideal_result = [
+            Statevector(ghz_circuit).expectation_value(observable).real
+            for observable in observables
+        ]
+        absolute_difference = [
+            abs(obs_theory - obs_exp) for obs_theory, obs_exp in zip(ideal_result, result.values)
+        ]
+        absolute_difference_dict = {
+            obs.paulis[0].to_label(): diff for obs, diff in zip(observables, absolute_difference)
+        }
+
+        print(
+            "absolute difference between theory and experiment expectation values: ",
+            absolute_difference_dict,
+        )
+        for diff in absolute_difference:
+            self.assertLess(diff, DIFFERENCE_THRESHOLD)
+
+    @run_integration_test
+    def test_estimator_qctrl_superposition(self, service):
+        """Test estimator qctrl small superposition"""
+        shots = 1000
+        num_qubits = 4
+        superposition_circuit = QuantumCircuit(num_qubits)
+        superposition_circuit.h(range(num_qubits))
+
+        # Measure some observables in the estimator
+        obs_labels = [["I"] * num_qubits for _ in range(num_qubits)]
+        for idx, obs in enumerate(obs_labels):
+            obs[idx] = "Z"
+        obs_labels = ["".join(obs) for obs in obs_labels]
+        observables = [SparsePauliOp(obs) for obs in obs_labels]
+
+        # Execute circuit in a session with estimator
+        with Session(service, backend=self.backend):
+            estimator = Estimator()
+
+            result = estimator.run(
+                [superposition_circuit] * len(observables), observables=observables, shots=shots
+            ).result()
+
+        ideal_result = [
+            Statevector(superposition_circuit).expectation_value(observable).real
+            for observable in observables
+        ]
+        absolute_difference = [
+            abs(obs_theory - obs_exp) for obs_theory, obs_exp in zip(ideal_result, result.values)
+        ]
+        # absolute_difference_dict = {
+        #     obs.paulis[0].to_label(): diff for obs, diff in zip(observables, absolute_difference)
+        # }
+
+        for diff in absolute_difference:
+            self.assertLess(diff, DIFFERENCE_THRESHOLD)
+
+    @run_integration_test
+    def test_estimator_qctrl_computational(self, service):
+        """Test estimator qctrl computational states"""
+        shots = 1000
+        num_qubits = 3
+        computational_states_circuits = []
+        for idx in range(2**num_qubits):
+            circuit = QuantumCircuit(num_qubits)
+            bitstring = f"{{0:0{num_qubits}b}}".format(idx)
+            for bit_pos, bit in enumerate(
+                bitstring[::-1]
+            ):  # convert to little-endian (qiskit convention)
+                if bit == "1":
+                    circuit.x(bit_pos)
+            computational_states_circuits.append(circuit)
+
+        # Measure some observables in the estimator
+        obs_labels = [["I"] * num_qubits for _ in range(num_qubits)]
+        for idx, obs in enumerate(obs_labels):
+            obs[idx] = "Z"
+        obs_labels = ["".join(obs) for obs in obs_labels]
+        observables = [SparsePauliOp(obs) for obs in obs_labels]
+
+        computational_states_circuits_estimator, observables_estimator = [], []
+        for circuit in computational_states_circuits:
+            computational_states_circuits_estimator += [circuit] * len(observables)
+            observables_estimator += observables
+
+        # Execute circuit in a session with estimator
+        with Session(service, self.backend):
+            estimator = Estimator()
+            result = estimator.run(
+                computational_states_circuits_estimator,
+                observables=observables_estimator,
+                shots=shots,
+            ).result()
+
+        ideal_result = [
+            Statevector(circuit).expectation_value(observable).real
+            for circuit, observable in zip(
+                computational_states_circuits_estimator, observables_estimator
+            )
+        ]
+        absolute_difference = [
+            abs(obs_theory - obs_exp) for obs_theory, obs_exp in zip(ideal_result, result.values)
+        ]
+
+        absolute_difference_dict = {}
+        for idx in range(2**num_qubits):
+            circuit = QuantumCircuit(num_qubits)
+            bitstring = f"{{0:0{num_qubits}b}}".format(idx)
+
+            absolute_difference_dict[bitstring] = {
+                obs.paulis[0].to_label(): diff
+                for obs, diff in zip(
+                    observables_estimator[idx * len(observables) : (idx + 1) * len(observables)],
+                    absolute_difference[idx * len(observables) : (idx + 1) * len(observables)],
+                )
+            }
+        for diff in absolute_difference:
+            self.assertLess(diff, DIFFERENCE_THRESHOLD)