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

How to use test_run method in yandex-tank

Best Python code snippet using yandex-tank

test_match.py

Source:test_match.py

1import sys2import os3import shutil4import subprocess5from django.core.management.base import BaseCommand6CWD = os.path.dirname(os.path.abspath(__file__))7sys.path.insert(1, os.path.join(CWD, '..', '..', '..'))8os.environ['DJANGO_SETTINGS_MODULE'] = 'settings'9from tests_app.models import Spider, TestRun, Report, ReportSearchterm,\10    LocalCache11from utils import test_run_to_dirname, get_output_fname, get_log_fname12sys.path.append(os.path.join(CWD, '..', '..', '..', '..', 'product-ranking'))13from debug_match_urls import match14def _create_proj_dir(test_run):15    dirname = test_run_to_dirname(test_run)16    if not dirname.endswith('/'):17        dirname += '/'18    if not os.path.exists(dirname):19        os.makedirs(dirname)20    return dirname21def _get_branches_dirs(test_run):22    base_dir = _create_proj_dir(test_run)23    dir1 = os.path.join(base_dir, test_run.branch1)24    dir2 = os.path.join(base_dir, test_run.branch2)25    return dir1, dir226def prepare_git_branches(test_run, copy_files=True, force=True):27    """ Creates 2 dirs under base path; each dir contains complete project28        with the specific branches """29    dir1, dir2 = _get_branches_dirs(test_run)30    if not os.path.exists(dir1):31        os.makedirs(dir1)32    if not os.path.exists(dir2):33        os.makedirs(dir2)34    # clone & checkout first dir35    this_repo_dir = os.path.abspath(os.path.join(CWD, '..', '..', '..', '..'))36    cmd_copy = 'cp -r "%s/." "%s"'37    cmd_fetch = 'cd %s; git fetch --all %s && git checkout %s %s && git pull origin %s %s'38    if copy_files:39        os.system(cmd_copy % (this_repo_dir, dir1))40        os.system(cmd_copy % (this_repo_dir, dir2))41    if force:42        force = ' --force '43    else:44        force = ''45    os.system(cmd_fetch % (dir1, force, test_run.branch1, force, test_run.branch1, force))46    os.system(cmd_fetch % (dir2, force, test_run.branch2, force, test_run.branch2, force))47def get_cache(searchterm, test_run, delete_expired=True):48    """ Returns valid cache for the specified searchterm and spider49    if it exists; None otherwise50    """51    cache = LocalCache.objects.filter(52        searchterm=searchterm, test_run=test_run, spider=test_run.spider53    ).order_by('-when_created')54    if delete_expired:55        for c in cache:56            if not c.is_valid():57                shutil.rmtree(c.get_path())58                print '    removing expired cache %s' % c59                c.delete()60    cache = LocalCache.objects.filter(61        searchterm=searchterm, test_run=test_run, spider=test_run.spider62    ).order_by('-when_created')63    if cache and os.path.exists(cache[0].get_path()):64        print '    returning cache %s' % cache[0]65        return cache[0]66    else:67        # file path does not exist - create new cache68        cache = LocalCache.objects.create(69            searchterm=searchterm, test_run=test_run, spider=test_run.spider)70        print '    created new cache %s' % cache71        return cache72def create_cache_path_if_doesnt_exist(cache):73    if not os.path.exists(cache.get_path()):74        print '    created cache path %s' % cache.get_path()75        os.makedirs(cache.get_path())76def test_match(test_run):77    prepare_git_branches(test_run, copy_files=True, force=True)78    cmd = ('cd "{branch_dir}/product-ranking/"; scrapy crawl {spider_name}'79           ' -a searchterms_str="{searchterm}" -a quantity={quantity}'80           ' -a enable_cache=True -s HTTPCACHE_DIR="{cache_dir}"'81           ' -s DOWNLOAD_DELAY=0.05 -s LOG_FILE={log_path} -o {output_path}')82    report = Report.objects.create(testrun=test_run)83    for searchterm in test_run.spider.searchterms.all():84        cache = get_cache(searchterm, test_run)85        create_cache_path_if_doesnt_exist(cache)86        print '    executing spider %s for ST %s' % (87            test_run.spider.name, searchterm.searchterm)88        output1 = get_output_fname(searchterm, test_run, test_run.branch1)89        output2 = get_output_fname(searchterm, test_run, test_run.branch2)90        log1 = get_log_fname(searchterm, test_run, test_run.branch1)91        log2 = get_log_fname(searchterm, test_run, test_run.branch2)92        os.system(cmd.format(93            branch_dir=_get_branches_dirs(test_run)[0],94            spider_name=test_run.spider.name,95            searchterm=searchterm.searchterm, quantity=searchterm.quantity,96            cache_dir=cache.get_path(), output_path=output1, log_path=log1))97        os.system(cmd.format(98            branch_dir=_get_branches_dirs(test_run)[1],99            spider_name=test_run.spider.name,100            searchterm=searchterm.searchterm, quantity=searchterm.quantity,101            cache_dir=cache.get_path(), output_path=output2, log_path=log2))102        if test_run.exclude_fields is None:103            test_run.exclude_fields = []104        diff = match(105            f1=output1, f2=output2,106            fields2exclude=test_run.exclude_fields,107            strip_get_args=test_run.strip_get_args,108            skip_urls=test_run.skip_urls,109            exclude_duplicates=test_run.exclude_duplicates,110            print_output=False111        )112        report_searchterm = ReportSearchterm.objects.create(113            report=report, searchterm=searchterm, total_urls=diff['total_urls'],114            matched_urls=diff['matched_urls'], diffs=diff['diff'])115def cleanup_files(test_run):116    for _dir in _get_branches_dirs(test_run):117        shutil.rmtree(_dir)118class Command(BaseCommand):119    can_import_settings = True120    def handle(self, *args, **options):121        # get a test run to check122        test_runs = TestRun.objects.filter(status='stopped').order_by('when_started')123        if test_runs:124            tr = test_runs[0]125            print 'Going to check test run %s' % tr126            tr.status = 'running'127            tr.save()128            test_match(tr)129            tr_reports = tr.testrun_reports.all().order_by('-when_created')130            if tr_reports:131                status = 'passed'132                if tr_reports[0].not_enough_matched_urls():133                    status = 'failed'134                if tr_reports[0].diffs_found():135                    status = 'failed'136            else:137                status = 'failed'138            tr.status = status139            tr.save()140            cleanup_files(tr)141        else:...

