Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_export(self):
patch_worker = mock.patch(
'opencensus.trace.exporters.transports.background_thread._Worker',
autospec=True)
exporter = mock.Mock()
with patch_worker:
transport = background_thread.BackgroundThreadTransport(exporter)
trace = {
'traceId': 'test',
'spans': [{}, {}],
}
transport.export(trace)
self.assertTrue(transport.worker.enqueue.called)
@mock.patch('snaps_common.ansible_snaps.ansible_utils.apply_playbook')
@mock.patch('time.sleep')
@mock.patch('drp_python.translation_layer.machines_translation.'
'MachineTranslation.add_machine_params')
@mock.patch('snaps_boot.provision.rebar_utils.__create_machine_params')
def test_setup_dhcp_service(self, m1, m2, m3, m4, m5, m6, m7, m8, m9, m10,
m11, m12):
"""
Tests the rebar_utils.
:return:
"""
conf_file = pkg_resources.resource_filename('tests.conf', 'hosts.yaml')
conf = file_utils.read_yaml(conf_file)
rebar_utils.install_config_drp(None, conf)
@patch("weibo_harvester.Weiboarc", autospec=True)
def test_search_topic(self, mock_weiboarc_class):
mock_weiboarc = MagicMock(spec=Weiboarc)
# search_topic Expecting 2 results. First returns 1tweets. Second returns none.
mock_weiboarc.search_topic.side_effect = [(weibo6, weibo7), ()]
# Return mock_weiboarc when instantiating a weiboarc.
mock_weiboarc_class.side_effect = [mock_weiboarc]
self.harvester.message = base_search_message
self.harvester.harvest_seeds()
query = self.harvester.message["seeds"][0]["token"]
self.assertDictEqual({"weibos": 2}, self.harvester.result.harvest_counter)
mock_weiboarc_class.assert_called_once_with(tests.WEIBO_ACCESS_TOKEN)
self.assertEqual([call(query, since_id=None)], mock_weiboarc.search_topic.mock_calls)
"""Patch open().
Patch open() to allow mocking both open() itself and the file that is
yielded.
Yields the mock for "open" and "file", respectively.
"""
mock_open = mock.MagicMock(spec=open)
mock_file = mock.MagicMock(spec=io.FileIO)
@contextlib.contextmanager
def stub_open(*args, **kwargs):
mock_open(*args, **kwargs)
yield mock_file
with mock.patch('builtins.open', stub_open):
yield mock_open, mock_file
@mock.patch('evelink.thirdparty.six.moves.urllib.request.urlopen')
def test_useragent(self, mock_urlopen):
mock_urlopen.return_value.read.return_value = self.test_xml
self.cache.get.return_value = None
self.api.get('foo/Bar', {'a':[1,2,3]})
test_useragent = '%s %s' % (evelink_api._user_agent, self.custom_useragent)
self.assertEqual(mock_urlopen.call_args[0][0].headers['User-agent'], test_useragent)
@mock.patch("geonode.services.serviceprocessors.wms.WebMapService",
autospec=True)
@mock.patch("geonode.services.serviceprocessors.wms.settings",
autospec=True)
def test_detects_cascaded_service(self, mock_settings, mock_wms):
mock_settings.DEFAULT_MAP_CRS = "EPSG:3857"
mock_layer_meta = mock.MagicMock(ContentMetadata)
mock_layer_meta.name = "phony_name"
mock_layer_meta.children = []
mock_layer_meta.crsOptions = ["epsg:4326"]
self.parsed_wms.contents = {
mock_layer_meta.name: mock_layer_meta,
}
mock_wms.return_value = (self.phony_url, self.parsed_wms)
handler = wms.WmsServiceHandler(self.phony_url)
self.assertEqual(handler.indexing_method, enumerations.CASCADED)
def test_get_record_ref_with_empty_server_name():
config = {}
with patch.dict(current_app.config, config, clear=True):
expected = 'http://inspirehep.net/api/endpoint/123'
result = get_record_ref(123, 'endpoint')
assert expected == result['$ref']
@patch('stretch.models.System')
def test_load_sources_receiver(self, system_mock):
system = Mock()
system_mock.objects.all.return_value = [system]
signals.load_sources.send(sender=Mock())
system.sync_source.assert_called_with()
@mock.patch.object(ResourceClient, 'get')
def test_get_with_uri_called_once(self, mock_get):
uri = '/rest/sas-logical-jbods-attachments/3518be0e-17c1-4189-8f81-83f3724f6155'
self._sas_logical_jbod_attachments.get(uri)
mock_get.assert_called_once_with(uri)
def test_event_transfer(client_alice, token_event):
url = reverse('event-list', kwargs={'version': 'v1'})
data = {
'events': token_event,
'project': 'ShipToken'
}
response = client_alice.post(url, data)
assert response.status_code == status.HTTP_403_FORBIDDEN
with patch('apps.eth.views.log_metric') as mocked:
# Set NGINX headers for engine auth
response = client_alice.post(url, data, X_NGINX_SOURCE='internal', X_SSL_CLIENT_VERIFY='SUCCESS',
X_SSL_CLIENT_DN='/CN=engine.test-internal')
AssertionHelper.HTTP_204(response)
assert mocked.call_count == 2