KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_side_effect method in molecule

Best Python code snippet using molecule_python

test_iot_ext_unit.py

Source:test_iot_ext_unit.py

1# coding=utf-82# --------------------------------------------------------------------------------------------3# Copyright (c) Microsoft Corporation. All rights reserved.4# Licensed under the MIT License. See License.txt in the project root for license information.5# --------------------------------------------------------------------------------------------6"""7NOTICE: These tests are to be phased out and introduced in more modern form.8        Try not to add any new content, only fixes if necessary.9        Look at IoT Hub jobs or configuration tests for a better example. Also use responses fixtures10        like mocked_response for http request mocking.11"""12import pytest13import json14import os15import responses16import re17from azext_iot.operations import hub as subject18from azext_iot.common.utility import read_file_content19from azext_iot.common.sas_token_auth import SasTokenAuthentication20from azext_iot.constants import TRACING_PROPERTY21from azext_iot.tests.generators import create_req_monitor_events, generate_generic_id22from knack.util import CLIError23from azext_iot.tests.conftest import (24    fixture_cmd,25    build_mock_response,26    path_service_client,27    mock_target,28    generate_cs,29)30from azext_iot.common.shared import DeviceAuthApiType31from pathlib import Path32CWD = os.path.dirname(os.path.abspath(__file__))33device_id = "mydevice"34child_device_id = "child_device1"35module_id = "mymod"36config_id = "myconfig"37message_etag = "3k28zb44-0d00-4ddd-ade3-6110eb94c476"38c2d_purge_response = {"deviceId": device_id, "moduleId": None, "totalMessagesPurged": 3}39generic_cs_template = "HostName={};SharedAccessKeyName={};SharedAccessKey={}"40mock_target["cs"] = generate_cs()41# TODO generalize all fixtures across DPS/Hub unit tests42def generate_device_create_req(43    ee=False,44    auth="shared_private_key",45    pk=None,46    sk=None,47    ptp="123",48    stp="321",49    status="enabled",50    status_reason=None,51    valid_days=None,52    output_dir=None,53    device_scope=None,54):55    return {56        "client": None,57        "device_id": device_id,58        "hub_name": mock_target["entity"],59        "ee": ee,60        "auth": auth,61        "pk": pk,62        "sk": sk,63        "ptp": ptp,64        "stp": stp,65        "status": status,66        "status_reason": status_reason,67        "valid_days": valid_days,68        "output_dir": output_dir,69        "device_scope": device_scope70    }71class TestDeviceCreate:72    @pytest.fixture(params=[200])73    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):74        service_client = mocker.patch(path_service_client)75        service_client.return_value = build_mock_response(mocker, request.param, {})76        return service_client77    @pytest.mark.parametrize(78        "req",79        [80            (generate_device_create_req()),81            (generate_device_create_req(pk=generate_generic_id(), sk=generate_generic_id())),82            (generate_device_create_req(ee=True)),83            (generate_device_create_req(ee=True, auth="x509_ca")),84            (generate_device_create_req(ee=True, auth="x509_thumbprint")),85            (generate_device_create_req(ee=True, auth="x509_thumbprint", stp=None)),86            (generate_device_create_req(auth="x509_ca")),87            (generate_device_create_req(auth="x509_thumbprint")),88            (generate_device_create_req(auth="x509_thumbprint", stp=None)),89            (90                generate_device_create_req(91                    auth="x509_thumbprint", ptp=None, stp=None, valid_days=3092                )93            ),94            (generate_device_create_req(status="disabled", status_reason="reasons")),95            (generate_device_create_req(device_scope=generate_generic_id()))96        ],97    )98    def test_device_create(self, serviceclient, req):99        subject.iot_device_create(100            fixture_cmd,101            req["device_id"],102            req["hub_name"],103            req["ee"],104            req["auth"],105            req["pk"],106            req["sk"],107            req["ptp"],108            req["stp"],109            req["status"],110            req["status_reason"],111            req["valid_days"],112            req["output_dir"],113            req["device_scope"]114        )115        args = serviceclient.call_args116        url = args[0][0].url117        assert "{}/devices/{}?".format(mock_target["entity"], device_id) in url118        assert args[0][0].method == "PUT"119        body = json.loads(args[0][0].body)120        assert body["deviceId"] == req["device_id"]121        assert body["status"] == req["status"]122        if req.get("status_reason"):123            assert body["statusReason"] == req["status_reason"]124        assert body["capabilities"]["iotEdge"] == req["ee"]125        if req.get("device_scope"):126            assert body["deviceScope"] == req["device_scope"]127        if req["auth"] == "shared_private_key":128            assert body["authentication"]["type"] == DeviceAuthApiType.sas.value129            if all([req["pk"], req["pk"]]):130                assert body["authentication"]["symmetricKey"]["primaryKey"] == req["pk"]131                assert body["authentication"]["symmetricKey"]["secondaryKey"] == req["sk"]132            else:133                assert body["authentication"]["symmetricKey"] == {}134        elif req["auth"] == "x509_ca":135            assert body["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value136            assert not body["authentication"].get("x509Thumbprint")137            assert not body["authentication"].get("symmetricKey")138        elif req["auth"] == "x509_thumbprint":139            assert body["authentication"]["type"] == DeviceAuthApiType.selfSigned.value140            x509tp = body["authentication"]["x509Thumbprint"]141            assert x509tp["primaryThumbprint"]142            if req["stp"] is None:143                assert x509tp.get("secondaryThumbprint") is None144            else:145                assert x509tp["secondaryThumbprint"] == req["stp"]146    @pytest.mark.parametrize(147        "req, exp",148        [149            (150                generate_device_create_req(ee=True, auth="x509_thumbprint", ptp=None),151                CLIError,152            ),153            (generate_device_create_req(auth="doesnotexist"), CLIError),154            (155                generate_device_create_req(auth="x509_thumbprint", ptp=None, stp=""),156                CLIError,157            ),158            (generate_device_create_req(pk=generate_generic_id()), CLIError),159            (generate_device_create_req(sk=generate_generic_id()), CLIError)160        ],161    )162    def test_device_create_invalid_args(self, serviceclient, req, exp):163        with pytest.raises(exp):164            subject.iot_device_create(165                cmd=None,166                device_id=req["device_id"],167                hub_name=req["hub_name"],168                edge_enabled=req["ee"],169                auth_method=req["auth"],170                primary_key=req["pk"],171                secondary_key=req["sk"],172                primary_thumbprint=req["ptp"],173                secondary_thumbprint=req["stp"],174                status=req["status"],175            )176    @pytest.mark.parametrize("req", [(generate_device_create_req())])177    def test_device_create_error(self, serviceclient_generic_error, req):178        with pytest.raises(CLIError):179            subject.iot_device_create(180                cmd=None,181                device_id=req["device_id"],182                hub_name=req["hub_name"],183                edge_enabled=req["ee"],184                auth_method=req["auth"],185                primary_key=req["pk"],186                secondary_key=req["sk"],187                primary_thumbprint=req["ptp"],188                secondary_thumbprint=req["stp"],189                status=req["status"],190            )191def generate_device_show(**kvp):192    payload = {193        "authentication": {194            "symmetricKey": {"primaryKey": None, "secondaryKey": None},195            "x509Thumbprint": {"primaryThumbprint": None, "secondaryThumbprint": None},196            "type": DeviceAuthApiType.sas.value,197        },198        "capabilities": {"iotEdge": True},199        "deviceId": device_id,200        "status": "disabled",201        "statusReason": "unknown reason",202    }203    for k in kvp:204        if payload.get(k):205            payload[k] = kvp[k]206    return payload207def device_update_con_arg(208    edge_enabled=None,209    status=None,210    status_reason=None,211    auth_method=None,212    primary_thumbprint=None,213    secondary_thumbprint=None,214    primary_key=None,215    secondary_key=None,216):217    return {218        "edge_enabled": edge_enabled,219        "status": status,220        "status_reason": status_reason,221        "auth_method": auth_method,222        "primary_thumbprint": primary_thumbprint,223        "secondary_thumbprint": secondary_thumbprint,224        "primary_key": primary_key,225        "secondary_key": secondary_key,226    }227class TestDeviceUpdate:228    @pytest.fixture(params=[200])229    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):230        service_client = mocker.patch(path_service_client)231        service_client.return_value = build_mock_response(mocker, request.param, {})232        return service_client233    # Update does a GET/SHOW first234    @pytest.mark.parametrize(235        "req",236        [237            (generate_device_show(capabilities={"iotEdge": False})),238            (generate_device_show(status="enabled")),239            (240                generate_device_show(241                    authentication={242                        "symmetricKey": {"primaryKey": "", "secondaryKey": ""},243                        "type": DeviceAuthApiType.sas.value,244                    }245                )246            ),247            (248                generate_device_show(249                    authentication={250                        "x509Thumbprint": {251                            "primaryThumbprint": "123",252                            "secondaryThumbprint": "321",253                        },254                        "type": DeviceAuthApiType.selfSigned.value,255                    }256                )257            ),258            (259                generate_device_show(260                    authentication={"type": DeviceAuthApiType.certificateAuthority.value},261                    etag=generate_generic_id(),262                )263            ),264        ],265    )266    def test_device_update(self, fixture_cmd, serviceclient, req):267        subject.iot_device_update(268            cmd=fixture_cmd,269            device_id=req["deviceId"],270            hub_name=mock_target["entity"],271            parameters=req,272            etag=req.get("etag"),273        )274        args = serviceclient.call_args275        assert (276            "{}/devices/{}?".format(mock_target["entity"], device_id) in args[0][0].url277        )278        assert args[0][0].method == "PUT"279        body = json.loads(args[0][0].body)280        assert body["deviceId"] == req["deviceId"]281        assert body["status"] == req["status"]282        assert body["capabilities"]["iotEdge"] == req["capabilities"]["iotEdge"]283        assert req["authentication"]["type"] == body["authentication"]["type"]284        if req["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value:285            assert not body["authentication"].get("x509Thumbprint")286            assert not body["authentication"].get("symmetricKey")287        elif req["authentication"]["type"] == DeviceAuthApiType.selfSigned.value:288            assert body["authentication"]["x509Thumbprint"]["primaryThumbprint"]289            assert body["authentication"]["x509Thumbprint"]["secondaryThumbprint"]290        headers = args[0][0].headers291        target_etag = req.get("etag")292        assert headers["If-Match"] == '"{}"'.format(target_etag if target_etag else "*")293    @pytest.mark.parametrize(294        "req, arg",295        [296            (297                generate_device_show(capabilities={"iotEdge": False}),298                device_update_con_arg(edge_enabled=True),299            ),300            (301                generate_device_show(status="disabled"),302                device_update_con_arg(status="enabled"),303            ),304            (generate_device_show(), device_update_con_arg(status_reason="test")),305            (306                generate_device_show(),307                device_update_con_arg(308                    auth_method="shared_private_key",309                    primary_key="primarykeyUpdated",310                    secondary_key="secondarykeyUpdated",311                ),312            ),313            (314                generate_device_show(315                    authentication={316                        "type": DeviceAuthApiType.selfSigned.value,317                        "symmetricKey": {"primaryKey": None, "secondaryKey": None},318                        "x509Thumbprint": {319                            "primaryThumbprint": "123",320                            "secondaryThumbprint": "321",321                        },322                    }323                ),324                device_update_con_arg(325                    auth_method="shared_private_key",326                    primary_key="primary_key",327                    secondary_key="secondary_key",328                ),329            ),330            (331                generate_device_show(332                    authentication={333                        "type": DeviceAuthApiType.certificateAuthority.value,334                        "symmetricKey": {"primaryKey": None, "secondaryKey": None},335                        "x509Thumbprint": {336                            "primaryThumbprint": None,337                            "secondaryThumbprint": None,338                        },339                    }340                ),341                device_update_con_arg(342                    auth_method="x509_thumbprint",343                    primary_thumbprint="primary_thumbprint",344                    secondary_thumbprint="secondary_thumbprint",345                ),346            ),347            (generate_device_show(), device_update_con_arg(auth_method="x509_ca",)),348            (generate_device_show(), device_update_con_arg(primary_key="secondary_key", secondary_key="primary_key")),349            (350                generate_device_show(351                    authentication={352                        "type": DeviceAuthApiType.selfSigned.value,353                        "symmetricKey": {"primaryKey": None, "secondaryKey": None},354                        "x509Thumbprint": {355                            "primaryThumbprint": "123",356                            "secondaryThumbprint": "321",357                        },358                    }359                ),360                device_update_con_arg(primary_thumbprint="321", secondary_thumbprint="123")361            )362        ],363    )364    def test_iot_device_custom(self, fixture_cmd, serviceclient, req, arg):365        instance = subject.update_iot_device_custom(366            instance=req,367            edge_enabled=arg["edge_enabled"],368            status=arg["status"],369            status_reason=arg["status_reason"],370            auth_method=arg["auth_method"],371            primary_thumbprint=arg["primary_thumbprint"],372            secondary_thumbprint=arg["secondary_thumbprint"],373            primary_key=arg["primary_key"],374            secondary_key=arg["secondary_key"],375        )376        if arg["edge_enabled"]:377            assert instance["capabilities"]["iotEdge"] == arg["edge_enabled"]378        if arg["status"]:379            assert instance["status"] == arg["status"]380        if arg["status_reason"]:381            assert instance["statusReason"] == arg["status_reason"]382        if arg["auth_method"]:383            if arg["auth_method"] == "shared_private_key":384                assert instance["authentication"]["type"] == DeviceAuthApiType.sas.value385                instance["authentication"]["symmetricKey"]["primaryKey"] == arg[386                    "primary_key"387                ]388                instance["authentication"]["symmetricKey"]["secondaryKey"] == arg[389                    "secondary_key"390                ]391            if arg["auth_method"] == "x509_thumbprint":392                assert instance["authentication"]["type"] == DeviceAuthApiType.selfSigned.value393                if arg["primary_thumbprint"]:394                    instance["authentication"]["x509Thumbprint"][395                        "primaryThumbprint"396                    ] = arg["primary_thumbprint"]397                if arg["secondary_thumbprint"]:398                    instance["authentication"]["x509Thumbprint"][399                        "secondaryThumbprint"400                    ] = arg["secondary_thumbprint"]401            if arg["auth_method"] == "x509_ca":402                assert instance["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value403    @pytest.mark.parametrize(404        "req, arg, exp",405        [406            (407                generate_device_show(),408                device_update_con_arg(409                    auth_method="shared_private_key", primary_key="primarykeyUpdated",410                ),411                CLIError,412            ),413            (414                generate_device_show(),415                device_update_con_arg(auth_method="x509_thumbprint",),416                CLIError,417            ),418            (419                generate_device_show(),420                device_update_con_arg(auth_method="Unknown",),421                ValueError,422            ),423            (424                generate_device_show(),425                device_update_con_arg(primary_thumbprint="newThumbprint",),426                ValueError,427            ),428            (429                generate_device_show(430                    authentication={431                        "type": DeviceAuthApiType.selfSigned.value,432                        "symmetricKey": {"primaryKey": None, "secondaryKey": None},433                        "x509Thumbprint": {434                            "primaryThumbprint": "123",435                            "secondaryThumbprint": "321",436                        },437                    }438                ),439                device_update_con_arg(primary_key='updated_key'),440                ValueError441            )442        ],443    )444    def test_iot_device_custom_invalid_args(self, serviceclient, req, arg, exp):445        with pytest.raises(exp):446            subject.update_iot_device_custom(447                instance=req,448                edge_enabled=arg["edge_enabled"],449                status=arg["status"],450                status_reason=arg["status_reason"],451                auth_method=arg["auth_method"],452                primary_thumbprint=arg["primary_thumbprint"],453                secondary_thumbprint=arg["secondary_thumbprint"],454                primary_key=arg["primary_key"],455                secondary_key=arg["secondary_key"],456            )457    @pytest.mark.parametrize(458        "req, exp",459        [460            (461                generate_device_show(462                    authentication={463                        "x509Thumbprint": {464                            "primaryThumbprint": "",465                            "secondaryThumbprint": "",466                        },467                        "type": DeviceAuthApiType.selfSigned.value,468                    }469                ),470                CLIError,471            ),472            (generate_device_show(authentication={"type": "doesnotexist"}), CLIError),473        ],474    )475    def test_device_update_invalid_args(self, serviceclient, req, exp):476        with pytest.raises(exp):477            subject.iot_device_update(478                cmd=fixture_cmd,479                device_id=req["deviceId"],480                hub_name=mock_target["entity"],481                parameters=req,482            )483    @pytest.mark.parametrize("req", [(generate_device_show())])484    def test_device_update_error(self, serviceclient_generic_error, req):485        with pytest.raises(CLIError):486            subject.iot_device_update(487                cmd=fixture_cmd,488                device_id=req["deviceId"],489                hub_name=mock_target["entity"],490                parameters=req,491            )492class TestDeviceRegenerateKey:493    @pytest.fixture(params=[200])494    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):495        service_client = mocker.patch(path_service_client)496        kvp = {}497        kvp.setdefault(498            "authentication",499            {500                "symmetricKey": {"primaryKey": "123", "secondaryKey": "321"},501                "type": DeviceAuthApiType.sas.value,502            },503        )504        test_side_effect = [505            build_mock_response(mocker, 200, generate_device_show(**kvp)),506            build_mock_response(mocker, 200, {}),507        ]508        service_client.side_effect = test_side_effect509        return service_client510    @pytest.mark.parametrize(511        "req, etag",512        [("primary", generate_generic_id()), ("secondary", None), ("swap", None)],513    )514    def test_device_key_regenerate(self, fixture_cmd, serviceclient, req, etag):515        subject.iot_device_key_regenerate(516            cmd=fixture_cmd,517            hub_name=mock_target["entity"],518            device_id=device_id,519            renew_key_type=req,520            etag=etag521        )522        args = serviceclient.call_args523        assert (524            "{}/devices/{}?".format(mock_target["entity"], device_id) in args[0][0].url525        )526        assert args[0][0].method == "PUT"527        body = json.loads(args[0][0].body)528        if req == "primary":529            assert body["authentication"]["symmetricKey"]["primaryKey"] != "123"530        if req == "secondary":531            assert body["authentication"]["symmetricKey"]["secondaryKey"] != "321"532        if req == "swap":533            assert body["authentication"]["symmetricKey"]["primaryKey"] == "321"534            assert body["authentication"]["symmetricKey"]["secondaryKey"] == "123"535        headers = args[0][0].headers536        assert headers["If-Match"] == '"{}"'.format(etag if etag else "*")537    @pytest.fixture(params=[200])538    def serviceclient_invalid_args(self, mocker, fixture_ghcs, fixture_sas, request):539        service_client = mocker.patch(path_service_client)540        kvp = {}541        kvp.setdefault("authentication", {"type": "test"})542        test_side_effect = [543            build_mock_response(mocker, 200, generate_device_show(**kvp))544        ]545        service_client.side_effect = test_side_effect546        return service_client547    @pytest.mark.parametrize(548        "req, exp", [("primary", CLIError), ("secondary", CLIError), ("swap", CLIError)]549    )550    def test_device_key_regenerate_invalid_args(551        self, fixture_cmd, serviceclient_invalid_args, req, exp552    ):553        with pytest.raises(exp):554            subject.iot_device_key_regenerate(555                cmd=fixture_cmd,556                hub_name=mock_target["entity"],557                device_id=device_id,558                renew_key_type=req,559            )560    @pytest.mark.parametrize("req", ["primary", "secondary", "swap"])561    def test_device_key_regenerate_error(self, serviceclient_generic_error, req):562        with pytest.raises(CLIError):563            subject.iot_device_key_regenerate(564                cmd=fixture_cmd,565                hub_name=mock_target["entity"],566                device_id=device_id,567                renew_key_type=req,568            )569class TestDeviceDelete:570    @pytest.fixture(params=[204])571    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):572        service_client = mocker.patch(path_service_client)573        test_side_effect = [574            build_mock_response(mocker, request.param),575        ]576        service_client.side_effect = test_side_effect577        return service_client578    @pytest.mark.parametrize("target_etag", [generate_generic_id(), None])579    def test_device_delete(self, serviceclient, target_etag):580        subject.iot_device_delete(581            cmd=fixture_cmd,582            device_id=device_id,583            hub_name=mock_target["entity"],584            etag=target_etag,585        )586        args = serviceclient.call_args587        url = args[0][0].url588        assert "{}/devices/{}?".format(mock_target["entity"], device_id) in url589        assert args[0][0].method == "DELETE"590        headers = args[0][0].headers591        assert headers["If-Match"] == '"{}"'.format(target_etag if target_etag else "*")592    def test_device_delete_error(self, serviceclient_generic_error):593        with pytest.raises(CLIError):594            subject.iot_device_delete(fixture_cmd, device_id, mock_target["entity"])595# Starting PoC for improving/simplyfing unit tests596class TestDeviceShow:597    def test_device_show(self, fixture_cmd, mocked_response, fixture_ghcs):598        device_id = generate_generic_id()599        mocked_response.add(600            method=responses.GET,601            url=re.compile(602                "https://{}/devices/{}".format(mock_target["entity"], device_id)603            ),604            body=json.dumps(generate_device_show(deviceId=device_id)),605            status=200,606            content_type="application/json",607            match_querystring=False,608        )609        result = subject.iot_device_show(610            cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"]611        )612        assert result["deviceId"] == device_id613    def test_device_show_error(self, fixture_cmd, service_client_generic_errors):614        with pytest.raises(CLIError):615            subject.iot_device_show(616                cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"]617            )618class TestDeviceList:619    @pytest.fixture(params=[10, 0])620    def service_client(self, mocked_response, fixture_ghcs, request):621        result = []622        size = request.param623        for _ in range(size):624            result.append(generate_device_show())625        mocked_response.add(626            method=responses.POST,627            url="https://{}/devices/query".format(mock_target["entity"]),628            body=json.dumps(result),629            headers={"x-ms-continuation": ""},630            status=200,631            content_type="application/json",632            match_querystring=False,633        )634        mocked_response.expected_size = size635        yield mocked_response636    @pytest.mark.parametrize("top, edge", [(10, True), (1000, False)])637    def test_device_list(self, fixture_cmd, service_client, top, edge):638        result = subject.iot_device_list(639            cmd=fixture_cmd, hub_name=mock_target["entity"], top=top, edge_enabled=edge640        )641        list_request = service_client.calls[0].request642        body = json.loads(list_request.body)643        headers = list_request.headers644        if edge:645            assert (646                body["query"]647                == "select * from devices where capabilities.iotEdge = true"648            )649        else:650            assert body["query"] == "select * from devices"651        assert json.dumps(result)652        assert len(result) == service_client.expected_size653        assert headers["x-ms-max-item-count"] == str(top)654    @pytest.mark.parametrize("top", [-2, 0])655    def test_device_list_invalid_args(self, fixture_cmd, top):656        with pytest.raises(CLIError):657            subject.iot_device_list(658                cmd=fixture_cmd, hub_name=mock_target["entity"], top=top659            )660    def test_device_list_error(self, fixture_cmd, service_client_generic_errors):661        service_client_generic_errors.assert_all_requests_are_fired = False662        with pytest.raises(CLIError):663            subject.iot_device_list(664                cmd=fixture_cmd, hub_name=mock_target["entity"],665            )666def generate_module_create_req(667    mid=module_id, **kwargs668):669    r = generate_device_create_req(**kwargs)670    r["module_id"] = mid671    return r672class TestDeviceModuleCreate:673    @pytest.fixture(params=[200])674    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):675        service_client = mocker.patch(path_service_client)676        service_client.return_value = build_mock_response(mocker, request.param, {})677        return service_client678    @pytest.mark.parametrize(679        "req",680        [681            (generate_module_create_req(auth="shared_private_key")),682            (generate_module_create_req(pk=generate_generic_id(), sk=generate_generic_id())),683            (generate_module_create_req(auth="x509_ca")),684            (generate_module_create_req(auth="x509_thumbprint")),685            (generate_module_create_req(auth="x509_thumbprint", stp=None)),686            (687                generate_module_create_req(688                    auth="x509_thumbprint", ptp=None, stp=None, valid_days=30689                )690            ),691        ],692    )693    def test_device_module_create(self, serviceclient, req):694        subject.iot_device_module_create(695            cmd=fixture_cmd,696            device_id=req["device_id"],697            hub_name=req["hub_name"],698            module_id=req["module_id"],699            auth_method=req["auth"],700            primary_key=req["pk"],701            secondary_key=req["sk"],702            primary_thumbprint=req["ptp"],703            secondary_thumbprint=req["stp"],704            valid_days=req.get("valid_days"),705        )706        args = serviceclient.call_args707        assert (708            "{}/devices/{}/modules/{}?".format(709                mock_target["entity"], device_id, module_id710            )711            in args[0][0].url712        )713        assert args[0][0].method == "PUT"714        body = json.loads(args[0][0].body)715        assert body["deviceId"] == req["device_id"]716        assert body["moduleId"] == req["module_id"]717        if req["auth"] == "shared_private_key":718            assert body["authentication"]["type"] == DeviceAuthApiType.sas.value719            if all([req["pk"], req["pk"]]):720                assert body["authentication"]["symmetricKey"]["primaryKey"] == req["pk"]721                assert body["authentication"]["symmetricKey"]["secondaryKey"] == req["sk"]722            else:723                assert body["authentication"]["symmetricKey"] == {}724        elif req["auth"] == "x509_ca":725            assert body["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value726            assert not body["authentication"].get("x509Thumbprint")727            assert not body["authentication"].get("symmetricKey")728        elif req["auth"] == "x509_thumbprint":729            assert body["authentication"]["type"] == DeviceAuthApiType.selfSigned.value730            x509tp = body["authentication"]["x509Thumbprint"]731            assert x509tp["primaryThumbprint"]732            if req["stp"] is None:733                assert x509tp.get("secondaryThumbprint") is None734            else:735                assert x509tp["secondaryThumbprint"] == req["stp"]736    @pytest.mark.parametrize(737        "req, exp",738        [739            (740                generate_module_create_req(ee=True, auth="x509_thumbprint", ptp=None),741                CLIError,742            ),743            (generate_module_create_req(auth="doesnotexist"), CLIError),744            (745                generate_module_create_req(auth="x509_thumbprint", ptp=None, stp=""),746                CLIError,747            ),748            (generate_module_create_req(pk=generate_generic_id()), CLIError),749            (generate_module_create_req(sk=generate_generic_id()), CLIError)750        ],751    )752    def test_device_module_create_invalid_args(self, serviceclient, req, exp):753        with pytest.raises(exp):754            subject.iot_device_module_create(755                cmd=fixture_cmd,756                device_id=req["device_id"],757                module_id=req["module_id"],758                hub_name=req["hub_name"],759                auth_method=req["auth"],760                primary_key=req["pk"],761                secondary_key=req["sk"],762                primary_thumbprint=req["ptp"],763            )764    @pytest.mark.parametrize("req", [(generate_module_create_req())])765    def test_device_module_create_error(self, serviceclient_generic_error, req):766        with pytest.raises(CLIError):767            subject.iot_device_module_create(768                cmd=fixture_cmd,769                device_id=req["device_id"],770                hub_name=req["hub_name"],771                module_id=req["module_id"],772            )773def generate_device_module_show(**kvp):774    payload = generate_device_show(**kvp)775    payload["moduleId"] = module_id776    return payload777class TestDeviceModuleUpdate:778    @pytest.fixture(params=[200])779    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):780        service_client = mocker.patch(path_service_client)781        service_client.return_value = build_mock_response(mocker, request.param, {})782        return service_client783    # Update does a GET/SHOW first784    @pytest.mark.parametrize(785        "req",786        [787            (788                generate_device_module_show(789                    authentication={790                        "symmetricKey": {"primaryKey": "", "secondaryKey": ""},791                        "type": DeviceAuthApiType.sas.value,792                    },793                    etag=generate_generic_id(),794                )795            ),796            (797                generate_device_module_show(798                    authentication={799                        "x509Thumbprint": {800                            "primaryThumbprint": "123",801                            "secondaryThumbprint": "321",802                        },803                        "type": DeviceAuthApiType.selfSigned.value,804                    }805                )806            ),807            (808                generate_device_module_show(809                    authentication={"type": DeviceAuthApiType.certificateAuthority.value}810                )811            ),812        ],813    )814    def test_device_module_update(self, serviceclient, req):815        subject.iot_device_module_update(816            cmd=fixture_cmd,817            device_id=req["deviceId"],818            module_id=req["moduleId"],819            hub_name=mock_target["entity"],820            parameters=req,821        )822        args = serviceclient.call_args823        url = args[0][0].url824        method = args[0][0].method825        body = json.loads(args[0][0].body)826        assert (827            "{}/devices/{}/modules/{}?".format(828                mock_target["entity"], device_id, module_id829            )830            in url831        )832        assert method == "PUT"833        assert body["deviceId"] == req["deviceId"]834        assert body["moduleId"] == req["moduleId"]835        assert not body.get("capabilities")836        assert req["authentication"]["type"] == body["authentication"]["type"]837        if req["authentication"]["type"] == DeviceAuthApiType.certificateAuthority.value:838            assert not body["authentication"].get("x509Thumbprint")839            assert not body["authentication"].get("symmetricKey")840        elif req["authentication"]["type"] == DeviceAuthApiType.selfSigned.value:841            assert body["authentication"]["x509Thumbprint"]["primaryThumbprint"]842            assert body["authentication"]["x509Thumbprint"]["secondaryThumbprint"]843        target_etag = req.get("etag")844        headers = args[0][0].headers845        assert headers["If-Match"] == '"{}"'.format(target_etag if target_etag else "*")846    @pytest.mark.parametrize(847        "req, exp",848        [849            (850                generate_device_module_show(851                    authentication={852                        "x509Thumbprint": {853                            "primaryThumbprint": "",854                            "secondaryThumbprint": "",855                        },856                        "type": DeviceAuthApiType.selfSigned.value,857                    }858                ),859                CLIError,860            ),861            (862                generate_device_module_show(authentication={"type": "doesnotexist"}),863                CLIError,864            ),865        ],866    )867    def test_device_module_update_invalid_args(self, serviceclient, req, exp):868        with pytest.raises(exp):869            subject.iot_device_module_update(870                cmd=fixture_cmd,871                device_id=req["deviceId"],872                module_id=req["moduleId"],873                hub_name=mock_target["entity"],874                parameters=req,875            )876    @pytest.mark.parametrize("req", [(generate_device_module_show())])877    def test_device_module_update_error(self, serviceclient_generic_error, req):878        with pytest.raises(CLIError):879            subject.iot_device_module_update(880                cmd=fixture_cmd,881                device_id=req["deviceId"],882                hub_name=mock_target["entity"],883                module_id=req["moduleId"],884                parameters=req,885            )886class TestDeviceModuleDelete:887    @pytest.fixture(params=[204])888    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):889        service_client = mocker.patch(path_service_client)890        test_side_effect = [891            build_mock_response(mocker, request.param),892        ]893        service_client.side_effect = test_side_effect894        return service_client895    @pytest.mark.parametrize("target_etag", [generate_generic_id(), None])896    def test_device_module_delete(self, serviceclient, target_etag):897        subject.iot_device_module_delete(898            cmd=fixture_cmd,899            device_id=device_id,900            module_id=module_id,901            hub_name=mock_target["entity"],902            etag=target_etag,903        )904        args = serviceclient.call_args905        url = args[0][0].url906        method = args[0][0].method907        headers = args[0][0].headers908        assert "devices/{}/modules/{}?".format(device_id, module_id) in url909        assert method == "DELETE"910        assert headers["If-Match"] == '"{}"'.format(target_etag if target_etag else "*")911    def test_device_module_delete_error(self, serviceclient_generic_error):912        with pytest.raises(CLIError):913            subject.iot_device_module_delete(914                cmd=fixture_cmd,915                device_id=device_id,916                module_id=module_id,917                hub_name=mock_target["entity"],918            )919class TestDeviceModuleShow:920    @pytest.fixture(params=[200])921    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):922        service_client = mocker.patch(path_service_client)923        service_client.return_value = build_mock_response(mocker, request.param, {})924        return service_client925    def test_device_module_show(self, serviceclient):926        result = subject.iot_device_module_show(927            cmd=fixture_cmd, device_id=device_id, module_id=module_id, hub_name=mock_target["entity"]928        )929        assert json.dumps(result)930        args = serviceclient.call_args931        url = args[0][0].url932        method = args[0][0].method933        assert "devices/{}/modules/{}?".format(device_id, module_id) in url934        assert method == "GET"935    def test_device_module_show_error(self, serviceclient_generic_error):936        with pytest.raises(CLIError):937            subject.iot_device_module_show(938                cmd=fixture_cmd,939                device_id=device_id,940                module_id=module_id,941                hub_name=mock_target["entity"],942            )943class TestDeviceModuleList:944    @pytest.fixture(params=[(200, 10), (200, 0)])945    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):946        service_client = mocker.patch(path_service_client)947        result = []948        size = request.param[1]949        for _ in range(size):950            result.append(generate_device_module_show())951        service_client.expected_size = size952        service_client.return_value = build_mock_response(953            mocker, request.param[0], result954        )955        return service_client956    @pytest.mark.parametrize("top", [10, 1000])957    def test_device_module_list(self, serviceclient, top):958        result = subject.iot_device_module_list(959            cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"], top=top960        )961        args = serviceclient.call_args962        url = args[0][0].url963        method = args[0][0].method964        assert method == "GET"965        assert "{}/devices/{}/modules?".format(mock_target["entity"], device_id) in url966        assert len(result) == serviceclient.expected_size967    def test_device_module_list_error(self, serviceclient_generic_error):968        with pytest.raises(CLIError):969            subject.iot_device_module_list(970                cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"]971            )972def change_dir():973    from inspect import getsourcefile974    os.chdir(os.path.dirname(os.path.abspath(getsourcefile(lambda: 0))))975def generate_device_twin_show(file_handle=False, **kvp):976    if file_handle:977        change_dir()978        path = os.path.join(Path(CWD).parent, "test_generic_twin.json")979        return path980    payload = {"deviceId": device_id}981    for k in kvp:982        payload[k] = kvp[k]983    return payload984class TestDeviceTwinShow:985    @pytest.fixture(params=[200])986    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):987        service_client = mocker.patch(path_service_client)988        service_client.return_value = build_mock_response(989            mocker, request.param, payload=generate_device_twin_show()990        )991        return service_client992    def test_device_twin_show(self, serviceclient):993        result = subject.iot_device_twin_show(994            cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"]995        )996        args = serviceclient.call_args997        url = args[0][0].url998        method = args[0][0].method999        assert json.dumps(result)1000        assert method == "GET"1001        assert "{}/twins/{}?".format(mock_target["entity"], device_id) in url1002    def test_device_twin_show_error(self, serviceclient_generic_error):1003        with pytest.raises(CLIError):1004            subject.iot_device_twin_show(1005                cmd=None, device_id=device_id, hub_name=mock_target["entity"]1006            )1007class TestDeviceTwinUpdate:1008    @pytest.fixture(params=[200])1009    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1010        service_client = mocker.patch(path_service_client)1011        service_client.return_value = build_mock_response(1012            mocker, request.param, payload=generate_device_twin_show()1013        )1014        return service_client1015    # Update does a GET/SHOW first1016    @pytest.mark.parametrize(1017        "req",1018        [1019            generate_device_twin_show(1020                properties={"desired": {"key": "value"}}, etag="abcd"1021            ),1022            generate_device_twin_show(properties={"desired": {"key": "value"}}),1023        ],1024    )1025    def test_device_twin_update(self, serviceclient, req):1026        subject.iot_device_twin_update(1027            cmd=fixture_cmd,1028            device_id=req["deviceId"],1029            hub_name=mock_target["entity"],1030            parameters=req,1031            etag=req.get("etag"),1032        )1033        args = serviceclient.call_args1034        body = json.loads(args[0][0].body)1035        assert body == req1036        assert "twins/{}".format(device_id) in args[0][0].url1037        target_etag = req.get("etag")1038        headers = args[0][0].headers1039        assert headers["If-Match"] == '"{}"'.format(target_etag if target_etag else "*")1040    @pytest.mark.parametrize("req", [generate_device_twin_show()])1041    def test_device_twin_update_error(self, serviceclient_generic_error, req):1042        with pytest.raises(CLIError):1043            subject.iot_device_twin_update(1044                cmd=fixture_cmd,1045                device_id=req["deviceId"],1046                hub_name=mock_target["entity"],1047                parameters=req,1048            )1049class TestDeviceTwinReplace:1050    @pytest.fixture(params=[200])1051    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1052        service_client = mocker.patch(path_service_client)1053        service_client.return_value = build_mock_response(1054            mocker, request.param, generate_device_twin_show(moduleId=module_id)1055        )1056        return service_client1057    # Replace does a GET/SHOW first1058    @pytest.mark.parametrize(1059        "req, isfile, etag",1060        [1061            (1062                generate_device_twin_show(moduleId=module_id),1063                False,1064                generate_generic_id(),1065            ),1066            (1067                generate_device_twin_show(1068                    moduleId=module_id, properties={"desired": {"key": "value"}}1069                ),1070                False,1071                None,1072            ),1073            (generate_device_twin_show(file_handle=True), True, None),1074        ],1075    )1076    def test_device_twin_replace(self, serviceclient, req, isfile, etag):1077        if not isfile:1078            req = json.dumps(req)1079        subject.iot_device_twin_replace(1080            cmd=fixture_cmd,1081            device_id=device_id,1082            hub_name=mock_target["entity"],1083            target_json=req,1084            etag=etag,1085        )1086        args = serviceclient.call_args1087        body = json.loads(args[0][0].body)1088        if isfile:1089            content = str(read_file_content(req))1090            assert body == json.loads(content)1091        else:1092            assert body == json.loads(req)1093        assert "{}/twins/{}?".format(mock_target["entity"], device_id) in args[0][0].url1094        assert args[0][0].method == "PUT"1095        headers = args[0][0].headers1096        assert headers["If-Match"] == '"{}"'.format(etag if etag else "*")1097    @pytest.mark.parametrize("req", [(generate_device_twin_show(moduleId=module_id))])1098    def test_device_twin_replace_error(self, serviceclient_generic_error, req):1099        with pytest.raises(CLIError):1100            subject.iot_device_twin_replace(1101                cmd=fixture_cmd,1102                device_id=device_id,1103                hub_name=mock_target["entity"],1104                target_json=json.dumps(req),1105            )1106class TestDeviceModuleTwinShow:1107    @pytest.fixture(params=[200])1108    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1109        service_client = mocker.patch(path_service_client)1110        service_client.return_value = build_mock_response(1111            mocker, request.param, payload=generate_device_twin_show()1112        )1113        return service_client1114    def test_device_module_twin_show(self, serviceclient):1115        result = subject.iot_device_module_twin_show(1116            cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"], module_id=module_id1117        )1118        args = serviceclient.call_args1119        assert "twins/{}".format(device_id) in args[0][0].url1120        assert "modules/{}".format(module_id) in args[0][0].url1121        assert json.dumps(result)1122    def test_device_module_twin_show_error(self, serviceclient_generic_error):1123        with pytest.raises(CLIError):1124            subject.iot_device_module_twin_show(1125                cmd=fixture_cmd,1126                device_id=device_id,1127                hub_name=mock_target["entity"],1128                module_id=module_id,1129            )1130class TestDeviceModuleTwinUpdate:1131    @pytest.fixture(params=[200])1132    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1133        service_client = mocker.patch(path_service_client)1134        service_client.return_value = build_mock_response(1135            mocker, request.param, payload=generate_device_twin_show(moduleId=module_id)1136        )1137        return service_client1138    # Update does a GET/SHOW first1139    @pytest.mark.parametrize(1140        "req",1141        [1142            (1143                generate_device_twin_show(1144                    moduleId=module_id,1145                    properties={"desired": {"key": "value"}},1146                    etag=generate_generic_id(),1147                )1148            ),1149            (1150                generate_device_twin_show(1151                    moduleId=module_id, properties={"desired": {"key": "value"}}1152                )1153            ),1154        ],1155    )1156    def test_device_module_twin_update(self, serviceclient, req):1157        subject.iot_device_module_twin_update(1158            cmd=fixture_cmd,1159            device_id=req["deviceId"],1160            hub_name=mock_target["entity"],1161            module_id=module_id,1162            parameters=req,1163            etag=req.get("etag"),1164        )1165        args = serviceclient.call_args1166        body = json.loads(args[0][0].body)1167        assert body == req1168        assert (1169            "twins/{}/modules/{}?".format(req["deviceId"], module_id) in args[0][0].url1170        )1171        target_etag = req.get("etag")1172        headers = args[0][0].headers1173        assert headers["If-Match"] == '"{}"'.format(target_etag if target_etag else "*")1174    @pytest.mark.parametrize("req", [(generate_device_twin_show(moduleId=module_id))])1175    def test_device_module_twin_update_error(self, serviceclient_generic_error, req):1176        with pytest.raises(CLIError):1177            subject.iot_device_module_twin_update(1178                cmd=fixture_cmd,1179                device_id=req["deviceId"],1180                hub_name=mock_target["entity"],1181                module_id=module_id,1182                parameters=req,1183            )1184class TestDeviceModuleTwinReplace:1185    @pytest.fixture(params=[200])1186    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1187        service_client = mocker.patch(path_service_client)1188        service_client.return_value = build_mock_response(1189            mocker, request.param, payload=generate_device_twin_show(moduleId=module_id)1190        )1191        return service_client1192    # Replace does a GET/SHOW first1193    @pytest.mark.parametrize(1194        "req, isfile, etag",1195        [1196            (1197                generate_device_twin_show(moduleId=module_id),1198                False,1199                generate_generic_id(),1200            ),1201            (1202                generate_device_twin_show(1203                    moduleId=module_id, properties={"desired": {"key": "value"}}1204                ),1205                False,1206                None,1207            ),1208            (generate_device_twin_show(file_handle=True), True, None),1209        ],1210    )1211    def test_device_module_twin_replace(self, serviceclient, req, isfile, etag):1212        if not isfile:1213            req = json.dumps(req)1214        subject.iot_device_module_twin_replace(1215            cmd=fixture_cmd,1216            device_id=device_id,1217            hub_name=mock_target["entity"],1218            module_id=module_id,1219            target_json=req,1220            etag=etag,1221        )1222        args = serviceclient.call_args1223        body = json.loads(args[0][0].body)1224        if isfile:1225            content = str(read_file_content(req))1226            assert body == json.loads(content)1227        else:1228            assert body == json.loads(req)1229        assert "twins/{}/modules/{}?".format(device_id, module_id) in args[0][0].url1230        assert args[0][0].method == "PUT"1231        headers = args[0][0].headers1232        assert headers["If-Match"] == '"{}"'.format(etag if etag else "*")1233    @pytest.mark.parametrize("req", [(generate_device_twin_show(moduleId=module_id))])1234    def test_device_module_twin_replace_error(self, serviceclient_generic_error, req):1235        with pytest.raises(CLIError):1236            subject.iot_device_module_twin_replace(1237                cmd=fixture_cmd,1238                device_id=device_id,1239                hub_name=mock_target["entity"],1240                module_id=module_id,1241                target_json=json.dumps(req),1242            )1243generic_query = "select * from devices"1244class TestQuery:1245    @pytest.fixture(params=[200])1246    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1247        service_client = mocker.patch(path_service_client)1248        return service_client1249    @pytest.mark.parametrize(1250        "query, servresult, servtotal, top",1251        [1252            (generic_query, [generate_device_twin_show()], 6, 3),1253            (1254                generic_query,1255                [generate_device_twin_show(), generate_device_twin_show()],1256                5,1257                2,1258            ),1259            (1260                generic_query,1261                [generate_device_twin_show(), generate_device_twin_show()],1262                6,1263                None,1264            ),1265            (generic_query, [generate_device_show() for i in range(0, 12)], 100, 51),1266            (generic_query, [generate_device_twin_show()], 1, 100),1267        ],1268    )1269    def test_query_basic(self, serviceclient, query, servresult, servtotal, top):1270        pagesize = len(servresult)1271        continuation = []1272        for i in range(int(servtotal / pagesize)):1273            continuation.append(generate_generic_id())1274        if servtotal % pagesize != 0:1275            continuation.append(generate_generic_id())1276        continuation[-1] = None1277        serviceclient.return_value = build_mock_response(1278            status_code=200, payload=servresult, headers_get_side_effect=continuation1279        )1280        result = subject.iot_query(1281            cmd=None, hub_name=mock_target["entity"], query_command=query, top=top1282        )1283        if top and top < servtotal:1284            targetcount = top1285        else:1286            targetcount = servtotal1287        assert len(result) == targetcount1288        if pagesize >= targetcount:1289            assert serviceclient.call_count == 11290        else:1291            if targetcount % pagesize == 0:1292                assert serviceclient.call_count == int(targetcount / pagesize)1293            else:1294                assert serviceclient.call_count == int(targetcount / pagesize) + 11295        args = serviceclient.call_args_list[0]1296        headers = args[0][0].headers1297        body = json.loads(args[0][0].body)1298        assert body["query"] == query1299        if top:1300            targetcount = top1301            if pagesize < top:1302                for i in range(1, len(serviceclient.call_args_list)):1303                    headers = serviceclient.call_args_list[i][0][0].headers1304                    targetcount = targetcount - pagesize1305                    assert headers["x-ms-max-item-count"] == str(targetcount)1306            else:1307                assert headers["x-ms-max-item-count"] == str(targetcount)1308        else:1309            assert not headers.get("x-ms-max-item-count")1310    @pytest.mark.parametrize("top", [-2, 0])1311    def test_query_invalid_args(self, top):1312        with pytest.raises(CLIError):1313            subject.iot_query(1314                cmd=None, hub_name=mock_target["entity"], query_command=generic_query, top=top1315            )1316    def test_query_error(self, serviceclient_generic_error):1317        with pytest.raises(CLIError):1318            subject.iot_query(1319                cmd=None, hub_name=mock_target["entity"], query_command=generic_query1320            )1321class TestDeviceMethodInvoke:1322    @pytest.fixture(params=[200])1323    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1324        service_client = mocker.patch(path_service_client)1325        service_client.return_value = build_mock_response(1326            mocker, request.param, {"payload": "value", "status": 0}1327        )1328        return service_client1329    @pytest.mark.parametrize("methodbody", ['{"key":"value"}', None])1330    def test_device_method(self, serviceclient, methodbody):1331        payload = methodbody1332        device_method = "mymethod"1333        timeout = 1001334        subject.iot_device_method(1335            cmd=fixture_cmd,1336            device_id=device_id,1337            hub_name=mock_target["entity"],1338            method_name=device_method,1339            method_payload=payload,1340            timeout=timeout,1341        )1342        args = serviceclient.call_args1343        body = json.loads(args[0][0].body)1344        url = args[0][0].url1345        method = args[0][0].method1346        assert method == "POST"1347        assert body["methodName"] == device_method1348        if methodbody:1349            assert body["payload"] == json.loads(payload)1350        else:1351            # We must ensure null is passed for payload.1352            assert body["payload"] is None1353        assert body["responseTimeoutInSeconds"] == timeout1354        assert body["connectTimeoutInSeconds"] == timeout1355        assert "{}/twins/{}/methods?".format(mock_target["entity"], device_id) in url1356    @pytest.mark.parametrize(1357        "req, etype, exp",1358        [1359            ("badformat", "payload", CLIError),1360            ('{"key":"valu', "payload", CLIError),1361            (1000, "timeout", CLIError),1362            (5, "timeout", CLIError),1363        ],1364    )1365    def test_device_method_invalid_args(self, serviceclient, req, etype, exp):1366        with pytest.raises(exp):1367            if etype == "payload":1368                subject.iot_device_method(1369                    fixture_cmd,1370                    device_id=device_id,1371                    hub_name=mock_target["entity"],1372                    method_name="mymethod",1373                    method_payload=req,1374                )1375            if etype == "timeout":1376                subject.iot_device_method(1377                    fixture_cmd,1378                    device_id=device_id,1379                    hub_name=mock_target["entity"],1380                    method_name="mymethod",1381                    method_payload='{"key":"value"}',1382                    timeout=req,1383                )1384    def test_device_method_error(self, serviceclient_generic_error):1385        with pytest.raises(CLIError):1386            subject.iot_device_method(1387                cmd=fixture_cmd,1388                device_id=device_id,1389                hub_name=mock_target["entity"],1390                method_name="mymethod",1391                method_payload='{"key":"value"}',1392            )1393class TestDeviceModuleMethodInvoke:1394    @pytest.fixture(params=[200])1395    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1396        service_client = mocker.patch(path_service_client)1397        service_client.return_value = build_mock_response(1398            mocker, request.param, {"payload": "value", "status": 0}1399        )1400        return service_client1401    @pytest.mark.parametrize("methodbody", ['{"key":"value"}', None])1402    def test_device_module_method(self, serviceclient, methodbody):1403        payload = methodbody1404        module_method = "mymethod"1405        timeout = 1001406        subject.iot_device_module_method(1407            cmd=fixture_cmd,1408            device_id=device_id,1409            module_id=module_id,1410            method_name=module_method,1411            hub_name=mock_target["entity"],1412            method_payload=payload,1413            timeout=timeout,1414        )1415        args = serviceclient.call_args1416        body = json.loads(args[0][0].body)1417        url = args[0][0].url1418        method = args[0][0].method1419        assert method == "POST"1420        assert body["methodName"] == module_method1421        if methodbody:1422            assert body["payload"] == json.loads(payload)1423        else:1424            # We must ensure null is passed for payload.1425            assert body["payload"] is None1426        assert body["responseTimeoutInSeconds"] == timeout1427        assert body["connectTimeoutInSeconds"] == timeout1428        assert (1429            "{}/twins/{}/modules/{}/methods?".format(1430                mock_target["entity"], device_id, module_id1431            )1432            in url1433        )1434    @pytest.mark.parametrize(1435        "req, etype, exp",1436        [1437            ("doesnotexist", "payload", CLIError),1438            ('{"key":"valu', "payload", CLIError),1439            (1000, "timeout", CLIError),1440            (5, "timeout", CLIError),1441        ],1442    )1443    def test_device_module_method_invalid_args(self, serviceclient, req, etype, exp):1444        with pytest.raises(exp):1445            if etype == "payload":1446                subject.iot_device_module_method(1447                    cmd=fixture_cmd,1448                    device_id=device_id,1449                    module_id=module_id,1450                    method_name="mymethod",1451                    hub_name=mock_target["entity"],1452                    method_payload=req,1453                )1454            if etype == "timeout":1455                subject.iot_device_module_method(1456                    cmd=fixture_cmd,1457                    device_id=device_id,1458                    module_id=module_id,1459                    method_name="mymethod",1460                    hub_name=mock_target["entity"],1461                    method_payload='{"key":"value"}',1462                    timeout=req,1463                )1464    def test_device_method_error(self, serviceclient_generic_error):1465        with pytest.raises(CLIError):1466            subject.iot_device_module_method(1467                cmd=fixture_cmd,1468                device_id=device_id,1469                module_id=module_id,1470                method_name="mymethod",1471                hub_name=mock_target["entity"],1472                method_payload='{"key":"value"}',1473            )1474class TestSasTokenAuth:1475    def test_generate_sas_token(self):1476        # Prepare parameters1477        uri = "iot-hub-for-test.azure-devices.net/devices/iot-device-for-test"1478        policy_name = "iothubowner"1479        access_key = "+XLy+MVZ+aTeOnVzN2kLeB16O+kSxmz6g3rS6fAf6rw="1480        expiry = 14719403631481        # Action1482        sas_auth = SasTokenAuthentication(uri, None, access_key, expiry)1483        token = sas_auth.generate_sas_token(absolute=True)1484        # Assertion1485        assert "SharedAccessSignature " in token1486        assert "sig=SIumZ1ACqqPJZ2okHDlW9MSYKykEpqsQY3z6FMOICd4%3D" in token1487        assert "se=1471940363" in token1488        assert (1489            "sr=iot-hub-for-test.azure-devices.net%2Fdevices%2Fiot-device-for-test"1490            in token1491        )1492        assert "skn=" not in token1493        # Prepare parameters1494        uri = "iot-hub-for-test.azure-devices.net"1495        # Action1496        sas_auth = SasTokenAuthentication(uri, policy_name, access_key, expiry)1497        token = sas_auth.generate_sas_token(absolute=True)1498        # Assertion1499        assert "SharedAccessSignature " in token1500        assert "sig=770sPjjYxRYpNz8%2FhEN7XR5XU5KDGYGTinSP8YyeTXw%3D" in token1501        assert "se=1471940363" in token1502        assert "sr=iot-hub-for-test.azure-devices.net" in token1503        assert "skn=iothubowner" in token1504        # Prepare parameters1505        uri = "iot-hub-for-test.azure-devices.net/devices/iot-device-for-test/modules/module-for-test"1506        # Action1507        sas_auth = SasTokenAuthentication(uri, policy_name, access_key, expiry)1508        token = sas_auth.generate_sas_token(absolute=True)1509        # Assertion1510        assert "SharedAccessSignature " in token1511        assert "sig=JwAxBBBPYA0E%2FTHdnrXzUfBfuZ7deH6pppCniJ23Uu0%3D" in token1512        assert "se=1471940363" in token1513        assert (1514            "sr=iot-hub-for-test.azure-devices.net%2Fdevices%2Fiot-device-for-test%2Fmodules%2Fmodule-for-test"1515            in token1516        )1517        assert "skn=iothubowner" in token1518class TestMonitorEvents:1519    @pytest.fixture(params=[200])1520    def serviceclient(self, mocker, fixture_ghcs, fixture_sas, request):1521        service_client = mocker.patch(path_service_client)1522        service_client.return_value = build_mock_response(mocker, request.param)1523        existing_target = fixture_ghcs.return_value1524        existing_target["events"] = {}1525        existing_target["events"]["partition_ids"] = []1526        return service_client1527    @pytest.mark.parametrize(1528        "req",1529        [1530            (1531                create_req_monitor_events(1532                    device_id="mydevice",1533                    device_query="select * from devices",1534                    consumer_group="group1",1535                    content_type="application/json",1536                    enqueued_time="54321",1537                    timeout="30",1538                    hub_name="myhub",1539                    resource_group_name="myrg",1540                    yes=True,1541                    properties="sys anno app",1542                    repair=True,1543                    login=mock_target["cs"],1544                )1545            )1546        ],1547    )1548    def test_monitor_events_entrypoint(1549        self, fixture_cmd, fixture_monitor_events_entrypoint, req1550    ):1551        subject.iot_hub_monitor_events(1552            cmd=fixture_cmd,1553            device_id=req["device_id"],1554            device_query=req["device_query"],1555            consumer_group=req["consumer_group"],1556            content_type=req["content_type"],1557            enqueued_time=req["enqueued_time"],1558            timeout=req["timeout"],1559            hub_name=req["hub_name"],1560            resource_group_name=req["resource_group_name"],1561            yes=req["yes"],1562            properties=req["properties"],1563            repair=req["repair"],1564            login=req["login"],1565        )1566        monitor_events_args = fixture_monitor_events_entrypoint.call_args[1]1567        attribute_set = [1568            "device_id",1569            "device_query",1570            "consumer_group",1571            "enqueued_time",1572            "content_type",1573            "timeout",1574            "login",1575            "hub_name",1576            "resource_group_name",1577            "yes",1578            "properties",1579            "repair",1580        ]1581        assert "interface" not in monitor_events_args1582        for attribute in attribute_set:1583            if req[attribute]:1584                assert monitor_events_args[attribute] == req[attribute]1585            else:1586                assert not monitor_events_args[attribute]1587    @pytest.mark.parametrize("timeout, exception", [(-1, CLIError)])1588    def test_monitor_events_invalid_args(1589        self, fixture_cmd, serviceclient, timeout, exception1590    ):1591        with pytest.raises(exception):1592            subject.iot_hub_monitor_events(1593                fixture_cmd, mock_target["entity"], device_id, timeout=timeout1594            )1595def generate_parent_device(**kvp):1596    payload = {1597        "etag": "abcd",1598        "capabilities": {"iotEdge": True},1599        "deviceId": device_id,1600        "status": "disabled",1601        "deviceScope": "ms-azure-iot-edge://{}-1234".format(device_id),1602        "parentScopes": ["ms-azure-iot-edge://{}-5678".format(device_id)],1603    }1604    for k in kvp:1605        if payload.get(k):1606            payload[k] = kvp[k]1607    return payload1608def generate_child_device(**kvp):1609    payload = {1610        "etag": "abcd",1611        "capabilities": {"iotEdge": False},1612        "deviceId": child_device_id,1613        "status": "disabled",1614        "deviceScope": "",1615    }1616    for k in kvp:1617        payload[k] = kvp[k]1618    return payload1619class TestEdgeOffline:1620    # get-parent1621    @pytest.fixture(params=[(200, 200)])1622    def sc_getparent(self, mocker, fixture_ghcs, fixture_sas, request):1623        service_client = mocker.patch(path_service_client)1624        child_kvp = {}1625        child_kvp.setdefault(1626            "parentScopes", [generate_parent_device().get("deviceScope")]1627        )1628        test_side_effect = [1629            build_mock_response(1630                mocker, request.param[0], generate_child_device(**child_kvp)1631            ),1632            build_mock_response(mocker, request.param[0], generate_parent_device()),1633        ]1634        service_client.side_effect = test_side_effect1635        return service_client1636    def test_device_get_parent(self, sc_getparent):1637        result = subject.iot_device_get_parent(1638            fixture_cmd, child_device_id, mock_target["entity"]1639        )1640        args = sc_getparent.call_args1641        url = args[0][0].url1642        assert "{}/devices/{}?".format(mock_target["entity"], device_id) in url1643        assert args[0][0].method == "GET"1644        assert json.dumps(result)1645    @pytest.fixture(params=[(200, 0), (200, 1)])1646    def sc_invalid_args_getparent(self, mocker, fixture_ghcs, fixture_sas, request):1647        service_client = mocker.patch(path_service_client)1648        child_kvp = {}1649        if request.param[1] == 0:1650            child_kvp.setdefault("parentScopes", [])1651        if request.param[1] == 1:1652            child_kvp.setdefault("deviceId", "")1653        test_side_effect = [1654            build_mock_response(1655                mocker, request.param[0], generate_child_device(**child_kvp)1656            )1657        ]1658        service_client.side_effect = test_side_effect1659        return service_client1660    @pytest.mark.parametrize("exp", [CLIError])1661    def test_device_getparent_invalid_args(self, sc_invalid_args_getparent, exp):1662        with pytest.raises(exp):1663            subject.iot_device_get_parent(fixture_cmd, device_id, mock_target["entity"])1664    def test_device_getparent_error(self, serviceclient_generic_error):1665        with pytest.raises(CLIError):1666            subject.iot_device_get_parent(fixture_cmd, device_id, mock_target["entity"])1667    # set-parent1668    @pytest.fixture(params=[(200, 0), (200, 1)])1669    def sc_setparent(self, mocker, fixture_ghcs, fixture_sas, request):1670        service_client = mocker.patch(path_service_client)1671        child_kvp = {}1672        if request.param[1] == 1:1673            child_kvp.setdefault("parentScopes", ["abcd"])1674        test_side_effect = [1675            build_mock_response(mocker, request.param[0], generate_parent_device()),1676            build_mock_response(1677                mocker, request.param[0], generate_child_device(**child_kvp)1678            ),1679            build_mock_response(mocker, request.param[0], {}),1680        ]1681        service_client.side_effect = test_side_effect1682        return service_client1683    def test_device_set_parent(self, sc_setparent):1684        subject.iot_device_set_parent(1685            fixture_cmd, child_device_id, device_id, True, mock_target["entity"]1686        )1687        args = sc_setparent.call_args1688        url = args[0][0].url1689        body = json.loads(args[0][0].body)1690        assert "{}/devices/{}?".format(mock_target["entity"], child_device_id) in url1691        assert args[0][0].method == "PUT"1692        assert body["deviceId"] == child_device_id1693        assert body["deviceScope"] == generate_parent_device().get("deviceScope")1694    @pytest.fixture(params=[(200, 0), (200, 1)])1695    def sc_invalid_args_setparent(self, mocker, fixture_ghcs, fixture_sas, request):1696        service_client = mocker.patch(path_service_client)1697        parent_kvp = {}1698        child_kvp = {}1699        if request.param[1] == 0:1700            parent_kvp.setdefault("capabilities", {"iotEdge": False})1701        if request.param[1] == 1:1702            child_kvp.setdefault("parentScopes", ["abcd"])1703        test_side_effect = [1704            build_mock_response(1705                mocker, request.param[0], generate_parent_device(**parent_kvp)1706            ),1707            build_mock_response(1708                mocker, request.param[0], generate_child_device(**child_kvp)1709            ),1710        ]1711        service_client.side_effect = test_side_effect1712        return service_client1713    @pytest.mark.parametrize("exp", [CLIError])1714    def test_device_setparent_invalid_args(self, sc_invalid_args_setparent, exp):1715        with pytest.raises(exp):1716            subject.iot_device_set_parent(1717                fixture_cmd, child_device_id, device_id, False, mock_target["entity"]1718            )1719    @pytest.fixture(params=[(200, 400), (200, 401), (200, 500)])1720    def sc_setparent_error(self, mocker, fixture_ghcs, fixture_sas, request):1721        service_client = mocker.patch(path_service_client)1722        test_side_effect = [1723            build_mock_response(mocker, request.param[0], generate_parent_device()),1724            build_mock_response(mocker, request.param[0], generate_child_device()),1725            build_mock_response(mocker, request.param[1], {}),1726        ]1727        service_client.side_effect = test_side_effect1728        return service_client1729    def test_device_setparent_error(self, sc_setparent_error):1730        with pytest.raises(CLIError):1731            subject.iot_device_set_parent(1732                fixture_cmd, child_device_id, device_id, False, mock_target["entity"]1733            )1734    @pytest.fixture(params=[(200, 200)])1735    def sc_invalid_etag_setparent(self, mocker, fixture_ghcs, fixture_sas, request):1736        service_client = mocker.patch(path_service_client)1737        child_kvp = {}1738        child_kvp.setdefault("etag", None)1739        test_side_effect = [1740            build_mock_response(mocker, request.param[0], generate_parent_device()),1741            build_mock_response(1742                mocker, request.param[0], generate_child_device(**child_kvp)1743            ),1744        ]1745        service_client.side_effect = test_side_effect1746        return service_client1747    @pytest.mark.parametrize("exp", [CLIError])1748    def test_device_setparent_invalid_etag(self, sc_invalid_etag_setparent, exp):1749        with pytest.raises(exp):1750            subject.iot_device_set_parent(1751                fixture_cmd, child_device_id, device_id, True, mock_target["entity"]1752            )1753    # add-children1754    @pytest.fixture(params=[(200, 0), (200, 1), (200, 2)])1755    def sc_addchildren(self, mocker, fixture_ghcs, fixture_sas, request):1756        service_client = mocker.patch(path_service_client)1757        child_kvp = {}1758        if request.param[1] == 1:1759            child_kvp.setdefault("parentScopes", ["abcd"])1760        if request.param[1] == 1:1761            child_kvp.setdefault("capabilities", {"iotEdge": True})1762        test_side_effect = [1763            build_mock_response(mocker, request.param[0], generate_parent_device()),1764            build_mock_response(1765                mocker, request.param[0], generate_child_device(**child_kvp)1766            ),1767            build_mock_response(mocker, request.param[0], {}),1768        ]1769        service_client.side_effect = test_side_effect1770        return service_client1771    def test_device_children_add(self, sc_addchildren):1772        subject.iot_device_children_add(1773            None, device_id, [child_device_id], True, mock_target["entity"]1774        )1775        args = sc_addchildren.call_args1776        url = args[0][0].url1777        body = json.loads(args[0][0].body)1778        assert "{}/devices/{}?".format(mock_target["entity"], child_device_id) in url1779        assert args[0][0].method == "PUT"1780        assert body["deviceId"] == child_device_id1781        assert body["deviceScope"] == generate_parent_device().get(1782            "deviceScope"1783        ) or body["parentScopes"] == [generate_parent_device().get("deviceScope")]1784    @pytest.fixture(params=[(200, 0), (200, 1)])1785    def sc_invalid_args_addchildren(self, mocker, fixture_ghcs, fixture_sas, request):1786        service_client = mocker.patch(path_service_client)1787        parent_kvp = {}1788        child_kvp = {}1789        if request.param[1] == 0:1790            parent_kvp.setdefault("capabilities", {"iotEdge": False})1791        if request.param[1] == 1:1792            child_kvp.setdefault("parentScopes", ["abcd"])1793        test_side_effect = [1794            build_mock_response(1795                mocker, request.param[0], generate_parent_device(**parent_kvp)1796            ),1797            build_mock_response(1798                mocker, request.param[0], generate_child_device(**child_kvp)1799            ),1800        ]1801        service_client.side_effect = test_side_effect1802        return service_client1803    @pytest.mark.parametrize("exp", [CLIError])1804    def test_device_addchildren_invalid_args(self, sc_invalid_args_addchildren, exp):1805        with pytest.raises(exp):1806            subject.iot_device_children_add(1807                fixture_cmd, device_id, [child_device_id], False, mock_target["entity"]1808            )1809    @pytest.fixture(params=[(200, 400), (200, 401), (200, 500)])1810    def sc_addchildren_error(self, mocker, fixture_ghcs, fixture_sas, request):1811        service_client = mocker.patch(path_service_client)1812        test_side_effect = [1813            build_mock_response(mocker, request.param[0], generate_parent_device()),1814            build_mock_response(mocker, request.param[0], generate_child_device()),1815            build_mock_response(mocker, request.param[1], {}),1816        ]1817        service_client.side_effect = test_side_effect1818        return service_client1819    def test_device_addchildren_error(self, sc_addchildren_error):1820        with pytest.raises(CLIError):1821            subject.iot_device_children_add(1822                fixture_cmd, device_id, [child_device_id], True, mock_target["entity"]1823            )1824    @pytest.fixture(params=[(200, 200)])1825    def sc_invalid_etag_addchildren(self, mocker, fixture_ghcs, fixture_sas, request):1826        service_client = mocker.patch(path_service_client)1827        child_kvp = {}1828        child_kvp.setdefault("etag", None)1829        test_side_effect = [1830            build_mock_response(mocker, request.param[0], generate_parent_device()),1831            build_mock_response(1832                mocker, request.param[0], generate_child_device(**child_kvp)1833            ),1834        ]1835        service_client.side_effect = test_side_effect1836        return service_client1837    @pytest.mark.parametrize("exp", [CLIError])1838    def test_device_addchildren_invalid_etag(self, sc_invalid_etag_setparent, exp):1839        with pytest.raises(exp):1840            subject.iot_device_children_add(1841                fixture_cmd, device_id, [child_device_id], True, mock_target["entity"]1842            )1843    # list-children1844    @pytest.fixture1845    def sc_listchildren(self, mocker, fixture_ghcs, fixture_sas, request):1846        service_client = mocker.patch(path_service_client)1847        child_kvp = {}1848        child_kvp.setdefault(1849            "parentScopes", [generate_parent_device().get("deviceScope")]1850        )1851        result = []1852        result.append(generate_child_device(**child_kvp))1853        test_side_effect = [1854            build_mock_response(mocker, 200, generate_parent_device()),1855            build_mock_response(mocker, 200, result, {"x-ms-continuation": None}),1856        ]1857        service_client.side_effect = test_side_effect1858        return service_client1859    def test_device_children_list(self, sc_listchildren):1860        result = subject.iot_device_children_list(1861            fixture_cmd, device_id, mock_target["entity"]1862        )1863        args = sc_listchildren.call_args1864        url = args[0][0].url1865        assert "{}/devices/query?".format(mock_target["entity"]) in url1866        assert args[0][0].method == "POST"1867        assert result == [child_device_id]1868    @pytest.fixture(params=[(200, 0), (200, 1)])1869    def sc_invalid_args_listchildren(self, mocker, fixture_ghcs, fixture_sas, request):1870        service_client = mocker.patch(path_service_client)1871        parent_kvp = {}1872        if request.param[1] == 0:1873            parent_kvp.setdefault("capabilities", {"iotEdge": False})1874        test_side_effect = [1875            build_mock_response(1876                mocker, request.param[0], generate_parent_device(**parent_kvp)1877            ),1878            build_mock_response(1879                mocker, request.param[0], [], {"x-ms-continuation": None}1880            ),1881        ]1882        service_client.side_effect = test_side_effect1883        return service_client1884    def test_device_listchildren_error(self, serviceclient_generic_error):1885        with pytest.raises(CLIError):1886            subject.iot_device_children_list(1887                fixture_cmd, device_id, mock_target["entity"]1888            )1889    # remove-children1890    @pytest.fixture(params=[(200, 200)])1891    def sc_removechildrenlist(self, mocker, fixture_ghcs, fixture_sas, request):1892        service_client = mocker.patch(path_service_client)1893        child_kvp = {}1894        child_kvp.setdefault(1895            "parentScopes", [generate_parent_device().get("deviceScope")]1896        )1897        test_side_effect = [1898            build_mock_response(mocker, request.param[0], generate_parent_device()),1899            build_mock_response(1900                mocker, request.param[0], generate_child_device(**child_kvp)1901            ),1902            build_mock_response(mocker, request.param[0], {}),1903        ]1904        service_client.side_effect = test_side_effect1905        return service_client1906    def test_device_children_remove_list(self, sc_removechildrenlist):1907        subject.iot_device_children_remove(1908            fixture_cmd, device_id, [child_device_id], False, mock_target["entity"]1909        )1910        args = sc_removechildrenlist.call_args1911        url = args[0][0].url1912        assert "{}/devices/{}?".format(mock_target["entity"], child_device_id) in url1913        assert args[0][0].method == "PUT"1914    @pytest.fixture(params=[(200, 0), (200, 1), (200, 2), (200, 3)])1915    def sc_invalid_args_removechildrenlist(1916        self, mocker, fixture_ghcs, fixture_sas, request1917    ):1918        service_client = mocker.patch(path_service_client)1919        parent_kvp = {}1920        child_kvp = {}1921        if request.param[1] == 0:1922            parent_kvp.setdefault("capabilities", {"iotEdge": False})1923        if request.param[1] == 2:1924            child_kvp.setdefault("parentScopes", [""])1925        if request.param[1] == 2:1926            child_kvp.setdefault("deviceScope", "")1927        if request.param[1] == 3:1928            child_kvp.setdefault("deviceScope", "other")1929        test_side_effect = [1930            build_mock_response(1931                mocker, request.param[0], generate_parent_device(**parent_kvp)1932            ),1933            build_mock_response(1934                mocker, request.param[0], generate_child_device(**child_kvp)1935            ),1936        ]1937        service_client.side_effect = test_side_effect1938        return service_client1939    @pytest.mark.parametrize("exp", [CLIError])1940    def test_device_removechildrenlist_invalid_args(1941        self, sc_invalid_args_removechildrenlist, exp1942    ):1943        with pytest.raises(exp):1944            subject.iot_device_children_remove(1945                fixture_cmd, device_id, [child_device_id], False, mock_target["entity"]1946            )1947    @pytest.fixture(params=[(200, 200)])1948    def sc_invalid_etag_removechildrenlist(1949        self, mocker, fixture_ghcs, fixture_sas, request1950    ):1951        service_client = mocker.patch(path_service_client)1952        child_kvp = {}1953        child_kvp.setdefault(1954            "parentScopes", [generate_parent_device().get("deviceScope")]1955        )1956        child_kvp.setdefault("etag", None)1957        test_side_effect = [1958            build_mock_response(mocker, request.param[0], generate_parent_device()),1959            build_mock_response(1960                mocker, request.param[0], generate_child_device(**child_kvp)1961            ),1962            build_mock_response(mocker, request.param[0], {}),1963        ]1964        service_client.side_effect = test_side_effect1965        return service_client1966    @pytest.mark.parametrize("exp", [CLIError])1967    def test_device_removechildrenlist_invalid_etag(1968        self, sc_invalid_etag_removechildrenlist, exp1969    ):1970        with pytest.raises(exp):1971            subject.iot_device_children_remove(1972                fixture_cmd, device_id, [child_device_id], False, mock_target["entity"]1973            )1974    @pytest.fixture(params=[(200, 400), (200, 401), (200, 500)])1975    def sc_removechildrenlist_error(self, mocker, fixture_ghcs, fixture_sas, request):1976        service_client = mocker.patch(path_service_client)1977        child_kvp = {}1978        child_kvp.setdefault(1979            "parentScopes", [generate_parent_device().get("deviceScope")]1980        )1981        test_side_effect = [1982            build_mock_response(mocker, request.param[0], generate_parent_device()),1983            build_mock_response(1984                mocker, request.param[0], generate_child_device(**child_kvp)1985            ),1986            build_mock_response(mocker, request.param[1], {}),1987        ]1988        service_client.side_effect = test_side_effect1989        return service_client1990    def test_device_removechildrenlist_error(self, sc_removechildrenlist_error):1991        with pytest.raises(CLIError):1992            subject.iot_device_children_remove(1993                fixture_cmd, device_id, child_device_id, False, mock_target["entity"]1994            )1995    @pytest.fixture(params=[(200, 200)])1996    def sc_removechildrenall(self, mocker, fixture_ghcs, fixture_sas, request):1997        service_client = mocker.patch(path_service_client)1998        child_kvp = {}1999        child_kvp.setdefault(2000            "parentScopes", [generate_parent_device().get("deviceScope")]2001        )2002        result = []2003        result.append(generate_child_device(**child_kvp))2004        test_side_effect = [2005            build_mock_response(mocker, request.param[0], generate_parent_device()),2006            build_mock_response(2007                mocker, request.param[0], result, {"x-ms-continuation": None}2008            ),2009            build_mock_response(2010                mocker, request.param[0], generate_child_device(**child_kvp)2011            ),2012            build_mock_response(mocker, request.param[0], {}),2013        ]2014        service_client.side_effect = test_side_effect2015        return service_client2016    def test_device_children_remove_all(self, sc_removechildrenall):2017        subject.iot_device_children_remove(2018            fixture_cmd, device_id, None, True, mock_target["entity"]2019        )2020        args = sc_removechildrenall.call_args2021        url = args[0][0].url2022        assert "{}/devices/{}?".format(mock_target["entity"], child_device_id) in url2023        assert args[0][0].method == "PUT"2024    @pytest.fixture(params=[(200, 0), (200, 1)])2025    def sc_invalid_args_removechildrenall(2026        self, mocker, fixture_ghcs, fixture_sas, request2027    ):2028        service_client = mocker.patch(path_service_client)2029        parent_kvp = {}2030        child_kvp = {}2031        child_kvp.setdefault(2032            "parentScopes", [generate_parent_device().get("deviceScope")]2033        )2034        result = []2035        result.append(generate_child_device(**child_kvp))2036        if request.param[1] == 0:2037            parent_kvp.setdefault("capabilities", {"iotEdge": False})2038        if request.param[1] == 1:2039            result = []2040        test_side_effect = [2041            build_mock_response(2042                mocker, request.param[0], generate_parent_device(**parent_kvp)2043            ),2044            build_mock_response(2045                mocker, request.param[0], result, {"x-ms-continuation": None}2046            ),2047        ]2048        service_client.side_effect = test_side_effect2049        return service_client2050    @pytest.mark.parametrize("exp", [CLIError])2051    def test_device_removechildrenall_invalid_args(2052        self, sc_invalid_args_removechildrenall, exp2053    ):2054        with pytest.raises(exp):2055            subject.iot_device_children_remove(2056                fixture_cmd, device_id, None, True, mock_target["entity"]2057            )2058    @pytest.fixture(params=[(200, 200)])2059    def sc_invalid_etag_removechildrenall(2060        self, mocker, fixture_ghcs, fixture_sas, request2061    ):2062        service_client = mocker.patch(path_service_client)2063        child_kvp = {}2064        child_kvp.setdefault(2065            "parentScopes", [generate_parent_device().get("deviceScope")]2066        )2067        child_kvp.setdefault("etag", None)2068        result = []2069        result.append(generate_child_device(**child_kvp))2070        test_side_effect = [2071            build_mock_response(mocker, request.param[0], generate_parent_device()),2072            build_mock_response(2073                mocker, request.param[0], result, {"x-ms-continuation": None}2074            ),2075            build_mock_response(2076                mocker, request.param[0], generate_child_device(**child_kvp)2077            ),2078            build_mock_response(mocker, request.param[0], {}),2079        ]2080        service_client.side_effect = test_side_effect2081        return service_client2082    @pytest.mark.parametrize("exp", [CLIError])2083    def test_device_removechildrenall_invalid_etag(2084        self, sc_invalid_etag_removechildrenall, exp2085    ):2086        with pytest.raises(exp):2087            subject.iot_device_children_remove(2088                fixture_cmd, device_id, None, True, mock_target["entity"]2089            )2090    @pytest.fixture(params=[(200, 400), (200, 401), (200, 500)])2091    def sc_removechildrenall_error(self, mocker, fixture_ghcs, fixture_sas, request):2092        service_client = mocker.patch(path_service_client)2093        child_kvp = {}2094        child_kvp.setdefault(2095            "parentScopes", [generate_parent_device().get("deviceScope")]2096        )2097        result = []2098        result.append(generate_child_device(**child_kvp))2099        test_side_effect = [2100            build_mock_response(mocker, request.param[0], generate_parent_device()),2101            build_mock_response(2102                mocker, request.param[0], result, {"x-ms-continuation": None}2103            ),2104            build_mock_response(2105                mocker, request.param[0], generate_child_device(**child_kvp)2106            ),2107            build_mock_response(mocker, request.param[1], {}),2108        ]2109        service_client.side_effect = test_side_effect2110        return service_client2111    def test_device_removechildrenall_error(self, sc_removechildrenall_error):2112        with pytest.raises(CLIError):2113            subject.iot_device_children_remove(2114                fixture_cmd, device_id, None, True, mock_target["entity"]2115            )2116class TestDeviceDistributedTracing:2117    @pytest.fixture(params=[(200, 200)])2118    def sc_distributed_tracing_show(self, mocker, fixture_ghcs, fixture_sas, request):2119        service_client = mocker.patch(path_service_client)2120        twin_kvp = {}2121        twin_kvp.setdefault("capabilities", {"iotEdge": False})2122        twin_kvp.setdefault(2123            "properties",2124            {2125                "desired": {2126                    TRACING_PROPERTY: {"sampling_mode": 1, "sampling_rate": 50}2127                },2128                "reported": {},2129            },2130        )2131        test_side_effect = [2132            build_mock_response(2133                mocker, request.param[0], payload=generate_device_twin_show(**twin_kvp)2134            )2135        ]2136        service_client.side_effect = test_side_effect2137        return service_client2138    def test_distributed_tracing_show(self, sc_distributed_tracing_show):2139        result = subject.iot_hub_distributed_tracing_show(2140            fixture_cmd, mock_target["entity"], device_id2141        )2142        args = sc_distributed_tracing_show.call_args2143        url = args[0][0].url2144        assert "{}/twins/{}?".format(mock_target["entity"], device_id) in url2145        assert args[0][0].method == "GET"2146        assert result["deviceId"] == device_id2147        assert result["samplingMode"] == "enabled"2148        assert result["samplingRate"] == "50%"2149        assert not result["isSynced"]2150        assert json.dumps(result)2151    @pytest.fixture(params=[(200, 0), (200, 1), (200, 2)])2152    def sc_invalid_args_distributed_tracing_show(2153        self, mocker, fixture_ghcs, fixture_sas, request2154    ):2155        service_client = mocker.patch(path_service_client)2156        twin_kvp = {}2157        twin_kvp.setdefault("capabilities", {"iotEdge": False})2158        if request.param[1] == 0:2159            mock_target["location"] = "westus"2160        if request.param[1] == 1:2161            mock_target["sku_tier"] = "Basic"2162        if request.param[1] == 2:2163            twin_kvp.setdefault("capabilities", {"iotEdge": True})2164        test_side_effect = [2165            build_mock_response(2166                mocker, request.param[0], payload=generate_device_twin_show(**twin_kvp)2167            )2168        ]2169        service_client.side_effect = test_side_effect2170        return service_client2171    @pytest.mark.parametrize("exp", [CLIError])2172    def test_distributed_tracing_show_invalid_args(2173        self, sc_invalid_args_distributed_tracing_show, exp2174    ):2175        with pytest.raises(exp):2176            subject.iot_hub_distributed_tracing_show(2177                fixture_cmd, mock_target["entity"], device_id2178            )2179    def test_distributed_tracing_show_error(self, serviceclient_generic_error):2180        with pytest.raises(CLIError):2181            subject.iot_hub_distributed_tracing_show(2182                fixture_cmd, mock_target["entity"], device_id2183            )2184    @pytest.fixture(params=[(200, 0), (200, 1), (200, 2)])2185    def sc_distributed_tracing_update(self, mocker, fixture_ghcs, fixture_sas, request):2186        service_client = mocker.patch(path_service_client)2187        twin_kvp = {}2188        twin_kvp.setdefault("capabilities", {"iotEdge": False})2189        mock_target["location"] = "westus2"2190        mock_target["sku_tier"] = "Standard"2191        if request.param[1] == 0:2192            twin_kvp.setdefault(2193                "properties",2194                {2195                    "desired": {2196                        TRACING_PROPERTY: {"sampling_mode": 1, "sampling_rate": 50}2197                    },2198                    "reported": {},2199                },2200            )2201        if request.param[1] == 1:2202            twin_kvp.setdefault("properties", {"desired": {}, "reported": {}})2203        if request.param[1] == 2:2204            twin_kvp.setdefault(2205                "properties",2206                {2207                    "desired": {2208                        TRACING_PROPERTY: {"sampling_mode": 2, "sampling_rate": 0}2209                    },2210                    "reported": {},2211                },2212            )2213        test_side_effect = [2214            build_mock_response(2215                mocker, request.param[0], payload=generate_device_twin_show(**twin_kvp)2216            ),2217            build_mock_response(2218                mocker,2219                request.param[0],2220                generate_device_twin_show(2221                    properties={2222                        "desired": {2223                            TRACING_PROPERTY: {"sampling_mode": 1, "sampling_rate": 58}2224                        },2225                        "reported": {},2226                    }2227                ),2228            ),2229        ]2230        service_client.side_effect = test_side_effect2231        return service_client2232    def test_distributed_tracing_update(self, sc_distributed_tracing_update):2233        result = subject.iot_hub_distributed_tracing_update(2234            fixture_cmd, mock_target["entity"], device_id, "on", 582235        )2236        args = sc_distributed_tracing_update.call_args2237        url = args[0][0].url2238        assert "{}/twins/{}?".format(mock_target["entity"], device_id) in url2239        assert args[0][0].method == "PATCH"2240        assert result["deviceId"] == device_id2241        assert result["samplingMode"] == "enabled"2242        assert result["samplingRate"] == "58%"2243        assert not result["isSynced"]2244        assert json.dumps(result)2245    @pytest.fixture(params=[(200, 0), (200, 1), (200, 2)])2246    def sc_invalid_args_distributed_tracing_update(2247        self, mocker, fixture_ghcs, fixture_sas, request2248    ):2249        service_client = mocker.patch(path_service_client)2250        twin_kvp = {}2251        twin_kvp.setdefault("capabilities", {"iotEdge": False})2252        if request.param[1] == 0:2253            mock_target["location"] = "westus"2254        if request.param[1] == 1:2255            mock_target["sku_tier"] = "Basic"2256        if request.param[1] == 2:2257            twin_kvp.setdefault("capabilities", {"iotEdge": True})2258        test_side_effect = [2259            build_mock_response(2260                mocker, request.param[0], payload=generate_device_twin_show(**twin_kvp)2261            )2262        ]2263        service_client.side_effect = test_side_effect2264        return service_client2265    @pytest.mark.parametrize("exp", [CLIError])2266    def test_distributed_tracing_update_invalid_args(2267        self, sc_invalid_args_distributed_tracing_update, exp2268    ):2269        with pytest.raises(exp):2270            subject.iot_hub_distributed_tracing_update(2271                fixture_cmd, mock_target["entity"], device_id, "on", 582272            )2273    def test_distributed_tracing_update_error(self, serviceclient_generic_error):2274        with pytest.raises(CLIError):2275            subject.iot_hub_distributed_tracing_update(2276                fixture_cmd, mock_target["entity"], device_id, "on", 58...

