Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def create(self, name, module, template, group_id):
if name == ".":
dir_path = os.getcwd()
else:
dir_path = os.path.join(os.getcwd(), name)
if not self.utility.is_dir_empty(dir_path):
raise ValueError("Directory is not empty. Run `iotedgedev iothub setup` to retrieve or create required Azure resources or clean the directory.")
self.output.header("CREATING AZURE IOT EDGE SOLUTION: {0}".format(name))
self.utility.ensure_dir(dir_path)
self.utility.copy_from_template_dir(Constants.default_deployment_template_file, dir_path, replacements={"%MODULE%": module})
self.utility.copy_from_template_dir(Constants.default_deployment_template_file, dir_path,
dest_file=Constants.default_deployment_debug_template_file, replacements={"%MODULE%": module})
self.utility.copy_from_template_dir(".gitignore", dir_path)
self.utility.copy_from_template_dir(".env.tmp", dir_path, dest_file=".env")
if template == "java":
mod_cmd = "iotedgedev solution add {0} --template {1} --group-id {2}".format(module, template, group_id)
else:
mod_cmd = "iotedgedev solution add {0} --template {1}".format(module, template)
self.output.header(mod_cmd)
self.utility.call_proc(mod_cmd.split(), cwd=name)
self.output.footer("Azure IoT Edge Solution Created")
if name != ".":
self.output.info("Execute 'cd {0}' to navigate to your new solution.".format(name))
def create(self, name, module, template, group_id):
if name == ".":
dir_path = os.getcwd()
else:
dir_path = os.path.join(os.getcwd(), name)
if not self.utility.is_dir_empty(dir_path):
raise ValueError("Directory is not empty. Run `iotedgedev iothub setup` to retrieve or create required Azure resources or clean the directory.")
self.output.header("CREATING AZURE IOT EDGE SOLUTION: {0}".format(name))
self.utility.ensure_dir(dir_path)
self.utility.copy_from_template_dir(Constants.default_deployment_template_file, dir_path, replacements={"%MODULE%": module})
self.utility.copy_from_template_dir(Constants.default_deployment_template_file, dir_path,
dest_file=Constants.default_deployment_debug_template_file, replacements={"%MODULE%": module})
self.utility.copy_from_template_dir(".gitignore", dir_path)
self.utility.copy_from_template_dir(".env.tmp", dir_path, dest_file=".env")
if template == "java":
mod_cmd = "iotedgedev solution add {0} --template {1} --group-id {2}".format(module, template, group_id)
else:
mod_cmd = "iotedgedev solution add {0} --template {1}".format(module, template)
self.output.header(mod_cmd)
self.utility.call_proc(mod_cmd.split(), cwd=name)
self.output.footer("Azure IoT Edge Solution Created")
if name != ".":
self.output.info("Execute 'cd {0}' to navigate to your new solution.".format(name))
def validate_deployment_template(self):
validation_success = True
try:
template_schema = json.loads(urlopen(Constants.deployment_template_schema_url).read().decode())
self._validate_json_schema(template_schema, self.json, "Deployment template")
except Exception as ex: # Ignore other non shcema validation errors
self.output.info("Unexpected error during deployment template schema validation, skip schema validation. Error:%s" % ex)
return validation_success
try:
self.DEVICE_CONNECTION_STRING = self.get_envvar("DEVICE_CONNECTION_STRING")
self.DEVICE_CONNECTION_INFO = None
if self.DEVICE_CONNECTION_STRING:
self.DEVICE_CONNECTION_INFO = DeviceConnectionString(self.DEVICE_CONNECTION_STRING)
except Exception as ex:
raise ValueError("Unable to parse DEVICE_CONNECTION_STRING Environment Variable. Please ensure that you have the right connection string set. {0}".format(str(ex)))
self.get_registries()
self.BYPASS_MODULES = self.get_envvar("BYPASS_MODULES", default="")
self.CONTAINER_TAG = self.get_envvar("CONTAINER_TAG", default="")
self.RUNTIME_TAG = self.get_envvar("RUNTIME_TAG", default="1.0")
self.CONFIG_OUTPUT_DIR = self.get_envvar("CONFIG_OUTPUT_DIR", default=Constants.default_config_folder)
self.DEPLOYMENT_CONFIG_TEMPLATE_FILE = self.get_envvar("DEPLOYMENT_CONFIG_TEMPLATE_FILE", default=Constants.default_deployment_template_file)
self.DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE = self.get_envvar("DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE", default=Constants.default_deployment_debug_template_file)
self.DEFAULT_PLATFORM = self.get_envvar("DEFAULT_PLATFORM", default=Constants.default_platform)
self.DEPLOYMENT_CONFIG_FILE = Utility.get_deployment_manifest_name(self.DEPLOYMENT_CONFIG_TEMPLATE_FILE, None, self.DEFAULT_PLATFORM)
self.MODULES_PATH = self.get_envvar("MODULES_PATH", default=Constants.default_modules_folder)
self.LOGS_PATH = self.get_envvar("LOGS_PATH", default="logs")
self.LOGS_CMD = self.get_envvar("LOGS_CMD", default="start /B start cmd.exe @cmd /k docker logs {0} -f")
self.SUBSCRIPTION_ID = self.get_envvar("SUBSCRIPTION_ID")
self.RESOURCE_GROUP_NAME = self.get_envvar("RESOURCE_GROUP_NAME")
self.RESOURCE_GROUP_LOCATION = self.get_envvar("RESOURCE_GROUP_LOCATION")
self.IOTHUB_NAME = self.get_envvar("IOTHUB_NAME")
self.IOTHUB_SKU = self.get_envvar("IOTHUB_SKU")
self.EDGE_DEVICE_ID = self.get_envvar("EDGE_DEVICE_ID")
self.CREDENTIALS = self.get_envvar("CREDENTIALS")
self.UPDATE_DOTENV = self.get_envvar("UPDATE_DOTENV")
except Exception as ex:
def get_deployment_manifest_name(template_file, template_schema_ver, platform):
if "DEPLOYMENT_CONFIG_FILE" in os.environ:
return os.environ["DEPLOYMENT_CONFIG_FILE"]
if template_schema_ver is None:
if os.path.exists(template_file):
json_ = json.loads(Utility.get_file_contents(template_file, expandvars=True))
template_schema_ver = json_.get("$schema-template", "")
else:
template_schema_ver = Constants.deployment_template_schema_version
platform = "." + platform if template_schema_ver > "0.0.1" else ""
prefix = os.path.basename(template_file)
if prefix.endswith(Constants.deployment_template_suffix):
prefix = prefix[:-len(Constants.deployment_template_suffix)]
elif prefix.endswith(".json"):
prefix = prefix[:-len(".json")]
else:
prefix = "deployment"
return "{0}{1}.json".format(prefix, platform)
self.DEVICE_CONNECTION_INFO = DeviceConnectionString(self.DEVICE_CONNECTION_STRING)
except Exception as ex:
raise ValueError("Unable to parse DEVICE_CONNECTION_STRING Environment Variable. Please ensure that you have the right connection string set. {0}".format(str(ex)))
self.get_registries()
self.BYPASS_MODULES = self.get_envvar("BYPASS_MODULES", default="")
self.CONTAINER_TAG = self.get_envvar("CONTAINER_TAG", default="")
self.RUNTIME_TAG = self.get_envvar("RUNTIME_TAG", default="1.0")
self.CONFIG_OUTPUT_DIR = self.get_envvar("CONFIG_OUTPUT_DIR", default=Constants.default_config_folder)
self.DEPLOYMENT_CONFIG_TEMPLATE_FILE = self.get_envvar("DEPLOYMENT_CONFIG_TEMPLATE_FILE", default=Constants.default_deployment_template_file)
self.DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE = self.get_envvar("DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE", default=Constants.default_deployment_debug_template_file)
self.DEFAULT_PLATFORM = self.get_envvar("DEFAULT_PLATFORM", default=Constants.default_platform)
self.DEPLOYMENT_CONFIG_FILE = Utility.get_deployment_manifest_name(self.DEPLOYMENT_CONFIG_TEMPLATE_FILE, None, self.DEFAULT_PLATFORM)
self.MODULES_PATH = self.get_envvar("MODULES_PATH", default=Constants.default_modules_folder)
self.LOGS_PATH = self.get_envvar("LOGS_PATH", default="logs")
self.LOGS_CMD = self.get_envvar("LOGS_CMD", default="start /B start cmd.exe @cmd /k docker logs {0} -f")
self.SUBSCRIPTION_ID = self.get_envvar("SUBSCRIPTION_ID")
self.RESOURCE_GROUP_NAME = self.get_envvar("RESOURCE_GROUP_NAME")
self.RESOURCE_GROUP_LOCATION = self.get_envvar("RESOURCE_GROUP_LOCATION")
self.IOTHUB_NAME = self.get_envvar("IOTHUB_NAME")
self.IOTHUB_SKU = self.get_envvar("IOTHUB_SKU")
self.EDGE_DEVICE_ID = self.get_envvar("EDGE_DEVICE_ID")
self.CREDENTIALS = self.get_envvar("CREDENTIALS")
self.UPDATE_DOTENV = self.get_envvar("UPDATE_DOTENV")
except Exception as ex:
msg = "Environment variables not configured correctly. Run `iotedgedev new` to create a new solution with sample .env file. "
"Please see README for variable configuration options. Tip: You might just need to restart your command prompt to refresh your Environment Variables. "
"Variable that caused exception: {0}".format(str(ex))
raise ValueError(msg)
def _validate_deployment_manifest_schema(self):
validation_success = True
try:
deployment_schema = json.loads(urlopen(Constants.deployment_manifest_schema_url).read())
self._validate_json_schema(deployment_schema, self.json, "Deployment manifest")
except Exception as ex: # Ignore other non schema validation errors
self.output.info("Unexpected error during deployment manifest schema validation, skip schema validation. Error:%s" % ex)
return validation_success
output.param("RESOURCE GROUP LOCATION", value, f("Setting Resource Group Location to '{value}'..."), azure_cli_processing_complete)
envvars.RESOURCE_GROUP_LOCATION = value
if param.name == "iothub_sku":
output.param("IOT HUB SKU", value, f("Setting IoT Hub SKU to '{value}'..."), azure_cli_processing_complete)
envvars.IOTHUB_SKU = value
if param.name == "iothub_name":
output.param("IOT HUB", value, f("Setting IoT Hub to '{value}'..."), azure_cli_processing_complete)
envvars.IOTHUB_NAME = value
if not azure_cli.extension_exists("azure-cli-iot-ext"):
try:
# Install fixed version of Azure CLI IoT extension
azure_cli.add_extension_with_source(Constants.azure_cli_iot_ext_source_url)
except Exception:
# Fall back to install latest Azure CLI IoT extension when fail
azure_cli.add_extension("azure-cli-iot-ext")
if not azure_cli.iothub_exists(value, envvars.RESOURCE_GROUP_NAME):
# check if the active subscription already contains a free IoT Hub
# if yes ask if the user wants to create an S1
# otherwise exit
if envvars.IOTHUB_SKU == "F1":
free_iot_name, free_iot_rg = azure_cli.get_free_iothub()
if free_iot_name:
output.info("You already have a Free IoT Hub SKU in your subscription, "
"so you must either use that existing IoT Hub or create a new S1 IoT Hub. "
"Enter (F) to use the existing Free IoT Hub or enter (S) to create a new S1 IoT Hub:")
user_response = sys.stdin.readline().strip().upper()
if user_response == "S":
envvars.IOTHUB_SKU = "S1"
deployment_manifest = DeploymentManifest(self.envvars, self.output, self.utility, template_file, True, True)
# get image tags for ${MODULES.modulename.xxx} placeholder
modules_path = os.path.join(template_file_folder, self.envvars.MODULES_PATH)
if os.path.isdir(modules_path):
for folder_name in os.listdir(modules_path):
project_folder = os.path.join(modules_path, folder_name)
if os.path.exists(os.path.join(project_folder, "module.json")):
module = Module(self.envvars, self.utility, project_folder)
self._update_module_maps("MODULES.{0}".format(folder_name), module, placeholder_tag_map, tag_build_profile_map, default_platform)
# get image tags for ${MODULEDIR.xxx} placeholder
user_modules = deployment_manifest.get_user_modules()
for module_name, module_info in user_modules.items():
image = module_info["settings"]["image"]
match_result = re.search(Constants.moduledir_placeholder_pattern, image)
if (match_result is not None):
module_dir = match_result.group(1)
module = Module(self.envvars, self.utility, os.path.join(template_file_folder, module_dir))
self._update_module_maps("MODULEDIR<{0}>".format(module_dir), module, placeholder_tag_map, tag_build_profile_map, default_platform)
replacements = {}
for module_name, module_info in user_modules.items():
image = module_info["settings"]["image"]
if image in placeholder_tag_map:
tag = placeholder_tag_map[image]
replacements[module_name] = tag
if not self.utility.in_asterisk_list(module_name, bypass_modules):
tags_to_build.add(tag)
if not no_build or not no_push:
docker = Docker(self.envvars, self.utility, self.output)