Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def start():
if environ.get('DEBUGGING'):
ptvsd.enable_attach(address=(
'127.0.0.1', int(environ.get('DEBUG_PORT'))))
print(ATTACH_DEBUGGER_EVENT)
t = Timer(int(environ.get("debugger_wait_time", 30)), _handle_detached)
t.start()
ptvsd.wait_for_attach()
t.cancel()
logger.debug('Starting grpc server..')
server = grpc.server(ThreadPoolExecutor(max_workers=1))
p = server.add_insecure_port('127.0.0.1:0')
handler = handlers.GrpcServiceHandler(server)
spg.add_RunnerServicer_to_server(handler, server)
logger.info('Listening on port:{}'.format(p))
server.start()
t = threading.Thread(name="listener", target=handler.wait_for_kill_event)
t.start()
t.join()
os._exit(0)
def _import_file(base_dir, file_path):
rel_path = os.path.normpath(file_path.replace(base_dir + os.path.sep, ''))
try:
module_name = os.path.splitext(rel_path.replace(os.path.sep, '.'))[0]
m = importlib.import_module(module_name)
# Get all classes in the imported module
classes = inspect.getmembers(m, lambda member: inspect.isclass(member) and member.__module__ == module_name)
if len(classes) > 0:
for c in classes:
file = inspect.getfile(c[1])
# Create instance of step implementation class.
if _has_methods_with_gauge_decoratores(c[1]):
update_step_resgistry_with_class(c[1](), file_path) # c[1]() will create a new instance of the class
except:
logger.error('Exception occurred while loading step implementations from file: {}.'.format(rel_path))
logger.error(traceback.format_exc())
def load_implementations():
d = get_step_impl_dirs()
logger.debug(
"Loading step implementations from {} dirs.".format(', '.join(d)))
for impl_dir in d:
if not path.exists(impl_dir):
logger.error('can not load implementations from {}. {} does not exist.'.format(
impl_dir, impl_dir))
load_files(d)
def load_impls(step_impl_dirs=impl_dirs):
os.chdir(project_root)
for impl_dir in step_impl_dirs:
if not os.path.isdir(impl_dir):
logger.error('Cannot import step implementations. Error: {} does not exist.'.format(step_impl_dirs))
logger.error('Make sure `STEP_IMPL_DIR` env var is set to a valid directory path.')
return
base_dir = project_root if impl_dir.startswith(project_root) else os.path.dirname(impl_dir)
_import_impl(base_dir, impl_dir)
def main():
logger.info("Python: {}".format(platform.python_version()))
if sys.argv[1] == "--init":
logger.debug("Initializing gauge project.")
copy_skel_files()
else:
load_implementations()
start()
def _take_screenshot():
temp_file = _uniqe_screenshot_file()
try:
call(['gauge_screenshot', temp_file])
return os.path.basename(temp_file)
except Exception as err:
logger.error(
"\nFailed to take screenshot using gauge_screenshot.\n{0}".format(err))
except:
logger.error("\nFailed to take screenshot using gauge_screenshot.\n{0}".format(
sys.exc_info()[0]))
return str.encode("")
def capture_to_file():
if not registry.is_screenshot_writer:
screenshot_file = _uniqe_screenshot_file()
content = registry.screenshot_provider()()
file = open(screenshot_file, "wb")
file.write(content)
file.close()
return os.path.basename(screenshot_file)
screenshot_file = registry.screenshot_provider()()
if(not os.path.isabs(screenshot_file)):
screenshot_file = os.path.join(_screenshots_dir(), screenshot_file)
if(not os.path.exists(screenshot_file)):
logger.warning("Screenshot file {0} does not exists.".format(screenshot_file))
return os.path.basename(screenshot_file)