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.

LambdaTest Learning Hubs:

• JUnit Tutorial
• TestNG Tutorial
• Webdriver Tutorial
• WebDriverIO Tutorial
• Protractor Tutorial
• Selenium 4 Tutorial
• Jenkins Tutorial
• NUnit Tutorial
• Jest Tutorial
• Playwright Tutorial
• Cypress Tutorial
• PyTest Tutorial

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run molecule automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

View sample repo View DocumentationSign up Free

›

_