KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use copy_config method in tempest

Best Python code snippet using tempest_python

content_copy.py

Source:content_copy.py

1import os.path as path2import sys3import traceback4import lib.util as util5import lib.command_line_interface as cli6import lib.operation_objects as op7import lib.bookmap as bkmap8import lib.role_updates as role9import re as regex10import subprocess11import signal12from lib.util import CCTError13from lib.http_util import credentials_valid14"""15This script is the main script of the content-copy-tool, it requires the16presence of the following utility files to execute properly.17configuration_objects.py18operation_objects.py19bookmap.py20role_updates.py21util.py22http_util.py23command_line_interface.py24"""25here = path.abspath(path.dirname(__file__))26version = {}27with open(path.join(here, '__version__.py')) as f:28  exec(f.read(), version)29VERSION = 'Content-Copy-Tool v%s' % version['__version__']30PRODUCTION = True31def run(settings, input_file, run_options):32    try:33        config = util.parse_json(settings)34    except Exception, e:35        print("There was an error reading the settings file: %s, confirm that the formatting compiles with "36                     "all json standards." % settings)37        print(traceback.format_exc())38        print("There was an error reading the settings file: %s, confirm that the formatting compiles with "39                     "all json standards." % settings)40        sys.exit(1)41    logfile = config['logfile']42    logger = util.init_logger(logfile, run_options.verboselevel)43    logger.debug("Logger is up and running.")44    logger.debugv("Verbose DEBUGV level for developers is activated.")45    # Bookmap46    bookmap_config = bkmap.BookmapConfiguration(str(config['chapter_number_column']),47                                                str(config['chapter_title_column']),48                                                str(config['module_title_column']),49                                                str(config['source_module_ID_column']),50                                                str(config['destination_module_ID_column']),51                                                str(config['destination_workgroup_column']),52                                                str(config['unit_number_column']),53                                                str(config['unit_title_column']),54                                                str(config['strip_section_numbers']))55    logger.debug("Bookmap configuration has been created")56    bookmap = bkmap.Bookmap(input_file, bookmap_config, run_options, logger)57    logger.debug("Bookmap has been created")58    # Copy Configuration and Copier59    source_server = str(config['source_server'])60    destination_server = str(config['destination_server'])61    # ensure server addresses have 'http[s]://' prepended62    if not regex.match(r'https?://', source_server):63        source_server = "http://%s" % source_server64    if not regex.match(r'https?://', destination_server):65        destination_server = "http://%s" % destination_server66    credentials = str(config['destination_credentials'])67    copy_config = op.CopyConfiguration(source_server, destination_server, credentials)68    copier = op.Copier(copy_config, bookmap.bookmap, str(config['path_to_tool']))69    logger.debug("Copier has been created")70    # Role Configuration71    role_config = role.RoleConfiguration(list(config['authors']),72                                         list(config['maintainers']),73                                         list(config['rightsholders']), config, credentials)74    role_updater = role.RoleUpdater(role_config)75    logger.debug("Role configuration has been created.")76    # Content_creator77    content_creator = op.ContentCreator(destination_server, credentials)78    logger.debug("ContentCreator has been created.")79    failures = []80    user_confirm(logger, copy_config, bookmap, run_options, role_config)  # Check before you run81    try:82        logger.debug("Beginning processing.")83        logger.debug("Testing credentials.")84        if (not credentials_valid(content_creator.credentials, content_creator.server) or85           not all(map(lambda user: credentials_valid(user, copy_config.destination_server), role_updater.get_users_of_roles()))):86            raise CCTError("Credentials invalid")87        if run_options.modules or run_options.workgroups:  # create placeholders88            create_placeholders(logger, bookmap, copy_config, run_options, content_creator, failures)89            output = bookmap.save(run_options.units)  # save output data90            logger.debug("Finished created placeholders, output has been saved: %s." % output)91        if run_options.copy:  # copy content92            copier.copy_content(role_config, run_options, logger, failures)93            logger.debug("Finished copying content.")94        if run_options.roles and not run_options.dryrun:  # accept all pending role requests95            role_updater.accept_roles(copy_config, logger, failures)96            logger.debug("Finished updating roles.")97        if run_options.collections:  # create and populate the collection98            create_populate_and_publish_collection(content_creator, copy_config, bookmap, run_options.units,99                                                   run_options.publish_collection, run_options.dryrun, logger, failures)100            logger.debug("Finished creating and populating the collection.")101        if run_options.publish:  # publish the modules102            publish_modules_post_copy(copier, content_creator, run_options, credentials, logger, failures)103            logger.debug("Finished publishing modules.")104    except (CCTError, util.TerminateError, util.SkipSignal) as e:105        output = bookmap.save(run_options.units, True)106        logger.error(e.msg)107    if run_options.modules or run_options.workgroups:108        logger.info("See output: \033[95m%s\033[0m" % output)109    print_failures(logger, failures)110    logger.info("------- Process completed --------")111    return bookmap.booktitle112def create_placeholders(logger, bookmap, copy_config, run_options, content_creator, failures):113    """114    Creates placeholder modules on the destination server (and workgroups if enables).115    Arguments:116        logger - the tool's logger117        bookmap - the bookmap of the input data parsed from the input file118        copy_config - the configuration of the copier with source and destination urls and credentials119        run_options - the input running options, what the tool should be doing120        content_creator - the content creator object121        failures - the list of failures to track failed placeholder creations122    Returns:123        None124    """125    if run_options.workgroups:126        logger.info("-------- Creating workgroups ------------------------")127        chapter_to_workgroup = {}128        workgroups_to_remove = []129        for workgroup in bookmap.bookmap.workgroups:130            try:131                content_creator.run_create_workgroup(workgroup, copy_config.destination_server, copy_config.credentials,132                                                     logger, dryrun=run_options.dryrun)133                logger.debug("[CREATED WORKGROUP] %s - %s" % (workgroup.title, workgroup.url))134            except util.TerminateError:135                raise util.TerminateError("Terminate Signaled")136            except (CCTError, util.SkipSignal, Exception) as e:137                if type(e) is not CCTError and type(e) is not util.SkipSignal:138                    logger.error("Problematic Error")139                    logger.debug(traceback.format_exc())140                if type(e) is util.SkipSignal:141                    logger.warn("User skipped creating workgroup.")142                logger.error("Workgroup %s failed to be created, skipping chapter %s" %143                             (workgroup.title, workgroup.chapter_number))144                workgroups_to_remove.append((workgroup, workgroup.chapter_number))145                for module in bookmap.bookmap.modules:146                    if module.chapter_number is workgroup.chapter_number:147                        module.valid = False148                        failures.append((module.full_title(), " creating placeholder"))149            chapter_to_workgroup[workgroup.chapter_number] = workgroup150        for workgroup, chapter in workgroups_to_remove:151            bookmap.chapters.remove(chapter)152            bookmap.bookmap.workgroups.remove(workgroup)153        logger.debug("Operating on these chapters now: %s" % bookmap.chapters)154    logger.info("-------- Creating modules -------------------------------")155    for module in bookmap.bookmap.modules:156        if module.valid and module.chapter_number in bookmap.chapters:157            workgroup_url = 'Members/'158            if run_options.workgroups:159                workgroup_url = chapter_to_workgroup[module.chapter_number].url160            if module.destination_workspace_url != "" and module.destination_workspace_url is not None:161                logger.debug("Adding module %s to existing workgroup %s" %162                             (module.title, module.destination_workspace_url))163                workgroup_url = module.destination_workspace_url164            try:165                content_creator.run_create_and_publish_module(module, copy_config.destination_server,166                                                              copy_config.credentials, logger, workgroup_url,167                                                              dryrun=run_options.dryrun)168                if run_options.workgroups:169                    chapter_to_workgroup[module.chapter_number].add_module(module)170                    chapter_to_workgroup[module.chapter_number].unit_number = module.unit_number171                logger.debug("[CREATED MODULE] %s - %s" % (module.title, module.destination_id))172            except util.TerminateError:173                raise util.TerminateError("Terminate Signaled")174            except (CCTError, Exception) as e:175                if type(e) is not CCTError and type(e) is not util.SkipSignal:176                    logger.error("Problematic Error")177                    logger.debug(traceback.format_exc())178                if type(e) is util.SkipSignal:179                    logger.warn("User skipped creating module.")180                logger.error("Module %s failed to be created. " % module.title)181                module.valid = False182                failures.append((module.full_title(), " creating placeholder"))183def create_populate_and_publish_collection(content_creator, copy_config, bookmap, units, publish_collection, dry_run,184                                           logger, failures):185    collection = None186    if not dry_run:187        try:188            logger.debug("Creating collection.")189            collection = content_creator.create_collection(copy_config.credentials, bookmap.booktitle,190                                                           copy_config.destination_server, logger)191        except util.TerminateError:192            raise util.TerminateError("Terminate Signaled")193        except (CCTError, Exception) as e:194            if type(e) is not CCTError and type(e) is not util.SkipSignal:195                    logger.error("Problematic Error")196                    logger.debug(traceback.format_exc())197            if type(e) is util.SkipSignal:198                logger.warn("User skipped creating collection.")199            logger.error("Failed to create the collection")200            failures.append(("%s" % bookmap.booktitle, "creating collection"))201            return None202    unit_numbers_and_title = set()203    units_map = {}204    if units:205        for module in bookmap.bookmap.modules:206            if module.chapter_number in bookmap.chapters \207                    and module.unit_number != 'APPENDIX' \208                    and module.unit_number != "":209                unit_numbers_and_title.add((module.unit_number, module.unit_title))210        as_list = list(unit_numbers_and_title)211        as_list.sort(key=lambda unit_number_and_title: unit_number_and_title[0])212        if not dry_run:213            for unit_number, unit_title in as_list:214                try:215                    unit_collection = content_creator.add_subcollections(["Unit %s. %s" % (unit_number, unit_title)],216                                                                         copy_config.destination_server,217                                                                         copy_config.credentials, collection, logger)218                    units_map[unit_number] = unit_collection[0]219                except util.TerminateError:220                    raise util.TerminateError("Terminate Signaled")221                except (CCTError, Exception) as e:222                    if type(e) is not CCTError and type(e) is not util.SkipSignal:223                        logger.error("Problematic Error")224                        logger.debug(traceback.format_exc())225                    if type(e) is util.SkipSignal:226                        logger.warn("User skipped creating subcollections for units.")227                    logger.error("Failed to create subcollections for units")228                    failures.append(("%s" % unit_number,229                                     "creating unit subcollections (those chapters were added to the collection %s)" %230                                     collection.title))231                    units_map[unit_number] = collection232    for workgroup in bookmap.bookmap.workgroups:233        if workgroup.chapter_number in bookmap.chapters:234            logger.debug("Added subcollections and modules to collection.")235            parent = collection236            if not dry_run:237                if units and workgroup.chapter_number != '0' \238                         and workgroup.unit_number != 'APPENDIX' \239                         and workgroup.unit_number != "":240                    parent = units_map[workgroup.unit_number]241                try:242                    if ((units and workgroup.unit_number != 'APPENDIX' and workgroup.unit_number != "") or not units)\243                            and workgroup.chapter_number != '0':244                        subcollections = content_creator.add_subcollections([workgroup.chapter_title],245                                                                            copy_config.destination_server,246                                                                            copy_config.credentials,247                                                                            parent, logger)248                        module_parent = subcollections[0]249                    else:250                        module_parent = collection251                except util.TerminateError:252                    raise util.TerminateError("Terminate Signaled")253                except (CCTError, Exception) as e:254                    if type(e) is not CCTError and type(e) is not util.SkipSignal:255                        logger.error("Problematic Error")256                        logger.debug(traceback.format_exc())257                    if type(e) is util.SkipSignal:258                        logger.warn("User skipped creating subcollections for chapters.")259                    logger.error("Failed to create subcollections for chapter %s, adding modules to %s" %260                                 (workgroup.chapter_number, collection.title))261                    failures.append(("%s" % workgroup.chapter_title,262                                     "creating subcollections (those modules were added to collection %s)" %263                                     collection.title))264                    module_parent = collection265                try:266                    content_creator.add_modules_to_collection(workgroup.modules, copy_config.destination_server,267                                                              copy_config.credentials, module_parent, logger, failures)268                except util.TerminateError:269                    raise util.TerminateError("Terminate Signaled")270                except (CCTError, Exception) as e:271                    if type(e) is not CCTError and type(e) is not util.SkipSignal:272                        logger.error("Problematic Error")273                        logger.debug(traceback.format_exc())274                    if type(e) is util.SkipSignal:275                        logger.warn("User skipped adding modules to subcollection.")276                    logger.error("Failed to add modules to chapter %s" % workgroup.chapter_number)277                    failures.append(("%s" % workgroup.modules, "adding modules to subcollections"))278    if not dry_run and publish_collection:279        try:280            content_creator.publish_collection(copy_config.destination_server, copy_config.credentials, collection,281                                               logger)282        except util.TerminateError:283            raise util.TerminateError("Terminate Signaled")284        except (CCTError, Exception) as e:285            if type(e) is not CCTError and type(e) is not util.SkipSignal:286                logger.error("Problematic Error")287                logger.debug(traceback.format_exc())288            if type(e) is util.SkipSignal:289                logger.warn("User skipped publishing collection.")290            logger.error("Failed to publish collection")291            failures.append(("%s" % collection.title, "publishing collection"))292            return None293def publish_modules_post_copy(copier, content_creator, run_options, credentials, logger, failures):294    """295    Publishes modules that has been copied to the destination server.296    Arguments:297        copier - the copier object that did the copying298        content_creator - the creator object to do the publishing299        run_options - the input running options, what will the tool do300        credentials - the user's credentials301        logger - the tool's logger302        failures - the working list of failures303    Returns:304        None305    """306    for module in copier.copy_map.modules:307        if module.valid and module.chapter_number in run_options.chapters:308            logger.info("Publishing module: %s - %s" % (module.destination_id, module.full_title()))309            if not run_options.dryrun:310                try:311                    content_creator.publish_module("%s/%s/" % (module.destination_workspace_url, module.destination_id),312                                                   credentials, logger, False)313                except util.TerminateError:314                    raise util.TerminateError("Terminate Signaled")315                except (CCTError, Exception) as e:316                    if type(e) is not CCTError and type(e) is not util.SkipSignal:317                        logger.error("Problematic Error")318                        logger.debug(traceback.format_exc())319                    if type(e) is util.SkipSignal:320                        logger.warn("User skipped publishing module.")321                    logger.error("Failed to publish module %s", module.destination_id)322                    module.valid = False323                    failures.append((module.full_title(), "publishing module"))324def print_failures(logger, failures):325    for failure in failures:326        logger.error("\033[95mFailed %s - \033[91m%s\033[0m", failure[1], failure[0])327def user_confirm(logger, copy_config, bookmap, run_options, role_config):328    """329    Prints a summary of the settings for the process that is about to run and330    asks for user confirmation.331    """332    logger.info("-------- Summary ---------------------------------------")333    if run_options.copy:  # confirm each entry in the bookmap has a source module ID.334        last_title = "!ITS THE FIRST ONE!"335        for module in bookmap.bookmap.modules:336            if module.chapter_number in bookmap.chapters and (module.source_id is '' or module.source_id is ' ' or337                                                              module.source_id is None):338                logger.warn("\033[91mInput file has missing source module ID for module [%s]"339                            "- the module after [%s].\033[0m" % (module.title, last_title))340            last_title = module.title341    logger.info("Source: \033[95m%s\033[0m" % copy_config.source_server)342    logger.info("Destination: \033[95m%s\033[0m" % copy_config.destination_server)343    if PRODUCTION:344        logger.info("User: \033[95m%s\033[0m" % copy_config.credentials.split(':')[0])345    else:346        logger.info("Destination Credentials: \033[95m%s\033[0m" % copy_config.credentials)347    logger.info("Content: \033[95m%s\033[0m" % bookmap.booktitle)348    logger.info("Which Chapters: \033[95m%s\033[0m" % ', '.join(bookmap.chapters))349    logger.info("Number of Modules: \033[95m%s\033[0m" %350                len([module for module in bookmap.bookmap.modules if module.chapter_number in bookmap.chapters]))351    logger.info("Create placeholders?: \033[95m%s\033[0m" % run_options.modules or run_options.workgroups)352    if run_options.modules:353        logger.info("Create workgroups? \033[95m%s\033[0m" % run_options.workgroups)354    logger.info("Copy content? \033[95m%s\033[0m" % run_options.copy)355    if run_options.copy:356        logger.info("Edit roles? \033[95m%s\033[0m" % run_options.roles)357    if run_options.roles:358            logger.info("Authors: \033[95m%s\033[0m" % ', '.join(role_config.creators))359            logger.info("Maintainers: \033[95m%s\033[0m" % ', '.join(role_config.maintainers))360            logger.info("Rightsholders: \033[95m%s\033[0m" % ', '.join(role_config.rightholders))361    logger.info("Create collections? \033[95m%s\033[0m" % run_options.collections)362    if run_options.collections:363        logger.info("Units? \033[95m%s\033[0m" % run_options.units)364    logger.info("Publish content? \033[95m%s\033[0m" % run_options.publish)365    if run_options.dryrun:366        logger.info("------------NOTE: \033[95mDRY RUN\033[0m-----------------")367    while True:368        var = raw_input("\33[95mPlease verify this information. If there are \033[91mwarnings\033[95m, "369                        "consider checking your data.\n"370                        "Enter:\n"371                        "    \033[92m1\033[0m - Proceed\n"372                        "    \033[91m2\033[0m - Cancel\n>>> ")373        if var is '1':374            break375        elif var is '2':376            sys.exit()377def main():378    args = cli.get_parser(VERSION).parse_args()379    cli.verify_args(args)380    if args.chapters:381        args.chapters.sort()382    run_options = op.RunOptions(args.modules, args.workgroups, args.copy, args.roles, args.collection,383                                args.units, args.publish, args.publish_collection, args.chapters, args.exclude,384                                args.dryrun, args.verboselevel)385    booktitle = ""386    signal.signal(signal.SIGINT, util.handle_terminate)387    signal.signal(signal.SIGTSTP, util.handle_user_skip)388    try:389        booktitle = run(args.settings, args.input_file, run_options)390    except Exception, e:391        print "Error: %s", e392        print(traceback.format_exc())393    print("Content Copy for {} has completed, see Terminal for results.".format(booktitle))394if __name__ == "__main__":...

__init__.py

Source:__init__.py

1import copy2from addict import Dict3from torch.utils.data import DataLoader4from .RecDataSet import RecDataLoader, RecTextLineDataset, RecLmdbDataset5from .DetDataSet import JsonDataset6from .RecCollateFn import RecCollateFn7from .DetCollateFN import DetCollectFN8__all__ = ['build_dataloader']9support_dataset = ['RecTextLineDataset', 'RecLmdbDataset', 'DetTextLineDataset','JsonDataset']10support_loader = ['RecDataLoader', 'DataLoader']11def build_dataset(config):12    """13    æ ¹æ®éç½®æé dataset14    :param config: æ°æ®éç¸å³çéç½®ï¼ä¸è¬ä¸º config['dataset']['train']['dataset] or config['dataset']['eval']['dataset]15    :return: æ ¹æ®éç½®æé å¥½ç DataSet ç±»å¯¹è±¡16    """17    dataset_type = config.pop('type')18    assert dataset_type in support_dataset, f'{dataset_type} is not developed yet!, only {support_dataset} are support now'19    dataset_class = eval(dataset_type)(config)20    return dataset_class21def build_loader(dataset, config):22    """23    æ ¹æ®éç½®æé  dataloader, åå«ä¸¤ä¸ªæ¥éª¤ï¼1. æé  collate_fn, 2. æé  dataloader24    :param dataset: ç»§æ¿èª torch.utils.data.DataSetçç±»å¯¹è±¡25    :param config: loader ç¸å³çéç½®ï¼ä¸è¬ä¸º config['dataset']['train']['loader] or config['dataset']['eval']['loader]26    :return: æ ¹æ®éç½®æé å¥½ç DataSet ç±»å¯¹è±¡27    """28    dataloader_type = config.pop('type')29    assert dataloader_type in support_loader, f'{dataloader_type} is not developed yet!, only {support_loader} are support now'30    # build collate_fn31    if 'collate_fn' in config:32        config['collate_fn']['dataset'] = dataset33        collate_fn = build_collate_fn(config.pop('collate_fn'))34    else:35        collate_fn = None36    dataloader_class = eval(dataloader_type)(dataset=dataset, collate_fn=collate_fn, **config)37    return dataloader_class38def build_collate_fn(config):39    """40    æ ¹æ®éç½®æé  collate_fn41    :param config: collate_fn ç¸å³çéç½®42    :return: æ ¹æ®éç½®æé å¥½ç collate_fn ç±»å¯¹è±¡43    """44    collate_fn_type = config.pop('type')45    if len(collate_fn_type) == 0:46        return None47    collate_fn_class = eval(collate_fn_type)(**config)48    return collate_fn_class49def build_dataloader(config):50    """51    æ ¹æ®éç½®æé  dataloader, åå«ä¸¤ä¸ªæ¥éª¤ï¼1. æé  dataset, 2. æé  dataloader52    :param config: æ°æ®éç¸å³çéç½®ï¼ä¸è¬ä¸º config['dataset']['train'] or config['dataset']['eval']53    :return: æ ¹æ®éç½®æé å¥½ç DataLoader ç±»å¯¹è±¡54    """55    # build dataset56    copy_config = copy.deepcopy(config)57    copy_config = Dict(copy_config)58    dataset = build_dataset(copy_config.dataset)59    # build loader60    loader = build_loader(dataset, copy_config.loader)...

import_cc_config.py

Source:import_cc_config.py

...27    if subsection:  section += "." + subsection28    if isinstance(value, list):29        value = " ".join(value) or "[]"30    cfg_client.set_config_option(section, name, str(value))31def copy_config(config_in, field, subsection, default=None, required=True):32    if field not in config_in and required:33        raise KeyError(f"Rquired field {field} not found")34    value = config_in.get(field, default)35    if value is not None:36        set_option(subsection, field, value)37        38def clear_section(section):39    section = CONFIG_SECTION_PREFIX if not section else CONFIG_SECTION_PREFIX + "." + section40    try:41        data = cfg_client.get_config(section)42    except ConfigNotFound:43        return44    if data:45        for k in data.keys():46            cfg_client.delete_config_option(section, k)47clear_section("")48clear_section("scanner")49clear_section("dbdump")50set_option("", "npartitions", defaults_in.get("partitions", 10))51if "scanner" in defaults_in:52    scanner_in = defaults_in["scanner"]53    54    copy_config(scanner_in, "recursion", "scanner")55    copy_config(scanner_in, "nworkers", "scanner", 10)56    copy_config(scanner_in, "timeout", "scanner", 60)57    copy_config(scanner_in, "server_root", "scanner", required = True)58    copy_config(scanner_in, "remove_prefix", "scanner", "/")59    copy_config(scanner_in, "add_prefix", "scanner", "/")60    61    roots_in = scanner_in.get("roots", {})62    set_option("scanner", "roots", json.dumps(roots_in))63    64if "dbdump" in defaults_in:65    dbdump_in = defaults_in["dbdump"]66    set_option("dbdump", "path_root", dbdump_in["path_root"])67    if "ignore" in dbdump_in:68        ignore = dbdump_in["ignore"]69        set_option("dbdump", "ignore", ignore)70#71# RSE specifics72#73rse_client = RSEClient()...

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.