Commit 155cc943 authored by Roque Porchetto's avatar Roque Porchetto

erp5_wendelin_telecom_ingestion: update unit tests

parent 0f36ef03
......@@ -73,7 +73,7 @@ def processFifData(file_name, data_array, data_descriptor):
try:
html_report = getHTMLReport(file_name, raw)
data_descriptor.setTextContent(html_report)
log("Data Descritor content saved")
log("Data Descriptor content saved")
except Exception as e:
log("Error handling Data Descriptor content: " + str(e))
......
import transaction
import time
import mne
import os
import numpy as np
from Products.ERP5Type.Log import log
def generateRawData(array):
info = mne.create_info(
ch_names=['MEG1', 'MEG2', 'EEG1', 'EEG2', 'EOG'],
ch_types=['grad', 'grad', 'eeg', 'eeg', 'eog'],
sfreq=100)
info = mne.create_info(array.shape[0],array.shape[1])
raw = mne.io.RawArray(array, info)
raw.save("test_raw_data.fif", overwrite=True)
with open("test_raw_data.fif", "rb") as file:
content = file.read()
return content
def generateRawData(sample_data_stream):
log("-TEST- Getting raw content from sample data stream...")
content = sample_data_stream.getData()
file_name = "sample_data_stream_raw.fif"
with open(file_name, 'wb') as fif_file:
fif_file.write(content)
raw = mne.io.read_raw_fif(file_name)
log("-TEST- Generating array from raw content...")
picks = mne.pick_types(raw.info)
data, times = raw[picks]
array = np.zeros(data.shape)
for pick in xrange(0, len(picks)):
data, times = raw[pick]
array[pick] = data[0]
if os.path.exists(file_name):
os.remove(file_name)
log("-TEST- Done.")
return content, array
......@@ -52,9 +52,7 @@
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple>
<string>W: 13, 42: Redefining built-in \'file\' (redefined-builtin)</string>
<string>W: 1, 0: Unused import transaction (unused-import)</string>
<string>W: 2, 0: Unused import time (unused-import)</string>
<string>W: 16, 8: Unused variable \'times\' (unused-variable)</string>
</tuple>
</value>
</item>
......
......@@ -43,8 +43,9 @@ class TestDataIngestion(SecurityTestCase):
yield l[i:i+n]
def generateRawDataBytesAndArray(self):
array = np.random.randn(5, 1000)
raw_data = self.context.generateRawData(array)
url = 'data_stream_module/mne_sample_for_test'
sample_data_stream = self.context.restrictedTraverse(url)
raw_data, array = self.context.generateRawData(sample_data_stream)
return raw_data, array
def getIngestionPolicy(self, reference, ingestion_script):
......@@ -118,7 +119,9 @@ class TestDataIngestion(SecurityTestCase):
self.simulateIngestionAlarm(ingestion_id, now)
data_array = self.getDataArray(ingestion_reference)
self.assertTrue(np.allclose(nparray, data_array.getArray()))
np.testing.assert_allclose(nparray, data_array.getArray()[:])
self.assertTrue(np.allclose(nparray, data_array.getArray()[:]))
def test_data_ingestion_splitted_file(self):
......@@ -135,28 +138,3 @@ class TestDataIngestion(SecurityTestCase):
data_stream = self.getDataStream(ingestion_reference)
self.assertEqual(len(data_chunk), len(data_stream.getData()))
self.assertEqual(data_chunk, data_stream.getData())
......@@ -46,8 +46,8 @@
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple>
<string>W:134, 4: Unused variable \'ingestion_id\' (unused-variable)</string>
<string>W:127, 16: Unused variable \'nparray\' (unused-variable)</string>
<string>W:137, 4: Unused variable \'ingestion_id\' (unused-variable)</string>
<string>W:130, 16: Unused variable \'nparray\' (unused-variable)</string>
<string>W: 9, 0: Unused log imported from Products.ERP5Type.Log (unused-import)</string>
</tuple>
</value>
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment