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.