main_view_handler.py

Source:main_view_handler.py

1# Copyright 2013 The Chromium Authors. All rights reserved.2# Use of this source code is governed by a BSD-style license that can be3# found in the LICENSE file.4"""Request handler to serve the main_view page."""5import jinja26import json7import os8import re9import sys10import webapp211import ispy_api12from common import constants13from common import ispy_utils14import gs_bucket15import views16JINJA = jinja2.Environment(17    loader=jinja2.FileSystemLoader(os.path.dirname(views.__file__)),18    extensions=['jinja2.ext.autoescape'])19class MainViewHandler(webapp2.RequestHandler):20  """Request handler to serve the main_view page."""21  def get(self):22    """Handles a get request to the main_view page.23    If the test_run parameter is specified, then a page displaying all of24    the failed runs in the test_run will be shown. Otherwise a view listing25    all of the test_runs available for viewing will be displayed.26    """27    test_run = self.request.get('test_run')28    bucket = gs_bucket.GoogleCloudStorageBucket(constants.BUCKET)29    ispy = ispy_utils.ISpyUtils(bucket)30    # Load the view.31    if test_run:32      self._GetForTestRun(test_run, ispy)33      return34    self._GetAllTestRuns(ispy)35  def _GetAllTestRuns(self, ispy):36    """Renders a list view of all of the test_runs available in GS.37    Args:38      ispy: An instance of ispy_api.ISpyApi.39    """40    template = JINJA.get_template('list_view.html')41    data = {}42    test_runs = set([path.lstrip('/').split('/')[1] for path in43                    ispy.GetAllPaths('failures/')])44    base_url = '/?test_run=%s'45    data['links'] = [(test_run, base_url % test_run) for test_run in test_runs]46    self.response.write(template.render(data))47  def _GetForTestRun(self, test_run, ispy):48    """Renders a sorted list of failure-rows for a given test_run.49    This method will produce a list of failure-rows that are sorted50    in descending order by number of different pixels.51    Args:52      test_run: The name of the test_run to render failure rows from.53      ispy: An instance of ispy_api.ISpyApi.54    """55    paths = set([path for path in ispy.GetAllPaths('failures/' + test_run)56                 if path.endswith('actual.png')])57    can_rebaseline = ispy_api.ISpyApi(58        ispy.cloud_bucket).CanRebaselineToTestRun(test_run)59    rows = [self._CreateRow(test_run, path, ispy) for path in paths]60    # Function that sorts by the different_pixels field in the failure-info.61    def _Sorter(a, b):62      return cmp(b['percent_different'],63                 a['percent_different'])64    template = JINJA.get_template('main_view.html')65    self.response.write(66        template.render({'comparisons': sorted(rows, _Sorter),67                         'test_run': test_run,68                         'can_rebaseline': can_rebaseline}))69  def _CreateRow(self, test_run, path, ispy):70    """Creates one failure-row.71    This method builds a dictionary with the data necessary to display a72    failure in the main_view html template.73    Args:74      test_run: The name of the test_run the failure is in.75      path: A path to the failure's actual.png file.76      ispy: An instance of ispy_api.ISpyApi.77    Returns:78      A dictionary with fields necessary to render a failure-row79        in the main_view html template.80    """81    res = {}82    res['expectation'] = path.lstrip('/').split('/')[2]83    res['test_run'] = test_run84    res['info'] = json.loads(ispy.cloud_bucket.DownloadFile(85        ispy_utils.GetFailurePath(res['test_run'], res['expectation'],86                                  'info.txt')))87    expected = ispy_utils.GetExpectationPath(88        res['expectation'], 'expected.png')89    diff = ispy_utils.GetFailurePath(test_run, res['expectation'], 'diff.png')90    res['percent_different'] = res['info']['fraction_different'] * 10091    res['expected_path'] = expected92    res['diff_path'] = diff93    res['actual_path'] = path94    res['expected'] = ispy.cloud_bucket.GetImageURL(expected)95    res['diff'] = ispy.cloud_bucket.GetImageURL(diff)96    res['actual'] = ispy.cloud_bucket.GetImageURL(path)...

