Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@unpack
def test_move_or_repackage_pmc_zip(self, input_zip_file_path, doi_id, workflow,
expected_zip_file, expected_zip_file_contents):
# create activity directories
self.activity.make_activity_directories()
# copy in some sample data
dest_input_zip_file_path = os.path.join(
self.activity.directories.get("INPUT_DIR"), input_zip_file_path.split('/')[-1])
shutil.copy(input_zip_file_path, dest_input_zip_file_path)
# call the activity function
self.activity.move_or_repackage_pmc_zip(doi_id, workflow)
# confirm the output
ftp_outbox_dir = self.activity.directories.get("FTP_TO_SOMEWHERE_DIR")
self.assertTrue(expected_zip_file in os.listdir(ftp_outbox_dir))
with zipfile.ZipFile(os.path.join(ftp_outbox_dir, expected_zip_file)) as zip_file:
self.assertEqual(sorted(zip_file.namelist()), sorted(expected_zip_file_contents))
def test_indicator_source_org_attr(self, data):
# Strip whitespace to avoid comparison issues
expected = "".join(data.indicator["sourceOrganization"].split())
result = "".join(data.dataset.indicator_source_org.split())
self.assertEqual(result, expected)
@ddt_idata((i, str(i)) for i in range(1, 16, 1)) # type: ignore
@ddt_unpack # type: ignore
def test_flags(self, flags: int, flags_qs: str) -> None:
# act
api_uri = Pr0gramm._get_api_uri(flags=flags, promoted=False)
(_, _, _, query_string, _) = urlsplit(api_uri)
query = parse_qs(query_string)
# assert
self.assertEqual([flags_qs], query['flags'])
@ddt.data(
(requests.delete, client._call_requests_delete),
(requests.delete, client.delete),
(requests.get, client._call_requests_get),
(requests.get, client.get),
(requests.head, client._call_requests_head),
(requests.head, client.head),
(requests.patch, client._call_requests_patch),
(requests.patch, client.patch),
(requests.post, client._call_requests_post),
(requests.post, client.post),
(requests.put, client._call_requests_put),
(requests.put, client.put),
(requests.request, client._call_requests_request),
(requests.request, client.request),
)
def test_identical_args(self, functions):
@ddt.data(fakes.MoverTestData().interface_name1,
fakes.MoverTestData().long_interface_name)
def test_get_mover_interface(self, interface_name):
self.hook.append(self.mover.resp_get_ref_succeed())
self.hook.append(self.mover.resp_get_succeed())
self.hook.append(self.mover.resp_get_succeed())
xml_connector = self.mover_interface_manager.xml_connector
xml_connector.post = utils.EMCMock(side_effect=self.hook)
interface = self.mover_interface_manager.get(
name=interface_name,
mover_name=self.mover.mover_name)
property_map = [
'name',
'mover_name',
'device',
@ddt.data((reporters.HTMLReporter, False),
(reporters.HTMLStaticReporter, True))
@ddt.unpack
def test_generate(self, cls, include_libs, mock_dumps, mock_ui_utils):
mock_render = mock_ui_utils.get_template.return_value.render
reporter = cls(get_verifications(), None)
self.assertEqual({"print": mock_render.return_value},
reporter.generate())
mock_render.assert_called_once_with(data=mock_dumps.return_value,
include_libs=include_libs)
mock_ui_utils.get_template.assert_called_once_with(
"verification/report.html")
self.assertEqual(1, mock_dumps.call_count)
args, kwargs = mock_dumps.call_args
@data('https://github.com/werwolfby/monitorrent/releases/tag/1.0.2',
'https://github.com/werwolfby/monitorrent/releases/tag/1.0.1',
'https://github.com/werwolfby/monitorrent/releases/tag/1.0.0')
def test_get_url(self, url):
new_version_checker = NewVersionChecker(Mock(), False)
new_version_resource = NewVersion(new_version_checker)
new_version_checker.new_version_url = url
self.api.add_route('/api/new_version', new_version_resource)
body = self.simulate_request("/api/new_version", decode='utf-8')
self.assertEqual(self.srmock.status, falcon.HTTP_OK)
self.assertTrue('application/json' in self.srmock.headers_dict['Content-Type'])
result = json.loads(body)
self.assertEqual(result, {'url': url})
@ddt.data(
{"new_size": 5},
{"new_size": 10}
)
def test__extend_share(self, new_size):
fake_share = mock.MagicMock()
self.scenario._extend_share(fake_share, new_size)
fake_share.extend.assert_called_with(new_size)
self.mock_wait_for_status.mock.assert_called_once_with(
fake_share,
ready_statuses=["available"],
update_resource=self.mock_get_from_manager.mock.return_value,
timeout=300, check_interval=3)
self.mock_get_from_manager.mock.assert_called_once_with()
@data(('off', True), ('json', True), ('html', True), ('foo', False))
def test_report(self, item):
"""Testing missing mandatory parameter."""
try:
options = {'definition': 'fake.yml'}
options.update({'report': item[0]})
ApplicationOptions(**options)
if not item[1]:
self.assertFalse("RuntimeError expected")
except RuntimeError as exception:
if item[1]:
self.assertFalse("Unexpected exception %s" % exception)
@ddt.data(({"max": 0, "min_iterations": 5, "sigmas": 2.5}, True),
({"max": -1}, False),
({"max": 0, "min_iterations": 2}, False),
({"max": 0, "sigmas": 0}, False),
({"foo": "bar"}, False))
@ddt.unpack
def test_validate(self, config, valid):
results = sla.SLA.validate("outliers", None, None, config)
if valid:
self.assertEqual([], results)
else:
self.assertEqual(1, len(results))