Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# check gif deposit
deposit = deposit_video_resolver(video['_deposit']['id'])
master_video = CDSVideosFilesIterator.get_master_video_file(deposit)
assert mock_args['master_id'] == master_video['version_id']
assert str(deposit.files.bucket.id) == mock_args['bucket']
# assert mock_args['bucket'].id == deposit.files.bucket.id
assert len(mock_args['frames']) == 10
assert 'output_dir' in mock_args
migration_streams = get_migration_streams(datadir=datadir)
with mock.patch.object(DataCiteProvider, 'register'), \
mock.patch.object(CDSRecordDumpLoader, '_create_frame',
side_effect=get_frames), \
mock.patch.object(CDSRecordDumpLoader, '_get_minimum_frames',
return_value=frames_required) as mock_frames, \
mock.patch.object(
ExtractFramesTask, '_create_gif') as mock_gif, \
mock.patch.object(
CDSRecordDumpLoader, '_get_migration_file_stream_and_size',
side_effect=migration_streams), \
mock.patch.object(CDSRecordDumpLoader, '_clean_file_list'):
video = CDSRecordDumpLoader.create(dump=dump)
assert mock_frames.called is True
db.session.add(video.model)
video_id = video.id
# check smil file
smil_obj = ObjectVersion.query.filter_by(
key='CERN-MOVIE-2012-193-001.smil', is_head=True).one()
storage = smil_obj.file.storage()
assert '
@mock.patch.object(middleware.LOG, 'info')
def test_init_formatter(self, mock_info, mock_RemoteControlDaemon,
mock_ControlDaemon, mock_find_entrypoint):
fake_formatter = mock_find_entrypoint.return_value
midware = middleware.TurnstileMiddleware('app',
dict(formatter='formatter'))
self.assertEqual(midware.app, 'app')
self.assertEqual(midware.limits, [])
self.assertEqual(midware.limit_sum, None)
self.assertEqual(midware.mapper, None)
self.assertIsInstance(midware.mapper_lock,
eventlet.semaphore.Semaphore)
self.assertEqual(midware.conf._config, {
None: {
'status': '413 Request Entity Too Large',
'formatter': 'formatter',
def test_dbcache_trim_age(pgdb, dbcache):
assert dbcache.size == 0
with mock.patch.object(initdb, "datetime") as mock_dt:
# create a few db with known dates
mock_dt.utcnow.return_value = TODAY
dbcache.add(pgdb, TEST_HASH1)
assert dbcache.size == 1
mock_dt.utcnow.return_value = TODAY_MINUS_2
dbcache.add(pgdb, TEST_HASH2)
assert dbcache.size == 2
mock_dt.utcnow.return_value = TODAY_MINUS_4
dbcache.add(pgdb, TEST_HASH3)
assert dbcache.size == 3
# get back to TODAY
mock_dt.utcnow.return_value = TODAY
# trim older than 5 days: no change
dbcache.trim_age(timedelta(days=5))
assert dbcache.size == 3
# trim older than 3 days: drop one
@mock.patch.object(ResourceClient, 'get')
def test_get_port_monitor_by_uri(self, mock_get):
uri_logical_interconnect = '/rest/logical-interconnects/be227eaf-3810-4b8a-b9ba-0af4479a9fe2'
uri_rest_call = uri_logical_interconnect + '/port-monitor'
self._logical_interconnect.get_port_monitor(uri_logical_interconnect)
mock_get.assert_called_once_with(uri_rest_call)
body = ''.join(nextline(self.fixture_file)) # exhaust the request either way
self.body = None
if self.method in ('PUT', 'POST'):
if 'Content-Type' in self.headers:
if 'application/xml' in self.headers['Content-Type']:
self.body = xml(body)
else:
self.body = body
# Set up the response returner.
sock = mock.Mock()
sock.makefile = mock.Mock(return_value=self.fixture_file)
response = httplib.HTTPResponse(sock, method=self.method)
response.begin()
self.response_context = mock.patch.object(httplib.HTTPConnection, 'getresponse', lambda self: response)
self.response_mock = self.response_context.__enter__()
return self
@mock.patch.object(dockercloud.api.http.Session, 'send')
def test_image_save(self, mock_send):
attribute = json.loads(
'{"base_image": false, "categories": [], "cluster_aware": true, "description": "description", "docker_registry": {"created": true, "host": "registry.hub.docker.com", "id": 5, "image_url": "/_static/assets/images/dockerregistries/docker.png", "is_ssl": true, "is_tutum_registry": false, "modified": true, "name": "Docker.io", "resource_uri": "/api/v1/registry/registry.hub.docker.com/", "uuid": "d533039e-c44c-4cdc-951b-e0e03b8410c6"}, "image_url": "", "imagetag_set": [{"full_name": "tifayuki/cadvisor:latest", "image": {"author": "fake ", "docker_id": "9e2907ef52bf811b4da100f50ba8f0908ccc610c7054bd69087f0a9f4703efdd", "entrypoint": "", "image_creation": "Fri, 15 Aug 2014 15:19:04 +0000", "imageenvvar_set": [{"key": "CADVISOR_TAG", "value": "0.2.2"}, {"key": "DB_NAME", "value": "cadvisor"}, {"key": "DB_PASS", "value": "root"}, {"key": "DB_USER", "value": "root"}, {"key": "HOME", "value": "/"}, {"key": "PATH", "value": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"}], "imageport_set": [], "run_command": "/run.sh"}, "image_info": "/api/v1/image/tifayuki/cadvisor/", "name": "latest", "resource_uri": "/api/v1/image/tifayuki/cadvisor/tag/latest/"}], "is_private_image": true, "name": "tifayuki/cadvisor", "public_url": "https://registry.hub.docker.com/u/tifayuki/cadvisor/", "resource_uri": "/api/v1/image/tifayuki/cadvisor/", "starred": false}'
)
mock_send.return_value = fake_resp(fake_image_save)
image = dockercloud.Repository.fetch('tifayuki/cadvisor')
image.description = 'descripiton'
self.assertTrue(image.save())
result = json.loads(json.dumps(image.get_all_attributes()))
target = json.loads(json.dumps(attribute))
self.assertDictEqual(target, result)
def test_load_calls_validate(tmpdir):
apply_fs(tmpdir, minimal_config_dir)
base = str(tmpdir)
with patch.object(BuildConfig, 'validate') as build_validate:
load(base, env_config)
assert build_validate.call_count == 1
def test_browse_directory_sort_date(config):
with mock.patch.object(APIClient, 'get_station_list', get_station_list_mock):
config['pandora']['sort_order'] = 'date'
backend = conftest.get_backend(config)
results = backend.library.browse(backend.library.root_directory.uri)
assert results[0].name == PandoraLibraryProvider.GENRE_DIR_NAME
assert results[1].name.startswith('QuickMix')
assert results[2].name == conftest.MOCK_STATION_NAME + ' 2'
assert results[3].name == conftest.MOCK_STATION_NAME + ' 1'
def test_domain_delete_existing_domain(self):
with patch.object(self.domain.connection, 'deprecate_domain'):
self.domain.delete()
@mock.patch.object(percona_utils, 'services')
@mock.patch.object(percona_utils, 'REQUIRED_INTERFACES')
@mock.patch.object(percona_utils, 'make_assess_status_func')
def test_assess_status_func(self,
make_assess_status_func,
REQUIRED_INTERFACES,
services):
services.return_value = ['mysql']
percona_utils.assess_status_func('test-config')
# ports=None whilst port checks are disabled.
make_assess_status_func.assert_called_once_with(
'test-config', REQUIRED_INTERFACES, charm_func=mock.ANY,
services=['mysql'], ports=None)
services.assert_called_once()