How to use _try_create_symlink method in Slash

Best Python code snippet using slash

log.py

Source:log.py

...129                self._create_last_failed_symlink_if_needed(result)130    def _create_last_failed_symlink_if_needed(self, result):131        assert result132        if result.is_error() or result.is_failure():133            self._try_create_symlink(result.get_log_path(), config.root.log.last_failed_symlink)134    @contextmanager135    def get_session_logging_context(self):136        assert self.session_log_handler is None137        with self._get_file_logging_context(138            config.root.log.session_subpath, config.root.log.last_session_symlink) as (handler, path):139            self.session_log_handler = handler140            self.session_log_path = path141            self.session.results.global_result.set_log_path(path)142            if config.root.log.last_session_dir_symlink is not None and self.session_log_path is not None:143                self._try_create_symlink(os.path.dirname(self.session_log_path), config.root.log.last_session_dir_symlink)144            yield path145    @contextmanager146    def _get_file_logging_context(self, filename_template, symlink):147        with ExitStack() as stack:148            handler = stack.enter_context(self._log_file_handler_context(filename_template, symlink, \149                                                                         use_compression=config.root.log.compression.enabled))150            stack.enter_context(handler.applicationbound())151            if config.root.log.compression.enabled and config.root.log.compression.use_rotating_raw_file:152                rotating_handler = stack.enter_context(self._log_file_handler_context(filename_template, symlink, bubble=True, use_rotation=True))153                stack.enter_context(rotating_handler.applicationbound())154            stack.enter_context(self.console_handler.applicationbound())155            stack.enter_context(self.warnings_handler.applicationbound())156            error_handler = stack.enter_context(self._get_error_logging_context())157            stack.enter_context(error_handler.applicationbound())158            stack.enter_context(self._get_silenced_logs_context())159            if config.root.log.unittest_mode:160                stack.enter_context(logbook.StreamHandler(sys.stderr, bubble=True, level=logbook.TRACE))161            for extra_handler in _extra_handlers:162                stack.enter_context(extra_handler.applicationbound())163            if config.root.log.unified_session_log and self.session_log_handler is not None:164                stack.enter_context(_make_bubbling_handler(self.session_log_handler))165            if config.root.run.capture.error_logs_as_errors:166                stack.enter_context(ErrorHandler().applicationbound())167            path = handler.stream.name if isinstance(handler, logbook.FileHandler) else None168            yield handler, path169    def _should_delete_log(self, result):170        return (not config.root.log.cleanup.keep_failed) or \171               (not result.is_global_result() and result.is_success(allow_skips=True)) or \172               (result.is_global_result() and self.session.results.is_success(allow_skips=True))173    @contextmanager174    def _get_error_logging_context(self):175        with ExitStack() as stack:176            path = config.root.log.highlights_subpath177            def _error_added_filter(record, handler): # pylint: disable=unused-argument178                return record.extra.get('highlight')179            handler = stack.enter_context(self._log_file_handler_context(path, symlink=None, bubble=True, filter=_error_added_filter))180            log_path = handler.stream.name if isinstance(handler, logbook.FileHandler) else None181            if log_path and self.session.results.current is self.session.results.global_result:182                self.session.results.global_result.add_extra_log_path(log_path)183            yield handler184    def _get_silenced_logs_context(self):185        if not config.root.log.silence_loggers:186            return ExitStack()187        return SilencedLoggersHandler(config.root.log.silence_loggers).applicationbound()188    def _get_log_file_path(self, subpath, use_compression):189        log_path = self._normalize_path(os.path.join(config.root.log.root, _format_log_path(subpath)))190        if use_compression:191            if config.root.log.compression.algorithm == "gzip":192                log_path += ".gz"193            elif config.root.log.compression.algorithm == "brotli":194                log_path += ".br"195            else:196                raise InvalidConfiguraion("Unsupported compression method: {}".format(config.root.log.compression.algorithm))197        return log_path198    def _create_log_file_handler(self, log_path, bubble=False, filter=_slash_logs_filter, use_compression=False, use_rotation=False):199        kwargs = {"bubble": bubble, "filter": filter}200        if use_compression:201            if config.root.log.compression.algorithm == "gzip":202                handler_class = logbook.GZIPCompressionHandler203            elif config.root.log.compression.algorithm == "brotli":204                handler_class = logbook.BrotliCompressionHandler205        elif use_rotation:206            kwargs.update({"max_size": 4*1024**2, "backup_count": 1})207            handler_class = logbook.RotatingFileHandler208        elif config.root.log.colorize:209            handler_class = ColorizedFileHandler210        else:211            handler_class = logbook.FileHandler212        return handler_class(log_path, **kwargs)213    @contextmanager214    def _log_file_handler_context(self, subpath, symlink, bubble=False, filter=_slash_logs_filter, use_compression=False, use_rotation=False):215        if subpath is None or config.root.log.root is None:216            yield NoopHandler() if bubble else logbook.NullHandler(filter=filter)217        else:218            log_path = self._get_log_file_path(subpath, use_compression)219            handler = self._log_path_to_handler.get(log_path, None)220            if handler is not None:221                yield handler222            else:223                result = context.result224                ensure_containing_directory(log_path)225                if symlink:226                    self._try_create_symlink(log_path, symlink)227                handler = self._create_log_file_handler(log_path, bubble=bubble, use_compression=use_compression,228                                                        use_rotation=use_rotation, filter=filter)229                try:230                    self._log_path_to_handler[log_path] = handler231                    self._set_formatting(handler, config.root.log.format)232                    with handling_exceptions():233                        yield handler234                finally:235                    handler.close()236                    self._log_path_to_handler[log_path] = None237                    with handling_exceptions(swallow=True):238                        hooks.log_file_closed(path=log_path, result=result)  # pylint: disable=no-member239                    if config.root.log.cleanup.enabled and self._should_delete_log(result):240                        with handling_exceptions(swallow=True):241                            os.remove(log_path)242                            dir_path = os.path.dirname(log_path)243                            if not os.listdir(dir_path) and dir_path != self._normalize_path(config.root.log.root):244                                os.rmdir(dir_path)245    def _normalize_path(self, p):246        return os.path.expanduser(p)247    def create_worker_symlink(self, worker_name, worker_session_id):248        if config.root.log.root is None:249            return250        symlink = os.path.join(self.session.id, worker_name)251        worker_dir = os.path.join(self._normalize_path(config.root.log.root), worker_session_id)252        self._try_create_symlink(worker_dir, symlink)253    def _try_create_symlink(self, path, symlink):254        if symlink is None or config.root.log.root is None or config.root.parallel.worker_id is not None:255            return256        symlink = self._normalize_path(symlink)257        if not os.path.isabs(symlink):258            symlink = os.path.join(self._normalize_path(config.root.log.root), symlink)259        try:260            ensure_containing_directory(symlink)261            if os.path.exists(symlink) or os.path.islink(symlink):262                os.unlink(symlink)263            os.symlink(path, symlink)264        except Exception:  # pylint: disable=broad-except265            _logger.debug("Failed to create symlink {0} --> {1}", path, symlink, exc_info=True)266    def _set_formatting(self, handler, fmt):267        if config.root.log.localtime:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.