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_startup(unittest):
import dtale.views as views
with pytest.raises(BaseException) as error:
views.startup(URL)
assert 'data loaded is None!' in str(error.value)
with pytest.raises(BaseException) as error:
views.startup(URL, dict())
assert 'data loaded must be one of the following types: pandas.DataFrame, pandas.Series, pandas.DatetimeIndex'\
in str(error.value)
test_data = pd.DataFrame([dict(date=pd.Timestamp('now'), security_id=1, foo=1.5)])
test_data = test_data.set_index(['date', 'security_id'])
instance = views.startup(URL, data_loader=lambda: test_data)
pdt.assert_frame_equal(instance.data, test_data.reset_index())
unittest.assertEqual(views.SETTINGS[instance._data_id], dict(locked=['date', 'security_id']), 'should lock index columns')
test_data = test_data.reset_index()
instance = views.startup(URL, data=test_data)
def mock_requests_get(url, verify=True):
if url.endswith('/health'):
is_ok = get_calls['ct'] > 0
get_calls['ct'] += 1
return getter(is_ok)
return getter(True)
stack.enter_context(mock.patch('requests.get', mock_requests_get))
instance = show(data=test_data, subprocess=False, name='foo')
assert instance.__repr__() == ''
assert instance.__str__() == ''
instance.adjust_cell_dimensions(height=600)
# cleanup
views.DATA = {}
views.DTYPES = {}
views.SETTINGS = {}
views.METADATA = {}
assert type(instance.__str__()).__name__ == 'str'
assert type(instance.__repr__()).__name__ == 'str'
def mock_run(self, *args, **kwargs):
assert self.jinja_env.auto_reload
assert self.config['TEMPLATES_AUTO_RELOAD']
with mock.patch('dtale.app.DtaleFlask.run', mock_run):
show(data=test_data, subprocess=False, port=9999, debug=True)
with mock.patch('dtale.app._thread.start_new_thread', mock.Mock()) as mock_thread:
show(data=test_data, subprocess=True)
mock_thread.assert_called()
# cleanup
views.DATA = {}
views.DTYPES = {}
views.SETTINGS = {}
views.METADATA = {}
mock_run = stack.enter_context(mock.patch('dtale.app.DtaleFlask.run', mock.Mock()))
mock_find_free_port = stack.enter_context(mock.patch('dtale.app.find_free_port', mock.Mock(return_value=9999)))
stack.enter_context(mock.patch('socket.gethostname', mock.Mock(return_value='localhost')))
stack.enter_context(mock.patch('dtale.app.is_up', mock.Mock(return_value=False)))
mock_requests = stack.enter_context(mock.patch('requests.get', mock.Mock()))
instance = show(data=test_data, subprocess=False, name='foo')
assert 'http://localhost:9999' == instance._url
mock_run.assert_called_once()
mock_find_free_port.assert_called_once()
pdt.assert_frame_equal(instance.data, test_data)
tmp = test_data.copy()
tmp['biz'] = 2.5
instance.data = tmp
unittest.assertEqual(
views.DTYPES[instance._data_id],
views.build_dtypes_state(tmp),
'should update app data/dtypes'
)
instance2 = get_instance(instance._data_id)
assert instance2._url == instance._url
assert instances()[instance._data_id]._url == instance._url
instance.kill()
mock_requests.assert_called_once()
mock_requests.call_args[0][0] == 'http://localhost:9999/shutdown'
assert views.METADATA['1']['name'] == 'foo'
with ExitStack() as stack:
mock_run = stack.enter_context(mock.patch('dtale.app.DtaleFlask.run', mock.Mock()))
mock_find_free_port = stack.enter_context(mock.patch('dtale.app.find_free_port', mock.Mock(return_value=9999)))
getter = namedtuple('get', 'ok')
def mock_requests_get(url, verify=True):
if url.endswith('/health'):
is_ok = get_calls['ct'] > 0
get_calls['ct'] += 1
return getter(is_ok)
return getter(True)
stack.enter_context(mock.patch('requests.get', mock_requests_get))
instance = show(data=test_data, subprocess=False, name='foo')
assert instance.__repr__() == ''
assert instance.__str__() == ''
instance.adjust_cell_dimensions(height=600)
# cleanup
views.DATA = {}
views.DTYPES = {}
views.SETTINGS = {}
views.METADATA = {}
pdt.assert_frame_equal(instance.data, test_data.reset_index())
unittest.assertEqual(views.SETTINGS[instance._data_id], dict(locked=['date', 'security_id']), 'should lock index columns')
test_data = test_data.reset_index()
instance = views.startup(URL, data=test_data)
pdt.assert_frame_equal(instance.data, test_data)
unittest.assertEqual(views.SETTINGS[instance._data_id], dict(locked=[]), 'no index = nothing locked')
test_data = pd.DataFrame([dict(date=pd.Timestamp('now'), security_id=1)])
test_data = test_data.set_index('security_id').date
instance = views.startup(URL, data_loader=lambda: test_data)
pdt.assert_frame_equal(instance.data, test_data.reset_index())
unittest.assertEqual(views.SETTINGS[instance._data_id], dict(locked=['security_id']), 'should lock index columns')
test_data = pd.DatetimeIndex([pd.Timestamp('now')], name='date')
instance = views.startup(URL, data_loader=lambda: test_data)
pdt.assert_frame_equal(instance.data, test_data.to_frame(index=False))
unittest.assertEqual(views.SETTINGS[instance._data_id], dict(locked=[]), 'should lock index columns')
test_data = pd.MultiIndex.from_arrays([[1, 2], [3, 4]], names=('a', 'b'))
instance = views.startup(URL, data_loader=lambda: test_data)
pdt.assert_frame_equal(instance.data, test_data.to_frame(index=False))
unittest.assertEqual(views.SETTINGS[instance._data_id], dict(locked=[]), 'should lock index columns')