Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use get_replication method in localstack

Best Python code snippet using localstack_python

check_solr.py

Source:check_solr.py

...20def get_cores(http, baseurl):21    path = '/admin/cores?action=status&wt=json'22    cores = get_data(http, baseurl, path)23    return cores24def get_replication(http, baseurl, core):25    path = '/{}/replication?command=details&wt=json'.format(core)26    27    repl = get_data(http, baseurl, path)28    return repl29def output_ping(http, baseurl, extended):30    # Should get cores first31    cores = get_cores(http, baseurl)32    state = 033    okCores = []34    nokCores = []35    outputE = ''36    for k, c in cores['status'].items():37        size = c['index']['sizeInBytes']38        numdocs = c['index']['numDocs']39        uptime = c['uptime']40        status = get_ping(http, baseurl, k)41        if status['status'] != 'OK':42            state = 243            nokCores.append(k)44        else:45            okCores.append(k)46        outputE += '\n{} - Status: {} | {}_size={}B {}_numdocs={} {}_uptime={}s'.format(k, status['status'], k, size, k, numdocs, k, uptime)47    if state == 2:48        print 'CRITICAL: couldn\'t ping {} cores'.format(','.join(nokCores))49    else:50        print 'OK: {} ping correctly'.format(','.join(okCores))51       52    if extended:53        print outputE54    55    sys.exit(state)56def output_replication_master(http, baseurl, extended):57    # Should get cores first58    cores = get_cores(http, baseurl)59    state = 060    okCores = []61    nokCores = []62    outputE = ''63    for k, c in cores['status'].items():64        repl = get_replication(http, baseurl, k)65        master = repl['details']['isMaster']66        enabled = repl['details']['master']['replicationEnabled']67        if master != "true" or enabled != "true":68            state = 269            nokCores.append(k)70        else:71            okCores.append(k)72        outputE += '\n{} - isMaster: {} -- replicationEnabled: {}'.format(k, master, enabled)73    if state == 2:74        print 'CRITICAL: {} are not masters/replication disabled'.format(','.join(nokCores))75    else:76        print 'OK: {} are all master & replicating!'.format(','.join(okCores))77    if extended:78        print outputE79    sys.exit(state)80def output_replication_slave(http, baseurl, extended, warn, critical):81    # Should get cores first82    cores = get_cores(http, baseurl)83    state = 084    okCores = []85    nokCores = []86    outputE = ''87    for k, c in cores['status'].items():88        repl = get_replication(http, baseurl, k)89        slave = repl['details']['isSlave']90        slaveIndex = repl['details']['indexVersion']91        slaveGen = repl['details']['generation']92        masterIndex = repl['details']['slave']['masterDetails']['master']['replicableVersion']93        masterGen = repl['details']['slave']['masterDetails']['master']['replicableGeneration']94        enabled = repl['details']['slave']['masterDetails']['master']['replicationEnabled']95        if (masterGen - slaveGen) >= critical or slave != "true" or enabled != "true":96            state = 297            nokCores.append(k)98        elif (masterGen - slaveGen) >= warn:99            nokCores.append(k)100            if state == 0:101                state = 1102        else:...

news_replication.py

Source:news_replication.py

...5import multiprocessing6from clustering import jaccard_distances7from text.nel import nel_from_news8n_jobs = multiprocessing.cpu_count()9def get_replication(news, sim_matrix, threshold=0.7):10    portals = set([article['portal'] for article in news])11    replication = dict(zip(portals, np.zeros(len(portals))))12    obs = dict(zip(portals, [[] for _ in range(0, len(portals))]))13    for idx, article in enumerate(news):14        portal = article['portal']15        for idx2, article2 in enumerate(news):16            if idx >= idx2 or portal != article2['portal']:17                continue18            if sim_matrix[idx][idx2] >= threshold:19                replication[portal] += 120                obs[portal].append((news[idx]['id'], news[idx2]['id']))21    return replication, obs22def load_cleaned_news(remove_stopwords=True, stem=False):23    news = load_news(fields=['id', 'title', 'subtitle', 'subject', 'portal', 'text'])24    results = Parallel(n_jobs=n_jobs)(delayed(cleanup_text)(article['text'],25                                                            article['title'],26                                                            article['subtitle'],27                                                            remove_stopwords,28                                                            stem) for article in news)29    for idx, article in enumerate(news):30        article['new_text'] = results[idx]31    return news32print("Loading news...")33news_nel = load_cleaned_news(remove_stopwords=False, stem=False)34corpus_nel = [article['text'] for article in news_nel]35coords = nel_from_news(corpus_nel, filename='data/vectors_nel.bin')36vectors_sim = (1 - jaccard_distances(coords))37print("Counting Breaking News with .8 threshold")38replication, obs = get_replication(news_nel, vectors_sim, threshold=0.8)39print(replication)40print("Counting Breaking News with .9 threshold")41replication, obs = get_replication(news_nel, vectors_sim, threshold=0.9)42print(replication)...

get_replication.py

Source:get_replication.py

1#!/usr/bin/python2DOCUMENTATION = '''3---4module: get_replication5short_description: generate mysql replication setting6set_facts: pillar.mysql_replication7'''8EXAMPLES = '''9- name: get_replication10  get_replication:11    master_hostvars: "{{ hostvars[pillar['mysql_replication']['master_host']] }}"12'''13def main():14    fields = {15        "master_hostvars": {"required": True, "type": "dict"},16    }17    module = AnsibleModule(argument_spec=fields)18    module.exit_json(changed=False)19from ansible.module_utils.basic import *20if __name__ == '__main__':...

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.