Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup(self):
configure_lang({})
opsdroid = mock.MagicMock()
loader = ld.Loader(opsdroid)
return opsdroid, loader
def test_import_module_new(self):
config = {}
config["module_path"] = "os"
config["name"] = ""
config["type"] = "system"
config["module"] = ""
module = ld.Loader.import_module(config)
self.assertIsInstance(module, ModuleType)
def setup(self):
configure_lang({})
opsdroid = mock.MagicMock()
loader = ld.Loader(opsdroid)
return opsdroid, loader
Args:
ctx (:obj:`click.Context`): The current click cli context.
path (string): a string representing the path to load the config,
obtained from `ctx.obj`.
value (string): the value of this parameter after invocation.
It is either "config" or "log" depending on the program
calling this function.
Returns:
int: the exit code. Always returns 0 in this case.
"""
with OpsDroid() as opsdroid:
loader = Loader(opsdroid)
config = load_config_file([path] if path else DEFAULT_CONFIG_LOCATIONS)
loader.load_modules_from_config(config)
click.echo("Configuration validated - No errors founds!")
ctx.exit(0)
module_spec = None
namespaces = [
config["module"],
config["module_path"] + "." + config["name"],
config["module_path"],
]
for namespace in namespaces:
try:
module_spec = importlib.util.find_spec(namespace)
if module_spec:
break
except (ImportError, AttributeError):
continue
if module_spec:
module = Loader.import_module_from_spec(module_spec)
_LOGGER.debug(_("Loaded %s: %s."), config["type"], config["module_path"])
return module
_LOGGER.error(
_("Failed to load %s: %s."), config["type"], config["module_path"]
)
return None
self._running = False
self.sys_status = 0
self.connectors = []
self.connector_tasks = []
self.eventloop = asyncio.get_event_loop()
if os.name != "nt":
for sig in (signal.SIGINT, signal.SIGTERM):
self.eventloop.add_signal_handler(
sig, lambda: asyncio.ensure_future(self.handle_signal())
)
self.eventloop.set_exception_handler(self.handle_async_exception)
self.skills = []
self.memory = Memory()
self.modules = {}
self.cron_task = None
self.loader = Loader(self)
if config is None:
self.config = {}
else:
self.config = config
self.stats = {
"messages_parsed": 0,
"webhooks_called": 0,
"total_response_time": 0,
"total_responses": 0,
}
self.web_server = None
self.stored_path = []
def git_pull(repository_path):
"""Pull the current branch of git repo forcing fast forward.
Args:
repository_path: Path to the module's local repository
"""
process = subprocess.Popen(
["git", "-C", repository_path, "pull", "--ff-only"],
shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
Loader._communicate_process(process)