Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use notify method in lisa
Best Python code snippet using lisa_python
tasks.py
Source:tasks.py 
...344 total_nums = response_list['increment_trades_get_response']['total_results']345 if total_nums > 0:346 notify_list = response_list['increment_trades_get_response']['notify_trades']['notify_trade']347 for notify in notify_list:348 TradeNotify.save_and_post_notify(notify)349 cur_nums = cur_page * settings.TAOBAO_PAGE_SIZE * 2350 has_next = cur_nums < total_nums351 cur_page += 1352 except Exception, exc:353 raise process_trade_interval_notify_task.retry(exc=exc, countdown=60)354 else:355 if not update_handler:356 user.trade_notify_updated = updated357 user.save()358############################ æ¹éä¸è½½åå主å¨æ¶æ¯å¤ç ###############################359@app.task(max_retries=3)360def process_item_interval_notify_task(user_id, update_from=None, update_to=None):361 update_handler = update_from and update_to362 try:363 user = User.objects.get(visitor_id=user_id)364 if not update_handler:365 update_from = user.trade_notify_updated366 dt = datetime.datetime.now()367 update_to = dt if dt.day == update_from.day else \368 datetime.datetime(update_from.year, update_from.month, update_from.day, 23, 59, 59)369 updated = dt if dt.day == update_from.day else \370 datetime.datetime(dt.year, dt.month, dt.day, 0, 0, 0)371 nick = user.nick372 update_from = update_from.strftime('%Y-%m-%d %H:%M:%S')373 update_to = update_to.strftime('%Y-%m-%d %H:%M:%S')374 has_next = True375 cur_page = 1376 while has_next:377 response_list = apis.taobao_increment_items_get(tb_user_id=user_id, nick=nick, page_no=cur_page378 , page_size=settings.TAOBAO_PAGE_SIZE * 2,379 start_modified=update_from, end_modified=update_to)380 total_nums = response_list['increment_items_get_response']['total_results']381 if total_nums > 0:382 notify_list = response_list['increment_items_get_response']['notify_items']['notify_item']383 for notify in notify_list:384 ItemNotify.save_and_post_notify(notify)385 cur_nums = cur_page * settings.TAOBAO_PAGE_SIZE * 2386 has_next = cur_nums < total_nums387 cur_page += 1388 except Exception, exc:389 raise process_item_interval_notify_task.retry(exc=exc, countdown=60)390 else:391 if not update_handler:392 user.item_notify_updated = updated393 user.save()394############################ æ¹éä¸è½½é款主å¨æ¶æ¯å¤ç ###############################395@app.task(max_retries=3)396def process_refund_interval_notify_task(user_id, update_from=None, update_to=None):397 update_handler = update_from and update_to398 try:399 user = User.objects.get(visitor_id=user_id)400 if not update_handler:401 update_from = user.trade_notify_updated402 dt = datetime.datetime.now()403 update_to = dt if dt.day == update_from.day else \404 datetime.datetime(update_from.year, update_from.month, update_from.day, 23, 59, 59)405 updated = dt if dt.day == update_from.day else \406 datetime.datetime(dt.year, dt.month, dt.day, 0, 0, 0)407 nick = user.nick408 update_from = update_from.strftime('%Y-%m-%d %H:%M:%S')409 update_to = update_to.strftime('%Y-%m-%d %H:%M:%S')410 has_next = True411 cur_page = 1412 while has_next:413 response_list = apis.taobao_increment_refunds_get(tb_user_id=user_id, nick=nick, page_no=cur_page414 , page_size=settings.TAOBAO_PAGE_SIZE * 2,415 start_modified=update_from, end_modified=update_to)416 total_nums = response_list['increment_refunds_get_response']['total_results']417 if total_nums > 0:418 notify_list = response_list['increment_refunds_get_response']['notify_refunds']['notify_refund']419 for notify in notify_list:420 RefundNotify.save_and_post_notify(notify)421 cur_nums = cur_page * settings.TAOBAO_PAGE_SIZE * 2422 has_next = cur_nums < total_nums423 cur_page += 1424 except Exception, exc:425 raise process_refund_interval_notify_task.retry(exc=exc, countdown=60)426 else:427 if not update_handler:428 user.refund_notify_updated = updated429 user.save()430############################ å¢é订å主å¨æ¶æ¯å¤ç ###############################431@app.task432def process_trade_increment_notify_task():433 users = User.objects.all()434 for user in users:...
test_notify.py
Source:test_notify.py
1"""Test the HMAC-based One Time Password (MFA) auth module."""2import asyncio3from unittest.mock import patch4from homeassistant import data_entry_flow5from homeassistant.auth import auth_manager_from_config, models as auth_models6from homeassistant.auth.mfa_modules import auth_mfa_module_from_config7from homeassistant.components.notify import NOTIFY_SERVICE_SCHEMA8from tests.common import MockUser, async_mock_service9MOCK_CODE = "123456"10MOCK_CODE_2 = "654321"11async def test_validating_mfa(hass):12 """Test validating mfa code."""13 notify_auth_module = await auth_mfa_module_from_config(hass, {"type": "notify"})14 await notify_auth_module.async_setup_user("test-user", {"notify_service": "dummy"})15 with patch("pyotp.HOTP.verify", return_value=True):16 assert await notify_auth_module.async_validate("test-user", {"code": MOCK_CODE})17async def test_validating_mfa_invalid_code(hass):18 """Test validating an invalid mfa code."""19 notify_auth_module = await auth_mfa_module_from_config(hass, {"type": "notify"})20 await notify_auth_module.async_setup_user("test-user", {"notify_service": "dummy"})21 with patch("pyotp.HOTP.verify", return_value=False):22 assert (23 await notify_auth_module.async_validate("test-user", {"code": MOCK_CODE})24 is False25 )26async def test_validating_mfa_invalid_user(hass):27 """Test validating an mfa code with invalid user."""28 notify_auth_module = await auth_mfa_module_from_config(hass, {"type": "notify"})29 await notify_auth_module.async_setup_user("test-user", {"notify_service": "dummy"})30 assert (31 await notify_auth_module.async_validate("invalid-user", {"code": MOCK_CODE})32 is False33 )34async def test_validating_mfa_counter(hass):35 """Test counter will move only after generate code."""36 notify_auth_module = await auth_mfa_module_from_config(hass, {"type": "notify"})37 await notify_auth_module.async_setup_user(38 "test-user", {"counter": 0, "notify_service": "dummy"}39 )40 async_mock_service(hass, "notify", "dummy")41 assert notify_auth_module._user_settings42 notify_setting = list(notify_auth_module._user_settings.values())[0]43 init_count = notify_setting.counter44 assert init_count is not None45 with patch("pyotp.HOTP.at", return_value=MOCK_CODE):46 await notify_auth_module.async_initialize_login_mfa_step("test-user")47 notify_setting = list(notify_auth_module._user_settings.values())[0]48 after_generate_count = notify_setting.counter49 assert after_generate_count != init_count50 with patch("pyotp.HOTP.verify", return_value=True):51 assert await notify_auth_module.async_validate("test-user", {"code": MOCK_CODE})52 notify_setting = list(notify_auth_module._user_settings.values())[0]53 assert after_generate_count == notify_setting.counter54 with patch("pyotp.HOTP.verify", return_value=False):55 assert (56 await notify_auth_module.async_validate("test-user", {"code": MOCK_CODE})57 is False58 )59 notify_setting = list(notify_auth_module._user_settings.values())[0]60 assert after_generate_count == notify_setting.counter61async def test_setup_depose_user(hass):62 """Test set up and despose user."""63 notify_auth_module = await auth_mfa_module_from_config(hass, {"type": "notify"})64 await notify_auth_module.async_setup_user("test-user", {})65 assert len(notify_auth_module._user_settings) == 166 await notify_auth_module.async_setup_user("test-user", {})67 assert len(notify_auth_module._user_settings) == 168 await notify_auth_module.async_depose_user("test-user")69 assert len(notify_auth_module._user_settings) == 070 await notify_auth_module.async_setup_user("test-user2", {"secret": "secret-code"})71 assert len(notify_auth_module._user_settings) == 172async def test_login_flow_validates_mfa(hass):73 """Test login flow with mfa enabled."""74 hass.auth = await auth_manager_from_config(75 hass,76 [77 {78 "type": "insecure_example",79 "users": [{"username": "test-user", "password": "test-pass"}],80 }81 ],82 [{"type": "notify"}],83 )84 user = MockUser(85 id="mock-user", is_owner=False, is_active=False, name="Paulus"86 ).add_to_auth_manager(hass.auth)87 await hass.auth.async_link_user(88 user,89 auth_models.Credentials(90 id="mock-id",91 auth_provider_type="insecure_example",92 auth_provider_id=None,93 data={"username": "test-user"},94 is_new=False,95 ),96 )97 notify_calls = async_mock_service(98 hass, "notify", "test-notify", NOTIFY_SERVICE_SCHEMA99 )100 await hass.auth.async_enable_user_mfa(101 user, "notify", {"notify_service": "test-notify"}102 )103 provider = hass.auth.auth_providers[0]104 result = await hass.auth.login_flow.async_init((provider.type, provider.id))105 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM106 result = await hass.auth.login_flow.async_configure(107 result["flow_id"], {"username": "incorrect-user", "password": "test-pass"}108 )109 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM110 assert result["errors"]["base"] == "invalid_auth"111 result = await hass.auth.login_flow.async_configure(112 result["flow_id"], {"username": "test-user", "password": "incorrect-pass"}113 )114 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM115 assert result["errors"]["base"] == "invalid_auth"116 with patch("pyotp.HOTP.at", return_value=MOCK_CODE):117 result = await hass.auth.login_flow.async_configure(118 result["flow_id"], {"username": "test-user", "password": "test-pass"}119 )120 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM121 assert result["step_id"] == "mfa"122 assert result["data_schema"].schema.get("code") == str123 # wait service call finished124 await hass.async_block_till_done()125 assert len(notify_calls) == 1126 notify_call = notify_calls[0]127 assert notify_call.domain == "notify"128 assert notify_call.service == "test-notify"129 message = notify_call.data["message"]130 message.hass = hass131 assert MOCK_CODE in message.async_render()132 with patch("pyotp.HOTP.verify", return_value=False):133 result = await hass.auth.login_flow.async_configure(134 result["flow_id"], {"code": "invalid-code"}135 )136 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM137 assert result["step_id"] == "mfa"138 assert result["errors"]["base"] == "invalid_code"139 # wait service call finished140 await hass.async_block_till_done()141 # would not send new code, allow user retry142 assert len(notify_calls) == 1143 # retry twice144 with patch("pyotp.HOTP.verify", return_value=False), patch(145 "pyotp.HOTP.at", return_value=MOCK_CODE_2146 ):147 result = await hass.auth.login_flow.async_configure(148 result["flow_id"], {"code": "invalid-code"}149 )150 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM151 assert result["step_id"] == "mfa"152 assert result["errors"]["base"] == "invalid_code"153 # after the 3rd failure, flow abort154 result = await hass.auth.login_flow.async_configure(155 result["flow_id"], {"code": "invalid-code"}156 )157 assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT158 assert result["reason"] == "too_many_retry"159 # wait service call finished160 await hass.async_block_till_done()161 # restart login162 result = await hass.auth.login_flow.async_init((provider.type, provider.id))163 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM164 with patch("pyotp.HOTP.at", return_value=MOCK_CODE):165 result = await hass.auth.login_flow.async_configure(166 result["flow_id"], {"username": "test-user", "password": "test-pass"}167 )168 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM169 assert result["step_id"] == "mfa"170 assert result["data_schema"].schema.get("code") == str171 # wait service call finished172 await hass.async_block_till_done()173 assert len(notify_calls) == 2174 notify_call = notify_calls[1]175 assert notify_call.domain == "notify"176 assert notify_call.service == "test-notify"177 message = notify_call.data["message"]178 message.hass = hass179 assert MOCK_CODE in message.async_render()180 with patch("pyotp.HOTP.verify", return_value=True):181 result = await hass.auth.login_flow.async_configure(182 result["flow_id"], {"code": MOCK_CODE}183 )184 assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY185 assert result["data"].id == "mock-id"186async def test_setup_user_notify_service(hass):187 """Test allow select notify service during mfa setup."""188 notify_calls = async_mock_service(hass, "notify", "test1", NOTIFY_SERVICE_SCHEMA)189 async_mock_service(hass, "notify", "test2", NOTIFY_SERVICE_SCHEMA)190 notify_auth_module = await auth_mfa_module_from_config(hass, {"type": "notify"})191 services = notify_auth_module.aync_get_available_notify_services()192 assert services == ["test1", "test2"]193 flow = await notify_auth_module.async_setup_flow("test-user")194 step = await flow.async_step_init()195 assert step["type"] == data_entry_flow.RESULT_TYPE_FORM196 assert step["step_id"] == "init"197 schema = step["data_schema"]198 schema({"notify_service": "test2"})199 with patch("pyotp.HOTP.at", return_value=MOCK_CODE):200 step = await flow.async_step_init({"notify_service": "test1"})201 assert step["type"] == data_entry_flow.RESULT_TYPE_FORM202 assert step["step_id"] == "setup"203 # wait service call finished204 await hass.async_block_till_done()205 assert len(notify_calls) == 1206 notify_call = notify_calls[0]207 assert notify_call.domain == "notify"208 assert notify_call.service == "test1"209 message = notify_call.data["message"]210 message.hass = hass211 assert MOCK_CODE in message.async_render()212 with patch("pyotp.HOTP.at", return_value=MOCK_CODE_2):213 step = await flow.async_step_setup({"code": "invalid"})214 assert step["type"] == data_entry_flow.RESULT_TYPE_FORM215 assert step["step_id"] == "setup"216 assert step["errors"]["base"] == "invalid_code"217 # wait service call finished218 await hass.async_block_till_done()219 assert len(notify_calls) == 2220 notify_call = notify_calls[1]221 assert notify_call.domain == "notify"222 assert notify_call.service == "test1"223 message = notify_call.data["message"]224 message.hass = hass225 assert MOCK_CODE_2 in message.async_render()226 with patch("pyotp.HOTP.verify", return_value=True):227 step = await flow.async_step_setup({"code": MOCK_CODE_2})228 assert step["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY229async def test_include_exclude_config(hass):230 """Test allow include exclude config."""231 async_mock_service(hass, "notify", "include1", NOTIFY_SERVICE_SCHEMA)232 async_mock_service(hass, "notify", "include2", NOTIFY_SERVICE_SCHEMA)233 async_mock_service(hass, "notify", "exclude1", NOTIFY_SERVICE_SCHEMA)234 async_mock_service(hass, "notify", "exclude2", NOTIFY_SERVICE_SCHEMA)235 async_mock_service(hass, "other", "include3", NOTIFY_SERVICE_SCHEMA)236 async_mock_service(hass, "other", "exclude3", NOTIFY_SERVICE_SCHEMA)237 notify_auth_module = await auth_mfa_module_from_config(238 hass, {"type": "notify", "exclude": ["exclude1", "exclude2", "exclude3"]}239 )240 services = notify_auth_module.aync_get_available_notify_services()241 assert services == ["include1", "include2"]242 notify_auth_module = await auth_mfa_module_from_config(243 hass, {"type": "notify", "include": ["include1", "include2", "include3"]}244 )245 services = notify_auth_module.aync_get_available_notify_services()246 assert services == ["include1", "include2"]247 # exclude has high priority than include248 notify_auth_module = await auth_mfa_module_from_config(249 hass,250 {251 "type": "notify",252 "include": ["include1", "include2", "include3"],253 "exclude": ["exclude1", "exclude2", "include2"],254 },255 )256 services = notify_auth_module.aync_get_available_notify_services()257 assert services == ["include1"]258async def test_setup_user_no_notify_service(hass):259 """Test setup flow abort if there is no available notify service."""260 async_mock_service(hass, "notify", "test1", NOTIFY_SERVICE_SCHEMA)261 notify_auth_module = await auth_mfa_module_from_config(262 hass, {"type": "notify", "exclude": "test1"}263 )264 services = notify_auth_module.aync_get_available_notify_services()265 assert services == []266 flow = await notify_auth_module.async_setup_flow("test-user")267 step = await flow.async_step_init()268 assert step["type"] == data_entry_flow.RESULT_TYPE_ABORT269 assert step["reason"] == "no_available_service"270async def test_not_raise_exception_when_service_not_exist(hass):271 """Test login flow will not raise exception when notify service error."""272 hass.auth = await auth_manager_from_config(273 hass,274 [275 {276 "type": "insecure_example",277 "users": [{"username": "test-user", "password": "test-pass"}],278 }279 ],280 [{"type": "notify"}],281 )282 user = MockUser(283 id="mock-user", is_owner=False, is_active=False, name="Paulus"284 ).add_to_auth_manager(hass.auth)285 await hass.auth.async_link_user(286 user,287 auth_models.Credentials(288 id="mock-id",289 auth_provider_type="insecure_example",290 auth_provider_id=None,291 data={"username": "test-user"},292 is_new=False,293 ),294 )295 await hass.auth.async_enable_user_mfa(296 user, "notify", {"notify_service": "invalid-notify"}297 )298 provider = hass.auth.auth_providers[0]299 result = await hass.auth.login_flow.async_init((provider.type, provider.id))300 assert result["type"] == data_entry_flow.RESULT_TYPE_FORM301 with patch("pyotp.HOTP.at", return_value=MOCK_CODE):302 result = await hass.auth.login_flow.async_configure(303 result["flow_id"], {"username": "test-user", "password": "test-pass"}304 )305 assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT306 assert result["reason"] == "unknown_error"307 # wait service call finished308 await hass.async_block_till_done()309async def test_race_condition_in_data_loading(hass):310 """Test race condition in the data loading."""311 counter = 0312 async def mock_load(_):313 """Mock homeassistant.helpers.storage.Store.async_load."""314 nonlocal counter315 counter += 1316 await asyncio.sleep(0)317 notify_auth_module = await auth_mfa_module_from_config(hass, {"type": "notify"})318 with patch("homeassistant.helpers.storage.Store.async_load", new=mock_load):319 task1 = notify_auth_module.async_validate("user", {"code": "value"})320 task2 = notify_auth_module.async_validate("user", {"code": "value"})321 results = await asyncio.gather(task1, task2, return_exceptions=True)322 assert counter == 1323 assert results[0] is False...
test_notify_csv.py
Source:test_notify_csv.py
...16 notify = mock.MagicMock()17 notify.send = mock.MagicMock()18 notify.admin_send = mock.MagicMock()19 with TempFile(data, suffix='.csv') as csv:20 with global_notify(notify):21 meta = parse(csv)22 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)23 assert meta['data_type'] == 'GENERAL'24 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))25 assert not notify.admin_send.called26def test_parse_with_invalid_csv():27 # Since we accept text columns, we presume that strings appear to be errors28 # only if it's less than 10% of them in a column. Hence it can be tested29 # only on a dataset with minimum 11 columns, where one contains string.30 #31 data = '1 2 3 4\n1 2 3 4\n5 6 a 8\n7 6 5 2\n8 8 8 8\n2 2 3 9\n5 6 7 8\n12 13 45 56\n12 43 6 7\n9 9 9 0\n1 2 5 0\n'32 log = """33 Parsing CSV with whitespace (tab) as delimiter.34 Found 4 fields in first row, assume all the rows have this number of fields.35 Parsing...36 Analyzing data...37 No header found, first row contains data.38 Found 1 row with invalid values:39 - row 3, column 340 Found 10 samples.41 """42 notify = mock.MagicMock()43 notify.send = mock.MagicMock()44 notify.admin_send = mock.MagicMock()45 with TempFile(data, suffix='.csv') as csv:46 with global_notify(notify):47 parse(csv)48 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)49 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))50 assert not notify.admin_send.called51def test_parse_with_invalid_csv_with_null_bytes():52 data = "1 2 3 4\n1 2 3 4\n5 6 \\x00 8\n7 6 5 2\n8 8 8 8\n2 2 3 9\n5 6 7 8\n12 13 45 56\n12 43 6 7\n9 9 9 0\n1 2 5 0\n"53 log = """54 Parsing CSV with whitespace (tab) as delimiter.55 Found 4 fields in first row, assume all the rows have this number of fields.56 Parsing...57 Analyzing data...58 No header found, first row contains data.59 Found 1 row with invalid values:60 - row 3, column 361 Found 10 samples.62 """63 notify = mock.MagicMock()64 notify.send = mock.MagicMock()65 notify.admin_send = mock.MagicMock()66 with TempFile(data, suffix='.csv') as csv:67 with global_notify(notify):68 parse(csv)69 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)70 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))71 assert not notify.admin_send.called72def test_parse_with_invalid_csv_other_file_type():73 data = '%PDF-1.4\n'74 log = """75 CSV doesn't contain a valid delimiter.76 This means your file isn't properly formatted77 (or you submitted another type of file).78 """79 notify = mock.MagicMock()80 notify.send = mock.MagicMock()81 notify.admin_send = mock.MagicMock()82 with TempFile(data, suffix='.csv') as csv:83 with global_notify(notify):84 with pytest.raises(InvalidCSV):85 parse(csv)86 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)87 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))88 assert not notify.admin_send.called89def test_parse_with_invalid_csv_other_file_type_valid_delimiter():90 data = '%PDF,-1.4\nadsadsadsad'91 log = """92 Parsing CSV with comma as delimiter.93 Found 2 fields in first row, assume all the rows have this number of fields.94 Parsing...95 Analyzing data...96 The dataset is empty or isn't properly formatted.97 """98 notify = mock.MagicMock()99 notify.send = mock.MagicMock()100 notify.admin_send = mock.MagicMock()101 with TempFile(data, suffix='.csv') as csv:102 with global_notify(notify):103 with pytest.raises(InvalidCSV):104 parse(csv)105 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)106 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))107def test_parse_invalid_delimiter():108 data = '3:4:5:4:dsf:/n'109 log = """110 CSV doesn't contain a valid delimiter.111 This means your file isn't properly formatted112 (or you submitted another type of file).113 """114 notify = mock.MagicMock()115 notify.send = mock.MagicMock()116 notify.admin_send = mock.MagicMock()117 with TempFile(data, suffix='.csv') as csv:118 with global_notify(notify):119 with pytest.raises(InvalidCSV):120 parse(csv)121 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)122 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))123 assert not notify.admin_send.called124def test_parse_invalid_first_line():125 data = '123,'126 log = """127 Parsing CSV with comma as delimiter.128 With selected delimiter found only 1 columns in first row, must be at least 2.129 This means your file isn't properly formatted130 (or you submitted another type of file).131 """132 notify = mock.MagicMock()133 notify.send = mock.MagicMock()134 notify.admin_send = mock.MagicMock()135 with TempFile(data, suffix='.csv') as csv:136 with global_notify(notify):137 with pytest.raises(InvalidCSV):138 parse(csv)139 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)140 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))141 assert not notify.admin_send.called142def test_parse_rows_non_consistent():143 # Warning! No error here: inconsisten rows ignored by np.genfromtxt144 data = '1,2,3,4,5,6,7,8\n8,7,6,5,4,3,2,1\n56,34,34\n'145 log = """146 Parsing CSV with comma as delimiter.147 Found 8 fields in first row, assume all the rows have this number of fields.148 Parsing...149 Analyzing data...150 No header found, first row contains data.151 Found 1 row with invalid values:152 - row 3, column 4153 Found 2 samples.154 """155 notify = mock.MagicMock()156 notify.send = mock.MagicMock()157 notify.admin_send = mock.MagicMock()158 with TempFile(data, suffix='.csv') as csv:159 with global_notify(notify):160 meta = parse(csv)161 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)162 assert meta['data_type'] == 'GENERAL'163 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))164 assert not notify.admin_send.called165def test_parse_only_two_row():166 data = ' \n4,\n'167 log = """168 First row is empty, it must contain headers or data.169 This means your file isn't properly formatted170 (or you submitted another type of file).171 """172 notify = mock.MagicMock()173 notify.send = mock.MagicMock()174 notify.admin_send = mock.MagicMock()175 with TempFile(data, suffix='.csv') as csv:176 with global_notify(notify):177 with pytest.raises(InvalidCSV):178 parse(csv)179 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)180 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))181 assert not notify.admin_send.called182def test_parse_different_delimiters_per_Row():183 data = '4 5\n4,6\n'184 log = """185 Parsing CSV with whitespace (tab) as delimiter.186 Found 2 fields in first row, assume all the rows have this number of fields.187 Parsing...188 Analyzing data...189 The dataset is empty or isn't properly formatted.190 """191 notify = mock.MagicMock()192 notify.send = mock.MagicMock()193 notify.admin_send = mock.MagicMock()194 with TempFile(data, suffix='.csv') as csv:195 with global_notify(notify):196 with pytest.raises(InvalidCSV):197 parse(csv)198 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)199 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))200 assert not notify.admin_send.called201def test_parse_no_data():202 data = ''203 log = """204 First row is empty, it must contain headers or data.205 This means your file isn't properly formatted206 (or you submitted another type of file).207 """208 notify = mock.MagicMock()209 notify.send = mock.MagicMock()210 notify.admin_send = mock.MagicMock()211 with TempFile(data, suffix='.csv') as csv:212 with global_notify(notify):213 with pytest.raises(InvalidCSV):214 parse(csv)215 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)216 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))217 assert not notify.admin_send.called218def test_parse_not_enough_columns():219 data = '3,\n 4,,5\n6,8,9\n'220 log = """221 Parsing CSV with comma as delimiter.222 With selected delimiter found only 1 columns in first row, must be at least 2.223 This means your file isn't properly formatted224 (or you submitted another type of file).225 """226 notify = mock.MagicMock()227 notify.send = mock.MagicMock()228 notify.admin_send = mock.MagicMock()229 with TempFile(data, suffix='.csv') as csv:230 with global_notify(notify):231 with pytest.raises(InvalidCSV):232 parse(csv)233 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)234 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))235 assert not notify.admin_send.called236def test_parse_bad_column_data_first_row_empty():237 data = '3,,,\n4,\n6,8,9\n'238 log = """239 Parsing CSV with comma as delimiter.240 Found 4 fields in first row, assume all the rows have this number of fields.241 Parsing...242 Analyzing data...243 The dataset is empty or isn't properly formatted.244 """245 notify = mock.MagicMock()246 notify.send = mock.MagicMock()247 notify.admin_send = mock.MagicMock()248 with TempFile(data, suffix='.csv') as csv:249 with global_notify(notify):250 with pytest.raises(InvalidCSV):251 parse(csv)252 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)253 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))254 assert not notify.admin_send.called255def test_parse_bad_column_data_first_row():256 data = '3,\\x00,4,3\n4,\n6,8,9\n'257 log = """258 Parsing CSV with comma as delimiter.259 Found 4 fields in first row, assume all the rows have this number of fields.260 Parsing...261 Analyzing data...262 The dataset is empty or isn't properly formatted.263 """264 notify = mock.MagicMock()265 notify.send = mock.MagicMock()266 notify.admin_send = mock.MagicMock()267 with TempFile(data, suffix='.csv') as csv:268 with global_notify(notify):269 with pytest.raises(InvalidCSV):270 parse(csv)271 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)272 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))273 assert not notify.admin_send.called274def test_parse_invalid_delimiter2():275 data = '3:4:5:4:dsf:/n/n/n/n/n/n/n/n/n/n/n/n/n/n/n/n/n'276 log = """277 CSV doesn't contain a valid delimiter.278 This means your file isn't properly formatted279 (or you submitted another type of file).280 """281 notify = mock.MagicMock()282 notify.send = mock.MagicMock()283 notify.admin_send = mock.MagicMock()284 with TempFile(data, suffix='.csv') as csv:285 with global_notify(notify):286 with pytest.raises(InvalidCSV):287 parse(csv)288 rval = '\n'.join(x[0][0] for x in notify.send.call_args_list)289 assert rval == '\n'.join(x.strip() for x in log.strip().split('\n'))...
notify.py
Source:notify.py
...254 raise NotImplementedError255 if self.handler is None:256 self.handler = self.route_map[self.notify.notify_type](self.notify, self.user)257 return self.handler258 def load_notify(self):259 # if self.notify.notify_type not in self.route_map:260 # raise NotImplementedError261 # _handler = self.route_map[self.notify.notify_type](self.notify)262 # return _handler.output()263 return self.get_handler().output()264 def check(self):265 return self.get_handler().check()266 def remove(self):267 return self.get_handler().remove()268 def read_notify(self, user_id):269 user = User.objects(id=user_id).first()270 if not user or self.notify not in user.notify_content:271 print 'fail to read notify', self.notify, user.notify_content272 return False273 print 'read', self.notify, type(self.notify)274 self.notify.read = True275 self.notify.save()...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
