Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
"""
Set up a folder structure containing one timepoint (0)
one channel (1) and two images in channel subfolder
"""
self.tempdir = TempDirectory()
self.temp_path = self.tempdir.path
self.tempdir.makedir('timepoint_0')
self.tempdir.makedir('timepoint_0/channel_1')
# Write images as bytes
im = np.zeros((15, 12), dtype=np.uint16)
res, im_encoded = cv2.imencode('.png', im)
im_encoded = im_encoded.tostring()
self.tempdir.write('timepoint_0/channel_1/im_0.png', im_encoded)
self.tempdir.write('timepoint_0/channel_1/im_1.png', im_encoded)
self.tempdir.listdir(recursive=True)
def tearDown(self):
TempDirectory.cleanup_all()
def setUp(self):
"""Set up a dir for tiling with flatfield"""
self.tempdir = TempDirectory()
self.temp_path = self.tempdir.path
# Start frames meta file
self.meta_name = 'frames_meta.csv'
frames_meta = aux_utils.make_dataframe()
# Write images
self.im = 127 * np.ones((15, 11), dtype=np.uint8)
self.im2 = 234 * np.ones((15, 11), dtype=np.uint8)
self.channel_idx = 1
self.time_idx = 5
self.pos_idx1 = 7
self.pos_idx2 = 8
self.int2str_len = 3
# Write test images with 4 z and 2 pos idx
for z in range(15, 20):
im_name = aux_utils.get_im_name(
def test_do_activity_success(self, test_data, fake_emit, fake_file_name,
fake_sqs_conn, fake_sqs_queue, fake_sqs_message):
directory = TempDirectory()
fake_sqs_conn.return_value = FakeSQSConn(directory)
fake_sqs_queue.return_value = FakeSQSQueue(directory)
fake_sqs_message.return_value = FakeSQSMessage(directory)
fake_file_name.return_value = "elife-00353-v1.xml"
# format the activity data
workflow_test_data = None
if test_data.get("set_workflow_data"):
workflow_test_data = workflow_data(WORKFLOW_DATA, test_data.get("run_type"))
activity_test_data = activity_data(
ACTIVITY_DATA, test_data.get("force"), workflow_test_data)
# run do_activity
result = self.activity.do_activity(activity_test_data)
# make assertions
self.assertEqual(result, True)
# read in the message body from the TempDirectory()
message_body = json.loads(directory.read("fake_sqs_body").decode())
def test_activity(self, mock_set_monitor_property, mock_emit_monitor_event, fake_s3_mock, fake_session, fake_key_mock):
directory = TempDirectory()
fake_session.return_value = FakeSession(data.session_example)
fake_key_mock.return_value = FakeKey(directory, data.bucket_dest_file_name,
data.RewriteEIF_json_input_string)
fake_s3_mock.return_value = FakeS3Connection()
success = self.activity_PreparePostEIF.do_activity(data.RewriteEIF_data)
self.assertEqual(True, success)
output_json = json.loads(directory.read(data.bucket_dest_file_name))
expected = data.RewriteEIF_json_output
self.assertDictEqual(output_json, expected)
def setUp(self):
"""
Set up a directory with some images to resample
"""
self.tempdir = TempDirectory()
self.temp_path = self.tempdir.path
self.mask_dir = os.path.join(self.temp_path, 'mask_dir')
self.tempdir.makedir('mask_dir')
self.input_dir = os.path.join(self.temp_path, 'input_dir')
self.tempdir.makedir('input_dir')
self.mask_channel = 1
self.slice_idx = 7
self.time_idx = 8
# Mask meta file
self.csv_name = 'mask_image_matchup.csv'
input_meta = aux_utils.make_dataframe()
# Make input meta
for c in range(4):
for p in range(10):
im_name = aux_utils.get_im_name(
channel_idx=c,
def test_activity(self, status_code, response_update_date, update_json, expected_update_date,
mock_sqs_message, mock_sqs_connect, mock_requests_put):
directory = TempDirectory()
mock_sqs_connect.return_value = FakeSQSConn(directory)
mock_sqs_message.return_value = FakeSQSMessage(directory)
mock_requests_put.return_value = classes_mock.FakeResponse(status_code, update_json)
success = self.activity_ApprovePublication.do_activity(
activity_data.ApprovePublication_data(response_update_date))
fake_sqs_queue = FakeSQSQueue(directory)
data_written_in_test_queue = fake_sqs_queue.read(activity_data.ApprovePublication_test_dir)
self.assertEqual(True, success)
output_json = json.loads(directory.read(activity_data.ApprovePublication_test_dir))
expected = activity_data.ApprovePublication_json_output_return_example(expected_update_date)
self.assertDictEqual(output_json, expected)
def setUp(self):
self.dir = TempDirectory()
self.missing = self.dir.getpath('missing')
self.path = search_path()
def tearDown(self):
TempDirectory.cleanup_all()
def setUp(self):
"""Initialise temporary directory."""
self.tempdir = TempDirectory(encoding='utf-8')
self.FolderClass = Folder