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_applybycols_with_drop():
"""Testing ApplyByCols pipeline stages."""
df = ph_df()
round_ph = ApplyByCols("ph", math.ceil, drop=False)
res_df = round_ph(df)
assert 'ph' in res_df.columns
assert 'ph_app' in res_df.columns
assert res_df.columns.get_loc('ph') == 0
assert res_df.columns.get_loc('ph_app') == 1
assert res_df['ph_app'][1] == 4
assert res_df['ph_app'][2] == 8
assert res_df['ph_app'][3] == 13
def test_applybycols_with_result_columns():
"""Testing ApplyByCols pipeline stages."""
df = ph_df()
round_ph = ApplyByCols("ph", math.ceil, result_columns='round_ph')
res_df = round_ph(df)
assert 'ph' not in res_df.columns
assert res_df.columns.get_loc('round_ph') == 0
assert res_df['round_ph'][1] == 4
assert res_df['round_ph'][2] == 8
assert res_df['round_ph'][3] == 13
def test_applybycols_with_bad_len_result_columns():
"""Testing ApplyByCols pipeline stages."""
with pytest.raises(ValueError):
ApplyByCols("ph", math.ceil, result_columns=['a', 'b'])
def test_applybycols_func_desc():
"""Testing ApplyByCols pipeline stages."""
df = ph_df()
round_ph = ApplyByCols("ph", math.ceil, func_desc='Round PH values')
res_df = round_ph(df)
assert res_df.columns.get_loc('ph') == 0
assert res_df['ph'][1] == 4
assert res_df['ph'][2] == 8
assert res_df['ph'][3] == 13
def test_applybycols():
"""Testing ApplyByCols pipeline stages."""
df = ph_df()
round_ph = ApplyByCols("ph", math.ceil)
res_df = round_ph(df)
assert res_df.columns.get_loc('ph') == 0
assert res_df['ph'][1] == 4
assert res_df['ph'][2] == 8
assert res_df['ph'][3] == 13
else:
self._result_columns = [col + suffix for col in self._columns]
else:
self._result_columns = _interpret_columns_param(result_columns)
if len(self._result_columns) != len(self._columns):
raise ValueError(
"columns and result_columns parameters must"
" be string lists of the same length!"
)
self._drop = drop
if func_desc is None:
func_desc = ""
self._func_desc = func_desc
col_str = _list_str(self._columns)
sfx = "s" if len(self._columns) > 1 else ""
base_str = ApplyByCols._BASE_STR.format(self._func_desc, sfx, col_str)
super_kwargs = {
"exmsg": base_str + ApplyByCols._DEF_EXC_MSG_SUFFIX,
"appmsg": base_str + ApplyByCols._DEF_APP_MSG_SUFFIX,
"desc": base_str + ApplyByCols._DEF_DESCRIPTION_SUFFIX,
}
super_kwargs.update(**kwargs)
super().__init__(**super_kwargs)
self._result_columns = _interpret_columns_param(result_columns)
if len(self._result_columns) != len(self._columns):
raise ValueError(
"columns and result_columns parameters must"
" be string lists of the same length!"
)
self._drop = drop
if func_desc is None:
func_desc = ""
self._func_desc = func_desc
col_str = _list_str(self._columns)
sfx = "s" if len(self._columns) > 1 else ""
base_str = ApplyByCols._BASE_STR.format(self._func_desc, sfx, col_str)
super_kwargs = {
"exmsg": base_str + ApplyByCols._DEF_EXC_MSG_SUFFIX,
"appmsg": base_str + ApplyByCols._DEF_APP_MSG_SUFFIX,
"desc": base_str + ApplyByCols._DEF_DESCRIPTION_SUFFIX,
}
super_kwargs.update(**kwargs)
super().__init__(**super_kwargs)
if len(self._result_columns) != len(self._columns):
raise ValueError(
"columns and result_columns parameters must"
" be string lists of the same length!"
)
self._drop = drop
if func_desc is None:
func_desc = ""
self._func_desc = func_desc
col_str = _list_str(self._columns)
sfx = "s" if len(self._columns) > 1 else ""
base_str = ApplyByCols._BASE_STR.format(self._func_desc, sfx, col_str)
super_kwargs = {
"exmsg": base_str + ApplyByCols._DEF_EXC_MSG_SUFFIX,
"appmsg": base_str + ApplyByCols._DEF_APP_MSG_SUFFIX,
"desc": base_str + ApplyByCols._DEF_DESCRIPTION_SUFFIX,
}
super_kwargs.update(**kwargs)
super().__init__(**super_kwargs)
else:
self._result_columns = _interpret_columns_param(result_columns)
if len(self._result_columns) != len(self._columns):
raise ValueError(
"columns and result_columns parameters must"
" be string lists of the same length!"
)
self._drop = drop
if func_desc is None:
func_desc = ""
self._func_desc = func_desc
col_str = _list_str(self._columns)
sfx = "s" if len(self._columns) > 1 else ""
base_str = ApplyByCols._BASE_STR.format(self._func_desc, sfx, col_str)
super_kwargs = {
"exmsg": base_str + ApplyByCols._DEF_EXC_MSG_SUFFIX,
"appmsg": base_str + ApplyByCols._DEF_APP_MSG_SUFFIX,
"desc": base_str + ApplyByCols._DEF_DESCRIPTION_SUFFIX,
}
super_kwargs.update(**kwargs)
super().__init__(**super_kwargs)