зеркало из
1
0
Форкнуть 0

Namespace adjustments + DPS Service Client (#6)

Namespace adjustments + DPS Service Client
This commit is contained in:
Carter Tinney 2018-10-18 12:20:50 -07:00 коммит произвёл GitHub
Родитель e868041ff0
Коммит caa8cfc4df
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
107 изменённых файлов: 4077 добавлений и 83 удалений

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -1,3 +1,8 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
__all__ = ["ConnectionString"]
CS_DELIMITER = ";"
@ -73,3 +78,9 @@ class ConnectionString(object):
def __repr__(self):
return self._strrep
def get(self, key, default=None):
try:
return self._dict[key]
except KeyError:
return default

Просмотреть файл

@ -1,3 +1,8 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import base64
import hmac
import hashlib
@ -31,7 +36,7 @@ class SasToken(object):
SasTokenError if trying to build a SasToken from invalid values
"""
_encoding_type = 'utf-8'
_encoding_type = "utf-8"
_service_token_format = "SharedAccessSignature sr={}&sig={}&se={}&skn={}"
_device_token_format = "SharedAccessSignature sr={}&sig={}&se={}"
@ -70,7 +75,5 @@ class SasToken(object):
self._uri, signature, str(self.expiry_time), self._key_name
)
else:
token = self._device_token_format.format(
self._uri, signature, str(self.expiry_time)
)
return token
token = self._device_token_format.format(self._uri, signature, str(self.expiry_time))
return token

Просмотреть файл

@ -0,0 +1,11 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
[bdist_wheel]
# This flag says to generate wheels that support both Python 2 and Python
# 3. If your code will not run unchanged on both Python 2 and 3, you will
# need to generate separate wheels for each Python version that you
# support.
universal=1

30
azure-iot-common/setup.py Normal file
Просмотреть файл

@ -0,0 +1,30 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from setuptools import setup, find_packages
setup(
name="azure-iot-common",
version="0.0.1",
description="Microsoft Azure IoT SDK Common",
license="MIT License",
author="Microsoft Corporation",
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Topic :: Software Development :: Build Tools",
"License :: OSI Approved :: MIT Software License",
"Programming Language :: Python",
"Programming Language :: Python :: 2",
"Programming Language :: Python :: 2.7",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
],
install_requires=["six"],
packages=find_packages(exclude=["tests"]),
)

Просмотреть файл

@ -1,5 +1,10 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import pytest
from connection_string import ConnectionString
from azure.iot.common.connection_string import ConnectionString
class TestConnectionStringInput(object):
@ -72,3 +77,24 @@ def test___getitem__item_does_not_exist():
"HostName=my.host.name;SharedAccessKeyName=mykeyname;SharedAccessKey=Zm9vYmFy"
)
cs["SharedAccessSignature"]
def test_get_item_exists():
cs = ConnectionString(
"HostName=my.host.name;SharedAccessKeyName=mykeyname;SharedAccessKey=Zm9vYmFy"
)
assert cs.get("HostName") == "my.host.name"
def test_get_item_does_not_exist_w_default():
cs = ConnectionString(
"HostName=my.host.name;SharedAccessKeyName=mykeyname;SharedAccessKey=Zm9vYmFy"
)
assert cs.get("invalidkey", "defaultval") == "defaultval"
def test_get_item_does_not_exist_no_given_default():
cs = ConnectionString(
"HostName=my.host.name;SharedAccessKeyName=mykeyname;SharedAccessKey=Zm9vYmFy"
)
assert cs.get("invalidkey") == None

Просмотреть файл

@ -1,8 +1,12 @@
# -*- coding: utf-8 -*-
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import pytest
import time
from sastoken import SasToken, SasTokenError
from azure.iot.common.sastoken import SasToken, SasTokenError
class TestCreateSasToken(object):

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -6,14 +6,7 @@
import os
import sys
# Temporary path hack (replace once monorepo path solution implemented)
if sys.platform == "darwin" or sys.platform == "linux" or sys.platform == "linux2":
sys.path.append(os.path.join(os.path.dirname(__file__), "../../python_shared_utils"))
else:
sys.path.append(os.path.join(os.path.dirname(__file__), "..\..\python_shared_utils"))
# ---------------------------------------------------------------------
from connection_string import (
from azure.iot.common.connection_string import (
ConnectionString,
HOST_NAME,
SHARED_ACCESS_KEY_NAME,
@ -23,7 +16,7 @@ from connection_string import (
MODULE_ID,
GATEWAY_HOST_NAME,
)
from sastoken import SasToken
from azure.iot.common.sastoken import SasToken
class SymmetricKeyAuthenticationProvider(object):

Просмотреть файл

@ -5,7 +5,7 @@
from .mqtt_provider import MQTTProvider
import types
from iothub_device_sdk.device.transport.abstract_transport import AbstractTransport
from azure.iot.hub.devicesdk.transport.abstract_transport import AbstractTransport
class MQTTTransport(AbstractTransport):

Просмотреть файл

@ -0,0 +1 @@
## README

Просмотреть файл

@ -4,8 +4,8 @@
# --------------------------------------------------------------------------------------------
import os
import logging
from iothub_device_sdk.device.device_client import DeviceClient
from iothub_device_sdk.device.transport.transport_config import TransportProtocol, TransportConfig
from azure.iot.hub.devicesdk.device_client import DeviceClient
from azure.iot.hub.devicesdk.transport.transport_config import TransportProtocol, TransportConfig
logging.basicConfig(level=logging.INFO)

Просмотреть файл

@ -0,0 +1,41 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from setuptools import setup, find_packages
with open("doc/package-readme.md", "r") as fh:
_long_description = fh.read()
setup(
name="azure-iot-hub-devicesdk",
version="0.0.1",
description="Microsoft Azure IoT Hub Device SDK",
license="MIT License",
url="https://github.com/Azure/azure-iot-sdk-python",
author="Microsoft Corporation",
long_description=_long_description,
long_description_content_type="text/markdown",
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Topic :: Software Development :: Build Tools",
"License :: OSI Approved :: MIT Software License",
"Programming Language :: Python :: 2",
"Programming Language :: Python :: 2.7",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
],
install_requires=[
"azure-iot-common",
"six",
"paho-mqtt==1.4.0",
"transitions==0.6.8",
"enum34==1.1.6",
],
packages=find_packages(exclude=["tests"]),
)

Просмотреть файл

@ -0,0 +1,15 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import pytest
from azure.iot.hub.devicesdk.transport.abstract_transport import AbstractTransport
def test_raises_exception():
with pytest.raises(TypeError) as error:
AbstractTransport()
msg = str(error.value)
expected_msg = "Can't instantiate abstract class AbstractTransport with abstract methods _get_connected_state_callback, connect, disconnect, send_event"
assert msg == expected_msg

Просмотреть файл

@ -1,7 +1,14 @@
from ..device.device_client import DeviceClient
from ..device.symmetric_key_authentication_provider import SymmetricKeyAuthenticationProvider
from iothub_device_sdk.device.transport.mqtt.mqtt_transport import MQTTTransport
from ..device.transport.transport_config import TransportConfig, TransportProtocol
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from azure.iot.hub.devicesdk.device_client import DeviceClient
from azure.iot.hub.devicesdk.symmetric_key_authentication_provider import (
SymmetricKeyAuthenticationProvider,
)
from azure.iot.hub.devicesdk.transport.mqtt.mqtt_transport import MQTTTransport
from azure.iot.hub.devicesdk.transport.transport_config import TransportConfig, TransportProtocol
import pytest
from six import add_move, MovedModule

Просмотреть файл

@ -1,4 +1,9 @@
from ..device.transport.mqtt.mqtt_provider import MQTTProvider
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from azure.iot.hub.devicesdk.transport.mqtt.mqtt_provider import MQTTProvider
import paho.mqtt.client as mqtt
from transitions import Machine
import os
@ -43,7 +48,7 @@ def test_create():
def test_on_enter_connecting(mocker):
mock_mqtt_client = MagicMock(spec=mqtt.Client)
mock_constructor_mqtt_client = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_provider.mqtt.Client"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_provider.mqtt.Client"
)
mock_constructor_mqtt_client.return_value = mock_mqtt_client
@ -88,7 +93,7 @@ def test_on_enter_disconnecting(mocker):
def test_connect(mocker):
mock_machine_from_real = create_from_real_state_machine()
mock_machine_constructor = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_provider.Machine"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_provider.Machine"
)
mock_machine_constructor.return_value = mock_machine_from_real
@ -101,7 +106,7 @@ def test_connect(mocker):
def test_disconnect(mocker):
mock_mqtt_client = MagicMock(spec=mqtt.Client)
mock_constructor_mqtt_client = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_provider.mqtt.Client"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_provider.mqtt.Client"
)
mock_constructor_mqtt_client.return_value = mock_mqtt_client
mocker.patch.object(mock_mqtt_client, "loop_stop")
@ -119,7 +124,7 @@ def test_publish(mocker):
mock_mqtt_client = MagicMock(spec=mqtt.Client)
mock_constructor_mqtt_client = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_provider.mqtt.Client"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_provider.mqtt.Client"
)
mock_constructor_mqtt_client.return_value = mock_mqtt_client
mocker.patch.object(mock_mqtt_client, "publish")
@ -136,7 +141,7 @@ def test_emit_connection_status(mocker):
mock_machine_from_real = create_from_real_state_machine()
mock_machine_constructor = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_provider.Machine"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_provider.Machine"
)
mock_machine_constructor.return_value = mock_machine_from_real
connected_state = "connected"

Просмотреть файл

@ -1,7 +1,14 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import pytest
from iothub_device_sdk.device.transport.mqtt.mqtt_transport import MQTTTransport
from iothub_device_sdk.device.transport.mqtt.mqtt_provider import MQTTProvider
from ..device.symmetric_key_authentication_provider import SymmetricKeyAuthenticationProvider
from azure.iot.hub.devicesdk.transport.mqtt.mqtt_transport import MQTTTransport
from azure.iot.hub.devicesdk.transport.mqtt.mqtt_provider import MQTTProvider
from azure.iot.hub.devicesdk.symmetric_key_authentication_provider import (
SymmetricKeyAuthenticationProvider,
)
from six import add_move, MovedModule
add_move(MovedModule("mock", "mock", "unittest.mock"))
@ -43,7 +50,7 @@ def test_create():
def test_connect_to_message_broker(mocker, transport):
mock_mqtt_provider = MagicMock(spec=MQTTProvider)
mock_mqtt_provider_constructor = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_transport.MQTTProvider"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_transport.MQTTProvider"
)
mock_mqtt_provider_constructor.return_value = mock_mqtt_provider
@ -59,7 +66,7 @@ def test_sendevent(mocker, transport):
mock_mqtt_provider = MagicMock(spec=MQTTProvider)
mock_mqtt_provider_constructor = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_transport.MQTTProvider"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_transport.MQTTProvider"
)
mock_mqtt_provider_constructor.return_value = mock_mqtt_provider
mocker.patch.object(mock_mqtt_provider, "connect")
@ -75,7 +82,7 @@ def test_sendevent(mocker, transport):
def test_disconnect_from_message_broker(mocker, transport):
mock_mqtt_provider = MagicMock(spec=MQTTProvider)
mock_mqtt_provider_constructor = mocker.patch(
"iothub_device_sdk.device.transport.mqtt.mqtt_transport.MQTTProvider"
"azure.iot.hub.devicesdk.transport.mqtt.mqtt_transport.MQTTProvider"
)
mock_mqtt_provider_constructor.return_value = mock_mqtt_provider
mocker.patch.object(mock_mqtt_provider, "disconnect")

Просмотреть файл

@ -1,4 +1,11 @@
from ..device.symmetric_key_authentication_provider import SymmetricKeyAuthenticationProvider
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from azure.iot.hub.devicesdk.symmetric_key_authentication_provider import (
SymmetricKeyAuthenticationProvider,
)
import pytest
@ -78,7 +85,7 @@ def test_create_from_module_gateway_connection_string():
def test_sastoken_key(mocker):
uri = hostname + "/devices/" + device_id
mock_sastoken = mocker.patch(
"iothub_device_sdk.device.symmetric_key_authentication_provider.SasToken"
"azure.iot.hub.devicesdk.symmetric_key_authentication_provider.SasToken"
)
dummy_value = "SharedAccessSignature sr=beauxbatons.academy-net%2Fdevices%2FMyPensieve&sig=zh8pwNIG56yUd3Nna7lyKA2HQAns84U3XwxyFQJqh48%3D&se=1539036534"
mock_sastoken.return_value.__str__.return_value = dummy_value
@ -96,7 +103,7 @@ def test_sastoken_key(mocker):
def test_sastoken_keyname(mocker):
uri = hostname + "/devices/" + device_id
mock_sastoken = mocker.patch(
"iothub_device_sdk.device.symmetric_key_authentication_provider.SasToken"
"azure.iot.hub.devicesdk.symmetric_key_authentication_provider.SasToken"
)
dummy_value = "SharedAccessSignature sr=beauxbatons.academy-net%2Fdevices%2FMyPensieve&sig=fT/nO0NA/25IKl0Ei2upxDDj6KnY6RPVIjlV84/9aR8%3D&se=1539043658&skn=alohomora"
mock_sastoken.return_value.__str__.return_value = dummy_value

Просмотреть файл

@ -1,6 +1,13 @@
from ..device.symmetric_key_authentication_provider import SymmetricKeyAuthenticationProvider
from iothub_device_sdk.device.transport.mqtt.mqtt_transport import MQTTTransport
from iothub_device_sdk.device.transport.transport_config import TransportConfig, TransportProtocol
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from azure.iot.hub.devicesdk.symmetric_key_authentication_provider import (
SymmetricKeyAuthenticationProvider,
)
from azure.iot.hub.devicesdk.transport.mqtt.mqtt_transport import MQTTTransport
from azure.iot.hub.devicesdk.transport.transport_config import TransportConfig, TransportProtocol
import pytest
connection_string_format = "HostName={};DeviceId={};SharedAccessKey={}"
@ -31,7 +38,7 @@ def test_create():
def test_create_specific_transport(mocker, authentication_provider):
mock_transport = mocker.patch(
"iothub_device_sdk.device.transport.transport_config.MQTTTransport"
"azure.iot.hub.devicesdk.transport.transport_config.MQTTTransport"
)
transport_config = TransportConfig(TransportProtocol.MQTT)

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -0,0 +1 @@
__path__ = __import__("pkgutil").extend_path(__path__, __name__)

Просмотреть файл

@ -0,0 +1,14 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from .client import ProvisioningServiceClient
from .models import (
QuerySpecification,
BulkEnrollmentOperation,
ProvisioningServiceErrorDetailsException,
)
from . import models
__all__ = ["ProvisioningServiceClient", "QuerySpecification", "BulkEnrollmentOperation", "models"]

Просмотреть файл

@ -0,0 +1,44 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
"""Provides authentication classes for use with the msrest library
"""
from msrest.authentication import Authentication
from azure.iot.common.connection_string import ConnectionString
from azure.iot.common.connection_string import HOST_NAME, SHARED_ACCESS_KEY_NAME, SHARED_ACCESS_KEY
from azure.iot.common.sastoken import SasToken
__all__ = ["ConnectionStringAuthentication"]
class ConnectionStringAuthentication(ConnectionString, Authentication):
"""ConnectionString class that can be used with msrest to provide SasToken authentication
:param connection_string: The connection string to generate SasToken with
"""
def __init__(self, connection_string):
super(ConnectionStringAuthentication, self).__init__(
connection_string
) # ConnectionString __init__
def signed_session(self, session=None):
"""Create requests session with any required auth headers applied.
If a session object is provided, configure it directly. Otherwise,
create a new session and return it.
:param session: The session to configure for authentication
:type session: requests.Session
:rtype: requests.Session
"""
session = super(ConnectionStringAuthentication, self).signed_session(session)
# Authorization header
sastoken = SasToken(self[HOST_NAME], self[SHARED_ACCESS_KEY], self[SHARED_ACCESS_KEY_NAME])
session.headers[self.header] = str(sastoken)
return session

Просмотреть файл

@ -0,0 +1,25 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
"""Redefine the generated ProvisioningServiceClient class via inheritance to allow for API
customization and authentication logic injection
"""
from .protocol import ProvisioningServiceClient as _ProvisioningServiceClient
from .auth import ConnectionStringAuthentication, HOST_NAME
class ProvisioningServiceClient(_ProvisioningServiceClient):
"""API for service operations with the Azure IoT Hub Device Provisioning Service
:ivar config: Configuration for client.
:vartype config: ProvisioningServiceClientConfiguration
:param str connection_string: Connection String for your Device Provisioning Service hub.
"""
def __init__(self, connection_string):
cs_auth = ConnectionStringAuthentication(connection_string)
super(ProvisioningServiceClient, self).__init__(cs_auth, "https://" + cs_auth[HOST_NAME])

Просмотреть файл

@ -0,0 +1,10 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
"""Imports the protocol.models package as a module for ease of use
and defines model convenience methods
"""
from .protocol.models import *

Просмотреть файл

@ -0,0 +1,13 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from .provisioning_service_client import ProvisioningServiceClient
from .version import VERSION
__all__ = ["ProvisioningServiceClient"]
__version__ = VERSION

Просмотреть файл

@ -0,0 +1,88 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
try:
from .provisioning_service_error_details_py3 import (
ProvisioningServiceErrorDetails,
ProvisioningServiceErrorDetailsException,
)
from .device_capabilities_py3 import DeviceCapabilities
from .device_registration_state_py3 import DeviceRegistrationState
from .tpm_attestation_py3 import TpmAttestation
from .x509_certificate_info_py3 import X509CertificateInfo
from .x509_certificate_with_info_py3 import X509CertificateWithInfo
from .x509_certificates_py3 import X509Certificates
from .x509_ca_references_py3 import X509CAReferences
from .x509_attestation_py3 import X509Attestation
from .symmetric_key_attestation_py3 import SymmetricKeyAttestation
from .attestation_mechanism_py3 import AttestationMechanism
from .metadata_py3 import Metadata
from .twin_collection_py3 import TwinCollection
from .initial_twin_properties_py3 import InitialTwinProperties
from .initial_twin_py3 import InitialTwin
from .reprovision_policy_py3 import ReprovisionPolicy
from .custom_allocation_definition_py3 import CustomAllocationDefinition
from .individual_enrollment_py3 import IndividualEnrollment
from .enrollment_group_py3 import EnrollmentGroup
from .bulk_enrollment_operation_py3 import BulkEnrollmentOperation
from .bulk_enrollment_operation_error_py3 import BulkEnrollmentOperationError
from .bulk_enrollment_operation_result_py3 import BulkEnrollmentOperationResult
from .query_specification_py3 import QuerySpecification
except (SyntaxError, ImportError):
from .provisioning_service_error_details import (
ProvisioningServiceErrorDetails,
ProvisioningServiceErrorDetailsException,
)
from .device_capabilities import DeviceCapabilities
from .device_registration_state import DeviceRegistrationState
from .tpm_attestation import TpmAttestation
from .x509_certificate_info import X509CertificateInfo
from .x509_certificate_with_info import X509CertificateWithInfo
from .x509_certificates import X509Certificates
from .x509_ca_references import X509CAReferences
from .x509_attestation import X509Attestation
from .symmetric_key_attestation import SymmetricKeyAttestation
from .attestation_mechanism import AttestationMechanism
from .metadata import Metadata
from .twin_collection import TwinCollection
from .initial_twin_properties import InitialTwinProperties
from .initial_twin import InitialTwin
from .reprovision_policy import ReprovisionPolicy
from .custom_allocation_definition import CustomAllocationDefinition
from .individual_enrollment import IndividualEnrollment
from .enrollment_group import EnrollmentGroup
from .bulk_enrollment_operation import BulkEnrollmentOperation
from .bulk_enrollment_operation_error import BulkEnrollmentOperationError
from .bulk_enrollment_operation_result import BulkEnrollmentOperationResult
from .query_specification import QuerySpecification
__all__ = [
"ProvisioningServiceErrorDetails",
"ProvisioningServiceErrorDetailsException",
"DeviceCapabilities",
"DeviceRegistrationState",
"TpmAttestation",
"X509CertificateInfo",
"X509CertificateWithInfo",
"X509Certificates",
"X509CAReferences",
"X509Attestation",
"SymmetricKeyAttestation",
"AttestationMechanism",
"Metadata",
"TwinCollection",
"InitialTwinProperties",
"InitialTwin",
"ReprovisionPolicy",
"CustomAllocationDefinition",
"IndividualEnrollment",
"EnrollmentGroup",
"BulkEnrollmentOperation",
"BulkEnrollmentOperationError",
"BulkEnrollmentOperationResult",
"QuerySpecification",
]

Просмотреть файл

@ -0,0 +1,41 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class AttestationMechanism(Model):
"""Attestation mechanism for individualEnrollment as well as enrollmentGroup.
All required parameters must be populated in order to send to Azure.
:param type: Required. Attestation Type. Possible values include: 'none',
'tpm', 'x509', 'symmetricKey'
:type type: str or ~protocol.models.enum
:param tpm: TPM attestation method.
:type tpm: ~protocol.models.TpmAttestation
:param x509: X509 attestation method.
:type x509: ~protocol.models.X509Attestation
:param symmetric_key: Symmetric Key attestation method.
:type symmetric_key: ~protocol.models.SymmetricKeyAttestation
"""
_validation = {"type": {"required": True}}
_attribute_map = {
"type": {"key": "type", "type": "str"},
"tpm": {"key": "tpm", "type": "TpmAttestation"},
"x509": {"key": "x509", "type": "X509Attestation"},
"symmetric_key": {"key": "symmetricKey", "type": "SymmetricKeyAttestation"},
}
def __init__(self, **kwargs):
super(AttestationMechanism, self).__init__(**kwargs)
self.type = kwargs.get("type", None)
self.tpm = kwargs.get("tpm", None)
self.x509 = kwargs.get("x509", None)
self.symmetric_key = kwargs.get("symmetric_key", None)

Просмотреть файл

@ -0,0 +1,41 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class AttestationMechanism(Model):
"""Attestation mechanism for individualEnrollment as well as enrollmentGroup.
All required parameters must be populated in order to send to Azure.
:param type: Required. Attestation Type. Possible values include: 'none',
'tpm', 'x509', 'symmetricKey'
:type type: str or ~protocol.models.enum
:param tpm: TPM attestation method.
:type tpm: ~protocol.models.TpmAttestation
:param x509: X509 attestation method.
:type x509: ~protocol.models.X509Attestation
:param symmetric_key: Symmetric Key attestation method.
:type symmetric_key: ~protocol.models.SymmetricKeyAttestation
"""
_validation = {"type": {"required": True}}
_attribute_map = {
"type": {"key": "type", "type": "str"},
"tpm": {"key": "tpm", "type": "TpmAttestation"},
"x509": {"key": "x509", "type": "X509Attestation"},
"symmetric_key": {"key": "symmetricKey", "type": "SymmetricKeyAttestation"},
}
def __init__(self, *, type, tpm=None, x509=None, symmetric_key=None, **kwargs) -> None:
super(AttestationMechanism, self).__init__(**kwargs)
self.type = type
self.tpm = tpm
self.x509 = x509
self.symmetric_key = symmetric_key

Просмотреть файл

@ -0,0 +1,33 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class BulkEnrollmentOperation(Model):
"""Bulk operation.
All required parameters must be populated in order to send to Azure.
:param enrollments: Required. Enrollment items
:type enrollments: list[~protocol.models.IndividualEnrollment]
:param mode: Required. Operation mode. Possible values include: 'create',
'update', 'updateIfMatchETag', 'delete'
:type mode: str or ~protocol.models.enum
"""
_validation = {"enrollments": {"required": True}, "mode": {"required": True}}
_attribute_map = {
"enrollments": {"key": "enrollments", "type": "[IndividualEnrollment]"},
"mode": {"key": "mode", "type": "str"},
}
def __init__(self, **kwargs):
super(BulkEnrollmentOperation, self).__init__(**kwargs)
self.enrollments = kwargs.get("enrollments", None)
self.mode = kwargs.get("mode", None)

Просмотреть файл

@ -0,0 +1,40 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class BulkEnrollmentOperationError(Model):
"""Bulk enrollment operation error.
All required parameters must be populated in order to send to Azure.
:param registration_id: Required. Device registration id.
:type registration_id: str
:param error_code: Required. Error code
:type error_code: int
:param error_status: Required. Error status
:type error_status: str
"""
_validation = {
"registration_id": {"required": True},
"error_code": {"required": True},
"error_status": {"required": True},
}
_attribute_map = {
"registration_id": {"key": "registrationId", "type": "str"},
"error_code": {"key": "errorCode", "type": "int"},
"error_status": {"key": "errorStatus", "type": "str"},
}
def __init__(self, **kwargs):
super(BulkEnrollmentOperationError, self).__init__(**kwargs)
self.registration_id = kwargs.get("registration_id", None)
self.error_code = kwargs.get("error_code", None)
self.error_status = kwargs.get("error_status", None)

Просмотреть файл

@ -0,0 +1,42 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class BulkEnrollmentOperationError(Model):
"""Bulk enrollment operation error.
All required parameters must be populated in order to send to Azure.
:param registration_id: Required. Device registration id.
:type registration_id: str
:param error_code: Required. Error code
:type error_code: int
:param error_status: Required. Error status
:type error_status: str
"""
_validation = {
"registration_id": {"required": True},
"error_code": {"required": True},
"error_status": {"required": True},
}
_attribute_map = {
"registration_id": {"key": "registrationId", "type": "str"},
"error_code": {"key": "errorCode", "type": "int"},
"error_status": {"key": "errorStatus", "type": "str"},
}
def __init__(
self, *, registration_id: str, error_code: int, error_status: str, **kwargs
) -> None:
super(BulkEnrollmentOperationError, self).__init__(**kwargs)
self.registration_id = registration_id
self.error_code = error_code
self.error_status = error_status

Просмотреть файл

@ -0,0 +1,33 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class BulkEnrollmentOperation(Model):
"""Bulk operation.
All required parameters must be populated in order to send to Azure.
:param enrollments: Required. Enrollment items
:type enrollments: list[~protocol.models.IndividualEnrollment]
:param mode: Required. Operation mode. Possible values include: 'create',
'update', 'updateIfMatchETag', 'delete'
:type mode: str or ~protocol.models.enum
"""
_validation = {"enrollments": {"required": True}, "mode": {"required": True}}
_attribute_map = {
"enrollments": {"key": "enrollments", "type": "[IndividualEnrollment]"},
"mode": {"key": "mode", "type": "str"},
}
def __init__(self, *, enrollments, mode, **kwargs) -> None:
super(BulkEnrollmentOperation, self).__init__(**kwargs)
self.enrollments = enrollments
self.mode = mode

Просмотреть файл

@ -0,0 +1,33 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class BulkEnrollmentOperationResult(Model):
"""Results of a bulk enrollment operation.
All required parameters must be populated in order to send to Azure.
:param is_successful: Required. Indicates if the operation was successful
in its entirety
:type is_successful: bool
:param errors: Registration errors
:type errors: list[~protocol.models.BulkEnrollmentOperationError]
"""
_validation = {"is_successful": {"required": True}}
_attribute_map = {
"is_successful": {"key": "isSuccessful", "type": "bool"},
"errors": {"key": "errors", "type": "[BulkEnrollmentOperationError]"},
}
def __init__(self, **kwargs):
super(BulkEnrollmentOperationResult, self).__init__(**kwargs)
self.is_successful = kwargs.get("is_successful", None)
self.errors = kwargs.get("errors", None)

Просмотреть файл

@ -0,0 +1,33 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class BulkEnrollmentOperationResult(Model):
"""Results of a bulk enrollment operation.
All required parameters must be populated in order to send to Azure.
:param is_successful: Required. Indicates if the operation was successful
in its entirety
:type is_successful: bool
:param errors: Registration errors
:type errors: list[~protocol.models.BulkEnrollmentOperationError]
"""
_validation = {"is_successful": {"required": True}}
_attribute_map = {
"is_successful": {"key": "isSuccessful", "type": "bool"},
"errors": {"key": "errors", "type": "[BulkEnrollmentOperationError]"},
}
def __init__(self, *, is_successful: bool, errors=None, **kwargs) -> None:
super(BulkEnrollmentOperationResult, self).__init__(**kwargs)
self.is_successful = is_successful
self.errors = errors

Просмотреть файл

@ -0,0 +1,35 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class CustomAllocationDefinition(Model):
"""Custom allocation definition.
All required parameters must be populated in order to send to Azure.
:param webhook_url: Required. The webhook URL used for allocation
requests.
:type webhook_url: str
:param api_version: Required. The API version of the provisioning service
types (such as IndividualEnrollment) sent in the custom allocation
request. Supported versions include: "2018-09-01-preview"
:type api_version: str
"""
_validation = {"webhook_url": {"required": True}, "api_version": {"required": True}}
_attribute_map = {
"webhook_url": {"key": "webhookUrl", "type": "str"},
"api_version": {"key": "apiVersion", "type": "str"},
}
def __init__(self, **kwargs):
super(CustomAllocationDefinition, self).__init__(**kwargs)
self.webhook_url = kwargs.get("webhook_url", None)
self.api_version = kwargs.get("api_version", None)

Просмотреть файл

@ -0,0 +1,35 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class CustomAllocationDefinition(Model):
"""Custom allocation definition.
All required parameters must be populated in order to send to Azure.
:param webhook_url: Required. The webhook URL used for allocation
requests.
:type webhook_url: str
:param api_version: Required. The API version of the provisioning service
types (such as IndividualEnrollment) sent in the custom allocation
request. Supported versions include: "2018-09-01-preview"
:type api_version: str
"""
_validation = {"webhook_url": {"required": True}, "api_version": {"required": True}}
_attribute_map = {
"webhook_url": {"key": "webhookUrl", "type": "str"},
"api_version": {"key": "apiVersion", "type": "str"},
}
def __init__(self, *, webhook_url: str, api_version: str, **kwargs) -> None:
super(CustomAllocationDefinition, self).__init__(**kwargs)
self.webhook_url = webhook_url
self.api_version = api_version

Просмотреть файл

@ -0,0 +1,27 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class DeviceCapabilities(Model):
"""Device capabilities.
All required parameters must be populated in order to send to Azure.
:param iot_edge: Required. If set to true, this device is an IoTEdge
device. Default value: False .
:type iot_edge: bool
"""
_validation = {"iot_edge": {"required": True}}
_attribute_map = {"iot_edge": {"key": "iotEdge", "type": "bool"}}
def __init__(self, **kwargs):
super(DeviceCapabilities, self).__init__(**kwargs)
self.iot_edge = kwargs.get("iot_edge", False)

Просмотреть файл

@ -0,0 +1,27 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class DeviceCapabilities(Model):
"""Device capabilities.
All required parameters must be populated in order to send to Azure.
:param iot_edge: Required. If set to true, this device is an IoTEdge
device. Default value: False .
:type iot_edge: bool
"""
_validation = {"iot_edge": {"required": True}}
_attribute_map = {"iot_edge": {"key": "iotEdge", "type": "bool"}}
def __init__(self, *, iot_edge: bool = False, **kwargs) -> None:
super(DeviceCapabilities, self).__init__(**kwargs)
self.iot_edge = iot_edge

Просмотреть файл

@ -0,0 +1,87 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class DeviceRegistrationState(Model):
"""Device registration state.
Variables are only populated by the server, and will be ignored when
sending a request.
:ivar registration_id: The registration ID is alphanumeric, lowercase, and
may contain hyphens.
:vartype registration_id: str
:ivar created_date_time_utc: Registration create date time (in UTC).
:vartype created_date_time_utc: datetime
:ivar assigned_hub: Assigned Azure IoT Hub.
:vartype assigned_hub: str
:ivar device_id: Device ID.
:vartype device_id: str
:ivar status: Enrollment status. Possible values include: 'unassigned',
'assigning', 'assigned', 'failed', 'disabled'
:vartype status: str or ~protocol.models.enum
:ivar substatus: Substatus for 'Assigned' devices. Possible values include
- 'initialAssignment': Device has been assigned to an IoT hub for the
first time, 'deviceDataMigrated': Device has been assigned to a different
IoT hub and its device data was migrated from the previously assigned IoT
hub. Device data was removed from the previously assigned IoT hub,
'deviceDataReset': Device has been assigned to a different IoT hub and
its device data was populated from the initial state stored in the
enrollment. Device data was removed from the previously assigned IoT hub.
Possible values include: 'initialAssignment', 'deviceDataMigrated',
'deviceDataReset'
:vartype substatus: str or ~protocol.models.enum
:ivar error_code: Error code.
:vartype error_code: int
:ivar error_message: Error message.
:vartype error_message: str
:ivar last_updated_date_time_utc: Last updated date time (in UTC).
:vartype last_updated_date_time_utc: datetime
:ivar etag: The entity tag associated with the resource.
:vartype etag: str
"""
_validation = {
"registration_id": {"readonly": True},
"created_date_time_utc": {"readonly": True},
"assigned_hub": {"readonly": True},
"device_id": {"readonly": True},
"status": {"readonly": True},
"substatus": {"readonly": True},
"error_code": {"readonly": True},
"error_message": {"readonly": True},
"last_updated_date_time_utc": {"readonly": True},
"etag": {"readonly": True},
}
_attribute_map = {
"registration_id": {"key": "registrationId", "type": "str"},
"created_date_time_utc": {"key": "createdDateTimeUtc", "type": "iso-8601"},
"assigned_hub": {"key": "assignedHub", "type": "str"},
"device_id": {"key": "deviceId", "type": "str"},
"status": {"key": "status", "type": "str"},
"substatus": {"key": "substatus", "type": "str"},
"error_code": {"key": "errorCode", "type": "int"},
"error_message": {"key": "errorMessage", "type": "str"},
"last_updated_date_time_utc": {"key": "lastUpdatedDateTimeUtc", "type": "iso-8601"},
"etag": {"key": "etag", "type": "str"},
}
def __init__(self, **kwargs):
super(DeviceRegistrationState, self).__init__(**kwargs)
self.registration_id = None
self.created_date_time_utc = None
self.assigned_hub = None
self.device_id = None
self.status = None
self.substatus = None
self.error_code = None
self.error_message = None
self.last_updated_date_time_utc = None
self.etag = None

Просмотреть файл

@ -0,0 +1,87 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class DeviceRegistrationState(Model):
"""Device registration state.
Variables are only populated by the server, and will be ignored when
sending a request.
:ivar registration_id: The registration ID is alphanumeric, lowercase, and
may contain hyphens.
:vartype registration_id: str
:ivar created_date_time_utc: Registration create date time (in UTC).
:vartype created_date_time_utc: datetime
:ivar assigned_hub: Assigned Azure IoT Hub.
:vartype assigned_hub: str
:ivar device_id: Device ID.
:vartype device_id: str
:ivar status: Enrollment status. Possible values include: 'unassigned',
'assigning', 'assigned', 'failed', 'disabled'
:vartype status: str or ~protocol.models.enum
:ivar substatus: Substatus for 'Assigned' devices. Possible values include
- 'initialAssignment': Device has been assigned to an IoT hub for the
first time, 'deviceDataMigrated': Device has been assigned to a different
IoT hub and its device data was migrated from the previously assigned IoT
hub. Device data was removed from the previously assigned IoT hub,
'deviceDataReset': Device has been assigned to a different IoT hub and
its device data was populated from the initial state stored in the
enrollment. Device data was removed from the previously assigned IoT hub.
Possible values include: 'initialAssignment', 'deviceDataMigrated',
'deviceDataReset'
:vartype substatus: str or ~protocol.models.enum
:ivar error_code: Error code.
:vartype error_code: int
:ivar error_message: Error message.
:vartype error_message: str
:ivar last_updated_date_time_utc: Last updated date time (in UTC).
:vartype last_updated_date_time_utc: datetime
:ivar etag: The entity tag associated with the resource.
:vartype etag: str
"""
_validation = {
"registration_id": {"readonly": True},
"created_date_time_utc": {"readonly": True},
"assigned_hub": {"readonly": True},
"device_id": {"readonly": True},
"status": {"readonly": True},
"substatus": {"readonly": True},
"error_code": {"readonly": True},
"error_message": {"readonly": True},
"last_updated_date_time_utc": {"readonly": True},
"etag": {"readonly": True},
}
_attribute_map = {
"registration_id": {"key": "registrationId", "type": "str"},
"created_date_time_utc": {"key": "createdDateTimeUtc", "type": "iso-8601"},
"assigned_hub": {"key": "assignedHub", "type": "str"},
"device_id": {"key": "deviceId", "type": "str"},
"status": {"key": "status", "type": "str"},
"substatus": {"key": "substatus", "type": "str"},
"error_code": {"key": "errorCode", "type": "int"},
"error_message": {"key": "errorMessage", "type": "str"},
"last_updated_date_time_utc": {"key": "lastUpdatedDateTimeUtc", "type": "iso-8601"},
"etag": {"key": "etag", "type": "str"},
}
def __init__(self, **kwargs) -> None:
super(DeviceRegistrationState, self).__init__(**kwargs)
self.registration_id = None
self.created_date_time_utc = None
self.assigned_hub = None
self.device_id = None
self.status = None
self.substatus = None
self.error_code = None
self.error_message = None
self.last_updated_date_time_utc = None
self.etag = None

Просмотреть файл

@ -0,0 +1,102 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class EnrollmentGroup(Model):
"""Enrollment group record.
Variables are only populated by the server, and will be ignored when
sending a request.
All required parameters must be populated in order to send to Azure.
:param enrollment_group_id: Required. Enrollment Group ID.
:type enrollment_group_id: str
:param attestation: Required. Attestation method used by the device.
:type attestation: ~protocol.models.AttestationMechanism
:param iot_hub_host_name: The Iot Hub host name.
:type iot_hub_host_name: str
:param initial_twin: Initial device twin.
:type initial_twin: ~protocol.models.InitialTwin
:param etag: The entity tag associated with the resource.
:type etag: str
:param provisioning_status: The provisioning status. Possible values
include: 'enabled', 'disabled'. Default value: "enabled" .
:type provisioning_status: str or ~protocol.models.enum
:param reprovision_policy: The behavior when a device is re-provisioned to
an IoT hub.
:type reprovision_policy: ~protocol.models.ReprovisionPolicy
:ivar created_date_time_utc: The DateTime this resource was created.
:vartype created_date_time_utc: datetime
:ivar last_updated_date_time_utc: The DateTime this resource was last
updated.
:vartype last_updated_date_time_utc: datetime
:param allocation_policy: The allocation policy of this resource. This
policy overrides the tenant level allocation policy for this individual
enrollment or enrollment group. Possible values include 'hashed': Linked
IoT hubs are equally likely to have devices provisioned to them,
'geoLatency': Devices are provisioned to an IoT hub with the lowest
latency to the device.If multiple linked IoT hubs would provide the same
lowest latency, the provisioning service hashes devices across those hubs,
'static' : Specification of the desired IoT hub in the enrollment list
takes priority over the service-level allocation policy, 'custom': Devices
are provisioned to an IoT hub based on your own custom logic. The
provisioning service passes information about the device to the logic, and
the logic returns the desired IoT hub as well as the desired initial
configuration. We recommend using Azure Functions to host your logic.
Possible values include: 'hashed', 'geoLatency', 'static', 'custom'
:type allocation_policy: str or ~protocol.models.enum
:param iot_hubs: The list of names of IoT hubs the device(s) in this
resource can be allocated to. Must be a subset of tenant level list of IoT
hubs.
:type iot_hubs: list[str]
:param custom_allocation_definition: Custom allocation definition.
:type custom_allocation_definition:
~protocol.models.CustomAllocationDefinition
"""
_validation = {
"enrollment_group_id": {"required": True},
"attestation": {"required": True},
"created_date_time_utc": {"readonly": True},
"last_updated_date_time_utc": {"readonly": True},
}
_attribute_map = {
"enrollment_group_id": {"key": "enrollmentGroupId", "type": "str"},
"attestation": {"key": "attestation", "type": "AttestationMechanism"},
"iot_hub_host_name": {"key": "iotHubHostName", "type": "str"},
"initial_twin": {"key": "initialTwin", "type": "InitialTwin"},
"etag": {"key": "etag", "type": "str"},
"provisioning_status": {"key": "provisioningStatus", "type": "str"},
"reprovision_policy": {"key": "reprovisionPolicy", "type": "ReprovisionPolicy"},
"created_date_time_utc": {"key": "createdDateTimeUtc", "type": "iso-8601"},
"last_updated_date_time_utc": {"key": "lastUpdatedDateTimeUtc", "type": "iso-8601"},
"allocation_policy": {"key": "allocationPolicy", "type": "str"},
"iot_hubs": {"key": "iotHubs", "type": "[str]"},
"custom_allocation_definition": {
"key": "customAllocationDefinition",
"type": "CustomAllocationDefinition",
},
}
def __init__(self, **kwargs):
super(EnrollmentGroup, self).__init__(**kwargs)
self.enrollment_group_id = kwargs.get("enrollment_group_id", None)
self.attestation = kwargs.get("attestation", None)
self.iot_hub_host_name = kwargs.get("iot_hub_host_name", None)
self.initial_twin = kwargs.get("initial_twin", None)
self.etag = kwargs.get("etag", None)
self.provisioning_status = kwargs.get("provisioning_status", "enabled")
self.reprovision_policy = kwargs.get("reprovision_policy", None)
self.created_date_time_utc = None
self.last_updated_date_time_utc = None
self.allocation_policy = kwargs.get("allocation_policy", None)
self.iot_hubs = kwargs.get("iot_hubs", None)
self.custom_allocation_definition = kwargs.get("custom_allocation_definition", None)

Просмотреть файл

@ -0,0 +1,116 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class EnrollmentGroup(Model):
"""Enrollment group record.
Variables are only populated by the server, and will be ignored when
sending a request.
All required parameters must be populated in order to send to Azure.
:param enrollment_group_id: Required. Enrollment Group ID.
:type enrollment_group_id: str
:param attestation: Required. Attestation method used by the device.
:type attestation: ~protocol.models.AttestationMechanism
:param iot_hub_host_name: The Iot Hub host name.
:type iot_hub_host_name: str
:param initial_twin: Initial device twin.
:type initial_twin: ~protocol.models.InitialTwin
:param etag: The entity tag associated with the resource.
:type etag: str
:param provisioning_status: The provisioning status. Possible values
include: 'enabled', 'disabled'. Default value: "enabled" .
:type provisioning_status: str or ~protocol.models.enum
:param reprovision_policy: The behavior when a device is re-provisioned to
an IoT hub.
:type reprovision_policy: ~protocol.models.ReprovisionPolicy
:ivar created_date_time_utc: The DateTime this resource was created.
:vartype created_date_time_utc: datetime
:ivar last_updated_date_time_utc: The DateTime this resource was last
updated.
:vartype last_updated_date_time_utc: datetime
:param allocation_policy: The allocation policy of this resource. This
policy overrides the tenant level allocation policy for this individual
enrollment or enrollment group. Possible values include 'hashed': Linked
IoT hubs are equally likely to have devices provisioned to them,
'geoLatency': Devices are provisioned to an IoT hub with the lowest
latency to the device.If multiple linked IoT hubs would provide the same
lowest latency, the provisioning service hashes devices across those hubs,
'static' : Specification of the desired IoT hub in the enrollment list
takes priority over the service-level allocation policy, 'custom': Devices
are provisioned to an IoT hub based on your own custom logic. The
provisioning service passes information about the device to the logic, and
the logic returns the desired IoT hub as well as the desired initial
configuration. We recommend using Azure Functions to host your logic.
Possible values include: 'hashed', 'geoLatency', 'static', 'custom'
:type allocation_policy: str or ~protocol.models.enum
:param iot_hubs: The list of names of IoT hubs the device(s) in this
resource can be allocated to. Must be a subset of tenant level list of IoT
hubs.
:type iot_hubs: list[str]
:param custom_allocation_definition: Custom allocation definition.
:type custom_allocation_definition:
~protocol.models.CustomAllocationDefinition
"""
_validation = {
"enrollment_group_id": {"required": True},
"attestation": {"required": True},
"created_date_time_utc": {"readonly": True},
"last_updated_date_time_utc": {"readonly": True},
}
_attribute_map = {
"enrollment_group_id": {"key": "enrollmentGroupId", "type": "str"},
"attestation": {"key": "attestation", "type": "AttestationMechanism"},
"iot_hub_host_name": {"key": "iotHubHostName", "type": "str"},
"initial_twin": {"key": "initialTwin", "type": "InitialTwin"},
"etag": {"key": "etag", "type": "str"},
"provisioning_status": {"key": "provisioningStatus", "type": "str"},
"reprovision_policy": {"key": "reprovisionPolicy", "type": "ReprovisionPolicy"},
"created_date_time_utc": {"key": "createdDateTimeUtc", "type": "iso-8601"},
"last_updated_date_time_utc": {"key": "lastUpdatedDateTimeUtc", "type": "iso-8601"},
"allocation_policy": {"key": "allocationPolicy", "type": "str"},
"iot_hubs": {"key": "iotHubs", "type": "[str]"},
"custom_allocation_definition": {
"key": "customAllocationDefinition",
"type": "CustomAllocationDefinition",
},
}
def __init__(
self,
*,
enrollment_group_id: str,
attestation,
iot_hub_host_name: str = None,
initial_twin=None,
etag: str = None,
provisioning_status="enabled",
reprovision_policy=None,
allocation_policy=None,
iot_hubs=None,
custom_allocation_definition=None,
**kwargs
) -> None:
super(EnrollmentGroup, self).__init__(**kwargs)
self.enrollment_group_id = enrollment_group_id
self.attestation = attestation
self.iot_hub_host_name = iot_hub_host_name
self.initial_twin = initial_twin
self.etag = etag
self.provisioning_status = provisioning_status
self.reprovision_policy = reprovision_policy
self.created_date_time_utc = None
self.last_updated_date_time_utc = None
self.allocation_policy = allocation_policy
self.iot_hubs = iot_hubs
self.custom_allocation_definition = custom_allocation_definition

Просмотреть файл

@ -0,0 +1,116 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class IndividualEnrollment(Model):
"""The device enrollment record.
Variables are only populated by the server, and will be ignored when
sending a request.
All required parameters must be populated in order to send to Azure.
:param capabilities: Capabilities of the device
:type capabilities: ~protocol.models.DeviceCapabilities
:param registration_id: Required. The registration ID is alphanumeric,
lowercase, and may contain hyphens.
:type registration_id: str
:param device_id: Desired IoT Hub device ID (optional).
:type device_id: str
:ivar registration_state: Current registration status.
:vartype registration_state: ~protocol.models.DeviceRegistrationState
:param attestation: Required. Attestation method used by the device.
:type attestation: ~protocol.models.AttestationMechanism
:param iot_hub_host_name: The Iot Hub host name.
:type iot_hub_host_name: str
:param initial_twin: Initial device twin.
:type initial_twin: ~protocol.models.InitialTwin
:param etag: The entity tag associated with the resource.
:type etag: str
:param provisioning_status: The provisioning status. Possible values
include: 'enabled', 'disabled'. Default value: "enabled" .
:type provisioning_status: str or ~protocol.models.enum
:param reprovision_policy: The behavior when a device is re-provisioned to
an IoT hub.
:type reprovision_policy: ~protocol.models.ReprovisionPolicy
:ivar created_date_time_utc: The DateTime this resource was created.
:vartype created_date_time_utc: datetime
:ivar last_updated_date_time_utc: The DateTime this resource was last
updated.
:vartype last_updated_date_time_utc: datetime
:param allocation_policy: The allocation policy of this resource. This
policy overrides the tenant level allocation policy for this individual
enrollment or enrollment group. Possible values include 'hashed': Linked
IoT hubs are equally likely to have devices provisioned to them,
'geoLatency': Devices are provisioned to an IoT hub with the lowest
latency to the device.If multiple linked IoT hubs would provide the same
lowest latency, the provisioning service hashes devices across those hubs,
'static' : Specification of the desired IoT hub in the enrollment list
takes priority over the service-level allocation policy, 'custom': Devices
are provisioned to an IoT hub based on your own custom logic. The
provisioning service passes information about the device to the logic, and
the logic returns the desired IoT hub as well as the desired initial
configuration. We recommend using Azure Functions to host your logic.
Possible values include: 'hashed', 'geoLatency', 'static', 'custom'
:type allocation_policy: str or ~protocol.models.enum
:param iot_hubs: The list of names of IoT hubs the device(s) in this
resource can be allocated to. Must be a subset of tenant level list of IoT
hubs.
:type iot_hubs: list[str]
:param custom_allocation_definition: Custom allocation definition.
:type custom_allocation_definition:
~protocol.models.CustomAllocationDefinition
"""
_validation = {
"registration_id": {"required": True},
"registration_state": {"readonly": True},
"attestation": {"required": True},
"created_date_time_utc": {"readonly": True},
"last_updated_date_time_utc": {"readonly": True},
}
_attribute_map = {
"capabilities": {"key": "capabilities", "type": "DeviceCapabilities"},
"registration_id": {"key": "registrationId", "type": "str"},
"device_id": {"key": "deviceId", "type": "str"},
"registration_state": {"key": "registrationState", "type": "DeviceRegistrationState"},
"attestation": {"key": "attestation", "type": "AttestationMechanism"},
"iot_hub_host_name": {"key": "iotHubHostName", "type": "str"},
"initial_twin": {"key": "initialTwin", "type": "InitialTwin"},
"etag": {"key": "etag", "type": "str"},
"provisioning_status": {"key": "provisioningStatus", "type": "str"},
"reprovision_policy": {"key": "reprovisionPolicy", "type": "ReprovisionPolicy"},
"created_date_time_utc": {"key": "createdDateTimeUtc", "type": "iso-8601"},
"last_updated_date_time_utc": {"key": "lastUpdatedDateTimeUtc", "type": "iso-8601"},
"allocation_policy": {"key": "allocationPolicy", "type": "str"},
"iot_hubs": {"key": "iotHubs", "type": "[str]"},
"custom_allocation_definition": {
"key": "customAllocationDefinition",
"type": "CustomAllocationDefinition",
},
}
def __init__(self, **kwargs):
super(IndividualEnrollment, self).__init__(**kwargs)
self.capabilities = kwargs.get("capabilities", None)
self.registration_id = kwargs.get("registration_id", None)
self.device_id = kwargs.get("device_id", None)
self.registration_state = None
self.attestation = kwargs.get("attestation", None)
self.iot_hub_host_name = kwargs.get("iot_hub_host_name", None)
self.initial_twin = kwargs.get("initial_twin", None)
self.etag = kwargs.get("etag", None)
self.provisioning_status = kwargs.get("provisioning_status", "enabled")
self.reprovision_policy = kwargs.get("reprovision_policy", None)
self.created_date_time_utc = None
self.last_updated_date_time_utc = None
self.allocation_policy = kwargs.get("allocation_policy", None)
self.iot_hubs = kwargs.get("iot_hubs", None)
self.custom_allocation_definition = kwargs.get("custom_allocation_definition", None)

Просмотреть файл

@ -0,0 +1,132 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class IndividualEnrollment(Model):
"""The device enrollment record.
Variables are only populated by the server, and will be ignored when
sending a request.
All required parameters must be populated in order to send to Azure.
:param capabilities: Capabilities of the device
:type capabilities: ~protocol.models.DeviceCapabilities
:param registration_id: Required. The registration ID is alphanumeric,
lowercase, and may contain hyphens.
:type registration_id: str
:param device_id: Desired IoT Hub device ID (optional).
:type device_id: str
:ivar registration_state: Current registration status.
:vartype registration_state: ~protocol.models.DeviceRegistrationState
:param attestation: Required. Attestation method used by the device.
:type attestation: ~protocol.models.AttestationMechanism
:param iot_hub_host_name: The Iot Hub host name.
:type iot_hub_host_name: str
:param initial_twin: Initial device twin.
:type initial_twin: ~protocol.models.InitialTwin
:param etag: The entity tag associated with the resource.
:type etag: str
:param provisioning_status: The provisioning status. Possible values
include: 'enabled', 'disabled'. Default value: "enabled" .
:type provisioning_status: str or ~protocol.models.enum
:param reprovision_policy: The behavior when a device is re-provisioned to
an IoT hub.
:type reprovision_policy: ~protocol.models.ReprovisionPolicy
:ivar created_date_time_utc: The DateTime this resource was created.
:vartype created_date_time_utc: datetime
:ivar last_updated_date_time_utc: The DateTime this resource was last
updated.
:vartype last_updated_date_time_utc: datetime
:param allocation_policy: The allocation policy of this resource. This
policy overrides the tenant level allocation policy for this individual
enrollment or enrollment group. Possible values include 'hashed': Linked
IoT hubs are equally likely to have devices provisioned to them,
'geoLatency': Devices are provisioned to an IoT hub with the lowest
latency to the device.If multiple linked IoT hubs would provide the same
lowest latency, the provisioning service hashes devices across those hubs,
'static' : Specification of the desired IoT hub in the enrollment list
takes priority over the service-level allocation policy, 'custom': Devices
are provisioned to an IoT hub based on your own custom logic. The
provisioning service passes information about the device to the logic, and
the logic returns the desired IoT hub as well as the desired initial
configuration. We recommend using Azure Functions to host your logic.
Possible values include: 'hashed', 'geoLatency', 'static', 'custom'
:type allocation_policy: str or ~protocol.models.enum
:param iot_hubs: The list of names of IoT hubs the device(s) in this
resource can be allocated to. Must be a subset of tenant level list of IoT
hubs.
:type iot_hubs: list[str]
:param custom_allocation_definition: Custom allocation definition.
:type custom_allocation_definition:
~protocol.models.CustomAllocationDefinition
"""
_validation = {
"registration_id": {"required": True},
"registration_state": {"readonly": True},
"attestation": {"required": True},
"created_date_time_utc": {"readonly": True},
"last_updated_date_time_utc": {"readonly": True},
}
_attribute_map = {
"capabilities": {"key": "capabilities", "type": "DeviceCapabilities"},
"registration_id": {"key": "registrationId", "type": "str"},
"device_id": {"key": "deviceId", "type": "str"},
"registration_state": {"key": "registrationState", "type": "DeviceRegistrationState"},
"attestation": {"key": "attestation", "type": "AttestationMechanism"},
"iot_hub_host_name": {"key": "iotHubHostName", "type": "str"},
"initial_twin": {"key": "initialTwin", "type": "InitialTwin"},
"etag": {"key": "etag", "type": "str"},
"provisioning_status": {"key": "provisioningStatus", "type": "str"},
"reprovision_policy": {"key": "reprovisionPolicy", "type": "ReprovisionPolicy"},
"created_date_time_utc": {"key": "createdDateTimeUtc", "type": "iso-8601"},
"last_updated_date_time_utc": {"key": "lastUpdatedDateTimeUtc", "type": "iso-8601"},
"allocation_policy": {"key": "allocationPolicy", "type": "str"},
"iot_hubs": {"key": "iotHubs", "type": "[str]"},
"custom_allocation_definition": {
"key": "customAllocationDefinition",
"type": "CustomAllocationDefinition",
},
}
def __init__(
self,
*,
registration_id: str,
attestation,
capabilities=None,
device_id: str = None,
iot_hub_host_name: str = None,
initial_twin=None,
etag: str = None,
provisioning_status="enabled",
reprovision_policy=None,
allocation_policy=None,
iot_hubs=None,
custom_allocation_definition=None,
**kwargs
) -> None:
super(IndividualEnrollment, self).__init__(**kwargs)
self.capabilities = capabilities
self.registration_id = registration_id
self.device_id = device_id
self.registration_state = None
self.attestation = attestation
self.iot_hub_host_name = iot_hub_host_name
self.initial_twin = initial_twin
self.etag = etag
self.provisioning_status = provisioning_status
self.reprovision_policy = reprovision_policy
self.created_date_time_utc = None
self.last_updated_date_time_utc = None
self.allocation_policy = allocation_policy
self.iot_hubs = iot_hubs
self.custom_allocation_definition = custom_allocation_definition

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class InitialTwin(Model):
"""Initial device twin. Contains a subset of the properties of Twin.
:param tags: Twin tags.
:type tags: ~protocol.models.TwinCollection
:param properties: Twin desired properties.
:type properties: ~protocol.models.InitialTwinProperties
"""
_attribute_map = {
"tags": {"key": "tags", "type": "TwinCollection"},
"properties": {"key": "properties", "type": "InitialTwinProperties"},
}
def __init__(self, **kwargs):
super(InitialTwin, self).__init__(**kwargs)
self.tags = kwargs.get("tags", None)
self.properties = kwargs.get("properties", None)

Просмотреть файл

@ -0,0 +1,22 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class InitialTwinProperties(Model):
"""Represents the initial properties that will be set on the device twin.
:param desired: Gets and sets the InitialTwin desired properties.
:type desired: ~protocol.models.TwinCollection
"""
_attribute_map = {"desired": {"key": "desired", "type": "TwinCollection"}}
def __init__(self, **kwargs):
super(InitialTwinProperties, self).__init__(**kwargs)
self.desired = kwargs.get("desired", None)

Просмотреть файл

@ -0,0 +1,22 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class InitialTwinProperties(Model):
"""Represents the initial properties that will be set on the device twin.
:param desired: Gets and sets the InitialTwin desired properties.
:type desired: ~protocol.models.TwinCollection
"""
_attribute_map = {"desired": {"key": "desired", "type": "TwinCollection"}}
def __init__(self, *, desired=None, **kwargs) -> None:
super(InitialTwinProperties, self).__init__(**kwargs)
self.desired = desired

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class InitialTwin(Model):
"""Initial device twin. Contains a subset of the properties of Twin.
:param tags: Twin tags.
:type tags: ~protocol.models.TwinCollection
:param properties: Twin desired properties.
:type properties: ~protocol.models.InitialTwinProperties
"""
_attribute_map = {
"tags": {"key": "tags", "type": "TwinCollection"},
"properties": {"key": "properties", "type": "InitialTwinProperties"},
}
def __init__(self, *, tags=None, properties=None, **kwargs) -> None:
super(InitialTwin, self).__init__(**kwargs)
self.tags = tags
self.properties = properties

Просмотреть файл

@ -0,0 +1,29 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class Metadata(Model):
"""Metadata for the TwinCollection.
:param last_updated: Last time the TwinCollection was updated
:type last_updated: datetime
:param last_updated_version: This SHOULD be null for Reported properties
metadata and MUST not be null for Desired properties metadata.
:type last_updated_version: long
"""
_attribute_map = {
"last_updated": {"key": "lastUpdated", "type": "iso-8601"},
"last_updated_version": {"key": "lastUpdatedVersion", "type": "long"},
}
def __init__(self, **kwargs):
super(Metadata, self).__init__(**kwargs)
self.last_updated = kwargs.get("last_updated", None)
self.last_updated_version = kwargs.get("last_updated_version", None)

Просмотреть файл

@ -0,0 +1,29 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class Metadata(Model):
"""Metadata for the TwinCollection.
:param last_updated: Last time the TwinCollection was updated
:type last_updated: datetime
:param last_updated_version: This SHOULD be null for Reported properties
metadata and MUST not be null for Desired properties metadata.
:type last_updated_version: long
"""
_attribute_map = {
"last_updated": {"key": "lastUpdated", "type": "iso-8601"},
"last_updated_version": {"key": "lastUpdatedVersion", "type": "long"},
}
def __init__(self, *, last_updated=None, last_updated_version: int = None, **kwargs) -> None:
super(Metadata, self).__init__(**kwargs)
self.last_updated = last_updated
self.last_updated_version = last_updated_version

Просмотреть файл

@ -0,0 +1,56 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
from msrest.exceptions import HttpOperationError
class ProvisioningServiceErrorDetails(Model):
"""Contains the properties of an error returned by the Azure IoT Hub
Provisioning Service .
:param error_code:
:type error_code: int
:param tracking_id:
:type tracking_id: str
:param message:
:type message: str
:param info:
:type info: dict[str, str]
:param timestamp_utc:
:type timestamp_utc: datetime
"""
_attribute_map = {
"error_code": {"key": "errorCode", "type": "int"},
"tracking_id": {"key": "trackingId", "type": "str"},
"message": {"key": "message", "type": "str"},
"info": {"key": "info", "type": "{str}"},
"timestamp_utc": {"key": "timestampUtc", "type": "iso-8601"},
}
def __init__(self, **kwargs):
super(ProvisioningServiceErrorDetails, self).__init__(**kwargs)
self.error_code = kwargs.get("error_code", None)
self.tracking_id = kwargs.get("tracking_id", None)
self.message = kwargs.get("message", None)
self.info = kwargs.get("info", None)
self.timestamp_utc = kwargs.get("timestamp_utc", None)
class ProvisioningServiceErrorDetailsException(HttpOperationError):
"""Server responsed with exception of type: 'ProvisioningServiceErrorDetails'.
:param deserialize: A deserializer
:param response: Server response to be deserialized.
"""
def __init__(self, deserialize, response, *args):
super(ProvisioningServiceErrorDetailsException, self).__init__(
deserialize, response, "ProvisioningServiceErrorDetails", *args
)

Просмотреть файл

@ -0,0 +1,65 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
from msrest.exceptions import HttpOperationError
class ProvisioningServiceErrorDetails(Model):
"""Contains the properties of an error returned by the Azure IoT Hub
Provisioning Service .
:param error_code:
:type error_code: int
:param tracking_id:
:type tracking_id: str
:param message:
:type message: str
:param info:
:type info: dict[str, str]
:param timestamp_utc:
:type timestamp_utc: datetime
"""
_attribute_map = {
"error_code": {"key": "errorCode", "type": "int"},
"tracking_id": {"key": "trackingId", "type": "str"},
"message": {"key": "message", "type": "str"},
"info": {"key": "info", "type": "{str}"},
"timestamp_utc": {"key": "timestampUtc", "type": "iso-8601"},
}
def __init__(
self,
*,
error_code: int = None,
tracking_id: str = None,
message: str = None,
info=None,
timestamp_utc=None,
**kwargs
) -> None:
super(ProvisioningServiceErrorDetails, self).__init__(**kwargs)
self.error_code = error_code
self.tracking_id = tracking_id
self.message = message
self.info = info
self.timestamp_utc = timestamp_utc
class ProvisioningServiceErrorDetailsException(HttpOperationError):
"""Server responsed with exception of type: 'ProvisioningServiceErrorDetails'.
:param deserialize: A deserializer
:param response: Server response to be deserialized.
"""
def __init__(self, deserialize, response, *args):
super(ProvisioningServiceErrorDetailsException, self).__init__(
deserialize, response, "ProvisioningServiceErrorDetails", *args
)

Просмотреть файл

@ -0,0 +1,26 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class QuerySpecification(Model):
"""A Json query request.
All required parameters must be populated in order to send to Azure.
:param query: Required. The query.
:type query: str
"""
_validation = {"query": {"required": True}}
_attribute_map = {"query": {"key": "query", "type": "str"}}
def __init__(self, **kwargs):
super(QuerySpecification, self).__init__(**kwargs)
self.query = kwargs.get("query", None)

Просмотреть файл

@ -0,0 +1,26 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class QuerySpecification(Model):
"""A Json query request.
All required parameters must be populated in order to send to Azure.
:param query: Required. The query.
:type query: str
"""
_validation = {"query": {"required": True}}
_attribute_map = {"query": {"key": "query", "type": "str"}}
def __init__(self, *, query: str, **kwargs) -> None:
super(QuerySpecification, self).__init__(**kwargs)
self.query = query

Просмотреть файл

@ -0,0 +1,44 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class ReprovisionPolicy(Model):
"""The behavior of the service when a device is re-provisioned to an IoT hub.
All required parameters must be populated in order to send to Azure.
:param update_hub_assignment: Required. When set to true (default), the
Device Provisioning Service will evaluate the device's IoT Hub assignment
and update it if necessary for any provisioning requests beyond the first
from a given device. If set to false, the device will stay assigned to its
current IoT hub. Default value: True .
:type update_hub_assignment: bool
:param migrate_device_data: Required. When set to true (default), the
Device Provisioning Service will migrate the device's data (twin, device
capabilities, and device ID) from one IoT hub to another during an IoT hub
assignment update. If set to false, the Device Provisioning Service will
reset the device's data to the initial desired configuration stored in the
corresponding enrollment list. Default value: True .
:type migrate_device_data: bool
"""
_validation = {
"update_hub_assignment": {"required": True},
"migrate_device_data": {"required": True},
}
_attribute_map = {
"update_hub_assignment": {"key": "updateHubAssignment", "type": "bool"},
"migrate_device_data": {"key": "migrateDeviceData", "type": "bool"},
}
def __init__(self, **kwargs):
super(ReprovisionPolicy, self).__init__(**kwargs)
self.update_hub_assignment = kwargs.get("update_hub_assignment", True)
self.migrate_device_data = kwargs.get("migrate_device_data", True)

Просмотреть файл

@ -0,0 +1,46 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class ReprovisionPolicy(Model):
"""The behavior of the service when a device is re-provisioned to an IoT hub.
All required parameters must be populated in order to send to Azure.
:param update_hub_assignment: Required. When set to true (default), the
Device Provisioning Service will evaluate the device's IoT Hub assignment
and update it if necessary for any provisioning requests beyond the first
from a given device. If set to false, the device will stay assigned to its
current IoT hub. Default value: True .
:type update_hub_assignment: bool
:param migrate_device_data: Required. When set to true (default), the
Device Provisioning Service will migrate the device's data (twin, device
capabilities, and device ID) from one IoT hub to another during an IoT hub
assignment update. If set to false, the Device Provisioning Service will
reset the device's data to the initial desired configuration stored in the
corresponding enrollment list. Default value: True .
:type migrate_device_data: bool
"""
_validation = {
"update_hub_assignment": {"required": True},
"migrate_device_data": {"required": True},
}
_attribute_map = {
"update_hub_assignment": {"key": "updateHubAssignment", "type": "bool"},
"migrate_device_data": {"key": "migrateDeviceData", "type": "bool"},
}
def __init__(
self, *, update_hub_assignment: bool = True, migrate_device_data: bool = True, **kwargs
) -> None:
super(ReprovisionPolicy, self).__init__(**kwargs)
self.update_hub_assignment = update_hub_assignment
self.migrate_device_data = migrate_device_data

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class SymmetricKeyAttestation(Model):
"""Attestation via SymmetricKey.
:param primary_key: Primary symmetric key.
:type primary_key: str
:param secondary_key: Secondary symmetric key.
:type secondary_key: str
"""
_attribute_map = {
"primary_key": {"key": "primaryKey", "type": "str"},
"secondary_key": {"key": "secondaryKey", "type": "str"},
}
def __init__(self, **kwargs):
super(SymmetricKeyAttestation, self).__init__(**kwargs)
self.primary_key = kwargs.get("primary_key", None)
self.secondary_key = kwargs.get("secondary_key", None)

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class SymmetricKeyAttestation(Model):
"""Attestation via SymmetricKey.
:param primary_key: Primary symmetric key.
:type primary_key: str
:param secondary_key: Secondary symmetric key.
:type secondary_key: str
"""
_attribute_map = {
"primary_key": {"key": "primaryKey", "type": "str"},
"secondary_key": {"key": "secondaryKey", "type": "str"},
}
def __init__(self, *, primary_key: str = None, secondary_key: str = None, **kwargs) -> None:
super(SymmetricKeyAttestation, self).__init__(**kwargs)
self.primary_key = primary_key
self.secondary_key = secondary_key

Просмотреть файл

@ -0,0 +1,32 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class TpmAttestation(Model):
"""Attestation via TPM.
All required parameters must be populated in order to send to Azure.
:param endorsement_key: Required.
:type endorsement_key: str
:param storage_root_key:
:type storage_root_key: str
"""
_validation = {"endorsement_key": {"required": True}}
_attribute_map = {
"endorsement_key": {"key": "endorsementKey", "type": "str"},
"storage_root_key": {"key": "storageRootKey", "type": "str"},
}
def __init__(self, **kwargs):
super(TpmAttestation, self).__init__(**kwargs)
self.endorsement_key = kwargs.get("endorsement_key", None)
self.storage_root_key = kwargs.get("storage_root_key", None)

Просмотреть файл

@ -0,0 +1,32 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class TpmAttestation(Model):
"""Attestation via TPM.
All required parameters must be populated in order to send to Azure.
:param endorsement_key: Required.
:type endorsement_key: str
:param storage_root_key:
:type storage_root_key: str
"""
_validation = {"endorsement_key": {"required": True}}
_attribute_map = {
"endorsement_key": {"key": "endorsementKey", "type": "str"},
"storage_root_key": {"key": "storageRootKey", "type": "str"},
}
def __init__(self, *, endorsement_key: str, storage_root_key: str = None, **kwargs) -> None:
super(TpmAttestation, self).__init__(**kwargs)
self.endorsement_key = endorsement_key
self.storage_root_key = storage_root_key

Просмотреть файл

@ -0,0 +1,37 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class TwinCollection(Model):
"""Represents a collection of properties within a Twin.
:param additional_properties: Unmatched properties from the message are
deserialized this collection
:type additional_properties: dict[str, object]
:param version: Version of the TwinCollection
:type version: long
:param count: Number of properties in the TwinCollection
:type count: int
:param metadata: Metadata for the TwinCollection
:type metadata: ~protocol.models.Metadata
"""
_attribute_map = {
"additional_properties": {"key": "", "type": "{object}"},
"version": {"key": "version", "type": "long"},
"count": {"key": "count", "type": "int"},
"metadata": {"key": "metadata", "type": "Metadata"},
}
def __init__(self, **kwargs):
super(TwinCollection, self).__init__(**kwargs)
self.additional_properties = kwargs.get("additional_properties", None)
self.version = kwargs.get("version", None)
self.count = kwargs.get("count", None)
self.metadata = kwargs.get("metadata", None)

Просмотреть файл

@ -0,0 +1,45 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class TwinCollection(Model):
"""Represents a collection of properties within a Twin.
:param additional_properties: Unmatched properties from the message are
deserialized this collection
:type additional_properties: dict[str, object]
:param version: Version of the TwinCollection
:type version: long
:param count: Number of properties in the TwinCollection
:type count: int
:param metadata: Metadata for the TwinCollection
:type metadata: ~protocol.models.Metadata
"""
_attribute_map = {
"additional_properties": {"key": "", "type": "{object}"},
"version": {"key": "version", "type": "long"},
"count": {"key": "count", "type": "int"},
"metadata": {"key": "metadata", "type": "Metadata"},
}
def __init__(
self,
*,
additional_properties=None,
version: int = None,
count: int = None,
metadata=None,
**kwargs
) -> None:
super(TwinCollection, self).__init__(**kwargs)
self.additional_properties = additional_properties
self.version = version
self.count = count
self.metadata = metadata

Просмотреть файл

@ -0,0 +1,32 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509Attestation(Model):
"""Attestation via X509.
:param client_certificates:
:type client_certificates: ~protocol.models.X509Certificates
:param signing_certificates:
:type signing_certificates: ~protocol.models.X509Certificates
:param ca_references:
:type ca_references: ~protocol.models.X509CAReferences
"""
_attribute_map = {
"client_certificates": {"key": "clientCertificates", "type": "X509Certificates"},
"signing_certificates": {"key": "signingCertificates", "type": "X509Certificates"},
"ca_references": {"key": "caReferences", "type": "X509CAReferences"},
}
def __init__(self, **kwargs):
super(X509Attestation, self).__init__(**kwargs)
self.client_certificates = kwargs.get("client_certificates", None)
self.signing_certificates = kwargs.get("signing_certificates", None)
self.ca_references = kwargs.get("ca_references", None)

Просмотреть файл

@ -0,0 +1,34 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509Attestation(Model):
"""Attestation via X509.
:param client_certificates:
:type client_certificates: ~protocol.models.X509Certificates
:param signing_certificates:
:type signing_certificates: ~protocol.models.X509Certificates
:param ca_references:
:type ca_references: ~protocol.models.X509CAReferences
"""
_attribute_map = {
"client_certificates": {"key": "clientCertificates", "type": "X509Certificates"},
"signing_certificates": {"key": "signingCertificates", "type": "X509Certificates"},
"ca_references": {"key": "caReferences", "type": "X509CAReferences"},
}
def __init__(
self, *, client_certificates=None, signing_certificates=None, ca_references=None, **kwargs
) -> None:
super(X509Attestation, self).__init__(**kwargs)
self.client_certificates = client_certificates
self.signing_certificates = signing_certificates
self.ca_references = ca_references

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509CAReferences(Model):
"""Primary and secondary CA references.
:param primary:
:type primary: str
:param secondary:
:type secondary: str
"""
_attribute_map = {
"primary": {"key": "primary", "type": "str"},
"secondary": {"key": "secondary", "type": "str"},
}
def __init__(self, **kwargs):
super(X509CAReferences, self).__init__(**kwargs)
self.primary = kwargs.get("primary", None)
self.secondary = kwargs.get("secondary", None)

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509CAReferences(Model):
"""Primary and secondary CA references.
:param primary:
:type primary: str
:param secondary:
:type secondary: str
"""
_attribute_map = {
"primary": {"key": "primary", "type": "str"},
"secondary": {"key": "secondary", "type": "str"},
}
def __init__(self, *, primary: str = None, secondary: str = None, **kwargs) -> None:
super(X509CAReferences, self).__init__(**kwargs)
self.primary = primary
self.secondary = secondary

Просмотреть файл

@ -0,0 +1,65 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509CertificateInfo(Model):
"""X509 certificate info.
All required parameters must be populated in order to send to Azure.
:param subject_name: Required.
:type subject_name: str
:param sha1_thumbprint: Required.
:type sha1_thumbprint: str
:param sha256_thumbprint: Required.
:type sha256_thumbprint: str
:param issuer_name: Required.
:type issuer_name: str
:param not_before_utc: Required.
:type not_before_utc: datetime
:param not_after_utc: Required.
:type not_after_utc: datetime
:param serial_number: Required.
:type serial_number: str
:param version: Required.
:type version: int
"""
_validation = {
"subject_name": {"required": True},
"sha1_thumbprint": {"required": True},
"sha256_thumbprint": {"required": True},
"issuer_name": {"required": True},
"not_before_utc": {"required": True},
"not_after_utc": {"required": True},
"serial_number": {"required": True},
"version": {"required": True},
}
_attribute_map = {
"subject_name": {"key": "subjectName", "type": "str"},
"sha1_thumbprint": {"key": "sha1Thumbprint", "type": "str"},
"sha256_thumbprint": {"key": "sha256Thumbprint", "type": "str"},
"issuer_name": {"key": "issuerName", "type": "str"},
"not_before_utc": {"key": "notBeforeUtc", "type": "iso-8601"},
"not_after_utc": {"key": "notAfterUtc", "type": "iso-8601"},
"serial_number": {"key": "serialNumber", "type": "str"},
"version": {"key": "version", "type": "int"},
}
def __init__(self, **kwargs):
super(X509CertificateInfo, self).__init__(**kwargs)
self.subject_name = kwargs.get("subject_name", None)
self.sha1_thumbprint = kwargs.get("sha1_thumbprint", None)
self.sha256_thumbprint = kwargs.get("sha256_thumbprint", None)
self.issuer_name = kwargs.get("issuer_name", None)
self.not_before_utc = kwargs.get("not_before_utc", None)
self.not_after_utc = kwargs.get("not_after_utc", None)
self.serial_number = kwargs.get("serial_number", None)
self.version = kwargs.get("version", None)

Просмотреть файл

@ -0,0 +1,77 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509CertificateInfo(Model):
"""X509 certificate info.
All required parameters must be populated in order to send to Azure.
:param subject_name: Required.
:type subject_name: str
:param sha1_thumbprint: Required.
:type sha1_thumbprint: str
:param sha256_thumbprint: Required.
:type sha256_thumbprint: str
:param issuer_name: Required.
:type issuer_name: str
:param not_before_utc: Required.
:type not_before_utc: datetime
:param not_after_utc: Required.
:type not_after_utc: datetime
:param serial_number: Required.
:type serial_number: str
:param version: Required.
:type version: int
"""
_validation = {
"subject_name": {"required": True},
"sha1_thumbprint": {"required": True},
"sha256_thumbprint": {"required": True},
"issuer_name": {"required": True},
"not_before_utc": {"required": True},
"not_after_utc": {"required": True},
"serial_number": {"required": True},
"version": {"required": True},
}
_attribute_map = {
"subject_name": {"key": "subjectName", "type": "str"},
"sha1_thumbprint": {"key": "sha1Thumbprint", "type": "str"},
"sha256_thumbprint": {"key": "sha256Thumbprint", "type": "str"},
"issuer_name": {"key": "issuerName", "type": "str"},
"not_before_utc": {"key": "notBeforeUtc", "type": "iso-8601"},
"not_after_utc": {"key": "notAfterUtc", "type": "iso-8601"},
"serial_number": {"key": "serialNumber", "type": "str"},
"version": {"key": "version", "type": "int"},
}
def __init__(
self,
*,
subject_name: str,
sha1_thumbprint: str,
sha256_thumbprint: str,
issuer_name: str,
not_before_utc,
not_after_utc,
serial_number: str,
version: int,
**kwargs
) -> None:
super(X509CertificateInfo, self).__init__(**kwargs)
self.subject_name = subject_name
self.sha1_thumbprint = sha1_thumbprint
self.sha256_thumbprint = sha256_thumbprint
self.issuer_name = issuer_name
self.not_before_utc = not_before_utc
self.not_after_utc = not_after_utc
self.serial_number = serial_number
self.version = version

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509CertificateWithInfo(Model):
"""Certificate and Certificate info.
:param certificate:
:type certificate: str
:param info:
:type info: ~protocol.models.X509CertificateInfo
"""
_attribute_map = {
"certificate": {"key": "certificate", "type": "str"},
"info": {"key": "info", "type": "X509CertificateInfo"},
}
def __init__(self, **kwargs):
super(X509CertificateWithInfo, self).__init__(**kwargs)
self.certificate = kwargs.get("certificate", None)
self.info = kwargs.get("info", None)

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509CertificateWithInfo(Model):
"""Certificate and Certificate info.
:param certificate:
:type certificate: str
:param info:
:type info: ~protocol.models.X509CertificateInfo
"""
_attribute_map = {
"certificate": {"key": "certificate", "type": "str"},
"info": {"key": "info", "type": "X509CertificateInfo"},
}
def __init__(self, *, certificate: str = None, info=None, **kwargs) -> None:
super(X509CertificateWithInfo, self).__init__(**kwargs)
self.certificate = certificate
self.info = info

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509Certificates(Model):
"""Primary and secondary certificates.
:param primary:
:type primary: ~protocol.models.X509CertificateWithInfo
:param secondary:
:type secondary: ~protocol.models.X509CertificateWithInfo
"""
_attribute_map = {
"primary": {"key": "primary", "type": "X509CertificateWithInfo"},
"secondary": {"key": "secondary", "type": "X509CertificateWithInfo"},
}
def __init__(self, **kwargs):
super(X509Certificates, self).__init__(**kwargs)
self.primary = kwargs.get("primary", None)
self.secondary = kwargs.get("secondary", None)

Просмотреть файл

@ -0,0 +1,28 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.serialization import Model
class X509Certificates(Model):
"""Primary and secondary certificates.
:param primary:
:type primary: ~protocol.models.X509CertificateWithInfo
:param secondary:
:type secondary: ~protocol.models.X509CertificateWithInfo
"""
_attribute_map = {
"primary": {"key": "primary", "type": "X509CertificateWithInfo"},
"secondary": {"key": "secondary", "type": "X509CertificateWithInfo"},
}
def __init__(self, *, primary=None, secondary=None, **kwargs) -> None:
super(X509Certificates, self).__init__(**kwargs)
self.primary = primary
self.secondary = secondary

Просмотреть файл

@ -0,0 +1,897 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
from msrest.service_client import SDKClient
from msrest import Configuration, Serializer, Deserializer
from .version import VERSION
from msrest.pipeline import ClientRawResponse
from . import models
class ProvisioningServiceClientConfiguration(Configuration):
"""Configuration for ProvisioningServiceClient
Note that all parameters used to create this instance are saved as instance
attributes.
:param credentials: Subscription credentials which uniquely identify
client subscription.
:type credentials: None
:param str base_url: Service URL
"""
def __init__(self, credentials, base_url=None):
if credentials is None:
raise ValueError("Parameter 'credentials' must not be None.")
if not base_url:
base_url = "https://localhost"
super(ProvisioningServiceClientConfiguration, self).__init__(base_url)
self.add_user_agent("provisioningserviceclient/{}".format(VERSION))
self.credentials = credentials
class ProvisioningServiceClient(SDKClient):
"""API for service operations with the Azure IoT Hub Device Provisioning Service
:ivar config: Configuration for client.
:vartype config: ProvisioningServiceClientConfiguration
:param credentials: Subscription credentials which uniquely identify
client subscription.
:type credentials: None
:param str base_url: Service URL
"""
def __init__(self, credentials, base_url=None):
self.config = ProvisioningServiceClientConfiguration(credentials, base_url)
super(ProvisioningServiceClient, self).__init__(self.config.credentials, self.config)
client_models = {k: v for k, v in models.__dict__.items() if isinstance(v, type)}
self.api_version = "2018-09-01-preview"
self._serialize = Serializer(client_models)
self._deserialize = Deserializer(client_models)
def get_individual_enrollment(self, id, custom_headers=None, raw=False, **operation_config):
"""Get a device enrollment record.
:param id: Registration ID.
:type id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: IndividualEnrollment or ClientRawResponse if raw=true
:rtype: ~protocol.models.IndividualEnrollment or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.get_individual_enrollment.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.get(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("IndividualEnrollment", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
get_individual_enrollment.metadata = {"url": "/enrollments/{id}"}
def create_or_update_individual_enrollment(
self, id, enrollment, if_match=None, custom_headers=None, raw=False, **operation_config
):
"""Create or update a device enrollment record.
:param id: The registration ID is alphanumeric, lowercase, and may
contain hyphens.
:type id: str
:param enrollment: The device enrollment record.
:type enrollment: ~protocol.models.IndividualEnrollment
:param if_match: The ETag of the enrollment record.
:type if_match: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: IndividualEnrollment or ClientRawResponse if raw=true
:rtype: ~protocol.models.IndividualEnrollment or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.create_or_update_individual_enrollment.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
header_parameters["Content-Type"] = "application/json; charset=utf-8"
if custom_headers:
header_parameters.update(custom_headers)
if if_match is not None:
header_parameters["If-Match"] = self._serialize.header("if_match", if_match, "str")
# Construct body
body_content = self._serialize.body(enrollment, "IndividualEnrollment")
# Construct and send request
request = self._client.put(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("IndividualEnrollment", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
create_or_update_individual_enrollment.metadata = {"url": "/enrollments/{id}"}
def delete_individual_enrollment(
self, id, if_match=None, custom_headers=None, raw=False, **operation_config
):
"""Delete a device enrollment record.
:param id: Registration ID.
:type id: str
:param if_match: The ETag of the enrollment record.
:type if_match: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: None or ClientRawResponse if raw=true
:rtype: None or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.delete_individual_enrollment.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
if custom_headers:
header_parameters.update(custom_headers)
if if_match is not None:
header_parameters["If-Match"] = self._serialize.header("if_match", if_match, "str")
# Construct and send request
request = self._client.delete(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [204]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
return client_raw_response
delete_individual_enrollment.metadata = {"url": "/enrollments/{id}"}
def get_enrollment_group(self, id, custom_headers=None, raw=False, **operation_config):
"""Get a device enrollment group.
:param id: Enrollment group ID.
:type id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: EnrollmentGroup or ClientRawResponse if raw=true
:rtype: ~protocol.models.EnrollmentGroup or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.get_enrollment_group.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.get(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("EnrollmentGroup", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
get_enrollment_group.metadata = {"url": "/enrollmentGroups/{id}"}
def create_or_update_enrollment_group(
self,
id,
enrollment_group,
if_match=None,
custom_headers=None,
raw=False,
**operation_config
):
"""Create or update a device enrollment group.
:param id: Enrollment group ID.
:type id: str
:param enrollment_group: The device enrollment group.
:type enrollment_group: ~protocol.models.EnrollmentGroup
:param if_match: The ETag of the enrollment record.
:type if_match: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: EnrollmentGroup or ClientRawResponse if raw=true
:rtype: ~protocol.models.EnrollmentGroup or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.create_or_update_enrollment_group.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
header_parameters["Content-Type"] = "application/json; charset=utf-8"
if custom_headers:
header_parameters.update(custom_headers)
if if_match is not None:
header_parameters["If-Match"] = self._serialize.header("if_match", if_match, "str")
# Construct body
body_content = self._serialize.body(enrollment_group, "EnrollmentGroup")
# Construct and send request
request = self._client.put(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("EnrollmentGroup", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
create_or_update_enrollment_group.metadata = {"url": "/enrollmentGroups/{id}"}
def delete_enrollment_group(
self, id, if_match=None, custom_headers=None, raw=False, **operation_config
):
"""Delete a device enrollment group.
:param id: Enrollment group ID.
:type id: str
:param if_match: The ETag of the enrollment group record.
:type if_match: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: None or ClientRawResponse if raw=true
:rtype: None or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.delete_enrollment_group.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
if custom_headers:
header_parameters.update(custom_headers)
if if_match is not None:
header_parameters["If-Match"] = self._serialize.header("if_match", if_match, "str")
# Construct and send request
request = self._client.delete(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [204]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
return client_raw_response
delete_enrollment_group.metadata = {"url": "/enrollmentGroups/{id}"}
def get_device_registration_state(self, id, custom_headers=None, raw=False, **operation_config):
"""Gets the device registration state.
:param id: Registration ID.
:type id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: DeviceRegistrationState or ClientRawResponse if raw=true
:rtype: ~protocol.models.DeviceRegistrationState or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.get_device_registration_state.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.get(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("DeviceRegistrationState", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
get_device_registration_state.metadata = {"url": "/registrations/{id}"}
def delete_device_registration_state(
self, id, if_match=None, custom_headers=None, raw=False, **operation_config
):
"""Deletes the device registration.
:param id: Registration ID.
:type id: str
:param if_match: The ETag of the registration status record.
:type if_match: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: None or ClientRawResponse if raw=true
:rtype: None or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.delete_device_registration_state.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
if custom_headers:
header_parameters.update(custom_headers)
if if_match is not None:
header_parameters["If-Match"] = self._serialize.header("if_match", if_match, "str")
# Construct and send request
request = self._client.delete(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [204]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
return client_raw_response
delete_device_registration_state.metadata = {"url": "/registrations/{id}"}
def run_bulk_enrollment_operation(
self, bulk_operation, custom_headers=None, raw=False, **operation_config
):
"""Bulk device enrollment operation.
:param bulk_operation: Bulk operation.
:type bulk_operation: ~protocol.models.BulkEnrollmentOperation
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: BulkEnrollmentOperationResult or ClientRawResponse if
raw=true
:rtype: ~protocol.models.BulkEnrollmentOperationResult or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.run_bulk_enrollment_operation.metadata["url"]
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
header_parameters["Content-Type"] = "application/json; charset=utf-8"
if custom_headers:
header_parameters.update(custom_headers)
# Construct body
body_content = self._serialize.body(bulk_operation, "BulkEnrollmentOperation")
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("BulkEnrollmentOperationResult", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
run_bulk_enrollment_operation.metadata = {"url": "/enrollments"}
def query_individual_enrollments(
self,
query_specification,
x_ms_max_item_count=None,
x_ms_continuation=None,
custom_headers=None,
raw=False,
**operation_config
):
"""Query the device enrollment records.
:param query_specification: The query specification.
:type query_specification: ~protocol.models.QuerySpecification
:param x_ms_max_item_count: pageSize
:type x_ms_max_item_count: int
:param x_ms_continuation: continuation token
:type x_ms_continuation: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: list or ClientRawResponse if raw=true
:rtype: list[~protocol.models.IndividualEnrollment] or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.query_individual_enrollments.metadata["url"]
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
header_parameters["Content-Type"] = "application/json; charset=utf-8"
if custom_headers:
header_parameters.update(custom_headers)
if x_ms_max_item_count is not None:
header_parameters["x-ms-max-item-count"] = self._serialize.header(
"x_ms_max_item_count", x_ms_max_item_count, "int"
)
if x_ms_continuation is not None:
header_parameters["x-ms-continuation"] = self._serialize.header(
"x_ms_continuation", x_ms_continuation, "str"
)
# Construct body
body_content = self._serialize.body(query_specification, "QuerySpecification")
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
header_dict = {}
if response.status_code == 200:
deserialized = self._deserialize("[IndividualEnrollment]", response)
header_dict = {
"x-ms-continuation": "str",
"x-ms-max-item-count": "int",
"x-ms-item-type": "str",
}
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
client_raw_response.add_headers(header_dict)
return client_raw_response
return deserialized
query_individual_enrollments.metadata = {"url": "/enrollments/query"}
def get_individual_enrollment_attestation_mechanism(
self, id, custom_headers=None, raw=False, **operation_config
):
"""Get the attestation mechanism in the device enrollment record.
:param id: Registration ID.
:type id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: AttestationMechanism or ClientRawResponse if raw=true
:rtype: ~protocol.models.AttestationMechanism or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.get_individual_enrollment_attestation_mechanism.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("AttestationMechanism", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
get_individual_enrollment_attestation_mechanism.metadata = {
"url": "/enrollments/{id}/attestationmechanism"
}
def query_enrollment_groups(
self,
query_specification,
x_ms_max_item_count=None,
x_ms_continuation=None,
custom_headers=None,
raw=False,
**operation_config
):
"""Query the device enrollment groups.
:param query_specification: The query specification.
:type query_specification: ~protocol.models.QuerySpecification
:param x_ms_max_item_count: pageSize
:type x_ms_max_item_count: int
:param x_ms_continuation: continuation token
:type x_ms_continuation: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: list or ClientRawResponse if raw=true
:rtype: list[~protocol.models.EnrollmentGroup] or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.query_enrollment_groups.metadata["url"]
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
header_parameters["Content-Type"] = "application/json; charset=utf-8"
if custom_headers:
header_parameters.update(custom_headers)
if x_ms_max_item_count is not None:
header_parameters["x-ms-max-item-count"] = self._serialize.header(
"x_ms_max_item_count", x_ms_max_item_count, "int"
)
if x_ms_continuation is not None:
header_parameters["x-ms-continuation"] = self._serialize.header(
"x_ms_continuation", x_ms_continuation, "str"
)
# Construct body
body_content = self._serialize.body(query_specification, "QuerySpecification")
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
header_dict = {}
if response.status_code == 200:
deserialized = self._deserialize("[EnrollmentGroup]", response)
header_dict = {
"x-ms-continuation": "str",
"x-ms-max-item-count": "int",
"x-ms-item-type": "str",
}
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
client_raw_response.add_headers(header_dict)
return client_raw_response
return deserialized
query_enrollment_groups.metadata = {"url": "/enrollmentGroups/query"}
def get_enrollment_group_attestation_mechanism(
self, id, custom_headers=None, raw=False, **operation_config
):
"""Get the attestation mechanism in the device enrollment group record.
:param id: Enrollment group ID
:type id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: AttestationMechanism or ClientRawResponse if raw=true
:rtype: ~protocol.models.AttestationMechanism or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.get_enrollment_group_attestation_mechanism.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("AttestationMechanism", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
get_enrollment_group_attestation_mechanism.metadata = {
"url": "/enrollmentGroups/{id}/attestationmechanism"
}
def query_device_registration_states(
self, id, custom_headers=None, raw=False, **operation_config
):
"""Gets the registration state of devices in this enrollmentGroup.
:param id: Enrollment group ID.
:type id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: list or ClientRawResponse if raw=true
:rtype: list[~protocol.models.DeviceRegistrationState] or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`ProvisioningServiceErrorDetailsException<protocol.models.ProvisioningServiceErrorDetailsException>`
"""
# Construct URL
url = self.query_device_registration_states.metadata["url"]
path_format_arguments = {"id": self._serialize.url("id", id, "str")}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
query_parameters["api-version"] = self._serialize.query(
"self.api_version", self.api_version, "str"
)
# Construct headers
header_parameters = {}
header_parameters["Accept"] = "application/json"
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.ProvisioningServiceErrorDetailsException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize("[DeviceRegistrationState]", response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
query_device_registration_states.metadata = {"url": "/registrations/{id}/query"}

Просмотреть файл

@ -0,0 +1,8 @@
# coding=utf-8
# --------------------------------------------------------------------------
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is
# regenerated.
# --------------------------------------------------------------------------
VERSION = "2018-09-01-preview"

Просмотреть файл

@ -0,0 +1,23 @@
# Azure IoT Hub Provisioning Service Client SDK
## How to Install
```
pip install azure-iot-provisioning-servicesdk
```
## Feature List
Use this SDK to:
* Manage Individual Enrollments and Enrollment Groups on the Azure IoT Hub Device Provisioning Service with CRUD operations
* Query Individual Enrollments, Enrollment Groups and Device Registration States [COMING SOON]
## User Guides
* Read the [Azure IoT Fundamentals][iot-fundamentals] guide to get an overview of what Azure IoT can do.
* Read the [Azure IoT Hub Device Provisioning Service][dps-doc] guide to understand how to enable zero-touch provisioning to IoT Hubs using this SDK.
## Examples
Please refer to our [sample repository][dps-service-samples] for examples of how to use the Azure IoT Hub Provisioning Device Client SDK.
[iot-fundamentals]: https://docs.microsoft.com/en-us/azure/iot-fundamentals/
[dps-doc]: https://docs.microsoft.com/en-us/azure/iot-dps/
[dps-service-samples]:https://github.com/Azure/azure-iot-sdk-python/tree/master/provisioning_service_client/samples

Просмотреть файл

@ -0,0 +1,2 @@
We do NOT maintain the service.json to generate code in this repository.
In order to generate, please acquire the file and place it in this directory, then run protocol-generator

Просмотреть файл

@ -0,0 +1,24 @@
# DPS Service Client
> see https://aka.ms/autorest
## Getting Started
To build the DPS Service Client, simply install AutoRest in Node.js via `npm` (`npm install -g autorest`) and then run:
> `autorest protocol-generator.md --use=@microsoft.azure/autorest.python@~3.0.56 --add-credentials`
To see additional help and options, run:
> `autorest --help`
For other options on installation see [Installing Autorest](https://aka.ms/autorest/install) on the AutoRest GitHub page
---
## Configuration
The following are the settings for using this API with AutoRest
```yaml
input-file: service.json
python:
namespace: protocol
output-folder: azure/iot/sdk/provisioning/service
```

Просмотреть файл

@ -0,0 +1,80 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import os
import argparse
from azure.iot.sdk.provisioning.service import ProvisioningServiceClient
from azure.iot.sdk.provisioning.service.models import (
BulkEnrollmentOperation,
IndividualEnrollment,
AttestationMechanism,
TpmAttestation,
QuerySpecification,
)
def run_sample(cs, ek):
client = ProvisioningServiceClient(connection_string=cs)
print("Creating Individual Enrollment with TPM Attestation...")
tpm = TpmAttestation(endorsement_key=ek)
am = AttestationMechanism(type="tpm", tpm=tpm)
ie = IndividualEnrollment(registration_id="reg-id", attestation=am)
ie = client.create_or_update_individual_enrollment(
id=ie.registration_id, enrollment=ie
) # returns like a get operation
print("Complete!")
print("Updating Individual Enrollment...")
ie.device_id = "dev-id"
ie = client.create_or_update_individual_enrollment(id=ie.registration_id, enrollment=ie)
print("Complete!")
print("Deleting Individual Enrollment...")
client.delete_individual_enrollment(id=ie.registration_id)
print("Complete!")
print("Running Bulk Operation - Create 10 Individual Enrollments...")
new_enrollments = []
for i in range(0, 10):
new_tpm = TpmAttestation(endorsement_key=ek)
new_am = AttestationMechanism(type="tpm", tpm=tpm)
new_ie = IndividualEnrollment(registration_id=("id-" + str(i)), attestation=new_am)
new_enrollments.append(new_ie)
bulk_op = BulkEnrollmentOperation(enrollments=new_enrollments, mode="create")
client.run_bulk_enrollment_operation(bulk_operation=bulk_op)
print("Complete!")
print("Running Bulk Operation - Delete 10 Individual Enrollments...")
bulk_op.mode = "delete"
client.run_bulk_enrollment_operation(bulk_operation=bulk_op)
print("Complete!")
if __name__ == "__main__":
connection_string_env = "ProvisioningServiceConnectionString"
endorsement_key_env = "ProvisioningTpmEndorsementKey"
parser = argparse.ArgumentParser("Run a Provisioning Service sample")
parser.add_argument(
"--connection_string",
"-cs",
default=os.environ.get(connection_string_env, None),
help="Provisioning Service Connection String. [default: {} environment variable".format(
connection_string_env
),
)
parser.add_argument(
"--endorsement_key",
"-ek",
default=os.environ.get(endorsement_key_env, None),
help="TPM Endorsement Key. [default: {} environment variable]".format(endorsement_key_env),
)
args = parser.parse_args()
try:
run_sample(args.connection_string, args.endorsement_key)
except Exception as e:
print("Error: {}".format(str(e)))

Просмотреть файл

@ -0,0 +1,11 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
[bdist_wheel]
# This flag says to generate wheels that support both Python 2 and Python
# 3. If your code will not run unchanged on both Python 2 and 3, you will
# need to generate separate wheels for each Python version that you
# support.
universal=1

Просмотреть файл

@ -0,0 +1,35 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from setuptools import setup, find_packages
with open("doc/package-readme.md", "r") as fh:
_long_description = fh.read()
setup(
name="azure-iot-provisioning-servicesdk",
version="1.1.0",
description="Microsoft Azure IoT Provisioning Service SDK",
license="MIT License",
url="https://github.com/Azure/azure-iot-sdk-python",
author="Microsoft Corporation",
long_description=_long_description,
long_description_content_type="text/markdown",
classifiers=[
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"Topic :: Software Development :: Build Tools",
"License :: OSI Approved :: MIT Software License",
"Programming Language :: Python :: 2",
"Programming Language :: Python :: 2.7",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
],
install_requires=["msrest", "azure-iot-common"],
packages=find_packages(exclude=["tests"]),
)

Просмотреть файл

@ -0,0 +1,226 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import sys
import os
import copy
import six
import pytest
import e2e_convenience
from azure.iot.provisioning.servicesdk import (
ProvisioningServiceClient,
QuerySpecification,
BulkEnrollmentOperation,
ProvisioningServiceErrorDetailsException,
)
from azure.iot.provisioning.servicesdk.models import (
IndividualEnrollment,
AttestationMechanism,
InitialTwin,
EnrollmentGroup,
DeviceCapabilities,
TwinCollection,
InitialTwinProperties,
)
e2e_convenience._patch_attestation_mechanism()
CONNECTION_STRING = os.environ["CONNECTION_STRING"]
ENDORSEMENT_KEY = os.environ["ENDORSEMENT_KEY"]
SIGNING_CERTIFICATE = os.environ["SIGNING_CERTIFICATE"]
CLIENT_CERTIFICATE = os.environ["CLIENT_CERTIFICATE"]
CA_REFERENCE = os.environ["CA_REFERENCE"]
REGISTRATION_ID = "e2e-test-reg-id"
GROUP_ID = "e2e-test-group-id"
TAGS = {"tag1": "val1"}
DESIRED_PROPERTIES = {"dp1": "val1", "dp2": {"dp3": "val2"}}
CREATE = "create"
DELETE = "delete"
BULK_SIZE = 10
@pytest.fixture(scope="module")
def client():
return ProvisioningServiceClient(CONNECTION_STRING)
@pytest.fixture
def tpm_attestation():
return AttestationMechanism.create_with_tpm(ENDORSEMENT_KEY)
@pytest.fixture
def x509_attestation_client_certs():
return AttestationMechanism.create_with_x509_client_certificates(CLIENT_CERTIFICATE)
@pytest.fixture
def x509_attestation_signing_certs():
return AttestationMechanism.create_with_x509_signing_certificates(SIGNING_CERTIFICATE)
@pytest.fixture
def x509_attestation_ca_refs():
return AttestationMechanism.create_with_x509_ca_references(CA_REFERENCE)
@pytest.fixture(params=[tpm_attestation, x509_attestation_client_certs])
def individual_enrollment(request):
attestation = request.param()
return IndividualEnrollment(registration_id=REGISTRATION_ID, attestation=attestation)
@pytest.fixture(params=[x509_attestation_signing_certs, x509_attestation_ca_refs])
def enrollment_group(request):
attestation = request.param()
return EnrollmentGroup(enrollment_group_id=GROUP_ID, attestation=attestation)
@pytest.fixture
def twin():
tags_collection = TwinCollection(additional_properties=TAGS)
dp_collection = TwinCollection(additional_properties=DESIRED_PROPERTIES)
properties = InitialTwinProperties(desired=dp_collection)
return InitialTwin(tags=tags_collection, properties=properties)
@pytest.fixture
def device_capabilities():
return DeviceCapabilities(iot_edge=True)
@pytest.fixture
def purge_individual_enrollments(client):
"""Delete all individual enrollments from the hub
"""
yield # teardown only
# Get all enrollments from the provisioning hub
enrollments = []
qs = QuerySpecification(query="*")
cont = ""
while cont != None:
qrr = client.query_individual_enrollments(
query_specification=qs, x_ms_continuation=cont, raw=True
)
enrollments.extend(qrr.output)
cont = qrr.headers.get("x-ms-continuation", None)
# delete enrollments
if enrollments:
bulk_op = BulkEnrollmentOperation(enrollments=enrollments, mode=DELETE)
client.run_bulk_enrollment_operation(bulk_op)
@pytest.fixture
def purge_enrollment_groups(client):
"""Delete all enrollment groups from the hub
"""
yield # teardown only
# Get all enrollments from the provisioning hub
enrollments = []
qs = QuerySpecification(query="*")
cont = ""
while cont != None:
qrr = client.query_enrollment_groups(
query_specification=qs, x_ms_continuation=cont, raw=True
)
enrollments.extend(qrr.output)
cont = qrr.headers.get("x-ms-continuation", None)
for enrollment in enrollments:
client.delete_enrollment_group(enrollment.enrollment_group_id)
@pytest.mark.usefixtures("purge_individual_enrollments")
class TestIndividualEnrollment(object):
def test_crud(self, client, individual_enrollment, twin, device_capabilities):
ie = individual_enrollment
# create
ret_ie = client.create_or_update_individual_enrollment(ie.registration_id, ie)
assert ret_ie.registration_id == REGISTRATION_ID
# update
ret_ie.initial_twin = twin
ret_ie.capabilities = device_capabilities
ret_ie = client.create_or_update_individual_enrollment(
ret_ie.registration_id, ret_ie, ret_ie.etag
)
assert ret_ie.registration_id == REGISTRATION_ID
assert ret_ie.initial_twin.tags.additional_properties == TAGS
assert ret_ie.initial_twin.properties.desired.additional_properties == DESIRED_PROPERTIES
assert ret_ie.capabilities.iot_edge == True
# get
ret_ie = client.get_individual_enrollment(REGISTRATION_ID)
assert ret_ie.registration_id == REGISTRATION_ID
assert ret_ie.initial_twin.tags.additional_properties == TAGS
assert ret_ie.initial_twin.properties.desired.additional_properties == DESIRED_PROPERTIES
assert ret_ie.capabilities.iot_edge == True
# delete
client.delete_individual_enrollment(REGISTRATION_ID)
with pytest.raises(ProvisioningServiceErrorDetailsException):
ret_ie = client.get_individual_enrollment(REGISTRATION_ID)
def test_bulk_operation(self, client, individual_enrollment):
# create
enrollments = []
for i in range(BULK_SIZE):
new = copy.copy(individual_enrollment)
new.registration_id = new.registration_id + str(i)
enrollments.append(new)
bulk_op = BulkEnrollmentOperation(enrollments=enrollments, mode=CREATE)
res = client.run_bulk_enrollment_operation(bulk_op)
assert res.is_successful
# delete
bulk_op = BulkEnrollmentOperation(enrollments=enrollments, mode=DELETE)
res = client.run_bulk_enrollment_operation(bulk_op)
assert res.is_successful
def test_query(self):
pass
@pytest.mark.usefixtures("purge_enrollment_groups")
class TestEnrollmentGroup(object):
def test_crud(self, client, enrollment_group, twin):
eg = enrollment_group
# create
ret_eg = client.create_or_update_enrollment_group(eg.enrollment_group_id, eg)
assert ret_eg.enrollment_group_id == GROUP_ID
# update
ret_eg.initial_twin = twin
ret_eg.capabilities = device_capabilities
ret_eg = client.create_or_update_enrollment_group(
ret_eg.enrollment_group_id, ret_eg, ret_eg.etag
)
assert ret_eg.enrollment_group_id == GROUP_ID
assert ret_eg.initial_twin.tags.additional_properties == TAGS
assert ret_eg.initial_twin.properties.desired.additional_properties == DESIRED_PROPERTIES
# get
ret_eg = client.get_enrollment_group(GROUP_ID)
assert ret_eg.enrollment_group_id == GROUP_ID
assert ret_eg.initial_twin.tags.additional_properties == TAGS
assert ret_eg.initial_twin.properties.desired.additional_properties == DESIRED_PROPERTIES
# delete
client.delete_enrollment_group(GROUP_ID)
with pytest.raises(ProvisioningServiceErrorDetailsException):
ret_eg = client.get_enrollment_group(GROUP_ID)
def test_query(self):
pass

Просмотреть файл

@ -0,0 +1,93 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
from azure.iot.provisioning.servicesdk.models import (
TpmAttestation,
X509CertificateWithInfo,
X509Attestation,
X509CAReferences,
AttestationMechanism,
X509Certificates,
)
def _patch_attestation_mechanism():
"""Add convenience methods to Attestation Mechanism for ease of use
"""
def create_with_tpm(cls, endorsement_key, storage_root_key=None):
"""Create an Attestation Mechanism using a TPM Attestation
:param str endorsement_key: Endorsement Key
:param str storage_root_key: Storage Root Key
:return: AttestationMechanism using TPM Attestation
:rtype: ~protocol.models.AttestationMechanism
"""
tpm = TpmAttestation(endorsement_key=endorsement_key, storage_root_key=storage_root_key)
return cls(type="tpm", tpm=tpm)
def _create_x509_certificates(primary_cert, secondary_cert=None):
"""Creates X509Certificates model
"""
primary = X509CertificateWithInfo(certificate=primary_cert)
secondary = None
if secondary_cert:
secondary = X509CertificateWithInfo(certificate=secondary_cert)
return X509Certificates(primary=primary, secondary=secondary)
def create_with_x509_client_certificates(cls, primary, secondary=None):
"""Create an Attestation Mechanism using a X509 Attestation with Client Certificates.
Only valid for IndividualEnrollment
:param str primary: Primary certificate (Base 64 encoded)
:param str secondary: Secondary certificate (Base64 encoded)
:return: AttestationMechanism using X509 Attestation wtih Client Certificates
:rtype: ~protocol.models.AttestationMechanism
"""
certs = _create_x509_certificates(primary, secondary)
x509 = X509Attestation(client_certificates=certs)
return cls(type="x509", x509=x509)
def create_with_x509_signing_certificates(cls, primary, secondary=None):
"""Create an Attestation Mechanism using a X509 Attestation with Signing Certificates.
Only valid for EnrollmentGroup
:param str primary: Primary certificate (Base 64 encoded)
:param str secondary: Secondary certificate (Base64 encoded)
:return: AttestationMechanism using X509 Attestation wtih Signing Certificates
:rtype: ~protocol.models.AttestationMechanism
"""
certs = _create_x509_certificates(primary, secondary)
x509 = X509Attestation(signing_certificates=certs)
return cls(type="x509", x509=x509)
def create_with_x509_ca_references(cls, primary, secondary=None):
"""Create an Attestation Mechanism using a X509 Attestation with CA References
:param str primary: Primary CA Reference
:param str secondary: Secondary CA Reference
:return: AttestationMechanism using X509 Attestation wtih CA References
:rtype: ~protocol.models.AttestationMechanism
"""
ca_refs = X509CAReferences(primary=primary, secondary=secondary)
x509 = X509Attestation(ca_references=ca_refs)
return cls(type="x509", x509=x509)
setattr(AttestationMechanism, "create_with_tpm", classmethod(create_with_tpm))
setattr(
AttestationMechanism,
"create_with_x509_client_certificates",
classmethod(create_with_x509_client_certificates),
)
setattr(
AttestationMechanism,
"create_with_x509_signing_certificates",
classmethod(create_with_x509_signing_certificates),
)
setattr(
AttestationMechanism,
"create_with_x509_ca_references",
classmethod(create_with_x509_ca_references),
)

Просмотреть файл

@ -0,0 +1,70 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import pytest
from pytest_mock import mocker
from azure.iot.provisioning.servicesdk.auth import (
ConnectionStringAuthentication,
HOST_NAME,
SHARED_ACCESS_KEY,
SHARED_ACCESS_KEY_NAME,
)
@pytest.fixture(scope="module")
def hostname():
return "my.host.name"
@pytest.fixture(scope="module")
def keyname():
return "mykeyname"
@pytest.fixture(scope="module")
def key():
return "Zm9vYmFy"
@pytest.fixture(scope="module")
def service_str(hostname, keyname, key):
return "HostName={};SharedAccessKeyName={};SharedAccessKey={}".format(hostname, keyname, key)
@pytest.fixture(scope="module")
def valid_cs_auth(service_str):
return ConnectionStringAuthentication(service_str)
def test___repr__(valid_cs_auth, service_str):
"""Test that a string representation of ConnectionStringAuthentication is the same as the
connection string given to it
"""
assert str(valid_cs_auth) == service_str
def test__getitem__(valid_cs_auth, hostname, keyname, key):
"""Test that __getitem__ syntax works
"""
assert valid_cs_auth[HOST_NAME] == hostname
assert valid_cs_auth[SHARED_ACCESS_KEY_NAME] == keyname
assert valid_cs_auth[SHARED_ACCESS_KEY] == key
with pytest.raises(KeyError):
valid_cs_auth["invalid"]
def test_signed_session(mocker, valid_cs_auth, hostname, keyname, key):
"""Test that a SasToken is created and added to the Authorization header
"""
mock_sas = mocker.patch("azure.iot.provisioning.servicesdk.auth.SasToken", autospec=True)
dummy_token = "DUMMY_SASTOKEN"
mock_sas.return_value.__str__.return_value = (
dummy_token
) # use __str__ instead of __repr__ because __repr__ is NonCallableMock
session = valid_cs_auth.signed_session()
mock_sas.assert_called_once_with(hostname, key, keyname)
assert session.headers["Authorization"] == dummy_token

Просмотреть файл

@ -0,0 +1,66 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import pytest
from pytest_mock import mocker
from azure.iot.provisioning.servicesdk import ProvisioningServiceClient
from azure.iot.provisioning.servicesdk.protocol import (
ProvisioningServiceClient as BaseProvisioningServiceClient,
)
from azure.iot.provisioning.servicesdk.auth import ConnectionStringAuthentication
from azure.iot.provisioning.servicesdk.models import (
IndividualEnrollment,
EnrollmentGroup,
AttestationMechanism,
)
@pytest.fixture(scope="module")
def service_str():
return "HostName=my.host.name;SharedAccessKeyName=mykeyname;SharedAccessKey=Zm9vYmFy"
@pytest.fixture(scope="module")
def service_client(service_str):
return ProvisioningServiceClient(service_str)
@pytest.fixture(scope="module")
def attestation_mechanism():
return AttestationMechanism.create_with_x509_ca_references("my-certificate-name")
@pytest.fixture() # don't scope, so changes aren't saved
def individual_enrollment(attestation_mechanism):
return IndividualEnrollment(
registration_id="registration_id", attestation=attestation_mechanism
)
@pytest.fixture() # don't scope, so changes aren't saved
def enrollment_group(attestation_mechanism):
return EnrollmentGroup(enrollment_group_id="group_id", attestation=attestation_mechanism)
@pytest.fixture(scope="module")
def etag():
return "my-etag"
def test_create(mocker, service_str):
"""Test that instantiation of the application ProvisioningServiceClient creates a ConnectionStringAuthentication from
the provided connection string, and then uses it along with an extracted hostname in the __init__ of the
superclass - the generated ProvisioningServiceClient from .protocol
"""
mock_parent_init = mocker.patch.object(BaseProvisioningServiceClient, "__init__", autospec=True)
auth = ConnectionStringAuthentication(service_str)
mock_auth = mocker.patch(
"azure.iot.provisioning.servicesdk.client.ConnectionStringAuthentication",
return_value=auth,
autospec=True,
)
client = ProvisioningServiceClient(service_str)
mock_auth.assert_called_once_with(service_str)
mock_parent_init.assert_called_once_with(client, mock_auth.return_value, "https://my.host.name")

38
dev_setup.py Normal file
Просмотреть файл

@ -0,0 +1,38 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
"""Prepare development environment
"""
import glob
import os
import sys
from subprocess import check_call, CalledProcessError
COMMON_PKG_NAME = "azure-iot-common"
def pip_command(command, error_ok=False):
try:
print("Executing: " + command)
check_call([sys.executable, "-m", "pip"] + command.split())
print()
except CalledProcessError as err:
print(err)
if not error_ok:
sys.exit(1)
if __name__ == "__main__":
packages = [os.path.dirname(p) for p in glob.glob("azure*/setup.py")]
# Ensure common is installed first
packages.remove(COMMON_PKG_NAME)
packages.insert(0, COMMON_PKG_NAME)
for package_name in packages:
pip_command("install -e {}".format(package_name))
pip_command("install -r requirements.txt")

Некоторые файлы не были показаны из-за слишком большого количества измененных файлов Показать больше