1# This code is part of Qiskit.
2#
3# (C) Copyright IBM 2018, 2019.
4#
5# This code is licensed under the Apache License, Version 2.0. You may
6# obtain a copy of this license in the LICENSE.txt file in the root directory
7# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
8#
9# Any modifications or derivative works of this code must retain this
10# copyright notice, and modified files need to carry a notice indicating
11# that they have been altered from the originals.
12
13# pylint: disable=missing-docstring
14
15"""
16Utilities for mocking the IBMQ provider, including job responses and backends.
17
18The module includes dummy provider, backends, and jobs. The purpose of
19these classes is to trick backends for testing purposes:
20testing local timeouts, arbitrary responses or behavior, etc.
21
22The mock devices are mainly for testing the compiler.
23"""
24
25import datetime
26import uuid
27import logging
28from concurrent import futures
29import time
30
31from qiskit.result import Result
32from qiskit.providers import BaseBackend, BaseJob
33from qiskit.providers.models import BackendProperties, BackendConfiguration
34from qiskit.providers.models.backendconfiguration import GateConfig
35from qiskit.qobj import (QasmQobj, QobjExperimentHeader, QobjHeader,
36                         QasmQobjInstruction, QasmQobjExperimentConfig,
37                         QasmQobjExperiment, QasmQobjConfig)
38from qiskit.providers.jobstatus import JobStatus
39from qiskit.providers.baseprovider import BaseProvider
40from qiskit.providers.exceptions import QiskitBackendNotFoundError
41from qiskit.providers.aer import AerError
42
43
44logger = logging.getLogger(__name__)
45
46
47class FakeProvider(BaseProvider):
48    """Dummy provider just for testing purposes.
49
50    Only filtering backends by name is implemented.
51    """
52
53    def get_backend(self, name=None, **kwargs):
54        backend = self._backends[0]
55        if name:
56            filtered_backends = [backend for backend in self._backends
57                                 if backend.name() == name]
58            if not filtered_backends:
59                raise QiskitBackendNotFoundError()
60            else:
61                backend = filtered_backends[0]
62        return backend
63
64    def backends(self, name=None, **kwargs):
65        return self._backends
66
67    def __init__(self):
68        # TODO Add the rest of simulators that we want to mock
69        self._backends = [FakeSuccessQasmSimulator(),
70                          FakeFailureQasmSimulator()]
71        super().__init__()
72
73
74class FakeBackend(BaseBackend):
75    """This is a dummy backend just for testing purposes."""
76
77    def __init__(self, configuration, time_alive=10):
78        """
79        Args:
80            configuration (BackendConfiguration): backend configuration
81            time_alive (int): time to wait before returning result
82        """
83        super().__init__(configuration)
84        self.time_alive = time_alive
85
86    def properties(self):
87        """Return backend properties"""
88        dummy_date = datetime.datetime.now().isoformat()
89        properties = {
90            'backend_name': self.name(),
91            'backend_version': self.configuration().backend_version,
92            'last_update_date': dummy_date,
93            'qubits': [[{'name': 'DUMMY', 'date': dummy_date,
94                         'unit': 'ms', 'value': 0}]],
95            'gates': [{'qubits': [0], 'gate': 'DUMMY',
96                       'parameters':
97                           [{'name': 'DUMMY', 'date': dummy_date,
98                             'unit': 'ms', 'value': 0}]}],
99            'general': []
100        }
101
102        return BackendProperties.from_dict(properties)
103
104    def run(self, qobj):
105        job_id = str(uuid.uuid4())
106        job = FakeJob(self, self.run_job, job_id, qobj)
107        job.submit()
108        return job
109
110    # pylint: disable=unused-argument
111    def run_job(self, job_id, qobj):
112        """Main dummy run loop"""
113        time.sleep(self.time_alive)
114
115        return Result.from_dict({
116            'job_id': job_id,
117            'backend_name': self.name(),
118            'backend_version': self.configuration().backend_version,
119            'qobj_id': qobj.qobj_id,
120            'results': [],
121            'status': 'COMPLETED',
122            'success': True
123        })
124
125
126class FakeSuccessQasmSimulator(FakeBackend):
127    """A fake QASM simulator backend that always returns SUCCESS"""
128
129    def __init__(self, time_alive=10):
130        configuration = BackendConfiguration(
131            backend_name='fake_success_qasm_simulator',
132            backend_version='0.0.0',
133            n_qubits=5,
134            basis_gates=['u1', 'u2', 'u3', 'cx', 'cz', 'id', 'x', 'y', 'z',
135                         'h', 's', 'sdg', 't', 'tdg', 'ccx', 'swap',
136                         'snapshot', 'unitary'],
137            simulator=True,
138            local=True,
139            conditional=True,
140            open_pulse=False,
141            memory=True,
142            max_shots=65536,
143            gates=[GateConfig(name='TODO', parameters=[], qasm_def='TODO')],
144            coupling_map=None
145        )
146
147        super().__init__(configuration, time_alive=time_alive)
148
149
150class FakeFailureQasmSimulator(FakeBackend):
151    """A fake simulator backend."""
152
153    def __init__(self, time_alive=10):
154        configuration = BackendConfiguration(
155            backend_name='fake_failure_qasm_simulator',
156            backend_version='0.0.0',
157            n_qubits=5,
158            basis_gates=['u1', 'u2', 'u3', 'cx', 'cz', 'id', 'x', 'y', 'z',
159                         'h', 's', 'sdg', 't', 'tdg', 'ccx', 'swap',
160                         'snapshot', 'unitary'],
161            simulator=True,
162            local=True,
163            conditional=True,
164            open_pulse=False,
165            memory=True,
166            max_shots=65536,
167            gates=[GateConfig(name='TODO', parameters=[], qasm_def='TODO')],
168            coupling_map=None
169        )
170
171        super().__init__(configuration, time_alive=time_alive)
172
173
174    # pylint: disable=unused-argument
175    def run_job(self, job_id, qobj):
176        """Main dummy run loop"""
177        time.sleep(self.time_alive)
178
179        raise AerError("Mocking a failure in the QASM Simulator")
180
181class FakeJob(BaseJob):
182    """Fake simulator job"""
183    _executor = futures.ThreadPoolExecutor()
184
185    def __init__(self, backend, fn, job_id, qobj):
186        super().__init__(backend, job_id)
187        self._backend = backend
188        self._job_id = job_id
189        self._qobj = qobj
190        self._future = None
191        self._future_callback = fn
192
193    def submit(self):
194        self._future = self._executor.submit(
195            self._future_callback, self._job_id, self._qobj
196        )
197
198    def result(self, timeout=None):
199        # pylint: disable=arguments-differ
200        return self._future.result(timeout=timeout)
201
202    def cancel(self):
203        return self._future.cancel()
204
205    def status(self):
206        if self._running:
207            _status = JobStatus.RUNNING
208        elif not self._done:
209            _status = JobStatus.QUEUED
210        elif self._cancelled:
211            _status = JobStatus.CANCELLED
212        elif self._done:
213            _status = JobStatus.DONE
214        elif self._error:
215            _status = JobStatus.ERROR
216        else:
217            raise Exception('Unexpected state of {0}'.format(
218                self.__class__.__name__))
219        _status_msg = None
220        return {'status': _status,
221                'status_msg': _status_msg}
222
223    def job_id(self):
224        return self._job_id
225
226    def backend(self):
227        return self._backend
228
229    @property
230    def _cancelled(self):
231        return self._future.cancelled()
232
233    @property
234    def _done(self):
235        return self._future.done()
236
237    @property
238    def _running(self):
239        return self._future.running()
240
241    @property
242    def _error(self):
243        return self._future.exception(timeout=0)
244
245
246def new_fake_qobj():
247    """Create fake `Qobj` and backend instances."""
248    backend = FakeQasmSimulator()
249    return QasmQobj(
250        qobj_id='test-id',
251        config=QasmQobjConfig(shots=1024, memory_slots=1, max_credits=100),
252        header=QobjHeader(backend_name=backend.name()),
253        experiments=[QasmQobjExperiment(
254            instructions=[
255                QasmQobjInstruction(name='barrier', qubits=[1])
256            ],
257            header=QobjExperimentHeader(),
258            config=QasmQobjExperimentConfig(seed_simulator=123456)
259        )]
260    )
261