test_pypeconfig_pluginregister.py

Source:test_pypeconfig_pluginregister.py

1# -*- coding: utf-8 -*-2"""$ pype pype-config plugin-register/plugin-unregister."""3import importlib4from pype.config import plugin_register, plugin_unregister5from tests import create_runner, create_test_env, invoke_runner, reload_config6class TestCLIPypePluginRegister:  # noqa: D1017    def test_register_without_name(self) -> None:  # noqa: D1028        test_run = invoke_runner(plugin_register.main)9        assert test_run.result.exit_code == 210        assert 'Missing option \'--name\'' in test_run.result.output11    def test_register_without_path(self) -> None:  # noqa: D10212        test_run = invoke_runner(13            plugin_register.main, ['--name', 'test'])14        assert test_run.result.exit_code == 215        assert 'Missing option \'--path\'' in test_run.result.output16    def test_register_without_create(self) -> None:  # noqa: D10217        test_run = invoke_runner(18            plugin_register.main,19            ['--name', 'plug', '--path', '%CONFIG_DIR%'])20        assert test_run.result.exit_code == 121        assert 'Could not find a python module' in test_run.result.output22    def test_register_with_create(self) -> None:  # noqa: D10223        test_run = invoke_runner(24            plugin_register.main,25            ['--name', 'plug', '--path', '%CONFIG_DIR%', '--create'])26        assert test_run.result.exit_code == 027        assert 'successfully created' in test_run.result.output28    def test_register_twice(self) -> None:  # noqa: D10229        with create_test_env() as test_env:30            # Register plugin31            test_run = create_runner(32                test_env,33                plugin_register.main,34                ['--name', 'plug', '--path', '%CONFIG_DIR%', '--create'])35            assert test_run.result.exit_code == 036            assert 'successfully created' in test_run.result.output37            result_configuration = reload_config(test_run)38            assert len(result_configuration['plugins']) == 139            assert result_configuration['plugins'][0]['name'] == 'plug'40            # Try to register again41            result = test_run.runner.invoke(42                plugin_register.main,43                ['--name', 'plug', '--path', test_run.test_env.config_dir])44            assert result.exit_code == 145            assert 'already a plugin named' in result.output46            assert len(result_configuration['plugins']) == 147    def test_register_unregister_and_reregister(self) -> None:  # noqa: D10248        with create_test_env() as test_env:49            # Register plugin50            test_run = create_runner(51                test_env,52                plugin_register.main,53                ['--name', 'plug', '--path', '%CONFIG_DIR%', '--create'])54            assert test_run.result.exit_code == 055            assert 'successfully created' in test_run.result.output56            result_configuration = reload_config(test_run)57            assert len(result_configuration['plugins']) == 158            assert result_configuration['plugins'][0]['name'] == 'plug'59            # Unregister plugin (doesn't delete)60            # Reload to activate current test_env61            importlib.reload(plugin_unregister)62            result = test_run.runner.invoke(63                plugin_unregister.main,64                ['--name', 'plug'])65            assert result.exit_code == 066            assert 'successfully unregistered' in result.output67            result_configuration = reload_config(test_run)68            assert len(result_configuration['plugins']) == 069            # Register plugin again70            result = test_run.runner.invoke(71                plugin_register.main,72                ['--name', 'plug', '--path', test_env.config_dir])73            assert result.exit_code == 074            assert 'successfully registered' in result.output75            result_configuration = reload_config(test_run)76            assert len(result_configuration['plugins']) == 1...