conftest.py

Source:conftest.py

...27        secret_key=faker.word(),28        autocreate_buckets=test_s3_autocreate_buckets,29    )30@pytest.fixture()31def test_side_effect(request: FixtureRequest) -> Optional[Exception]:32    if hasattr(request, "param"):33        return cast(Exception, request.param)34    return None35@pytest.fixture()36def test_exception(request: FixtureRequest) -> Optional[Exception]:37    if hasattr(request, "param"):38        return cast(Exception, request.param)39    return None40@pytest.fixture(scope="class")41def test_object_owner() -> Dict[str, str]:42    return {"DisplayName": "hello-friend", "ID": "hello-friend"}43@pytest.fixture(scope="class")44def test_object_dict(test_object_owner: Dict[str, str]) -> Dict[str, Any]:45    return {...

tests.py

Source:tests.py

1from django.test import TestCase2from apps.drug_side_effects.models import drugs3import time4class TestingViews(TestCase):5    def test_home(self):6        response = self.client.get('/drug-side-effects/')7        self.assertEquals(response.status_code, 200)8        self.assertTemplateUsed(response ,'drug_side_effects_home.html')9    def test_add_drug(self):10        response = self.client.post('/drug-side-effects/', {"add-drug": "add-drug", "drug_name": time.time(), "side_effects": "test_side_effect"})11        self.assertEquals(response.status_code, 200)12        self.assertTemplateUsed(response,'drug_side_effects_home.html')13    def test_side_effects_drug(self):14        newDrug = drugs(drug_name = "test_drug", side_effects="test_side_effect")15        newDrug.save()16        response = self.client.post('/drug-side-effects/', {"get-side-effects": "get-side-effects", "drug_name": "test_drug"})17        self.assertEquals(response.status_code, 200)18        self.assertTemplateUsed(response,'drug_side_effects_home.html')19    20    def test_external_api(self):21        response = self.client.post('/drug-side-effects/', {"drug_name": "test_drug"})22        self.assertEquals(response.status_code, 200)23        self.assertTemplateUsed(response,'drug_side_effects_home.html')24class ModelTest(TestCase):25    def test_model(self):26        newdrug = drugs.objects.create(27            drug_name = "test_drug",28            side_effects = "test_side_effect",29        )...

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.