KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use load_module_from_file method in autotest
Best Python code snippet using autotest_python
test_services.py
Source:test_services.py 
...6from surrogate import surrogate7from mqttwarn.util import load_module_by_name, load_module_from_file8from mqttwarn.model import ProcessorItem as Item9def test_alexa_notify_me_success(srv, caplog):10 module = load_module_from_file("mqttwarn/services/alexa-notify-me.py")11 accessCode = "myToken"12 item = Item(addrs=[accessCode], message="â½ Notification message â½")13 with caplog.at_level(logging.DEBUG):14 with mock.patch("requests.post") as requests_mock:15 outcome = module.plugin(srv, item)16 requests_mock.assert_called_once_with(17 url="https://api.notifymyecho.com/v1/NotifyMe",18 data='{"notification": "\\u26bd Notification message \\u26bd", "accessCode": "myToken"}',19 )20 assert outcome is True21 assert "Sending to NotifyMe service" in caplog.text22 assert "Successfully sent to NotifyMe service" in caplog.text23def test_alexa_notify_me_real_auth_failure(srv, caplog):24 module = load_module_from_file("mqttwarn/services/alexa-notify-me.py")25 accessCode = "myToken"26 item = Item(addrs=[accessCode], message="â½ Notification message â½")27 with caplog.at_level(logging.DEBUG):28 outcome = module.plugin(srv, item)29 assert outcome is False30 assert "Sending to NotifyMe service" in caplog.text31 assert "Failed to send message to NotifyMe service" in caplog.text32@surrogate("puka")33@mock.patch("puka.Client", create=True)34def test_amqp_success(mock_puka_client, srv, caplog):35 module = load_module_by_name("mqttwarn.services.amqp")36 exchange, routing_key = ["name_of_exchange", "my_routing_key"]37 item = Item(38 config={"uri": "amqp://user:password@localhost:5672/"},39 target="test",40 addrs=[exchange, routing_key],41 message="â½ Notification message â½",42 )43 with caplog.at_level(logging.DEBUG):44 outcome = module.plugin(srv, item)45 assert mock_puka_client.mock_calls == [46 mock.call("amqp://user:password@localhost:5672/"),47 call().connect(),48 call().wait(mock.ANY),49 call().basic_publish(50 exchange="name_of_exchange",51 routing_key="my_routing_key",52 headers={53 "content_type": "text/plain",54 "x-agent": "mqttwarn",55 "delivery_mode": 1,56 },57 body="â½ Notification message â½",58 ),59 call().wait(mock.ANY),60 call().close(),61 ]62 assert outcome is True63 assert "AMQP publish to test [name_of_exchange/my_routing_key]" in caplog.text64 assert "Successfully published AMQP notification" in caplog.text65@surrogate("puka")66def test_amqp_failure(srv, caplog):67 module = load_module_by_name("mqttwarn.services.amqp")68 exchange, routing_key = ["name_of_exchange", "my_routing_key"]69 item = Item(70 config={"uri": "amqp://user:password@localhost:5672/"},71 target="test",72 addrs=[exchange, routing_key],73 message="â½ Notification message â½",74 )75 with caplog.at_level(logging.DEBUG):76 mock_connection = mock.MagicMock()77 # Make the call to `basic_publish` raise an exception.78 def error(*args, **kwargs):79 raise Exception("something failed")80 mock_connection.basic_publish = error81 with mock.patch(82 "puka.Client", side_effect=[mock_connection], create=True83 ) as mock_client:84 outcome = module.plugin(srv, item)85 assert mock_client.mock_calls == [86 mock.call("amqp://user:password@localhost:5672/"),87 ]88 assert mock_connection.mock_calls == [89 call.connect(),90 call.wait(mock.ANY),91 ]92 assert outcome is False93 assert (94 "AMQP publish to test [name_of_exchange/my_routing_key]" in caplog.text95 )96 assert (97 "Error on AMQP publish to test [name_of_exchange/my_routing_key]: something failed"98 in caplog.text99 )100@surrogate("apns")101@mock.patch("apns.APNs", create=True)102@mock.patch("apns.Payload", create=True)103def test_apns_success(mock_apns_payload, mock_apns, srv, caplog):104 with caplog.at_level(logging.DEBUG):105 module = load_module_from_file("mqttwarn/services/apns.py")106 cert_file, key_file = ["cert_file", "key_file"]107 item = Item(108 target="test",109 addrs=[cert_file, key_file],110 message="â½ Notification message â½",111 data={"apns_token": "foobar", "payload": "{}"},112 )113 outcome = module.plugin(srv, item)114 assert mock_apns_payload.mock_calls == [115 call(116 alert="â½ Notification message â½",117 custom={},118 sound="default",119 badge=1,120 ),121 ]122 assert mock_apns.mock_calls == [123 mock.call(use_sandbox=False, cert_file="cert_file", key_file="key_file"),124 call().gateway_server.send_notification("foobar", mock.ANY),125 ]126 assert outcome is True127 assert "Successfully published APNS notification to foobar" in caplog.text128@surrogate("apns")129@mock.patch("apns.APNs", create=True)130@mock.patch("apns.Payload", create=True)131def test_apns_success_no_payload(mock_apns_payload, mock_apns, srv, caplog):132 with caplog.at_level(logging.DEBUG):133 module = load_module_from_file("mqttwarn/services/apns.py")134 cert_file, key_file = ["cert_file", "key_file"]135 item = Item(136 target="test",137 addrs=[cert_file, key_file],138 message="â½ Notification message â½",139 data={"apns_token": "foobar"},140 )141 outcome = module.plugin(srv, item)142 assert mock_apns_payload.mock_calls == [143 call(144 alert="â½ Notification message â½",145 custom={},146 sound="default",147 badge=1,148 ),149 ]150 assert mock_apns.mock_calls == [151 mock.call(use_sandbox=False, cert_file="cert_file", key_file="key_file"),152 call().gateway_server.send_notification("foobar", mock.ANY),153 ]154 assert outcome is True155 assert "Successfully published APNS notification to foobar" in caplog.text156@surrogate("apns")157@mock.patch("apns.APNs", create=True)158@mock.patch("apns.Payload", create=True)159def test_apns_success_custom_payload(mock_apns_payload, mock_apns, srv, caplog):160 with caplog.at_level(logging.DEBUG):161 module = load_module_from_file("mqttwarn/services/apns.py")162 cert_file, key_file = ["cert_file", "key_file"]163 item = Item(164 target="test",165 addrs=[cert_file, key_file],166 message="â½ Notification message â½",167 data={168 "apns_token": "foobar",169 "payload": '{"custom": {"baz": "qux"}}',170 },171 )172 outcome = module.plugin(srv, item)173 assert mock_apns_payload.mock_calls == [174 call(175 alert="â½ Notification message â½",176 custom={"baz": "qux"},177 sound="default",178 badge=1,179 ),180 ]181 assert mock_apns.mock_calls == [182 mock.call(use_sandbox=False, cert_file="cert_file", key_file="key_file"),183 call().gateway_server.send_notification("foobar", mock.ANY),184 ]185 assert outcome is True186 assert "Successfully published APNS notification to foobar" in caplog.text187@surrogate("apns")188@mock.patch("apns.APNs", create=True)189@mock.patch("apns.Payload", create=True)190def test_apns_failure_invalid_config(mock_apns_payload, mock_apns, srv, caplog):191 with caplog.at_level(logging.DEBUG):192 module = load_module_from_file("mqttwarn/services/apns.py")193 item = Item(194 target="test",195 addrs=[None],196 message="â½ Notification message â½",197 data={"apns_token": "foobar", "payload": "{}"},198 )199 outcome = module.plugin(srv, item)200 assert outcome is False201 assert "Incorrect service configuration" in caplog.text202@surrogate("apns")203@mock.patch("apns.APNs", create=True)204@mock.patch("apns.Payload", create=True)205def test_apns_failure_apns_token_missing(mock_apns_payload, mock_apns, srv, caplog):206 with caplog.at_level(logging.DEBUG):207 module = load_module_from_file("mqttwarn/services/apns.py")208 cert_file, key_file = ["cert_file", "key_file"]209 item = Item(210 target="test",211 addrs=[cert_file, key_file],212 message="â½ Notification message â½",213 data={},214 )215 outcome = module.plugin(srv, item)216 assert outcome is False217 assert "Cannot notify via APNS: apns_token is missing" in caplog.text218@surrogate("apprise")219@mock.patch("apprise.Apprise", create=True)220@mock.patch("apprise.AppriseAsset", create=True)221def test_apprise_success(apprise_asset, apprise_mock, srv, caplog):222 with caplog.at_level(logging.DEBUG):223 module = load_module_from_file("mqttwarn/services/apprise.py")224 item = Item(225 config={226 "baseuri": "mailtos://smtp_username:smtp_password@mail.example.org"227 },228 target="test",229 addrs=["foo@example.org", "bar@example.org"],230 title="â½ Message title â½",231 message="â½ Notification message â½",232 )233 outcome = module.plugin(srv, item)234 assert apprise_mock.mock_calls == [235 call(asset=mock.ANY),236 call().add(237 "mailtos://smtp_username:smtp_password@mail.example.org?from=None&to=foo@example.org,bar@example.org"238 ),239 call().notify(body="â½ Notification message â½", title="â½ Message title â½"),240 call().notify().__bool__(),241 ]242 assert outcome is True243 assert (244 "Sending notification to Apprise test, addresses: ['foo@example.org', 'bar@example.org']"245 in caplog.text246 )247 assert "Successfully sent message using Apprise" in caplog.text248@surrogate("apprise")249@mock.patch("apprise.Apprise", create=True)250@mock.patch("apprise.AppriseAsset", create=True)251def test_apprise_failure_no_addresses(apprise_asset, apprise_mock, srv, caplog):252 with caplog.at_level(logging.DEBUG):253 module = load_module_from_file("mqttwarn/services/apprise.py")254 item = Item(255 config={256 "baseuri": "mailtos://smtp_username:smtp_password@mail.example.org"257 },258 target="test",259 addrs=[],260 title="â½ Message title â½",261 message="â½ Notification message â½",262 )263 outcome = module.plugin(srv, item)264 assert apprise_mock.mock_calls == []265 assert outcome is False266 assert (267 "Skipped sending notification to Apprise test, no addresses configured"268 in caplog.text269 )270@surrogate("apprise")271def test_apprise_failure_notify(srv, caplog):272 with caplog.at_level(logging.DEBUG):273 mock_connection = mock.MagicMock()274 # Make the call to `notify` signal failure.275 def error(*args, **kwargs):276 return False277 mock_connection.notify = error278 with mock.patch(279 "apprise.Apprise", side_effect=[mock_connection], create=True280 ) as mock_client:281 with mock.patch("apprise.AppriseAsset", create=True) as mock_asset:282 module = load_module_from_file("mqttwarn/services/apprise.py")283 item = Item(284 config={285 "baseuri": "mailtos://smtp_username:smtp_password@mail.example.org"286 },287 target="test",288 addrs=["foo@example.org", "bar@example.org"],289 title="â½ Message title â½",290 message="â½ Notification message â½",291 )292 outcome = module.plugin(srv, item)293 assert mock_client.mock_calls == [294 mock.call(asset=mock.ANY),295 ]296 assert mock_connection.mock_calls == [297 call.add(298 "mailtos://smtp_username:smtp_password@mail.example.org?from=None&to=foo@example.org,bar@example.org"299 ),300 ]301 assert outcome is False302 assert (303 "Sending notification to Apprise test, addresses: ['foo@example.org', 'bar@example.org']"304 in caplog.text305 )306 assert "Sending message using Apprise failed" in caplog.text307@surrogate("apprise")308def test_apprise_error(srv, caplog):309 with caplog.at_level(logging.DEBUG):310 mock_connection = mock.MagicMock()311 # Make the call to `notify` raise an exception.312 def error(*args, **kwargs):313 raise Exception("something failed")314 mock_connection.notify = error315 with mock.patch(316 "apprise.Apprise", side_effect=[mock_connection], create=True317 ) as mock_client:318 with mock.patch("apprise.AppriseAsset", create=True) as mock_asset:319 module = load_module_from_file("mqttwarn/services/apprise.py")320 item = Item(321 config={322 "baseuri": "mailtos://smtp_username:smtp_password@mail.example.org"323 },324 target="test",325 addrs=["foo@example.org", "bar@example.org"],326 title="â½ Message title â½",327 message="â½ Notification message â½",328 )329 outcome = module.plugin(srv, item)330 assert mock_client.mock_calls == [331 mock.call(asset=mock.ANY),332 ]333 assert mock_connection.mock_calls == [334 call.add(335 "mailtos://smtp_username:smtp_password@mail.example.org?from=None&to=foo@example.org,bar@example.org"336 ),337 ]338 assert outcome is False339 assert (340 "Sending notification to Apprise test, addresses: ['foo@example.org', 'bar@example.org']"341 in caplog.text342 )343 assert "Error sending message to test: something failed" in caplog.text344@surrogate("apprise")345@mock.patch("apprise.Apprise", create=True)346@mock.patch("apprise.AppriseAsset", create=True)347def test_apprise_success_with_sender(apprise_asset, apprise_mock, srv, caplog):348 with caplog.at_level(logging.DEBUG):349 module = load_module_from_file("mqttwarn/services/apprise.py")350 item = Item(351 config={352 "baseuri": "mailtos://smtp_username:smtp_password@mail.example.org",353 "sender": "example@example.org",354 "sender_name": "Max Mustermann",355 },356 target="test",357 addrs=["foo@example.org", "bar@example.org"],358 title="â½ Message title â½",359 message="â½ Notification message â½",360 )361 outcome = module.plugin(srv, item)362 assert apprise_mock.mock_calls == [363 call(asset=mock.ANY),364 call().add(365 "mailtos://smtp_username:smtp_password@mail.example.org?from=example@example.org&to=foo@example.org,bar@example.org&name=Max Mustermann"366 ),367 call().notify(body="â½ Notification message â½", title="â½ Message title â½"),368 call().notify().__bool__(),369 ]370 assert outcome is True371 assert "Successfully sent message using Apprise" in caplog.text372@surrogate("asterisk.manager")373@mock.patch("asterisk.manager.Manager", create=True)374def test_asterisk_success(asterisk_mock, srv, caplog):375 with caplog.at_level(logging.DEBUG):376 attrs = {"login.return_value": 42, "originate.return_value": 42}377 asterisk_mock.return_value = mock.MagicMock(**attrs)378 module = load_module_from_file("mqttwarn/services/asterisk.py")379 item = Item(380 config={381 "host": "asterisk.example.org",382 "port": 5038,383 "username": "foobar",384 "password": "bazqux",385 "extension": 2222,386 "context": "default",387 },388 target="test",389 addrs=["SIP/avaya/", "0123456789"],390 message="â½ Notification message â½",391 )392 outcome = module.plugin(srv, item)393 assert asterisk_mock.mock_calls == [394 call(),395 call().connect("asterisk.example.org", 5038),396 call().login("foobar", "bazqux"),397 call().originate(398 "SIP/avaya/0123456789",399 2222,400 context="default",401 priority="1",402 caller_id=2222,403 variables={"text": "â½ Notification message â½"},404 ),405 call().logoff(),406 call().close(),407 ]408 assert outcome is True409 assert "Authentication 42" in caplog.text410 assert "Call 42" in caplog.text411class ManagerSocketException(Exception):412 pass413class ManagerAuthException(Exception):414 pass415class ManagerException(Exception):416 pass417@surrogate("asterisk.manager")418@mock.patch("asterisk.manager.Manager", create=True)419@mock.patch(420 "asterisk.manager.ManagerSocketException", ManagerSocketException, create=True421)422def test_asterisk_failure_no_connection(asterisk_mock, srv, caplog):423 with caplog.at_level(logging.DEBUG):424 attrs = {"connect.side_effect": ManagerSocketException("something failed")}425 asterisk_mock.return_value = mock.MagicMock(**attrs)426 module = load_module_from_file("mqttwarn/services/asterisk.py")427 item = Item(428 config={429 "host": "asterisk.example.org",430 "port": 5038,431 "username": "foobar",432 "password": "bazqux",433 "extension": 2222,434 "context": "default",435 },436 target="test",437 addrs=["SIP/avaya/", "0123456789"],438 message="â½ Notification message â½",439 )440 outcome = module.plugin(srv, item)441 assert asterisk_mock.mock_calls == [442 call(),443 call().connect("asterisk.example.org", 5038),444 call().close(),445 ]446 assert outcome is False447 assert "Error connecting to the manager: something failed" in caplog.text448@surrogate("asterisk.manager")449@mock.patch("asterisk.manager.Manager", create=True)450@mock.patch(451 "asterisk.manager.ManagerSocketException", ManagerSocketException, create=True452)453@mock.patch("asterisk.manager.ManagerAuthException", ManagerAuthException, create=True)454def test_asterisk_failure_login_invalid(asterisk_mock, srv, caplog):455 with caplog.at_level(logging.DEBUG):456 attrs = {"login.side_effect": ManagerAuthException("something failed")}457 asterisk_mock.return_value = mock.MagicMock(**attrs)458 module = load_module_from_file("mqttwarn/services/asterisk.py")459 item = Item(460 config={461 "host": "asterisk.example.org",462 "port": 5038,463 "username": "foobar",464 "password": "bazqux",465 "extension": 2222,466 "context": "default",467 },468 target="test",469 addrs=["SIP/avaya/", "0123456789"],470 message="â½ Notification message â½",471 )472 outcome = module.plugin(srv, item)473 assert asterisk_mock.mock_calls == [474 call(),475 call().connect("asterisk.example.org", 5038),476 call().login("foobar", "bazqux"),477 call().close(),478 ]479 assert outcome is False480 assert "Error logging in to the manager: something failed" in caplog.text481@surrogate("asterisk.manager")482@mock.patch("asterisk.manager.Manager", create=True)483@mock.patch(484 "asterisk.manager.ManagerSocketException", ManagerSocketException, create=True485)486@mock.patch("asterisk.manager.ManagerAuthException", ManagerAuthException, create=True)487@mock.patch("asterisk.manager.ManagerException", ManagerException, create=True)488def test_asterisk_failure_originate_croaks(asterisk_mock, srv, caplog):489 with caplog.at_level(logging.DEBUG):490 attrs = {491 "login.return_value": 42,492 "originate.side_effect": ManagerException("something failed"),493 }494 asterisk_mock.return_value = mock.MagicMock(**attrs)495 module = load_module_from_file("mqttwarn/services/asterisk.py")496 item = Item(497 config={498 "host": "asterisk.example.org",499 "port": 5038,500 "username": "foobar",501 "password": "bazqux",502 "extension": 2222,503 "context": "default",504 },505 target="test",506 addrs=["SIP/avaya/", "0123456789"],507 message="â½ Notification message â½",508 )509 outcome = module.plugin(srv, item)510 assert asterisk_mock.mock_calls == [511 call(),512 call().connect("asterisk.example.org", 5038),513 call().login("foobar", "bazqux"),514 call().originate(515 "SIP/avaya/0123456789",516 2222,517 context="default",518 priority="1",519 caller_id=2222,520 variables={"text": "â½ Notification message â½"},521 ),522 call().close(),523 ]524 assert outcome is False525 assert "Error: something failed" in caplog.text526@surrogate("asterisk.manager")527@mock.patch("asterisk.manager.Manager", create=True)528@mock.patch(529 "asterisk.manager.ManagerSocketException", ManagerSocketException, create=True530)531@mock.patch("asterisk.manager.ManagerAuthException", ManagerAuthException, create=True)532@mock.patch("asterisk.manager.ManagerException", ManagerException, create=True)533def test_asterisk_success_with_broken_close(asterisk_mock, srv, caplog):534 with caplog.at_level(logging.DEBUG):535 attrs = {536 "login.return_value": 42,537 "originate.return_value": 42,538 "close.side_effect": ManagerSocketException("something failed"),539 }540 asterisk_mock.return_value = mock.MagicMock(**attrs)541 module = load_module_from_file("mqttwarn/services/asterisk.py")542 item = Item(543 config={544 "host": "asterisk.example.org",545 "port": 5038,546 "username": "foobar",547 "password": "bazqux",548 "extension": 2222,549 "context": "default",550 },551 target="test",552 addrs=["SIP/avaya/", "0123456789"],553 message="â½ Notification message â½",554 )555 outcome = module.plugin(srv, item)556 assert asterisk_mock.mock_calls == [557 call(),558 call().connect("asterisk.example.org", 5038),559 call().login("foobar", "bazqux"),560 call().originate(561 "SIP/avaya/0123456789",562 2222,563 context="default",564 priority="1",565 caller_id=2222,566 variables={"text": "â½ Notification message â½"},567 ),568 call().logoff(),569 call().close(),570 ]571 assert outcome is True572def test_autoremote_success(srv, caplog):573 item = Item(574 target="test",575 addrs=["ApiKey", "Password", "Target", "Group", "TTL"],576 topic="autoremote/user",577 message="â½ Notification message â½",578 )579 with caplog.at_level(logging.DEBUG):580 module = load_module_from_file("mqttwarn/services/autoremote.py")581 with mock.patch("requests.get") as requests_mock:582 outcome = module.plugin(srv, item)583 requests_mock.assert_called_once_with(584 "https://autoremotejoaomgcd.appspot.com/sendmessage",585 params={586 "key": "ApiKey",587 "message": "â½ Notification message â½",588 "target": "Target",589 "sender": "autoremote/user",590 "password": "Password",591 "ttl": "TTL",592 "collapseKey": "Group",593 },594 )595 assert outcome is True596 assert "Sending to autoremote service" in caplog.text597 assert "Successfully sent to autoremote service" in caplog.text598def test_autoremote_failure(srv, caplog):599 item = Item(600 target="test",601 addrs=["ApiKey", "Password", "Target", "Group", "TTL"],602 topic="autoremote/user",603 message="â½ Notification message â½",604 )605 with caplog.at_level(logging.DEBUG):606 module = load_module_from_file("mqttwarn/services/autoremote.py")607 with mock.patch(608 "requests.get", side_effect=Exception("something failed")609 ) as requests_mock:610 outcome = module.plugin(srv, item)611 requests_mock.assert_called_once_with(612 "https://autoremotejoaomgcd.appspot.com/sendmessage",613 params={614 "key": "ApiKey",615 "message": "â½ Notification message â½",616 "target": "Target",617 "sender": "autoremote/user",618 "password": "Password",619 "ttl": "TTL",620 "collapseKey": "Group",621 },622 )623 assert outcome is False624 assert "Sending to autoremote service" in caplog.text625 assert (626 "Failed to send message to autoremote service: something failed"627 in caplog.text628 )629def test_azure_iot_success_string(srv, caplog):630 item = Item(631 config={"iothubname": "acmehub"},632 target="test",633 addrs=["device-id", "SharedAccessSignature sr=..."],634 message="â½ Notification message â½",635 )636 with caplog.at_level(logging.DEBUG):637 module = load_module_from_file("mqttwarn/services/azure_iot.py")638 mqtt_publish_mock = mock.MagicMock()639 module.mqtt = mqtt_publish_mock640 outcome = module.plugin(srv, item)641 mqtt_publish_mock.single.assert_called_once_with(642 "devices/device-id/messages/events/",643 bytearray(b"\xe2\x9a\xbd Notification message \xe2\x9a\xbd"),644 auth={645 "username": "acmehub.azure-devices.net/device-id/?api-version=2018-06-30",646 "password": "SharedAccessSignature sr=...",647 },648 tls={649 "ca_certs": None,650 "certfile": None,651 "keyfile": None,652 "tls_version": mock.ANY,653 "ciphers": None,654 "cert_reqs": mock.ANY,655 },656 hostname="acmehub.azure-devices.net",657 port=8883,658 protocol=4,659 qos=0,660 retain=False,661 client_id="device-id",662 )663 assert outcome is True664 assert (665 "Publishing to Azure IoT Hub for target=test (device-id): devices/device-id/messages/events/ 'bytearray(b'\\xe2\\x9a\\xbd Notification message \\xe2\\x9a\\xbd')'"666 in caplog.text667 )668def test_azure_iot_success_bytes(srv, caplog):669 item = Item(670 config={"iothubname": "acmehub"},671 target="test",672 addrs=["device-id", "SharedAccessSignature sr=..."],673 message=b"### Notification message ###",674 )675 with caplog.at_level(logging.DEBUG):676 module = load_module_from_file("mqttwarn/services/azure_iot.py")677 mqtt_publish_mock = mock.MagicMock()678 module.mqtt = mqtt_publish_mock679 outcome = module.plugin(srv, item)680 mqtt_publish_mock.single.assert_called_once_with(681 "devices/device-id/messages/events/",682 bytearray(b"### Notification message ###"),683 auth={684 "username": "acmehub.azure-devices.net/device-id/?api-version=2018-06-30",685 "password": "SharedAccessSignature sr=...",686 },687 tls={688 "ca_certs": None,689 "certfile": None,690 "keyfile": None,691 "tls_version": mock.ANY,692 "ciphers": None,693 "cert_reqs": mock.ANY,694 },695 hostname="acmehub.azure-devices.net",696 port=8883,697 protocol=4,698 qos=0,699 retain=False,700 client_id="device-id",701 )702 assert outcome is True703 assert (704 "Publishing to Azure IoT Hub for target=test (device-id): devices/device-id/messages/events/ 'b'### Notification message ###''"705 in caplog.text706 )707def test_azure_iot_failure_wrong_qos(srv, caplog):708 item = Item(709 config={"iothubname": "acmehub", "qos": 999},710 target="test",711 addrs=["device-id", "SharedAccessSignature sr=..."],712 message="â½ Notification message â½",713 )714 with caplog.at_level(logging.DEBUG):715 module = load_module_from_file("mqttwarn/services/azure_iot.py")716 outcome = module.plugin(srv, item)717 assert outcome is False718 assert "Only QoS 0 or 1 allowed for Azure IoT Hub, not '999'" in caplog.text719def test_azure_iot_failure_invalid_message(srv, caplog):720 item = Item(721 config={"iothubname": "acmehub"},722 target="test",723 addrs=["device-id", "SharedAccessSignature sr=..."],724 )725 with mock.patch.object(Item, "message", new_callable=PropertyMock) as msg_mock:726 msg_mock.side_effect = Exception("something failed")727 with caplog.at_level(logging.DEBUG):728 module = load_module_from_file("mqttwarn/services/azure_iot.py")729 outcome = module.plugin(srv, item)730 assert outcome is False731 assert (732 "Unable to prepare message for target=test: something failed"733 in caplog.text734 )735def test_azure_iot_failure_mqtt_publish(srv, caplog):736 item = Item(737 config={"iothubname": "acmehub"},738 target="test",739 addrs=["device-id", "SharedAccessSignature sr=..."],740 message="â½ Notification message â½",741 )742 with caplog.at_level(logging.DEBUG):743 module = load_module_from_file("mqttwarn/services/azure_iot.py")744 mqtt_publish_mock = mock.MagicMock(side_effect=Exception("something failed"))745 module.mqtt.single = mqtt_publish_mock746 outcome = module.plugin(srv, item)747 assert outcome is False748 assert (749 "Unable to publish to Azure IoT Hub for target=test (device-id): something failed"750 in caplog.text751 )752def test_carbon_success_metric_value_timestamp(srv, caplog):753 item = Item(754 target="test", addrs=["localhost", 2003], message="foo 42.42 1623887596", data={}755 )756 with caplog.at_level(logging.DEBUG):757 module = load_module_from_file("mqttwarn/services/carbon.py")758 socket_mock = mock.MagicMock()759 module.socket.socket = socket_mock760 outcome = module.plugin(srv, item)761 assert socket_mock.mock_calls == [762 call(),763 call().connect(("localhost", 2003)),764 call().sendall("foo 42.42 1623887596\n"),765 call().close(),766 ]767 assert outcome is True768 assert "Sending to carbon: foo 42.42 1623887596" in caplog.text769def test_carbon_success_metric_value(srv, caplog):770 item = Item(771 target="test", addrs=["localhost", 2003], message="foo 42.42", data={}772 )773 with caplog.at_level(logging.DEBUG):774 module = load_module_from_file("mqttwarn/services/carbon.py")775 socket_mock = mock.MagicMock()776 module.socket.socket = socket_mock777 outcome = module.plugin(srv, item)778 assert socket_mock.mock_calls == [779 call(),780 call().connect(("localhost", 2003)),781 call().sendall(mock.ANY),782 call().close(),783 ]784 assert outcome is True785 assert "Sending to carbon: foo 42.42" in caplog.text786def test_carbon_success_value(srv, caplog):787 item = Item(788 target="test", addrs=["localhost", 2003], message="42.42", data={}789 )790 with caplog.at_level(logging.DEBUG):791 module = load_module_from_file("mqttwarn/services/carbon.py")792 socket_mock = mock.MagicMock()793 module.socket.socket = socket_mock794 outcome = module.plugin(srv, item)795 assert socket_mock.mock_calls == [796 call(),797 call().connect(("localhost", 2003)),798 call().sendall(mock.ANY),799 call().close(),800 ]801 assert outcome is True802 assert "Sending to carbon: ohno 42.42" in caplog.text803def test_carbon_success_value_metric_from_topic(srv, caplog):804 item = Item(805 target="test", addrs=["localhost", 2003], message="42.42", data={"topic": "foo/bar"}806 )807 with caplog.at_level(logging.DEBUG):808 module = load_module_from_file("mqttwarn/services/carbon.py")809 socket_mock = mock.MagicMock()810 module.socket.socket = socket_mock811 outcome = module.plugin(srv, item)812 assert socket_mock.mock_calls == [813 call(),814 call().connect(("localhost", 2003)),815 call().sendall(mock.ANY),816 call().close(),817 ]818 assert outcome is True819 assert "Sending to carbon: foo.bar 42.42" in caplog.text820def test_carbon_success_value_metric_from_topic_with_leading_slash(srv, caplog):821 item = Item(822 target="test", addrs=["localhost", 2003], message="42.42", data={"topic": "/foo/bar"}823 )824 with caplog.at_level(logging.DEBUG):825 module = load_module_from_file("mqttwarn/services/carbon.py")826 socket_mock = mock.MagicMock()827 module.socket.socket = socket_mock828 outcome = module.plugin(srv, item)829 assert socket_mock.mock_calls == [830 call(),831 call().connect(("localhost", 2003)),832 call().sendall(mock.ANY),833 call().close(),834 ]835 assert outcome is True836 assert "Sending to carbon: foo.bar 42.42" in caplog.text837def test_carbon_failure_invalid_configuration(srv, caplog):838 item = Item(target="test", addrs=["172.16.153.110", "foobar"])839 with caplog.at_level(logging.DEBUG):840 module = load_module_from_file("mqttwarn/services/carbon.py")841 outcome = module.plugin(srv, item)842 assert outcome is False843 assert "Configuration for target `carbon' is incorrect" in caplog.text844def test_carbon_failure_empty_message(srv, caplog):845 item = Item(target="test", addrs=["172.16.153.110", 2003])846 with caplog.at_level(logging.DEBUG):847 module = load_module_from_file("mqttwarn/services/carbon.py")848 outcome = module.plugin(srv, item)849 assert outcome is False850 assert "target `carbon': cannot split string" in caplog.text851def test_carbon_failure_invalid_message_format(srv, caplog):852 item = Item(853 target="test",854 addrs=["172.16.153.110", 2003],855 message="foo bar baz qux",856 data={},857 )858 with caplog.at_level(logging.DEBUG):859 module = load_module_from_file("mqttwarn/services/carbon.py")860 outcome = module.plugin(srv, item)861 assert outcome is False862 assert "target `carbon': error decoding message" in caplog.text863def test_carbon_failure_connect(srv, caplog):864 item = Item(865 target="test", addrs=["localhost", 2003], message="foo 42.42 1623887596", data={}866 )867 with caplog.at_level(logging.DEBUG):868 module = load_module_from_file("mqttwarn/services/carbon.py")869 attrs = {"connect.side_effect": Exception("something failed")}870 socket_mock = mock.MagicMock()871 socket_mock.return_value = mock.MagicMock(**attrs)872 module.socket.socket = socket_mock873 outcome = module.plugin(srv, item)874 assert socket_mock.mock_calls == [call(), call().connect(("localhost", 2003))]875 assert outcome is False876 assert "Sending to carbon: foo 42.42 1623887596" in caplog.text877 assert (878 "Cannot send to carbon service localhost:2003: something failed"879 in caplog.text...
userbot.py
Source:userbot.py
...42 super().__init__(session, **kwargs)43 self._event_builders = Reverse()44 self.loop.run_until_complete(self._async_init(bot_token=bot_token))45 core_module = Path(__file__).parent / "core.py"46 self.load_module_from_file(core_module)47 for a_module_path in Path().glob(f"{self._module_path}/*.py"):48 self.load_module_from_file(a_module_path)49 LOAD = self.env.LOAD50 NO_LOAD = self.env.NO_LOAD51 if LOAD or NO_LOAD:52 to_load = LOAD53 if to_load:54 self._logger.info("Modules to LOAD: ")55 self._logger.info(to_load)56 if NO_LOAD:57 for module_name in NO_LOAD:58 if module_name in self._modules:59 self.remove_module(module_name)60 async def _async_init(self, **kwargs):61 await self.start(**kwargs)62 self.me = await self.get_me()63 self.uid = telethon.utils.get_peer_id(self.me)64 self._logger.info(f"Logged in as {self.uid}")65 def load_module(self, shortname):66 self.load_module_from_file(f"{self._module_path}/{shortname}.py")67 def load_module_from_file(self, path):68 path = Path(path)69 shortname = path.stem70 name = f"_UserbotModules.{self._name}.{shortname}"71 spec = importlib.util.spec_from_file_location(name, path)72 mod = importlib.util.module_from_spec(spec)73 mod.events = _events74 mod.client = self75 mod.humanbytes = humanbytes76 mod.progress = progress77 mod.time_formatter = time_formatter78 mod.build = f"The-TG-Bot-v3{time.strftime('%d%m%Y', time.localtime(os.stat('./').st_mtime))}"79 mod.me = self.me80 mod.logger = logging.getLogger(shortname)81 mod.ENV = self.env...
tests.py
Source:tests.py
1from unittest import TestCase2class TextAttackTestCase(TestCase):3 def test_load_model_from_file(self):4 from textattack.commands.attack.attack_args_helpers import load_module_from_file5 attack_module = load_module_from_file('../model1.py')...
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 autotest 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?