test_DataBrowserWidget.py

Source:test_DataBrowserWidget.py

1# -*- coding: iso-8859-1 -*-2"""3    MoinMoin - MoinMoin.widget.browser Tests4    @copyright: 2010 by MoinMoin:ReimarBauer5    @license: GNU GPL, see COPYING for details.6"""7import py8from MoinMoin.util.dataset import TupleDataset, Column9from MoinMoin.widget.browser import DataBrowserWidget10class Test_DataBrowserWidget_sort_table(object):11    def setup_class(self):12        # check if state of example changes during tests13        example = [['L1', (u'c', u'c'), (u'1', u'1'),   (u'2', u'2'),           (u'a', u'a'),  (u'4', u'4'),   (u'5', u'5')],14                   ['L2', (u'b', u'b'), (u'10', u'10'), (u'21', u'21'),         (u'B', u'B'),  (u'40', u'40'), (u'10', u'10')],15                   ['L3', (u'b', u'b'), (u'2', u'2'),   (u'3.14', u'3.14'),     (u'c', u'c'),  (u'54', u'54'), (u'50', u'50')],16                   ['L4', (u'b', u'b'), (u'90', u'90'), (u'-2.240', u'-2.240'), (u'D', u'D'),  (u'40', u'40'), (u'5', u'5')],17                   ['L5', (u'a', u'a'), (u'95', u'95'), (u'20', u'20'),         (u'e', u'e'),  (u'40', u'40'), (u'10', u'10')],18                  ]19        self.example = example20        data = TupleDataset()21        data.columns = []22        data.columns.extend([Column('TEST', label='TEST')])23        for line in self.example:24            data.addRow([line[0]] + line[1:])25            data.columns.extend([Column(line[0], label=line[0])])26        self.data = data27        self.table = DataBrowserWidget(self.request)28    def test_tablecreation(self):29        """30        tests input data not changed31        """32        self.table.setData(self.data)33        result = self.table.data.data34        example = self.example35        assert result == example36    def test_sort_one_column_alphas(self):37        """38        tests one column sorted alphabetically39        """40        self.table.setData(self.data, sort_columns=[4])41        test_run = self.table.data.data42        result = [result[0] for result in test_run]43        assert result == ['L2', 'L4', 'L1', 'L3', 'L5']44    def test_sort_one_column_integers(self):45        """46        tests one column sorted numerical47        """48        self.table.setData(self.data, sort_columns=[2])49        test_run = self.table.data.data50        result = [result[0] for result in test_run]51        assert result == ['L1', 'L3', 'L2', 'L4', 'L5']52    def test_sort_one_column_floats(self):53        """54        tests one column sorted numerical for floating point values55        """56        self.table.setData(self.data, sort_columns=[3])57        test_run = self.table.data.data58        result = [result[0] for result in test_run]59        assert result == ['L4', 'L1', 'L3', 'L5', 'L2']60    def test_n_sort(self):61        """62        tests n_sort63        """64        self.table.setData(self.data, sort_columns=[1, 2])65        test_run = self.table.data.data66        result = [result[0] for result in test_run]67        assert result == ['L5', 'L3', 'L2', 'L4', 'L1']68        self.table.setData(self.data, sort_columns=[5, 6, 3])69        test_run = self.table.data.data70        result = [result[0] for result in test_run]71        assert result == ['L1', 'L4', 'L5', 'L2', 'L3']72    def test_reverse_sort(self):73        """74        tests reverse sort75        """76        self.table.setData(self.data, sort_columns=[0], reverse=True)77        test_run = self.table.data.data78        result = [result[0] for result in test_run]79        assert result == ['L5', 'L4', 'L3', 'L2', 'L1']80    def test_sort_and_reverse_by_a_different_column(self):81        """82        tests reverse sort by a different column as the one to sort83        """84        self.table.setData(self.data, sort_columns=[6], reverse=[0])85        test_run = self.table.data.data86        result = [result[0] for result in test_run]...

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.