зеркало из https://github.com/Azure/iotedgedev.git
Add command to add layered deployments to IoT Hub (#498)
* Add push and build for layered deployments * Use setup.py * Remove redudant command * Update flake8 version to fix whitespace formating * Determine layered deployment by schema layout * Add additional test cases * Refactor splitting on dot * Fix pystyle code version * Stop pip from installing 100s of libs to resolve conficts By pinning all versions * Remove test * Attempt to fix running all tests * Fix tests * Clean up after tests * Remove invalid and add new test case * Add layered deployment deployment * readd missing deploy command * Use login parameter * Add test cases * Add an additional test case * Remove unused method * Move deployments to iothub from edge * Use correct version * Fix test * Install missing extension * Use new CLI naming and refactor * Fix issue with import being cached across tests * Skip instead of commenting out test * Use variables * Refactor * Fix issue with tests failing due to previous tests * Update changelog * Fix issue with invalid manifest name for deployment * Skip test on windows
This commit is contained in:
Eliise
GitHub
Родитель
ef8e584d68
Коммит
174aa64ec4
11
.env.tmp
11
.env.tmp
|
|
@ -50,6 +50,17 @@ ACTIVE_DOCKER_PLATFORMS=""
|
|||
|
||||
CONTAINER_TAG=""
|
||||
|
||||
#
|
||||
# LAYERED DEPLOYMENTS
|
||||
#
|
||||
|
||||
IOTHUB_DEPLOYMENT_TARGET_CONDITION=""
|
||||
# To specifiy the target condition of a IoT Hub deployment
|
||||
# Required when creating a deployment on IoT Hub
|
||||
# Examples:
|
||||
# IOTHUB_DEPLOYMENT_TARGET_CONDITION="tags.environment='dev'"
|
||||
# IOTHUB_DEPLOYMENT_TARGET_CONDITION="tags.building=9 and tags.environment='test'"
|
||||
|
||||
#
|
||||
# SOLUTION SETTINGS
|
||||
#
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
# Changelog
|
||||
All notable changes to this project since 0.82.0 will be documented in this file.
|
||||
|
||||
## [3.3.0] - 2021-8-24
|
||||
- Add support for layered deployment manifests when building, pushing and generating
|
||||
- Add support for creating deployments in IoTHub via the new command: `iotedgedev iothub deploy`
|
||||
|
||||
## [3.2.0] - 2021-7-30
|
||||
- Added support for IoT Edge Runtime version 1.2
|
||||
- Enable change of edgehub and edgeagent schema versions
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
from iotedgedev.connectionstring import IoTHubConnectionString
|
||||
import json
|
||||
import os
|
||||
import signal
|
||||
|
|
@ -20,7 +21,7 @@ def get_query_argument_for_id_and_name(token):
|
|||
|
||||
|
||||
class AzureCli:
|
||||
def __init__(self, output, envvars, cli=get_default_cli()):
|
||||
def __init__(self, output, envvars, cli=get_default_cli()):
|
||||
self.output = output
|
||||
self.envvars = envvars
|
||||
self.az_cli = cli
|
||||
|
|
@ -37,7 +38,7 @@ class AzureCli:
|
|||
if suppress_output:
|
||||
args.extend(["--query", "\"[?n]|[0]\""])
|
||||
|
||||
az_args = ["az"]+args
|
||||
az_args = ["az"] + args
|
||||
return az_args
|
||||
|
||||
def invoke_az_cli_outproc(self, args, error_message=None, stdout_io=None, stderr_io=None, suppress_output=False, timeout=None):
|
||||
|
|
@ -240,13 +241,13 @@ class AzureCli:
|
|||
def login_account(self, username, password):
|
||||
|
||||
return self.invoke_az_cli_outproc(["login", "-u", username,
|
||||
"-p", password],
|
||||
"-p", password],
|
||||
"Error while trying to login to Azure. Make sure your account credentials are correct", suppress_output=True)
|
||||
|
||||
def login_sp(self, username, password, tenant):
|
||||
|
||||
return self.invoke_az_cli_outproc(["login", "--service-principal", "-u", username,
|
||||
"-p", password, "--tenant", tenant],
|
||||
"-p", password, "--tenant", tenant],
|
||||
"Error while trying to login to Azure. Make sure your service principal credentials are correct.", suppress_output=True)
|
||||
|
||||
def login_interactive(self):
|
||||
|
|
@ -361,19 +362,43 @@ class AzureCli:
|
|||
|
||||
return result
|
||||
|
||||
def set_modules(self, device_id, connection_string, config):
|
||||
self.output.status(f("Deploying '{config}' to '{device_id}'..."))
|
||||
def set_modules(self, config: str, device_id: str, connection_string: IoTHubConnectionString):
|
||||
self.output.status(f"Deploying '{config}' to '{device_id}'...")
|
||||
|
||||
config = os.path.join(os.getcwd(), config)
|
||||
|
||||
if not os.path.exists(config):
|
||||
raise FileNotFoundError('Deployment manifest file "{0}" not found. Please run `iotedgedev build` first'.format(config))
|
||||
raise FileNotFoundError(f"Deployment manifest file '{config}' not found. Please run `iotedgedev build` first")
|
||||
|
||||
telemetry.add_extra_props({'iothubhostname': connection_string.iothub_host.name_hash, 'iothubhostnamesuffix': connection_string.iothub_host.name_suffix})
|
||||
|
||||
return self.invoke_az_cli_outproc(["iot", "edge", "set-modules", "-d", device_id, "-n", connection_string.iothub_host.hub_name, "-k", config, "-l", connection_string.connection_string],
|
||||
error_message=f("Failed to deploy '{config}' to '{device_id}'..."), suppress_output=True)
|
||||
|
||||
def create_deployment(self,
|
||||
config: str,
|
||||
connection_string: IoTHubConnectionString,
|
||||
deployment_name: str,
|
||||
target_condition: str,
|
||||
priority: str) -> bool:
|
||||
self.output.status(f"Deploying '{config}' to '{connection_string.iothub_host.hub_name}'...")
|
||||
|
||||
config = os.path.join(os.getcwd(), config)
|
||||
|
||||
if not os.path.exists(config):
|
||||
raise FileNotFoundError(f"Deployment manifest file '{config}' not found. Please run `iotedgedev build` first")
|
||||
|
||||
telemetry.add_extra_props({'iothubhostname': connection_string.iothub_host.name_hash, 'iothubhostnamesuffix': connection_string.iothub_host.name_suffix})
|
||||
with output_io_cls() as io:
|
||||
result = self.invoke_az_cli_outproc(["iot", "edge", "deployment", "create", "-d", deployment_name, "-l", connection_string.connection_string, "--content", config,
|
||||
"--target-condition", target_condition, "--priority", priority],
|
||||
error_message=f"Failed to deploy '{config}' to '{connection_string.iothub_host.hub_name}'...", stderr_io=io)
|
||||
if io.getvalue():
|
||||
self.output.error(io.getvalue())
|
||||
self.output.line()
|
||||
|
||||
return result
|
||||
|
||||
def monitor_events(self, device_id, connection_string, hub_name, timeout=300):
|
||||
return self.invoke_az_cli_outproc(["iot", "hub", "monitor-events", "-d", device_id, "-n", hub_name, "-l", connection_string, '-t', str(timeout), '-y'],
|
||||
error_message=f("Failed to start monitoring events."), suppress_output=False, timeout=timeout)
|
||||
|
|
|
|||
|
|
@ -248,6 +248,40 @@ def deploy(manifest_file):
|
|||
main.add_command(deploy)
|
||||
|
||||
|
||||
@iothub.command(
|
||||
name="deploy",
|
||||
context_settings=CONTEXT_SETTINGS,
|
||||
help="Create a deployment in IoT Hub")
|
||||
@click.option("--file",
|
||||
"-f",
|
||||
"manifest_file",
|
||||
default=envvars.DEPLOYMENT_CONFIG_FILE_PATH,
|
||||
show_default=True,
|
||||
required=False,
|
||||
help="Specify the deployment manifest file")
|
||||
@click.option("--name",
|
||||
"-n",
|
||||
required=True,
|
||||
help="Specify the deployment name")
|
||||
@click.option("--priority",
|
||||
"-p",
|
||||
required=True,
|
||||
help="Specify the deployment priority")
|
||||
@click.option("--target-condition",
|
||||
"--tc",
|
||||
"-t",
|
||||
"target_condition",
|
||||
default=envvars.get_envvar("IOTHUB_DEPLOYMENT_TARGET_CONDITION"),
|
||||
show_default=True,
|
||||
required=False,
|
||||
help="Specify the deployment target condition")
|
||||
@with_telemetry
|
||||
def iothub_deploy(manifest_file, name, priority, target_condition):
|
||||
ensure_azure_cli_iot_ext()
|
||||
iothub = IoTHub(envvars, output, None, azure_cli)
|
||||
iothub.deploy(manifest_file, name, priority, target_condition)
|
||||
|
||||
|
||||
@solution.command(context_settings=CONTEXT_SETTINGS,
|
||||
help="Expand environment variables and placeholders in deployment manifest template file and copy to config folder",
|
||||
# hack to prevent Click truncating help messages
|
||||
|
|
@ -424,7 +458,7 @@ def modulecred(local, output_file):
|
|||
def monitor(timeout):
|
||||
ensure_azure_cli_iot_ext()
|
||||
utility = Utility(envvars, output)
|
||||
ih = IoTHub(envvars, utility, output, azure_cli)
|
||||
ih = IoTHub(envvars, output, utility, azure_cli)
|
||||
ih.monitor_events(timeout)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,15 @@
|
|||
from iotedgedev.output import Output
|
||||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.azurecli import AzureCli
|
||||
|
||||
|
||||
class Edge:
|
||||
def __init__(self, envvars, output, azure_cli):
|
||||
def __init__(self, envvars: EnvVars, output: Output, azure_cli: AzureCli):
|
||||
self.envvars = envvars
|
||||
self.output = output
|
||||
self.azure_cli = azure_cli
|
||||
|
||||
def deploy(self, manifest_file):
|
||||
def deploy(self, manifest_file: str):
|
||||
|
||||
self.output.header("DEPLOYING CONFIGURATION")
|
||||
|
||||
|
|
@ -12,5 +17,7 @@ class Edge:
|
|||
self.envvars.verify_envvar_has_val("DEVICE_CONNECTION_STRING", self.envvars.DEVICE_CONNECTION_INFO)
|
||||
self.envvars.verify_envvar_has_val("DEPLOYMENT_CONFIG_FILE", self.envvars.DEPLOYMENT_CONFIG_FILE)
|
||||
|
||||
if self.azure_cli.set_modules(self.envvars.DEVICE_CONNECTION_INFO.device_id, self.envvars.IOTHUB_CONNECTION_INFO, manifest_file):
|
||||
if self.azure_cli.set_modules(config=manifest_file,
|
||||
connection_string=self.envvars.IOTHUB_CONNECTION_INFO,
|
||||
device_id=self.envvars.DEVICE_CONNECTION_INFO.device_id):
|
||||
self.output.footer("DEPLOYMENT COMPLETE")
|
||||
|
|
|
|||
|
|
@ -1,13 +1,33 @@
|
|||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.output import Output
|
||||
from iotedgedev.azurecli import AzureCli
|
||||
from .utility import Utility
|
||||
from .version import PY35
|
||||
|
||||
|
||||
class IoTHub:
|
||||
def __init__(self, envvars, utility, output, azure_cli):
|
||||
def __init__(self, envvars: EnvVars, output: Output, utility: Utility, azure_cli: AzureCli):
|
||||
self.envvars = envvars
|
||||
self.output = output
|
||||
self.utility = utility
|
||||
self.azure_cli = azure_cli
|
||||
|
||||
def deploy(self, manifest_file: str, layered_deployment_name: str, priority: str, target_condition: str):
|
||||
|
||||
self.output.header("DEPLOYING CONFIGURATION")
|
||||
|
||||
self.envvars.verify_envvar_has_val("IOTHUB_CONNECTION_STRING", self.envvars.IOTHUB_CONNECTION_INFO)
|
||||
if not target_condition:
|
||||
target_condition = self.envvars.get_envvar("IOTHUB_DEPLOYMENT_TARGET_CONDITION", True)
|
||||
self.envvars.verify_envvar_has_val("DEPLOYMENT_CONFIG_FILE", self.envvars.DEPLOYMENT_CONFIG_FILE)
|
||||
|
||||
if self.azure_cli.create_deployment(config=manifest_file,
|
||||
connection_string=self.envvars.IOTHUB_CONNECTION_INFO,
|
||||
deployment_name=layered_deployment_name,
|
||||
target_condition=target_condition,
|
||||
priority=priority):
|
||||
self.output.footer("DEPLOYMENT COMPLETE")
|
||||
|
||||
def monitor_events(self, timeout=0):
|
||||
|
||||
self.envvars.verify_envvar_has_val("IOTHUB_CONNECTION_STRING", self.envvars.IOTHUB_CONNECTION_STRING)
|
||||
|
|
|
|||
|
|
@ -2,25 +2,28 @@ import os
|
|||
import re
|
||||
import shutil
|
||||
import sys
|
||||
from io import BytesIO
|
||||
from urllib.request import urlopen
|
||||
from zipfile import ZipFile
|
||||
|
||||
import commentjson
|
||||
from io import BytesIO
|
||||
from urllib.request import urlopen
|
||||
|
||||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.output import Output
|
||||
|
||||
from . import telemetry
|
||||
from .buildoptionsparser import BuildOptionsParser
|
||||
from .buildprofile import BuildProfile
|
||||
from .constants import Constants
|
||||
from .deploymentmanifest import DeploymentManifest
|
||||
from .dockercls import Docker
|
||||
from .dotnet import DotNet
|
||||
from .module import Module
|
||||
from .utility import Utility
|
||||
from .constants import Constants
|
||||
|
||||
|
||||
class Modules:
|
||||
def __init__(self, envvars, output):
|
||||
def __init__(self, envvars: EnvVars, output: Output):
|
||||
self.envvars = envvars
|
||||
self.output = output
|
||||
self.utility = Utility(self.envvars, self.output)
|
||||
|
|
|
|||
|
|
@ -31,9 +31,10 @@ class Solution:
|
|||
if(runtime_tag == "1.2"):
|
||||
edgeagent_schema_version = "1.1"
|
||||
edgehub_schema_version = "1.2"
|
||||
|
||||
self.utility.copy_from_template_dir(".env.tmp", dir_path, dest_file=".env", replacements={"%EDGE_RUNTIME_VERSION%": runtime_tag, "%EDGEAGENT_SCHEMA_VERSION%": edgeagent_schema_version, "%EDGEHUB_SCHEMA_VERSION%": edgehub_schema_version})
|
||||
|
||||
|
||||
self.utility.copy_from_template_dir(".env.tmp", dir_path, dest_file=".env", replacements={
|
||||
"%EDGE_RUNTIME_VERSION%": runtime_tag, "%EDGEAGENT_SCHEMA_VERSION%": edgeagent_schema_version, "%EDGEHUB_SCHEMA_VERSION%": edgehub_schema_version})
|
||||
|
||||
if template == "java":
|
||||
mod_cmd = "iotedgedev solution add {0} --template {1} --group-id {2}".format(module, template, group_id)
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -1,57 +1,50 @@
|
|||
import os
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.output import Output
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
||||
def test_valid_get_envvar():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
def test_get_envvar__valid():
|
||||
envvars = EnvVars(Output())
|
||||
deployment_template = envvars.get_envvar("DEPLOYMENT_CONFIG_TEMPLATE_FILE")
|
||||
assert deployment_template is not None
|
||||
|
||||
|
||||
def test_invalid_get_envvar():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
def test_get_envvar__invalid():
|
||||
envvars = EnvVars(Output())
|
||||
testerval = envvars.get_envvar("TESTER")
|
||||
assert not testerval
|
||||
|
||||
|
||||
def test_valid_load():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
def test_load_valid():
|
||||
envvars = EnvVars(Output())
|
||||
envvars.load()
|
||||
assert envvars.DEPLOYMENT_CONFIG_TEMPLATE_FILE == "deployment.template.json"
|
||||
|
||||
|
||||
def test_valid_verify_envvar_has_val():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
def test_verify_envvar_has_val__valid():
|
||||
envvars = EnvVars(Output())
|
||||
envvars.load()
|
||||
result = envvars.verify_envvar_has_val("DEPLOYMENT_CONFIG_TEMPLATE_FILE", envvars.DEPLOYMENT_CONFIG_TEMPLATE_FILE)
|
||||
assert not result
|
||||
|
||||
|
||||
def test_valid_get_envvar_key_if_val():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
def test_get_envvar_key_if_val__valid():
|
||||
envvars = EnvVars(Output())
|
||||
assert envvars.get_envvar_key_if_val("DEPLOYMENT_CONFIG_TEMPLATE_FILE")
|
||||
|
||||
|
||||
def test_invalid_get_envvar_key_if_val():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
def test_get_envvar_key_if_val__invalid():
|
||||
envvars = EnvVars(Output())
|
||||
assert not envvars.get_envvar_key_if_val("TESTER")
|
||||
|
||||
|
||||
def test_set_envvar():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars = EnvVars(Output())
|
||||
registry_server = envvars.get_envvar("DEPLOYMENT_CONFIG_TEMPLATE_FILE")
|
||||
envvars.set_envvar("DEPLOYMENT_CONFIG_TEMPLATE_FILE", "deployment.template_new.json")
|
||||
new_registry_server = envvars.get_envvar("DEPLOYMENT_CONFIG_TEMPLATE_FILE")
|
||||
|
|
@ -60,135 +53,100 @@ def test_set_envvar():
|
|||
|
||||
|
||||
def test_envvar_clean():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
EnvVars(Output())
|
||||
envvar_clean_name = u"IOTEDGEDEV_ENVVAR_CLEAN_TEST"
|
||||
os.environ[envvar_clean_name] = u"test unicode string"
|
||||
|
||||
|
||||
def test_in_command_list_true_1():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert envvars.in_command_list("solution new test_solution", ["init", "e2e", "solution new", "new", "simulator stop"])
|
||||
@pytest.mark.parametrize(
|
||||
"command, command_list",
|
||||
[
|
||||
("solution new test_solution", ["init", "e2e", "solution new", "new", "simulator stop"]),
|
||||
("solution new", ["init", "e2e", "solution new", "new", "simulator stop"]),
|
||||
("", ["init", "e2e", "", "new", "simulator stop"]),
|
||||
]
|
||||
)
|
||||
def test_in_command_list_true(command, command_list):
|
||||
envvars = EnvVars(Output())
|
||||
assert envvars.in_command_list(command, command_list)
|
||||
|
||||
|
||||
def test_in_command_list_true_2():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert envvars.in_command_list("solution new", ["init", "e2e", "solution new", "new", "simulator stop"])
|
||||
@pytest.mark.parametrize(
|
||||
"command, command_list",
|
||||
[
|
||||
("solution add filtermodule", ["init", "e2e", "solution new", "new", "simulator stop"]),
|
||||
("solution addotherstuff filtermodule", ["init", "e2e", "solution add", "new", "simulator stop"]),
|
||||
("", ["init", "e2e", "solution new", "new", "simulator stop"]),
|
||||
("solution new test_solution", ["init", "e2e", "", "new", "simulator stop"])
|
||||
]
|
||||
)
|
||||
def test_in_command_list_false(command, command_list):
|
||||
envvars = EnvVars(Output())
|
||||
assert not envvars.in_command_list(command, command_list)
|
||||
|
||||
|
||||
def test_in_command_list_false_1():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert not envvars.in_command_list("solution add filtermodule", ["init", "e2e", "solution new", "new", "simulator stop"])
|
||||
|
||||
|
||||
def test_in_command_list_false_2():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert not envvars.in_command_list("solution addotherstuff filtermodule", ["init", "e2e", "solution add", "new", "simulator stop"])
|
||||
|
||||
|
||||
def test_in_command_list_empty_1():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert not envvars.in_command_list("", ["init", "e2e", "solution new", "new", "simulator stop"])
|
||||
|
||||
|
||||
def test_in_command_list_empty_2():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert not envvars.in_command_list("solution new test_solution", ["init", "e2e", "", "new", "simulator stop"])
|
||||
|
||||
|
||||
def test_in_command_list_empty_3():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert envvars.in_command_list("", ["init", "e2e", "", "new", "simulator stop"])
|
||||
|
||||
|
||||
def test_is_terse_command_true():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert envvars.is_terse_command("iothub setup --update-dotenv")
|
||||
@pytest.mark.parametrize(
|
||||
"command",
|
||||
[
|
||||
"iothub setup --update-dotenv",
|
||||
""
|
||||
]
|
||||
)
|
||||
def test_is_terse_command_true(command):
|
||||
envvars = EnvVars(Output())
|
||||
assert envvars.is_terse_command(command)
|
||||
|
||||
|
||||
def test_is_terse_command_false():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars = EnvVars(Output())
|
||||
assert not envvars.is_terse_command("solution add")
|
||||
|
||||
|
||||
def test_is_terse_command_empty():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
assert envvars.is_terse_command("")
|
||||
|
||||
|
||||
def test_default_container_registry_server_key_exists():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars = EnvVars(Output())
|
||||
envvars.load()
|
||||
assert "CONTAINER_REGISTRY_SERVER" in os.environ
|
||||
|
||||
|
||||
def test_default_container_registry_server_value_exists():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
server = envvars.get_envvar("CONTAINER_REGISTRY_SERVER")
|
||||
@pytest.mark.parametrize(
|
||||
"envvar",
|
||||
[
|
||||
"CONTAINER_REGISTRY_SERVER",
|
||||
"CONTAINER_REGISTRY_USERNAME",
|
||||
"CONTAINER_REGISTRY_PASSWORD"
|
||||
|
||||
]
|
||||
)
|
||||
def test_default_envvar_value_exists(envvar):
|
||||
envvars = EnvVars(Output())
|
||||
server = envvars.get_envvar(envvar)
|
||||
assert server is not None
|
||||
|
||||
|
||||
def test_default_container_registry_username_value_exists_or_returns_empty_string():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
username = envvars.get_envvar("CONTAINER_REGISTRY_USERNAME")
|
||||
assert username is not None
|
||||
|
||||
|
||||
def test_default_container_registry_password_value_exists_or_returns_empty_string():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
password = envvars.get_envvar("CONTAINER_REGISTRY_PASSWORD")
|
||||
assert password is not None
|
||||
|
||||
|
||||
def test_container_registry_server_key_missing_sys_exit():
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars = EnvVars(Output())
|
||||
with pytest.raises(ValueError):
|
||||
envvars.get_envvar("CONTAINER_REGISTRY_SERVER_UNITTEST", required=True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def setup_test_env(request):
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars.set_envvar("CONTAINER_REGISTRY_SERVER_UNITTEST", 'unittest.azurecr.io')
|
||||
envvars.set_envvar("CONTAINER_REGISTRY_USERNAME_UNITTEST", 'username')
|
||||
envvars.set_envvar("CONTAINER_REGISTRY_PASSWORD_UNITTEST", 'password')
|
||||
@pytest.mark.parametrize(
|
||||
"envvar",
|
||||
[
|
||||
"CONTAINER_REGISTRY_SERVER",
|
||||
"CONTAINER_REGISTRY_USERNAME",
|
||||
"CONTAINER_REGISTRY_PASSWORD"
|
||||
|
||||
def clean():
|
||||
os.environ.pop("CONTAINER_REGISTRY_SERVER_UNITTEST")
|
||||
os.environ.pop("CONTAINER_REGISTRY_USERNAME_UNITTEST")
|
||||
os.environ.pop("CONTAINER_REGISTRY_PASSWORD_UNITTEST")
|
||||
request.addfinalizer(clean)
|
||||
|
||||
return
|
||||
|
||||
|
||||
def test_unique_container_registry_server_tokens():
|
||||
]
|
||||
)
|
||||
def test_unique_envvar_tokens(envvar):
|
||||
unique = set()
|
||||
length_container_registry_server = len('container_registry_server')
|
||||
envvar_lenght = len(envvar)
|
||||
is_unique = True
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars = EnvVars(Output())
|
||||
envvars.load()
|
||||
for key in os.environ:
|
||||
key = key.lower()
|
||||
if key.startswith('container_registry_server'):
|
||||
token = key[length_container_registry_server:]
|
||||
if key.startswith(envvar):
|
||||
token = key[envvar_lenght:]
|
||||
if token not in unique:
|
||||
unique.add(token)
|
||||
else:
|
||||
|
|
@ -196,45 +154,13 @@ def test_unique_container_registry_server_tokens():
|
|||
assert is_unique
|
||||
|
||||
|
||||
def test_unique_container_registry_username_tokens():
|
||||
unique = set()
|
||||
length_container_registry_username = len('container_registry_username')
|
||||
is_unique = True
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars.load()
|
||||
for key in os.environ:
|
||||
key = key.lower()
|
||||
if key.startswith('container_registry_username'):
|
||||
token = key[length_container_registry_username:]
|
||||
if token not in unique:
|
||||
unique.add(token)
|
||||
else:
|
||||
is_unique = False
|
||||
assert is_unique
|
||||
|
||||
|
||||
def test_unique_container_registry_password_tokens():
|
||||
unique = set()
|
||||
length_container_registry_password = len('container_registry_password')
|
||||
is_unique = True
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
envvars.load()
|
||||
for key in os.environ:
|
||||
key = key.lower()
|
||||
if key.startswith('container_registry_password'):
|
||||
token = key[length_container_registry_password:]
|
||||
if token not in unique:
|
||||
unique.add(token)
|
||||
else:
|
||||
is_unique = False
|
||||
assert is_unique
|
||||
|
||||
|
||||
def test_additional_container_registry_map_has_val(setup_test_env):
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
@mock.patch.dict(os.environ, {
|
||||
"CONTAINER_REGISTRY_SERVER_UNITTEST": "unittest.azurecr.io",
|
||||
"CONTAINER_REGISTRY_USERNAME_UNITTEST": "username",
|
||||
"CONTAINER_REGISTRY_PASSWORD_UNITTEST": "password"
|
||||
})
|
||||
def test_additional_container_registry_map_is_set_from_environ():
|
||||
envvars = EnvVars(Output())
|
||||
envvars.load()
|
||||
assert len(envvars.CONTAINER_REGISTRY_MAP) == 2
|
||||
assert 'UNITTEST' in envvars.CONTAINER_REGISTRY_MAP.keys()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,115 @@
|
|||
import os
|
||||
import uuid
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from iotedgedev.azurecli import AzureCli
|
||||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.output import Output
|
||||
|
||||
from .utility import get_platform_type, runner_invoke
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
test_solution_shared_lib_dir = os.path.join(os.getcwd(), "tests", "assets", "test_solution_shared_lib")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config, manifest",
|
||||
[
|
||||
("layered_deployment.flattened_props.template.json", "layered_deployment.flattened_props.json"),
|
||||
("layered_deployment.no_modules.template.json", "layered_deployment.no_modules.json"),
|
||||
("deployment.template.json", f"deployment.{get_platform_type()}.json")
|
||||
]
|
||||
)
|
||||
# Test that cmd line target condition (-t) overrides target condition from .env
|
||||
@ mock.patch.dict(os.environ, {"IOTHUB_DEPLOYMENT_TARGET_CONDITION": "invalid_target"})
|
||||
def test_iothub_deploy(config, manifest):
|
||||
# Arrange
|
||||
deployment_name = f'test-{uuid.uuid4()}'
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
# Act
|
||||
result = runner_invoke(['build', '-f', config, '-P', get_platform_type()])
|
||||
result = runner_invoke(['iothub', 'deploy',
|
||||
'-f', f'config/{manifest}',
|
||||
'-n', deployment_name,
|
||||
'-p', '10',
|
||||
'-t', "tags.environment='dev'"
|
||||
])
|
||||
|
||||
# Assert
|
||||
assert 'DEPLOYMENT COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
azure_cli = AzureCli(output, envvars)
|
||||
assert azure_cli.invoke_az_cli_outproc(["iot", "edge", "deployment", "delete", "-d", deployment_name, "-l", envvars.get_envvar("IOTHUB_CONNECTION_STRING")])
|
||||
|
||||
|
||||
@ mock.patch.dict(os.environ, {"IOTHUB_DEPLOYMENT_TARGET_CONDITION": "tags.environment='dev'"})
|
||||
def test_iothub_deploy_with_target_group_set_from_dotenv():
|
||||
# Arrange
|
||||
deployment_name = f'test-layered{uuid.uuid4()}'
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
# Act
|
||||
result = runner_invoke(['build', '-f', "layered_deployment.flattened_props.template.json", '-P', get_platform_type()])
|
||||
result = runner_invoke(['iothub', 'deploy',
|
||||
'-f', 'config/layered_deployment.flattened_props.json',
|
||||
'-n', deployment_name,
|
||||
'-p', '10',
|
||||
])
|
||||
|
||||
# Assert
|
||||
assert 'DEPLOYMENT COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
azure_cli = AzureCli(output, envvars)
|
||||
|
||||
assert azure_cli.invoke_az_cli_outproc(["iot", "edge", "deployment", "delete", "-d", deployment_name, "-l", envvars.get_envvar("IOTHUB_CONNECTION_STRING")])
|
||||
|
||||
|
||||
def test_iothub_deploy_error_from_az_cli_bubbled_up():
|
||||
# Arrange
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
# Act
|
||||
result = runner_invoke(['build', '-f', "layered_deployment.flattened_props.template.json", '-P', get_platform_type()])
|
||||
result = runner_invoke(['iothub', 'deploy',
|
||||
'-f', 'config/layered_deployment.flattened_props.json',
|
||||
'-n', "test-layered-deployment",
|
||||
'-p', '10',
|
||||
'-t', "invalid_target_group"
|
||||
])
|
||||
|
||||
# Assert
|
||||
assert "ERROR: {'Message': 'ErrorCode:ArgumentInvalid;BadRequest'," in result.output
|
||||
|
||||
|
||||
def test_iothub_deploy_error_missing_name():
|
||||
# Act
|
||||
with pytest.raises(Exception) as context:
|
||||
runner_invoke(['iothub', 'deploy', '-n', "test"])
|
||||
|
||||
# Assert
|
||||
assert "Error: Missing option '--priority' / '-p'." in str(context)
|
||||
|
||||
|
||||
def test_iothub_deploy_error_missing_priority():
|
||||
# Act
|
||||
with pytest.raises(Exception) as context:
|
||||
runner_invoke(['iothub', 'deploy', '-n', "test"])
|
||||
|
||||
# Assert
|
||||
assert "Error: Missing option '--priority' / '-p'." in str(context)
|
||||
|
||||
|
||||
def test_iothub_deploy_error_missing_target_condition():
|
||||
# Act
|
||||
with pytest.raises(Exception) as context:
|
||||
runner_invoke(['iothub', 'deploy', '-n', "test", '-p', '10'])
|
||||
|
||||
# Assert
|
||||
assert "ERROR: Environment Variable IOTHUB_DEPLOYMENT_TARGET_CONDITION not set. Either add to .env file or to your system's Environment Variables" in str(context.value)
|
||||
|
|
@ -1,38 +0,0 @@
|
|||
import json
|
||||
import os
|
||||
import pytest
|
||||
from .utility import (
|
||||
get_platform_type,
|
||||
runner_invoke,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
test_solution_shared_lib_dir = os.path.join(os.getcwd(), "tests", "assets", "test_solution_shared_lib")
|
||||
|
||||
def test_build_and_push():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '--push', '-f', "layered_deployment.template_with_flattened_props.json", '-P', get_platform_type()])
|
||||
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
def test_build_and_push_with_no_modules():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '--push', '-f', "layered_deployment.template_with_no_modules.json", '-P', get_platform_type()])
|
||||
|
||||
new_config_deployment_name = 'layered_deployment.template_with_no_modules.json'
|
||||
new_config_deployment_path = os.path.join(test_solution_shared_lib_dir, 'config', new_config_deployment_name)
|
||||
|
||||
with open(new_config_deployment_path, "r") as f:
|
||||
content = json.load(f)
|
||||
|
||||
set_property = content["content"]["modulesContent"]["exampleModule"]["properties.desired"]["foo"]
|
||||
|
||||
assert 'ERROR' not in result.output
|
||||
assert 'bar-1.2' == set_property
|
||||
|
|
@ -2,7 +2,6 @@ import os
|
|||
import pytest
|
||||
import shutil
|
||||
import time
|
||||
import sys
|
||||
|
||||
from iotedgedev.version import PY35
|
||||
from iotedgedev.envvars import EnvVars
|
||||
|
|
@ -99,7 +98,7 @@ def test_start_solution(capfd):
|
|||
def test_start_solution_with_setup(capfd):
|
||||
result = runner_invoke(['simulator', 'start', '--setup', '-s', '-b', '-f', 'deployment.template.json'])
|
||||
out, err = capfd.readouterr()
|
||||
|
||||
|
||||
assert 'Setup IoT Edge Simulator successfully.' in result.output
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'IoT Edge Simulator has been started in solution mode.' in out
|
||||
|
|
@ -129,6 +128,6 @@ def test_start_solution_with_deployment(capfd):
|
|||
deployment_file_path = os.path.join(test_solution_dir, 'config', 'deployment.' + platform_type + '.json')
|
||||
runner_invoke(['simulator', 'start', '-f', deployment_file_path])
|
||||
out, err = capfd.readouterr()
|
||||
|
||||
|
||||
assert 'IoT Edge Simulator has been started in solution mode.' in out
|
||||
test_monitor(capfd)
|
||||
|
|
@ -1,26 +1,21 @@
|
|||
import json
|
||||
import os
|
||||
import platform
|
||||
import pytest
|
||||
import shutil
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
from iotedgedev.version import PY35
|
||||
import pytest
|
||||
from iotedgedev.connectionstring import (DeviceConnectionString,
|
||||
IoTHubConnectionString)
|
||||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.output import Output
|
||||
from iotedgedev.version import PY35
|
||||
|
||||
from .utility import (assert_json_file_equal,
|
||||
get_platform_type,
|
||||
get_file_content,
|
||||
get_all_docker_images,
|
||||
get_all_docker_containers,
|
||||
remove_docker_container,
|
||||
remove_docker_image,
|
||||
get_docker_os_type,
|
||||
runner_invoke,
|
||||
update_file_content)
|
||||
from .utility import (assert_json_file_equal, get_all_docker_containers,
|
||||
get_all_docker_images, get_docker_os_type,
|
||||
get_platform_type, remove_docker_container,
|
||||
remove_docker_image, runner_invoke)
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
|
|
@ -56,12 +51,6 @@ def create_solution(template, custom_module_name=None):
|
|||
return result
|
||||
|
||||
|
||||
def add_module(template):
|
||||
module_name = template + "module"
|
||||
result = runner_invoke(["solution", "add", module_name, '--template', template])
|
||||
return result
|
||||
|
||||
|
||||
def clean_folder(folder_path):
|
||||
os.chdir(tests_dir)
|
||||
time.sleep(5)
|
||||
|
|
@ -88,47 +77,6 @@ def prepare_solution_with_env():
|
|||
return
|
||||
|
||||
|
||||
def assert_solution_folder_structure(template):
|
||||
module_name = template + "module"
|
||||
|
||||
expected_files = [".env", "deployment.template.json", "deployment.debug.template.json",
|
||||
os.path.join(".vscode", "launch.json"),
|
||||
os.path.join("modules", module_name, "Dockerfile.amd64"),
|
||||
os.path.join("modules", module_name, "Dockerfile.amd64.debug"),
|
||||
os.path.join("modules", module_name, "Dockerfile.arm32v7"),
|
||||
os.path.join("modules", module_name, "module.json")]
|
||||
for expected_file in expected_files:
|
||||
assert os.path.exists(os.path.join(test_solution_dir, expected_file))
|
||||
|
||||
expected_template_files = [os.environ["DEPLOYMENT_CONFIG_TEMPLATE_FILE"], os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"]]
|
||||
|
||||
for expected_template_file in expected_template_files:
|
||||
with open(os.path.join(test_solution_dir, expected_template_file)) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"]
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert module_name in content["modulesContent"]["$edgeHub"]["properties.desired"]["routes"]["sensorTo" + module_name]
|
||||
assert module_name in content["modulesContent"]["$edgeHub"]["properties.desired"]["routes"][module_name + "ToIoTHub"]
|
||||
|
||||
|
||||
def assert_module_folder_structure(template):
|
||||
module_name = template + "module"
|
||||
expected_template_files = [os.environ["DEPLOYMENT_CONFIG_TEMPLATE_FILE"], os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"]]
|
||||
|
||||
for expected_template_file in expected_template_files:
|
||||
with open(expected_template_file) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"]
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert module_name in content["modulesContent"]["$edgeHub"]["properties.desired"]["routes"][module_name + "ToIoTHub"]
|
||||
|
||||
if expected_template_file == os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"]:
|
||||
if module_name in ["cmodule", "pythonmodule", "nodejsmodule", "javamodule"]:
|
||||
assert "HostConfig" in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["createOptions"]
|
||||
|
||||
|
||||
def test_solution_create_in_non_empty_current_path(prepare_solution_with_env):
|
||||
result = runner_invoke(['solution', 'new', '.'], True)
|
||||
|
||||
|
|
@ -187,8 +135,8 @@ def test_module_add(prepare_solution_with_env):
|
|||
if (template == "nodejs") and (platform.system().lower() != 'windows'):
|
||||
launch_file = launch_json_file_without_nodejs
|
||||
else:
|
||||
result = add_module(template)
|
||||
module_name = template + "module"
|
||||
result = runner_invoke(["solution", "add", module_name, '--template', template])
|
||||
assert 'ADD COMPLETE' in result.output
|
||||
assert os.path.exists(os.path.join(os.environ["MODULES_PATH"], module_name))
|
||||
assert_module_folder_structure(template)
|
||||
|
|
@ -196,6 +144,47 @@ def test_module_add(prepare_solution_with_env):
|
|||
assert_json_file_equal(os.path.join(test_solution_dir, ".vscode", "launch.json"), launch_file)
|
||||
|
||||
|
||||
def assert_solution_folder_structure(template):
|
||||
module_name = template + "module"
|
||||
|
||||
expected_files = [".env", "deployment.template.json", "deployment.debug.template.json",
|
||||
os.path.join(".vscode", "launch.json"),
|
||||
os.path.join("modules", module_name, "Dockerfile.amd64"),
|
||||
os.path.join("modules", module_name, "Dockerfile.amd64.debug"),
|
||||
os.path.join("modules", module_name, "Dockerfile.arm32v7"),
|
||||
os.path.join("modules", module_name, "module.json")]
|
||||
for expected_file in expected_files:
|
||||
assert os.path.exists(os.path.join(test_solution_dir, expected_file))
|
||||
|
||||
expected_template_files = [os.environ["DEPLOYMENT_CONFIG_TEMPLATE_FILE"], os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"]]
|
||||
|
||||
for expected_template_file in expected_template_files:
|
||||
with open(os.path.join(test_solution_dir, expected_template_file)) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"]
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert module_name in content["modulesContent"]["$edgeHub"]["properties.desired"]["routes"]["sensorTo" + module_name]
|
||||
assert module_name in content["modulesContent"]["$edgeHub"]["properties.desired"]["routes"][module_name + "ToIoTHub"]
|
||||
|
||||
|
||||
def assert_module_folder_structure(template):
|
||||
module_name = template + "module"
|
||||
expected_template_files = [os.environ["DEPLOYMENT_CONFIG_TEMPLATE_FILE"], os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"]]
|
||||
|
||||
for expected_template_file in expected_template_files:
|
||||
with open(expected_template_file) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"]
|
||||
assert module_name in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert module_name in content["modulesContent"]["$edgeHub"]["properties.desired"]["routes"][module_name + "ToIoTHub"]
|
||||
|
||||
if expected_template_file == os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"]:
|
||||
if module_name in ["cmodule", "pythonmodule", "nodejsmodule", "javamodule"]:
|
||||
assert "HostConfig" in content["modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["createOptions"]
|
||||
|
||||
|
||||
def test_module_add_invalid_name(prepare_solution_with_env):
|
||||
"""Test the addmodule command with invalid module name"""
|
||||
|
||||
|
|
@ -271,116 +260,6 @@ def test_valid_env_device_connectionstring():
|
|||
assert connectionstring.device_id
|
||||
|
||||
|
||||
def test_solution_build_and_push_with_platform():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'PUSH COMPLETE' not in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
result = runner_invoke(['push', '--no-build', '-P', get_platform_type()])
|
||||
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'BUILD COMPLETE' not in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
def test_solution_build_and_push_with_different_cwd():
|
||||
cwd = os.path.join(test_solution_shared_lib_dir, 'config')
|
||||
if not os.path.exists(cwd):
|
||||
os.makedirs(cwd)
|
||||
os.chdir(cwd)
|
||||
|
||||
result = runner_invoke(['build', '-f', '../deployment.template.json', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
result = runner_invoke(['push', '-f', '../deployment.template.json', '--no-build', '-P', get_platform_type()])
|
||||
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
@pytest.mark.skipif(platform.system().lower() != 'windows', reason='The path is not valid in non windows platform')
|
||||
def test_solution_build_and_push_with_escapedpath():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '-f', 'deployment.escapedpath.template.json', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
result = runner_invoke(['push', '--no-build', '-P', get_platform_type()])
|
||||
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
def test_solution_build_with_version_and_build_options():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
module_json_file_path = os.path.join(test_solution_shared_lib_dir, "modules", "sample_module", "module.json")
|
||||
module_2_json_file_path = os.path.join(test_solution_shared_lib_dir, "sample_module_2", "module.json")
|
||||
try:
|
||||
envvars.set_envvar("VERSION", "0.0.2")
|
||||
update_file_content(module_json_file_path, '"version": "0.0.1-RC"', '"version": "${VERSION}"')
|
||||
update_file_content(module_json_file_path, '"buildOptions": (.*),', '"buildOptions": [ "--add-host=github.com:192.30.255.112", "--build-arg a=b" ],')
|
||||
update_file_content(module_2_json_file_path, '"version": "0.0.1-RC"', '"version": "${VERSION}"')
|
||||
update_file_content(module_2_json_file_path, '"buildOptions": (.*),', '"buildOptions": [ "--add-host=github.com:192.30.255.112", "--build-arg a=b" ],')
|
||||
|
||||
result = runner_invoke(['build', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'sample_module:0.0.2' in result.output
|
||||
assert 'sample_module_2:0.0.2' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
assert '0.0.2' in get_all_docker_images()
|
||||
|
||||
finally:
|
||||
update_file_content(module_json_file_path, '"version": "(.*)"', '"version": "0.0.1-RC"')
|
||||
update_file_content(module_json_file_path, '"buildOptions": (.*),', '"buildOptions": [],')
|
||||
update_file_content(module_2_json_file_path, '"version": "(.*)"', '"version": "0.0.1-RC"')
|
||||
update_file_content(module_2_json_file_path, '"buildOptions": (.*),', '"buildOptions": [],')
|
||||
del os.environ["VERSION"]
|
||||
|
||||
|
||||
def test_solution_build_without_schema_template():
|
||||
try:
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
os.rename('deployment.template.json', 'deployment.template.backup.json')
|
||||
template_without_schema_version = os.path.join(tests_dir, "assets", "deployment.template_without_schema_template.json")
|
||||
shutil.copyfile(template_without_schema_version, 'deployment.template.json')
|
||||
|
||||
update_file_content('deployment.template.json', '"image": "(.*)MODULES.sample_module}",', '"image": "${MODULES.sample_module.' + get_platform_type() + '}",')
|
||||
|
||||
result = runner_invoke(['build'])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
config_file_path = os.path.join(test_solution_shared_lib_dir, "config", "deployment.json")
|
||||
assert os.path.exists(config_file_path)
|
||||
|
||||
content = get_file_content(config_file_path)
|
||||
assert "sample_module:0.0.1-RC-" + get_platform_type() in content
|
||||
finally:
|
||||
os.remove('deployment.template.json')
|
||||
os.rename('deployment.template.backup.json', 'deployment.template.json')
|
||||
|
||||
|
||||
def test_create_new_solution():
|
||||
os.chdir(tests_dir)
|
||||
clean_folder(test_solution_dir)
|
||||
|
|
@ -398,46 +277,6 @@ def test_create_new_solution():
|
|||
clean_folder(test_solution_dir)
|
||||
|
||||
|
||||
def test_solution_build_with_default_platform(prepare_solution_with_env):
|
||||
result = runner_invoke(['build'])
|
||||
|
||||
module_name = "filtermodule"
|
||||
test_solution_config_dir = os.path.join('config', 'deployment.' + get_platform_type() + '.json')
|
||||
env_container_registry_server = os.getenv("CONTAINER_REGISTRY_SERVER")
|
||||
with open(test_solution_config_dir) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
assert env_container_registry_server + "/" + module_name + ":0.0.1-" + get_platform_type() in content[
|
||||
"modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert module_name in get_all_docker_images()
|
||||
|
||||
|
||||
@pytest.mark.skipif(get_docker_os_type() == 'windows', reason='Debugger does not support C# in windows container')
|
||||
def test_solution_build_with_debug_template():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '-f', os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"], '-P', get_platform_type()])
|
||||
|
||||
module_name = "sample_module"
|
||||
module_2_name = "sample_module_2"
|
||||
test_solution_shared_debug_config = os.path.join('config', 'deployment.debug.' + get_platform_type() + '.json')
|
||||
env_container_registry_server = os.getenv("CONTAINER_REGISTRY_SERVER")
|
||||
with open(test_solution_shared_debug_config) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
assert env_container_registry_server + "/" + module_name + ":0.0.1-RC-" + get_platform_type() + ".debug" in content[
|
||||
"modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert env_container_registry_server + "/" + module_2_name + ":0.0.1-RC-" + get_platform_type() + ".debug" in content[
|
||||
"modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_2_name]["settings"]["image"]
|
||||
all_docker_images = get_all_docker_images()
|
||||
assert module_name in all_docker_images
|
||||
assert module_2_name in all_docker_images
|
||||
|
||||
|
||||
def test_solution_push_with_default_platform(prepare_solution_with_env):
|
||||
result = runner_invoke(['push'])
|
||||
|
||||
|
|
@ -519,12 +358,12 @@ def test_validate_deployment_template_and_manifest_failed():
|
|||
os.remove(os.path.join(tests_assets_dir, env_file_name))
|
||||
|
||||
|
||||
@mock.patch.dict(os.environ, {"CONTAINER_REGISTRY_PASSWORD": "nonempty"})
|
||||
def test_validate_deployment_template_and_manifest_success():
|
||||
try:
|
||||
deployment_file_name = "deployment.template.json"
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
shutil.copyfile(env_file_path, os.path.join(test_solution_shared_lib_dir, env_file_name))
|
||||
os.environ["CONTAINER_REGISTRY_PASSWORD"] = "nonempty"
|
||||
|
||||
if get_docker_os_type() == "windows":
|
||||
result = runner_invoke(['genconfig', '-P', get_platform_type(), '-f', deployment_file_name])
|
||||
|
|
@ -561,31 +400,31 @@ def test_validate_create_options_failed():
|
|||
assert "Warning: Errors found during createOptions validation" in result.output
|
||||
|
||||
|
||||
def test_fail_gen_config_on_validation_error():
|
||||
@pytest.mark.parametrize(
|
||||
"deployment_file_name",
|
||||
["deployment.manifest_invalid.json", "deployment.manifest_invalid_schema.json", "deployment.manifest_invalid_createoptions.json"]
|
||||
)
|
||||
def test_fail_gen_config_on_validation_error(deployment_file_name):
|
||||
os.chdir(tests_assets_dir)
|
||||
test_files = ["deployment.manifest_invalid.json", "deployment.manifest_invalid_schema.json", "deployment.manifest_invalid_createoptions.json"]
|
||||
for deployment_file_name in test_files:
|
||||
try:
|
||||
if get_docker_os_type() == "windows":
|
||||
result = runner_invoke(['genconfig', '-P', get_platform_type(), '-f', deployment_file_name, '--fail-on-validation-error'])
|
||||
else:
|
||||
result = runner_invoke(['genconfig', '-f', deployment_file_name, '--fail-on-validation-error'])
|
||||
raise Exception("genconfig command should fail in %s" % deployment_file_name)
|
||||
except Exception as err:
|
||||
assert "ERROR: Deployment manifest validation failed. Please see previous logs for more details." in "%s" % err
|
||||
assert "genconfig command should fail" not in "%s" % err
|
||||
|
||||
with pytest.raises(Exception) as context:
|
||||
if get_docker_os_type() == "windows":
|
||||
result = runner_invoke(['genconfig', '-P', get_platform_type(), '-f', deployment_file_name, '--fail-on-validation-error'])
|
||||
else:
|
||||
result = runner_invoke(['genconfig', '-f', deployment_file_name, '--fail-on-validation-error'])
|
||||
|
||||
if get_docker_os_type() == "windows":
|
||||
result = runner_invoke(['genconfig', '-P', get_platform_type(), '-f', deployment_file_name])
|
||||
else:
|
||||
result = runner_invoke(['genconfig', '-f', deployment_file_name])
|
||||
|
||||
assert "ERROR: Deployment manifest validation failed. Please see previous logs for more details." in str(context.value)
|
||||
assert "ERROR" not in result.output
|
||||
|
||||
|
||||
@mock.patch.dict(os.environ, {"TTL": "7200"})
|
||||
def test_gen_config_with_non_string_placeholder():
|
||||
os.chdir(tests_assets_dir)
|
||||
os.environ["TTL"] = "7200"
|
||||
deployment_file_name = "deployment.template.non_str_placeholder.json"
|
||||
if get_docker_os_type() == "windows":
|
||||
result = runner_invoke(['genconfig', '-P', get_platform_type(), '-f', deployment_file_name, '--fail-on-validation-error'])
|
||||
|
|
@ -595,42 +434,24 @@ def test_gen_config_with_non_string_placeholder():
|
|||
assert "ERROR" not in result.output
|
||||
|
||||
|
||||
@mock.patch.dict(os.environ, {"CONTAINER_REGISTRY_SERVER": "localhost:5000"})
|
||||
@pytest.mark.skipif(get_docker_os_type() == 'windows', reason='windows container does not support local registry image')
|
||||
def test_push_modules_to_local_registry(prepare_solution_with_env):
|
||||
env_container_registry_server = os.getenv("CONTAINER_REGISTRY_SERVER")
|
||||
try:
|
||||
module_name = "filtermodule"
|
||||
|
||||
if module_name in get_all_docker_images():
|
||||
remove_docker_image(module_name)
|
||||
|
||||
local_registry = "localhost:5000"
|
||||
envvars.set_envvar("CONTAINER_REGISTRY_SERVER", local_registry)
|
||||
|
||||
result = runner_invoke(['push', '-P', get_platform_type()])
|
||||
if result.exit_code == 0:
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
assert local_registry + "/" + module_name in get_all_docker_images()
|
||||
else:
|
||||
raise Exception(result.stdout)
|
||||
|
||||
assert 'ERROR' not in result.output
|
||||
assert result.exit_code == 0
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert f"localhost:5000/{module_name in get_all_docker_images()}"
|
||||
finally:
|
||||
envvars.set_envvar("CONTAINER_REGISTRY_SERVER", env_container_registry_server)
|
||||
if "registry" in get_all_docker_containers():
|
||||
remove_docker_container("registry")
|
||||
if "registry" in get_all_docker_images():
|
||||
remove_docker_image("registry:2")
|
||||
|
||||
# # TODO: The output of docker build logs is not captured by pytest, need to capture this before enable this test
|
||||
# def test_docker_build_status_output():
|
||||
# prune_docker_images()
|
||||
# prune_docker_containers()
|
||||
# prune_docker_build_cache()
|
||||
# remove_docker_image("sample_module:0.0.1-RC")
|
||||
|
||||
# os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
# result = runner_invoke(['build', '-P', get_platform_type()])
|
||||
|
||||
# assert re.match('\\[=*>\\s*\\]', result.output) is not None
|
||||
|
|
@ -0,0 +1,246 @@
|
|||
import json
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.output import Output
|
||||
|
||||
from .utility import (get_all_docker_images, get_docker_os_type,
|
||||
get_file_content, get_platform_type, runner_invoke,
|
||||
update_file_content)
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
output = Output()
|
||||
envvars = EnvVars(output)
|
||||
|
||||
tests_dir = os.path.join(os.getcwd(), "tests")
|
||||
tests_assets_dir = os.path.join(tests_dir, "assets")
|
||||
|
||||
env_file_name = envvars.get_dotenv_file()
|
||||
env_file_path = envvars.get_dotenv_path(env_file_name)
|
||||
|
||||
launch_json_file = os.path.join(tests_dir, "assets", "launch.json")
|
||||
launch_json_file_without_nodejs = os.path.join(tests_assets_dir, "launch_without_nodejs.json")
|
||||
test_solution_shared_lib_dir = os.path.join(tests_assets_dir, "test_solution_shared_lib")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def prepare_solution_with_env():
|
||||
os.chdir(tests_dir)
|
||||
test_solution = "test_solution"
|
||||
test_solution_dir = os.path.join(tests_dir, test_solution)
|
||||
template = "csharp"
|
||||
module_name = "filtermodule"
|
||||
result = runner_invoke(['new', test_solution, '-m', module_name, '-t', template])
|
||||
|
||||
if 'AZURE IOT EDGE SOLUTION CREATED' not in result.output:
|
||||
raise Exception(result.stdout)
|
||||
|
||||
shutil.copyfile(env_file_path, os.path.join(test_solution_dir, env_file_name))
|
||||
|
||||
os.chdir(test_solution_dir)
|
||||
|
||||
yield prepare_solution_with_env
|
||||
os.chdir(tests_dir)
|
||||
time.sleep(5)
|
||||
shutil.rmtree(test_solution_dir, ignore_errors=True)
|
||||
|
||||
return
|
||||
|
||||
|
||||
def test_solution_build_and_push_with_layered_deployment():
|
||||
# Arrange
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
new_config_deployment_path = os.path.join(test_solution_shared_lib_dir, 'config', "layered_deployment.flattened_props.json")
|
||||
|
||||
# Act
|
||||
result = runner_invoke(['build', '--push', '-f', "layered_deployment.flattened_props.template.json", '-P', get_platform_type()])
|
||||
|
||||
# Assert
|
||||
assert os.path.exists(new_config_deployment_path)
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
def test_solution_build_and_push_with_layered_deployment_no_modules():
|
||||
# Arrange
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
new_config_deployment_path = os.path.join(test_solution_shared_lib_dir, 'config', 'layered_deployment.no_modules.json')
|
||||
|
||||
# Act
|
||||
result = runner_invoke(['build', '--push', '-f', "layered_deployment.no_modules.template.json", '-P', get_platform_type()])
|
||||
|
||||
# Assert
|
||||
|
||||
with open(new_config_deployment_path, "r") as f:
|
||||
content = json.load(f)
|
||||
|
||||
set_property = content["content"]["modulesContent"]["exampleModule"]["properties.desired"]["foo"]
|
||||
|
||||
assert 'ERROR' not in result.output
|
||||
assert 'bar-1.2' == set_property
|
||||
|
||||
|
||||
def test_solution_build_and_push_with_platform():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'PUSH COMPLETE' not in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
result = runner_invoke(['push', '--no-build', '-P', get_platform_type()])
|
||||
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'BUILD COMPLETE' not in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
def test_solution_build_and_push_with_different_cwd():
|
||||
cwd = os.path.join(test_solution_shared_lib_dir, 'config')
|
||||
if not os.path.exists(cwd):
|
||||
os.makedirs(cwd)
|
||||
os.chdir(cwd)
|
||||
|
||||
result = runner_invoke(['build', '-f', '../deployment.template.json', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
result = runner_invoke(['push', '-f', '../deployment.template.json', '--no-build', '-P', get_platform_type()])
|
||||
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'sample_module:0.0.1-RC' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
@pytest.mark.skipif(platform.system().lower() != 'windows', reason='The path is not valid in non windows platform')
|
||||
def test_solution_build_and_push_with_escapedpath():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '-f', 'deployment.escapedpath.template.json', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
result = runner_invoke(['push', '--no-build', '-P', get_platform_type()])
|
||||
|
||||
assert 'PUSH COMPLETE' in result.output
|
||||
assert 'sample_module_2:0.0.1-RC' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
|
||||
@mock.patch.dict(os.environ, {"VERSION": "0.0.2"})
|
||||
def test_solution_build_with_version_and_build_options():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
module_json_file_path = os.path.join(test_solution_shared_lib_dir, "modules", "sample_module", "module.json")
|
||||
module_2_json_file_path = os.path.join(test_solution_shared_lib_dir, "sample_module_2", "module.json")
|
||||
try:
|
||||
update_file_content(module_json_file_path, '"version": "0.0.1-RC"', '"version": "${VERSION}"')
|
||||
update_file_content(module_json_file_path, '"buildOptions": (.*),', '"buildOptions": [ "--add-host=github.com:192.30.255.112", "--build-arg a=b" ],')
|
||||
update_file_content(module_2_json_file_path, '"version": "0.0.1-RC"', '"version": "${VERSION}"')
|
||||
update_file_content(module_2_json_file_path, '"buildOptions": (.*),', '"buildOptions": [ "--add-host=github.com:192.30.255.112", "--build-arg a=b" ],')
|
||||
|
||||
result = runner_invoke(['build', '-P', get_platform_type()])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'sample_module:0.0.2' in result.output
|
||||
assert 'sample_module_2:0.0.2' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
assert '0.0.2' in get_all_docker_images()
|
||||
|
||||
finally:
|
||||
update_file_content(module_json_file_path, '"version": "(.*)"', '"version": "0.0.1-RC"')
|
||||
update_file_content(module_json_file_path, '"buildOptions": (.*),', '"buildOptions": [],')
|
||||
update_file_content(module_2_json_file_path, '"version": "(.*)"', '"version": "0.0.1-RC"')
|
||||
update_file_content(module_2_json_file_path, '"buildOptions": (.*),', '"buildOptions": [],')
|
||||
|
||||
|
||||
def test_solution_build_without_schema_template():
|
||||
try:
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
os.rename('deployment.template.json', 'deployment.template.backup.json')
|
||||
template_without_schema_version = os.path.join(tests_dir, "assets", "deployment.template_without_schema_template.json")
|
||||
shutil.copyfile(template_without_schema_version, 'deployment.template.json')
|
||||
|
||||
update_file_content('deployment.template.json', '"image": "(.*)MODULES.sample_module}",', '"image": "${MODULES.sample_module.' + get_platform_type() + '}",')
|
||||
|
||||
result = runner_invoke(['build'])
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
|
||||
config_file_path = os.path.join(test_solution_shared_lib_dir, "config", "deployment.json")
|
||||
assert os.path.exists(config_file_path)
|
||||
|
||||
content = get_file_content(config_file_path)
|
||||
assert "sample_module:0.0.1-RC-" + get_platform_type() in content
|
||||
finally:
|
||||
os.remove('deployment.template.json')
|
||||
os.rename('deployment.template.backup.json', 'deployment.template.json')
|
||||
|
||||
|
||||
def test_solution_build_with_default_platform(prepare_solution_with_env):
|
||||
result = runner_invoke(['build'])
|
||||
|
||||
module_name = "filtermodule"
|
||||
test_solution_config_dir = os.path.join('config', 'deployment.' + get_platform_type() + '.json')
|
||||
env_container_registry_server = os.getenv("CONTAINER_REGISTRY_SERVER")
|
||||
with open(test_solution_config_dir) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
assert env_container_registry_server + "/" + module_name + ":0.0.1-" + get_platform_type() in content[
|
||||
"modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert module_name in get_all_docker_images()
|
||||
|
||||
|
||||
@pytest.mark.skipif(get_docker_os_type() == 'windows', reason='Debugger does not support C# in windows container')
|
||||
def test_solution_build_with_debug_template():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '-f', os.environ["DEPLOYMENT_CONFIG_DEBUG_TEMPLATE_FILE"], '-P', get_platform_type()])
|
||||
|
||||
module_name = "sample_module"
|
||||
module_2_name = "sample_module_2"
|
||||
test_solution_shared_debug_config = os.path.join('config', 'deployment.debug.' + get_platform_type() + '.json')
|
||||
env_container_registry_server = os.getenv("CONTAINER_REGISTRY_SERVER")
|
||||
with open(test_solution_shared_debug_config) as f:
|
||||
content = json.load(f)
|
||||
|
||||
assert 'BUILD COMPLETE' in result.output
|
||||
assert 'ERROR' not in result.output
|
||||
assert env_container_registry_server + "/" + module_name + ":0.0.1-RC-" + get_platform_type() + ".debug" in content[
|
||||
"modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_name]["settings"]["image"]
|
||||
assert env_container_registry_server + "/" + module_2_name + ":0.0.1-RC-" + get_platform_type() + ".debug" in content[
|
||||
"modulesContent"]["$edgeAgent"]["properties.desired"]["modules"][module_2_name]["settings"]["image"]
|
||||
all_docker_images = get_all_docker_images()
|
||||
assert module_name in all_docker_images
|
||||
assert module_2_name in all_docker_images
|
||||
|
||||
|
||||
@pytest.mark.skipif(get_docker_os_type() == 'windows', reason='The output of docker build logs is not captured by logs on windows, need to capture this before enabling this test')
|
||||
def test_verify_docker_build_status_in_output():
|
||||
os.chdir(test_solution_shared_lib_dir)
|
||||
|
||||
result = runner_invoke(['build', '-f' 'deployment.debug.template.json', '-P', get_platform_type()])
|
||||
|
||||
assert all(x in result.output for x in ["--->", "Step 1/", "Successfully tagged"])
|
||||
|
|
@ -1,12 +1,13 @@
|
|||
import os
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from iotedgedev.envvars import EnvVars
|
||||
from iotedgedev.output import Output
|
||||
from iotedgedev.utility import Utility
|
||||
|
||||
from .utility import assert_list_equal, assert_file_equal, assert_json_file_equal
|
||||
from .utility import (assert_file_equal, assert_json_file_equal,
|
||||
assert_list_equal)
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
|
@ -59,12 +60,12 @@ def test_copy_template(utility, tmpdir):
|
|||
assert_json_file_equal(test_file_2, dest)
|
||||
|
||||
|
||||
@mock.patch.dict(os.environ, {"CONTAINER_REGISTRY_SERVER": "localhost:5000"})
|
||||
def test_copy_template_expandvars(utility, tmpdir):
|
||||
replacements = {
|
||||
"${MODULES.csharpmodule.amd64}": "${CONTAINER_REGISTRY_SERVER}/csharpmodule:0.0.1-amd64",
|
||||
"${MODULES.csharpfunction.amd64.debug}": "${CONTAINER_REGISTRY_SERVER}/csharpfunction:0.0.1-amd64.debug"
|
||||
}
|
||||
os.environ["CONTAINER_REGISTRY_SERVER"] = "localhost:5000"
|
||||
dest = tmpdir.join("deployment_template_2.dest.json").strpath
|
||||
dest2 = tmpdir.join("deployment_template_2.dest2.json").strpath
|
||||
utility.copy_template(test_file_1, dest, replacements=replacements, expandvars=True)
|
||||
|
|
@ -116,6 +117,11 @@ def test_get_sha256_hash():
|
|||
assert Utility.get_sha256_hash("foo") == "2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae"
|
||||
|
||||
|
||||
@mock.patch.dict(os.environ, {"DEPLOYMENT_CONFIG_FILE": "foo.json"})
|
||||
def test_get_deployment_manifest_name__set_from_envvar():
|
||||
assert Utility.get_deployment_manifest_name("deployment.debug.template.json", "1.0.0", "amd64") == "foo.json"
|
||||
|
||||
|
||||
def test_get_deployment_manifest_name():
|
||||
assert Utility.get_deployment_manifest_name("config/deployment.template.json", "0.0.1", "amd64") == "deployment.json"
|
||||
assert Utility.get_deployment_manifest_name("deployment.template.json", "0.0.1", "amd64") == "deployment.json"
|
||||
|
|
@ -124,6 +130,3 @@ def test_get_deployment_manifest_name():
|
|||
assert Utility.get_deployment_manifest_name("deployment.template.json", "1.0.0", "amd64") == "deployment.amd64.json"
|
||||
assert Utility.get_deployment_manifest_name("deployment.debug.template.json", "1.0.0", "amd64") == "deployment.debug.amd64.json"
|
||||
assert Utility.get_deployment_manifest_name("", "", "") == "deployment.json"
|
||||
|
||||
os.environ["DEPLOYMENT_CONFIG_FILE"] = "foo.json"
|
||||
assert Utility.get_deployment_manifest_name("deployment.debug.template.json", "1.0.0", "amd64") == "foo.json"
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from click.testing import CliRunner
|
||||
from iotedgedev.dockercls import Docker
|
||||
|
|
@ -87,7 +88,10 @@ def remove_docker_image(image_name):
|
|||
def runner_invoke(args, expect_failure=False):
|
||||
runner = CliRunner()
|
||||
with runner.isolation(env={"DEFAULT_PLATFORM": get_platform_type()}):
|
||||
cli = __import__("iotedgedev.cli", fromlist=['main'])
|
||||
iotedgedev_import = "iotedgedev.cli"
|
||||
cli = __import__(iotedgedev_import, fromlist=['main'])
|
||||
# Remove "iotedgedev.cli" import from cache, to prevent variables being saved across tests
|
||||
del sys.modules[iotedgedev_import]
|
||||
result = runner.invoke(cli.main, args)
|
||||
if (result.exit_code == 0) or (expect_failure is True):
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ steps:
|
|||
pip install --upgrade tox
|
||||
sudo npm i -g iothub-explorer
|
||||
az --version
|
||||
az extension add --name azure-iot
|
||||
displayName: "Update and install required tools"
|
||||
|
||||
- script: |
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ steps:
|
|||
- pwsh: |
|
||||
npm i -g iothub-explorer yo generator-azure-iot-edge-module
|
||||
az --version
|
||||
az extension add --name azure-iot
|
||||
& "C:\Program Files (x86)\Microsoft SDKs\Azure\CLI2\python.exe" -m pip install -U setuptools==52.0.0
|
||||
displayName: "Install IoT Hub explorer, Yeoman and Azure IoT Edge Node.js module generator packages"
|
||||
|
||||
|
|
@ -19,8 +20,8 @@ steps:
|
|||
mkdir C:\registry
|
||||
docker run -d -p 5000:5000 --restart=always --name registry -v C:\registry:C:\registry stefanscherer/registry-windows:2.6.2
|
||||
displayName: "Pull and run local registry containers"
|
||||
|
||||
|
||||
- pwsh: |
|
||||
pip install tox
|
||||
tox -e "$(TOXENV)"
|
||||
displayName: "Run tests against iotedgedev source code"
|
||||
displayName: "Run tests against iotedgedev source code"
|
||||
|
|
|
|||
Загрузка…
Ссылка в новой задаче