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

How to use redirected_to method in Playwright Python

Best Python code snippet using playwright-python

admin.py

Source:admin.py

1# -*- coding: utf-8 -*-2#3# Copyright (C) 2007-2010 Edgewall Software4# All rights reserved.5#6# This software is licensed as described in the file COPYING, which7# you should have received as part of this distribution. The terms8# are also available at http://bitten.edgewall.org/wiki/License.9import shutil10import tempfile11import unittest12from trac.core import TracError13from trac.db import DatabaseManager14from trac.perm import PermissionCache, PermissionError, PermissionSystem15from trac.resource import Resource16from trac.test import EnvironmentStub, Mock17from trac.web.href import Href18from trac.web.main import RequestDone19from bitten.main import BuildSystem20from bitten.model import BuildConfig, TargetPlatform, schema21from bitten.admin import BuildMasterAdminPageProvider, \22                         BuildConfigurationsAdminPageProvider23try:24    from trac.perm import DefaultPermissionPolicy25except ImportError:26    DefaultPermissionPolicy = None27class BuildMasterAdminPageProviderTestCase(unittest.TestCase):28    def setUp(self):29        self.env = EnvironmentStub(enable=['trac.*', 'bitten.*'])30        self.env.path = tempfile.mkdtemp()31        # Create tables32        db = self.env.get_db_cnx()33        cursor = db.cursor()34        connector, _ = DatabaseManager(self.env)._get_connector()35        for table in schema:36            for stmt in connector.to_sql(table):37                cursor.execute(stmt)38        # Set up permissions39        self.env.config.set('trac', 'permission_store',40                            'DefaultPermissionStore')41        PermissionSystem(self.env).grant_permission('joe', 'BUILD_ADMIN')42        if DefaultPermissionPolicy is not None and hasattr(DefaultPermissionPolicy, "CACHE_EXPIRY"):43            self.old_perm_cache_expiry = DefaultPermissionPolicy.CACHE_EXPIRY44            DefaultPermissionPolicy.CACHE_EXPIRY = -145        # Hook up a dummy repository46        self.repos = Mock(47            get_node=lambda path, rev=None: Mock(get_history=lambda: [],48                                                 isdir=True),49            normalize_path=lambda path: path,50            sync=lambda: None51        )52        self.env.get_repository = lambda authname=None: self.repos # 0.1153        try: # 0.12+54            from trac.core import Component, implements55            from trac.versioncontrol.api import IRepositoryConnector, \56                                                IRepositoryProvider57            class DummyRepos(Component):58                implements(IRepositoryConnector, IRepositoryProvider)59                def get_supported_types(self):60                    yield ('dummy', 9)61                def get_repository(this, repos_type, repos_dir, params):62                    return self.repos # Note: 'this' vs 'self' usage63                def get_repositories(self):64                    yield ('', {'dir': 'dummy_dir', 'type': 'dummy'})65            self.dummy = DummyRepos66        except ImportError:67            self.dummy = None # not supported, will use get_repository()68    def tearDown(self):69        if self.dummy: # remove from components list + interfaces dict70            self.env.__metaclass__._components.remove(self.dummy)71            for key in self.env.__metaclass__._registry.keys():72                if self.dummy in self.env.__metaclass__._registry[key]:73                    self.env.__metaclass__._registry[key].remove(self.dummy)74        if DefaultPermissionPolicy is not None and hasattr(DefaultPermissionPolicy, "CACHE_EXPIRY"):75            DefaultPermissionPolicy.CACHE_EXPIRY = self.old_perm_cache_expiry76        shutil.rmtree(self.env.path)77    def test_get_admin_panels(self):78        provider = BuildMasterAdminPageProvider(self.env)79        req = Mock(perm=PermissionCache(self.env, 'joe'))80        self.assertEqual([('bitten', 'Builds', 'master', 'Master Settings')],81                         list(provider.get_admin_panels(req)))82        PermissionSystem(self.env).revoke_permission('joe', 'BUILD_ADMIN')83        req = Mock(perm=PermissionCache(self.env, 'joe'))84        self.assertEqual([], list(provider.get_admin_panels(req)))85    def test_process_get_request(self):86        req = Mock(method='GET', chrome={}, href=Href('/'),87                   perm=PermissionCache(self.env, 'joe'))88        provider = BuildMasterAdminPageProvider(self.env)89        template_name, data = provider.render_admin_panel(90            req, 'bitten', 'master', ''91        )92        self.assertEqual('bitten_admin_master.html', template_name)93        assert 'master' in data94        master = data['master']95        self.assertEqual(3600, master.slave_timeout)96        self.assertEqual(0, master.stabilize_wait)97        assert not master.adjust_timestamps98        assert not master.build_all99        self.assertEqual('log/bitten', master.logs_dir)100    def test_process_config_changes(self):101        redirected_to = []102        def redirect(url):103            redirected_to.append(url)104            raise RequestDone105        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),106                   abs_href=Href('http://example.org/'), redirect=redirect,107                   args={'slave_timeout': '60', 'adjust_timestamps': ''})108        provider = BuildMasterAdminPageProvider(self.env)109        try:110            provider.render_admin_panel(req, 'bitten', 'master', '')111            self.fail('Expected RequestDone')112        except RequestDone:113            self.assertEqual('http://example.org/admin/bitten/master',114                             redirected_to[0])115            section = self.env.config['bitten']116            self.assertEqual(60, section.getint('slave_timeout'))117            self.assertEqual(True, section.getbool('adjust_timestamps'))118            self.assertEqual(False, section.getbool('build_all'))119class BuildConfigurationsAdminPageProviderTestCase(unittest.TestCase):120    def setUp(self):121        self.env = EnvironmentStub(enable=['trac.*', 'bitten.*'])122        self.env.path = tempfile.mkdtemp()123        # Create tables124        db = self.env.get_db_cnx()125        cursor = db.cursor()126        connector, _ = DatabaseManager(self.env)._get_connector()127        for table in schema:128            for stmt in connector.to_sql(table):129                cursor.execute(stmt)130        # Set up permissions131        self.env.config.set('trac', 'permission_store',132                            'DefaultPermissionStore')133        PermissionSystem(self.env).grant_permission('joe', 'BUILD_CREATE')134        PermissionSystem(self.env).grant_permission('joe', 'BUILD_DELETE')135        PermissionSystem(self.env).grant_permission('joe', 'BUILD_MODIFY')136        if DefaultPermissionPolicy is not None and hasattr(DefaultPermissionPolicy, "CACHE_EXPIRY"):137            self.old_perm_cache_expiry = DefaultPermissionPolicy.CACHE_EXPIRY138            DefaultPermissionPolicy.CACHE_EXPIRY = -1139        # Hook up a dummy repository140        self.repos = Mock(141            get_node=lambda path, rev=None: Mock(get_history=lambda: [],142                                                 isdir=True),143            normalize_path=lambda path: path,144            sync=lambda: None,145            resource=Resource('repository', None)146        )147        self.env.get_repository = lambda authname=None: self.repos # 0.11148        try: # 0.12+149            from trac.core import Component, implements150            from trac.versioncontrol.api import IRepositoryConnector, \151                                                IRepositoryProvider152            class DummyRepos(Component):153                implements(IRepositoryConnector, IRepositoryProvider)154                def get_supported_types(self):155                    yield ('dummy', 9)156                def get_repository(this, repos_type, repos_dir, params):157                    return self.repos # Note: 'this' vs 'self' usage158                def get_repositories(self):159                    yield ('', {'dir': 'dummy_dir', 'type': 'dummy'})160            self.dummy = DummyRepos161        except ImportError:162            self.dummy = None # not supported, will use get_repository()163    def tearDown(self):164        if self.dummy: # remove from components list + interfaces dict165            self.env.__metaclass__._components.remove(self.dummy)166            for key in self.env.__metaclass__._registry.keys():167                if self.dummy in self.env.__metaclass__._registry[key]:168                    self.env.__metaclass__._registry[key].remove(self.dummy)169        if DefaultPermissionPolicy is not None and hasattr(DefaultPermissionPolicy, "CACHE_EXPIRY"):170            DefaultPermissionPolicy.CACHE_EXPIRY = self.old_perm_cache_expiry171        shutil.rmtree(self.env.path)172    def test_get_admin_panels(self):173        provider = BuildConfigurationsAdminPageProvider(self.env)174        req = Mock(perm=PermissionCache(self.env, 'joe'))175        self.assertEqual([('bitten', 'Builds', 'configs', 'Configurations')],176                         list(provider.get_admin_panels(req)))177        PermissionSystem(self.env).revoke_permission('joe', 'BUILD_MODIFY')178        req = Mock(perm=PermissionCache(self.env, 'joe'))179        self.assertEqual([], list(provider.get_admin_panels(req)))180    def test_process_view_configs_empty(self):181        req = Mock(method='GET', chrome={}, href=Href('/'),182                   perm=PermissionCache(self.env, 'joe'))183        provider = BuildConfigurationsAdminPageProvider(self.env)184        template_name, data = provider.render_admin_panel(185            req, 'bitten', 'configs', ''186        )187        self.assertEqual('bitten_admin_configs.html', template_name)188        self.assertEqual([], data['configs'])189    def test_process_view_configs(self):190        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',191                    active=True).insert()192        BuildConfig(self.env, name='bar', label='Bar', path='branches/bar',193                    min_rev='123', max_rev='456').insert()194        req = Mock(method='GET', chrome={}, href=Href('/'),195                   perm=PermissionCache(self.env, 'joe'))196        provider = BuildConfigurationsAdminPageProvider(self.env)197        template_name, data = provider.render_admin_panel(198            req, 'bitten', 'configs', ''199        )200        self.assertEqual('bitten_admin_configs.html', template_name)201        assert 'configs' in data202        configs = data['configs']203        self.assertEqual(2, len(configs))204        self.assertEqual({205            'name': 'bar', 'href': '/admin/bitten/configs/bar',206            'label': 'Bar', 'min_rev': '123', 'max_rev': '456',207            'path': 'branches/bar', 'active': False, 'recipe': False208        }, configs[0])209        self.assertEqual({210            'name': 'foo', 'href': '/admin/bitten/configs/foo',211            'label': 'Foo', 'min_rev': None, 'max_rev': None,212            'path': 'branches/foo', 'active': True, 'recipe': False213        }, configs[1])214    def test_process_view_config(self):215        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',216                    active=True).insert()217        TargetPlatform(self.env, config='foo', name='any').insert()218        req = Mock(method='GET', chrome={}, href=Href('/'),219                   perm=PermissionCache(self.env, 'joe'))220        provider = BuildConfigurationsAdminPageProvider(self.env)221        template_name, data = provider.render_admin_panel(222            req, 'bitten', 'configs', 'foo'223        )224        self.assertEqual('bitten_admin_configs.html', template_name)225        assert 'config' in data226        config = data['config']227        self.assertEqual({228            'name': 'foo', 'label': 'Foo', 'description': '', 'recipe': '',229            'path': 'branches/foo', 'min_rev': None, 'max_rev': None,230            'active': True, 'platforms': [{231                'href': '/admin/bitten/configs/foo/1',232                'name': 'any', 'id': 1, 'rules': []233            }]234        }, config)235    def test_process_activate_config(self):236        BuildConfig(self.env, name='foo', path='branches/foo').insert()237        BuildConfig(self.env, name='bar', path='branches/bar').insert()238        redirected_to = []239        def redirect(url):240            redirected_to.append(url)241            raise RequestDone242        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),243                   abs_href=Href('http://example.org/'), redirect=redirect,244                   authname='joe',245                   args={'apply': '', 'active': ['foo']})246        provider = BuildConfigurationsAdminPageProvider(self.env)247        try:248            provider.render_admin_panel(req, 'bitten', 'configs', '')249            self.fail('Expected RequestDone')250        except RequestDone:251            self.assertEqual('http://example.org/admin/bitten/configs',252                             redirected_to[0])253            config = BuildConfig.fetch(self.env, name='foo')254            self.assertEqual(True, config.active)255    def test_process_deactivate_config(self):256        BuildConfig(self.env, name='foo', path='branches/foo',257                    active=True).insert()258        BuildConfig(self.env, name='bar', path='branches/bar',259                    active=True).insert()260        redirected_to = []261        def redirect(url):262            redirected_to.append(url)263            raise RequestDone264        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),265                   abs_href=Href('http://example.org/'), redirect=redirect,266                   authname='joe',267                   args={'apply': ''})268        provider = BuildConfigurationsAdminPageProvider(self.env)269        try:270            provider.render_admin_panel(req, 'bitten', 'configs', '')271            self.fail('Expected RequestDone')272        except RequestDone:273            self.assertEqual('http://example.org/admin/bitten/configs',274                             redirected_to[0])275            config = BuildConfig.fetch(self.env, name='foo')276            self.assertEqual(False, config.active)277            config = BuildConfig.fetch(self.env, name='bar')278            self.assertEqual(False, config.active)279    def test_process_add_config(self):280        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',281                    active=True).insert()282        redirected_to = []283        def redirect(url):284            redirected_to.append(url)285            raise RequestDone286        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),287                   abs_href=Href('http://example.org/'), redirect=redirect,288                   authname='joe',289                   args={'add': '', 'name': 'bar', 'label': 'Bar'})290        provider = BuildConfigurationsAdminPageProvider(self.env)291        try:292            provider.render_admin_panel(req, 'bitten', 'configs', '')293            self.fail('Expected RequestDone')294        except RequestDone:295            self.assertEqual('http://example.org/admin/bitten/configs/bar',296                             redirected_to[0])297            config = BuildConfig.fetch(self.env, name='bar')298            self.assertEqual('Bar', config.label)299    def test_process_add_config_cancel(self):300        redirected_to = []301        def redirect(url):302            redirected_to.append(url)303            raise RequestDone304        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),305                   abs_href=Href('http://example.org/'), redirect=redirect,306                   args={'cancel': '', 'name': 'bar', 'label': 'Bar'})307        provider = BuildConfigurationsAdminPageProvider(self.env)308        try:309            provider.render_admin_panel(req, 'bitten', 'configs', '')310            self.fail('Expected RequestDone')311        except RequestDone:312            self.assertEqual('http://example.org/admin/bitten/configs',313                             redirected_to[0])314            configs = list(BuildConfig.select(self.env, include_inactive=True))315            self.assertEqual(0, len(configs))316    def test_process_add_config_no_name(self):317        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),318                   chrome={'warnings': []}, href=Href('/'),319                   authname='joe', args={'add': ''})320        provider = BuildConfigurationsAdminPageProvider(self.env)321        try:322            provider.render_admin_panel(req, 'bitten', 'configs', '')323            self.fail('Expected TracError')324        except TracError, e:325            self.assertEqual('Missing required field "name".', e.message)326            self.assertEqual('Add Configuration', e.title)327    def test_process_add_config_invalid_name(self):328        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),329                   chrome={'warnings': []}, href=Href('/'), authname='joe',330                   args={'add': '', 'name': 'no spaces allowed'})331        provider = BuildConfigurationsAdminPageProvider(self.env)332        try:333            provider.render_admin_panel(req, 'bitten', 'configs', '')334            self.fail('Expected TracError')335        except TracError, e:336            self.assertEqual('The field "name" may only contain letters, '337                             'digits, periods, or dashes.', e.message)338            self.assertEqual('Add Configuration', e.title)339    def test_new_config_submit_with_invalid_path(self):340        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),341                   authname='joe',342                   args={'add': '', 'name': 'foo', 'path': 'invalid/path'})343        def get_node(path, rev=None):344            raise TracError('No such node')345        self.repos.get_node = get_node346        provider = BuildConfigurationsAdminPageProvider(self.env)347        try:348            provider.render_admin_panel(req, 'bitten', 'configs', '')349            self.fail('Expected TracError')350        except TracError, e:351            self.failUnless('Invalid Repository Path' in e.message)352    def test_process_add_config_no_perms(self):353        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',354                    active=True).insert()355        PermissionSystem(self.env).revoke_permission('joe', 'BUILD_CREATE')356        req = Mock(method='POST',357                   perm=PermissionCache(self.env, 'joe'),358                   args={'add': '', 'name': 'bar', 'label': 'Bar'})359        provider = BuildConfigurationsAdminPageProvider(self.env)360        self.assertRaises(PermissionError, provider.render_admin_panel, req,361                          'bitten', 'configs', '')362    def test_process_remove_config(self):363        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',364                    active=True).insert()365        BuildConfig(self.env, name='bar', label='Bar', path='branches/bar',366                    min_rev='123', max_rev='456').insert()367        redirected_to = []368        def redirect(url):369            redirected_to.append(url)370            raise RequestDone371        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),372                   abs_href=Href('http://example.org/'), redirect=redirect,373                   args={'remove': '', 'sel': 'bar'})374        provider = BuildConfigurationsAdminPageProvider(self.env)375        try:376            provider.render_admin_panel(req, 'bitten', 'configs', '')377            self.fail('Expected RequestDone')378        except RequestDone:379            self.assertEqual('http://example.org/admin/bitten/configs',380                             redirected_to[0])381            assert not BuildConfig.fetch(self.env, name='bar')382    def test_process_remove_config_cancel(self):383        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',384                    active=True).insert()385        BuildConfig(self.env, name='bar', label='Bar', path='branches/bar',386                    min_rev='123', max_rev='456').insert()387        redirected_to = []388        def redirect(url):389            redirected_to.append(url)390            raise RequestDone391        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),392                   abs_href=Href('http://example.org/'), redirect=redirect,393                   args={'cancel': '', 'sel': 'bar'})394        provider = BuildConfigurationsAdminPageProvider(self.env)395        try:396            provider.render_admin_panel(req, 'bitten', 'configs', '')397            self.fail('Expected RequestDone')398        except RequestDone:399            self.assertEqual('http://example.org/admin/bitten/configs',400                             redirected_to[0])401            configs = list(BuildConfig.select(self.env, include_inactive=True))402            self.assertEqual(2, len(configs))403    def test_process_remove_config_no_selection(self):404        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',405                    active=True).insert()406        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),407                   args={'remove': ''})408        provider = BuildConfigurationsAdminPageProvider(self.env)409        try:410            provider.render_admin_panel(req, 'bitten', 'configs', '')411            self.fail('Expected TracError')412        except TracError, e:413            self.assertEqual('No configuration selected', e.message)414    def test_process_remove_config_bad_selection(self):415        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',416                    active=True).insert()417        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),418                   args={'remove': '', 'sel': 'baz'})419        provider = BuildConfigurationsAdminPageProvider(self.env)420        try:421            provider.render_admin_panel(req, 'bitten', 'configs', '')422            self.fail('Expected TracError')423        except TracError, e:424            self.assertEqual("Configuration 'baz' not found", e.message)425    def test_process_remove_config_no_perms(self):426        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',427                    active=True).insert()428        PermissionSystem(self.env).revoke_permission('joe', 'BUILD_DELETE')429        req = Mock(method='POST',430                   perm=PermissionCache(self.env, 'joe'),431                   args={'remove': '', 'sel': 'bar'})432        provider = BuildConfigurationsAdminPageProvider(self.env)433        self.assertRaises(PermissionError, provider.render_admin_panel, req,434                          'bitten', 'configs', '')435    def test_process_update_config(self):436        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',437                    active=True).insert()438        redirected_to = []439        def redirect(url):440            redirected_to.append(url)441            raise RequestDone442        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),443                   abs_href=Href('http://example.org/'), redirect=redirect,444                   authname='joe', chrome={'warnings': [], 'notices': []},445                   href=Href('/'),446                   args={'save': '', 'name': 'foo', 'label': 'Foobar',447                         'description': 'Thanks for all the fish!'})448        provider = BuildConfigurationsAdminPageProvider(self.env)449        try:450            provider.render_admin_panel(req, 'bitten', 'configs', 'foo')451            self.fail('Expected RequestDone')452        except RequestDone:453            self.assertEqual(['Configuration Saved.'], req.chrome['notices'])454            self.assertEqual('http://example.org/admin/bitten/configs/foo',455                             redirected_to[0])456            config = BuildConfig.fetch(self.env, name='foo')457            self.assertEqual('Foobar', config.label)458            self.assertEqual('Thanks for all the fish!', config.description)459    def test_process_update_config_no_name(self):460        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',461                    active=True).insert()462        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),463                   args={'save': '', 'name': ''}, authname='joe',464                   chrome={'warnings':[]}, href=Href('/'))465        provider = BuildConfigurationsAdminPageProvider(self.env)466        provider.render_admin_panel(req, 'bitten', 'configs', 'foo')467        self.failUnless(req.chrome['warnings'], "No warnings?")468        self.assertEquals(['Missing required field "name".'],469                            req.chrome['warnings'])470    def test_process_update_config_invalid_name(self):471        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',472                    active=True).insert()473        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),474                   args={'save': '', 'name': 'no spaces allowed'},475                   authname='joe', chrome={'warnings': []},476                   href=Href('/'))477        provider = BuildConfigurationsAdminPageProvider(self.env)478        provider.render_admin_panel(req, 'bitten', 'configs', 'foo')479        self.failUnless(req.chrome['warnings'], "No warnings?")480        self.assertEquals(req.chrome['warnings'], ['The field "name" may ' \481                        'only contain letters, digits, periods, or dashes.'])482    def test_process_update_config_invalid_path(self):483        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',484                    active=True).insert()485        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),486                   authname='joe', chrome={'warnings': []}, href=Href('/'),487                   args={'save': '', 'name': 'foo', 'path': 'invalid/path'})488        def get_node(path, rev=None):489            raise TracError('No such node')490        self.repos.get_node = get_node491        provider = BuildConfigurationsAdminPageProvider(self.env)492        provider.render_admin_panel(req, 'bitten', 'configs', 'foo')493        494        self.failUnless(req.chrome['warnings'], "No warnings?")495        self.assertTrue(req.chrome['warnings'] in (496                    # single repos / 0.11497                    ['Invalid Repository Path: "invalid/path" does not exist '498                     'within the "(default)" repository.'],499                     # multi-repos / 0.12+500                    ['Invalid Repository Path "invalid/path".']),501                req.chrome['warnings'])502    def test_process_update_config_non_wellformed_recipe(self):503        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',504                    active=True).insert()505        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),506                   authname='joe', chrome={'warnings': []}, href=Href('/'),507                   args={'save': '', 'name': 'foo', 'recipe': 'not_xml'})508        provider = BuildConfigurationsAdminPageProvider(self.env)509        provider.render_admin_panel(req, 'bitten', 'configs', 'foo')510        self.failUnless(req.chrome['warnings'], "No warnings?")511        self.assertEquals(['Failure parsing recipe: syntax error: line 1, ' \512                        'column 0.'], req.chrome['warnings'])513    def test_process_update_config_invalid_recipe(self):514        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',515                    active=True).insert()516        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),517                   authname='joe', chrome={'warnings': []}, href=Href('/'),518                   args={'save': '', 'name': 'foo',519                         'recipe': '<build><step /></build>'})520        provider = BuildConfigurationsAdminPageProvider(self.env)521        provider.render_admin_panel(req, 'bitten', 'configs', 'foo')522        523        self.failUnless(req.chrome['warnings'], "No warnings?")524        self.assertEquals(req.chrome['warnings'],525                    ['Invalid Recipe: Steps must have an "id" attribute.'])526    def test_process_new_platform_no_name(self):527        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',528                    active=True).insert()529        data = {}530        req = Mock(method='POST', chrome={}, hdf=data, href=Href('/'),531                   perm=PermissionCache(self.env, 'joe'),532                   args={'new': '', 'platform_name': ''})533        provider = BuildConfigurationsAdminPageProvider(self.env)534        try:535            provider.render_admin_panel(req, 'bitten', 'configs', 'foo')536            self.fail("No TracError?")537        except Exception, e:538            self.assertEquals(e.message, 'Missing required field "name"')539            self.assertEquals(e.title, 'Missing field')540    def test_process_new_platform(self):541        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',542                    active=True).insert()543        redirected_to = []544        def redirect(url):545            redirected_to.append(url)546            raise RequestDone547        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),548                   abs_href=Href('http://example.org/'), redirect=redirect,549                   authname='joe',550                   args={'add': '', 'new': '', 'platform_name': 'Test'})551        provider = BuildConfigurationsAdminPageProvider(self.env)552        try:553            provider.render_admin_panel(req, 'bitten', 'configs', 'foo')554            self.fail('Expected RequestDone')555        except RequestDone:556            self.assertEqual('http://example.org/admin/bitten/configs/foo/1',557                             redirected_to[0])558            platforms = list(TargetPlatform.select(self.env, config='foo'))559            self.assertEqual(1, len(platforms))560            self.assertEqual('Test', platforms[0].name)561            self.assertEqual([], platforms[0].rules)562    def test_process_remove_platforms(self):563        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',564                    active=True).insert()565        platform = TargetPlatform(self.env, config='foo', name='any')566        platform.insert()567        redirected_to = []568        def redirect(url):569            redirected_to.append(url)570            raise RequestDone571        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),572                   abs_href=Href('http://example.org/'), redirect=redirect,573                   authname='joe', chrome={'warnings': [], 'notices': []},574                   href=Href('/'),575                   args={'remove': '', 'sel': str(platform.id)})576        provider = BuildConfigurationsAdminPageProvider(self.env)577        try:578            provider.render_admin_panel(req, 'bitten', 'configs', 'foo')579            self.fail('Expected RequestDone')580        except RequestDone:581            self.assertEquals(['Target Platform(s) Removed.'],582                               req.chrome['notices'])583            self.assertEqual('http://example.org/admin/bitten/configs/foo',584                             redirected_to[0])585            platforms = list(TargetPlatform.select(self.env, config='foo'))586            self.assertEqual(0, len(platforms))587    def test_process_remove_platforms_no_selection(self):588        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',589                    active=True).insert()590        platform = TargetPlatform(self.env, config='foo', name='any')591        platform.insert()592        redirected_to = []593        def redirect(url):594            redirected_to.append(url)595            raise RequestDone596        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),597                   abs_href=Href('http://example.org/'), redirect=redirect,598                   authname='joe',599                   args={'remove': ''})600        provider = BuildConfigurationsAdminPageProvider(self.env)601        try:602            provider.render_admin_panel(req, 'bitten', 'configs', 'foo')603            self.fail('Expected TracError')604        except TracError, e:605            self.assertEqual('No platform selected', e.message)606    def test_process_edit_platform(self):607        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',608                    active=True).insert()609        platform = TargetPlatform(self.env, config='foo', name='any')610        platform.insert()611        req = Mock(method='GET', chrome={}, href=Href('/'),612                   perm=PermissionCache(self.env, 'joe'), args={})613        provider = BuildConfigurationsAdminPageProvider(self.env)614        template_name, data = provider.render_admin_panel(615            req, 'bitten', 'configs', 'foo/%d' % platform.id616        )617        self.assertEqual('bitten_admin_configs.html', template_name)618        assert 'platform' in data619        platform = data['platform']620        self.assertEqual({621            'id': 1, 'exists': True, 'name': 'any', 'rules': [('', '')],622        }, platform)623    def test_process_update_platform(self):624        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',625                    active=True).insert()626        platform = TargetPlatform(self.env, config='foo', name='any')627        platform.insert()628        redirected_to = []629        def redirect(url):630            redirected_to.append(url)631            raise RequestDone632        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),633                   abs_href=Href('http://example.org/'), redirect=redirect,634                   authname='joe',635                   args={'save': '', 'edit': '', 'name': 'Test',636                         'property_0': 'family', 'pattern_0': 'posix'})637        provider = BuildConfigurationsAdminPageProvider(self.env)638        try:639            provider.render_admin_panel(req, 'bitten', 'configs',640                                           'foo/%d' % platform.id)641            self.fail('Expected RequestDone')642        except RequestDone:643            self.assertEqual('http://example.org/admin/bitten/configs/foo',644                             redirected_to[0])645            platforms = list(TargetPlatform.select(self.env, config='foo'))646            self.assertEqual(1, len(platforms))647            self.assertEqual('Test', platforms[0].name)648            self.assertEqual([('family', 'posix')], platforms[0].rules)649    def test_process_update_platform_cancel(self):650        BuildConfig(self.env, name='foo', label='Foo', path='branches/foo',651                    active=True).insert()652        platform = TargetPlatform(self.env, config='foo', name='any')653        platform.insert()654        redirected_to = []655        def redirect(url):656            redirected_to.append(url)657            raise RequestDone658        req = Mock(method='POST', perm=PermissionCache(self.env, 'joe'),659                   abs_href=Href('http://example.org/'), redirect=redirect,660                   authname='joe',661                   args={'cancel': '', 'edit': '', 'name': 'Changed',662                         'property_0': 'family', 'pattern_0': 'posix'})663        provider = BuildConfigurationsAdminPageProvider(self.env)664        try:665            provider.render_admin_panel(req, 'bitten', 'configs',666                                           'foo/%d' % platform.id)667            self.fail('Expected RequestDone')668        except RequestDone:669            self.assertEqual('http://example.org/admin/bitten/configs/foo',670                             redirected_to[0])671            platforms = list(TargetPlatform.select(self.env, config='foo'))672            self.assertEqual(1, len(platforms))673            self.assertEqual('any', platforms[0].name)674            self.assertEqual([], platforms[0].rules)675def suite():676    suite = unittest.TestSuite()677    suite.addTest(unittest.makeSuite(678        BuildMasterAdminPageProviderTestCase, 'test'679    ))680    suite.addTest(unittest.makeSuite(681        BuildConfigurationsAdminPageProviderTestCase, 'test'682    ))683    return suite684if __name__ == '__main__':...

redirector.py

Source:redirector.py

1from datetime import datetime2from logging import Logger, getLogger3from math import floor4from typing import Optional, Any5from aiohttp import ClientSession6from discord import Embed, User, HTTPException7from discord.ext.commands import Cog, command, Option, Context8from discord.ext.menus import ListPageSource, ViewMenuPages9from discord.ext.tasks import loop10from motor.motor_asyncio import AsyncIOMotorCollection11from utils.bots import BOT_TYPES, CustomContext12from utils.database import Document13from utils.misc import random_string, UTC_OFFSET14API_URL: str = "https://redirector.regulad.xyz/"15SHORT_URL: str = "https://crud.space/"16logger: Logger = getLogger(__name__)17class LoggedEvent:18    """A logged redirector event."""19    def __init__(self, link_id: str, redirected_to: str, redirected_at: datetime, remote: str,20                 user_agent: Optional[str], *, document: Document):21        self.link_id: str = link_id22        self.redirected_to: str = redirected_to23        self.redirected_at: datetime = redirected_at24        self.remote: str = remote25        self.document: Document = document26        self.user_agent: Optional[str] = user_agent27    __slots__ = ('link_id', 'redirected_to', 'redirected_at', 'remote', 'document', 'user_agent')28class CampaignSource(ListPageSource):29    def __init__(self, campaigns: list[tuple[str, datetime]], *, per_page: int = 10) -> None:30        super().__init__(sorted(campaigns, key=lambda x: x[1], reverse=True), per_page=per_page)31    async def format_page(self, menu, entries):32        offset = menu.current_page * self.per_page33        embed: Embed = Embed(title=f"All campaigns", color=0x00ff00)34        embed.set_footer(text=f"Page {menu.current_page + 1}/{self.get_max_pages()}")35        for iteration, entry in enumerate(entries, start=offset):36            embed.add_field(37                name=f"{iteration + 1}. {entry[0]}",38                value=f"<t:{floor((entry[1] - UTC_OFFSET).timestamp())}>",39                inline=False,40            )41        return embed42class HitSource(ListPageSource):43    def __init__(self, hits: list[LoggedEvent], campaign_id: str, *, per_page: int = 4) -> None:44        self.campaign_id: str = campaign_id45        super().__init__(sorted(hits, key=lambda x: x.redirected_at, reverse=True), per_page=per_page)46    async def format_page(self, menu, entries):47        offset = menu.current_page * self.per_page48        embed: Embed = Embed(49            title=f"Campaign {self.campaign_id} summary", color=0x00ff00, description=f"{len(self.entries)} total hits"50        )51        embed.set_footer(text=f"Page {menu.current_page + 1}/{self.get_max_pages()}")52        for iteration, entry in enumerate(entries, start=offset):53            entry: LoggedEvent54            iteration: int55            embed.add_field(56                name=f"{iteration + 1}. ",57                value=f"Occurred at <t:{floor((entry.redirected_at - UTC_OFFSET).timestamp())}>\n"58                      f"Redirected to [{entry.redirected_to}]({entry.redirected_to})\n"59                      f"IP Address: `{entry.remote}`"60                      + (f"\nUser Agent: `{entry.user_agent}`" if entry.user_agent is not None else ""),61                inline=False,62            )63        return embed64async def get_listen_doc(collection: AsyncIOMotorCollection, link_id: str) -> Document:65    return await Document.get_document(collection, {"_id": link_id})66def get_collection(bot: BOT_TYPES) -> AsyncIOMotorCollection:67    """Returns the database for the Redirector cog."""68    return bot.database["redirector"]69class Redirector(Cog):70    """A cog that allows you to use Redirector, an IP logging and analytics service."""71    def __init__(self, bot: BOT_TYPES) -> None:72        self.bot: BOT_TYPES = bot73        self.client_session: Optional[ClientSession] = None74    @Cog.listener()75    async def on_ready(self) -> None:76        # TODO: Webhook implementation77        self.check_for_redirections.start()78    @loop(seconds=60)79    async def check_for_redirections(self) -> None:80        await self.get_client()81        async for raw_document in get_collection(self.bot).find({}):82            document: Document = Document(raw_document, collection=get_collection(self.bot),83                                          query={"_id": raw_document["_id"]})84            if document.get("listening_user") is None:85                continue86            else:87                async with self.client_session.get(f"{API_URL}hits/{document['_id']}") as response:88                    if response.status == 200:89                        data: list[dict[str, Any]] = await response.json()90                        for entry in data:91                            timestamp: datetime = datetime.fromisoformat(entry["timestamp"])92                            if timestamp > document["last_checked"]:93                                self.bot.dispatch(94                                    "redirection",95                                    LoggedEvent(96                                        entry["link_id"],97                                        entry["redirected_to"],98                                        timestamp,99                                        entry["remote"],100                                        entry.get("user_agent"),101                                        document=document,102                                    )103                                )104                        await document.update_db({"$currentDate": {"last_checked": True}})105                    else:106                        logger.warning(f"Failed to get redirections for {document['_id']}")107    async def get_client(self) -> None:108        if self.client_session is None:109            self.client_session = ClientSession()110    async def cog_before_invoke(self, ctx: Context) -> None:111        await self.get_client()112    def cog_unload(self) -> None:113        if self.client_session is not None:114            self.bot.loop.create_task(self.client_session.close())115        if self.check_for_redirections.is_running():116            self.check_for_redirections.stop()117    @Cog.listener()118    async def on_redirection(self, logged_event: LoggedEvent) -> None:119        """A listener that is called when a redirection is logged."""120        listening_document: Document = logged_event.document121        if listening_document.get("listening_user") is not None:122            listening_user: User = (123                    self.bot.get_user(listening_document["listening_user"])124                    or await self.bot.fetch_user(listening_document["listening_user"])125            )126            embed: Embed = (127                Embed(128                    title="Result",129                    description=f"Regarding campaign ID `{logged_event.link_id}`"130                )131                    .add_field(132                    name="Redirected to:",133                    value=f"[{logged_event.redirected_to}]({logged_event.redirected_to})"134                )135                    .add_field(136                    name="Redirected at:",137                    value=f"<t:{floor((logged_event.redirected_at - UTC_OFFSET).timestamp())}>"138                )139                    .add_field(140                    name="IP Address:",141                    value=f"`{logged_event.remote}`"142                )143            )144            if logged_event.user_agent is not None:145                embed: Embed = embed.add_field(146                    name="User Agent:",147                    value=f"`{logged_event.user_agent}`"148                )149            await listening_user.send(150                "An IP address has been grabbed!",151                embed=embed152            )153    @command()154    async def stopgrabbing(155            self,156            ctx: CustomContext,157            link_id: str = Option(name="campaign_id", description="The campaign ID to stop listening to.")158    ) -> None:159        """Stops listening to a campaign ID."""160        await ctx.defer(ephemeral=True)161        listening_document: Document = await get_listen_doc(get_collection(self.bot), link_id)162        if listening_document.get("listening_user") is None:163            await ctx.send("That campaign ID is not being listened to.", ephemeral=True)164            return165        if listening_document["listening_user"] != ctx.author.id:166            await ctx.send("You are not listening to that campaign ID.", ephemeral=True)167            return168        await listening_document.delete_db()169        await ctx.send("You are no longer listening to that campaign ID.", ephemeral=True)170    @command(aliases=["register_listener", "make_campaign"])171    async def ipgrab(172            self,173            ctx: CustomContext,174            destination: str = Option(description="The URL to redirect the victim to."),175            link_id: Optional[str] = Option(name="campaign_id", description="The campaign ID to register.")176    ) -> None:177        """178        Registers a redirector campaign.179        This can be used for IP grabbing, analytics, link shortening, or anything else that needs to be redirected.180        """181        await ctx.defer(ephemeral=True)182        link_id: str = link_id or random_string(length=6)183        # Start listening184        listening_document: Document = await get_listen_doc(get_collection(self.bot), link_id)185        await listening_document.update_db(186            {"$set": {"listening_user": ctx.author.id}, "$currentDate": {"last_checked": True, "created_at": True}}187        )188        async with self.client_session.post(f"{API_URL}{link_id}", data=destination) as response:189            if response.status != 201:190                raise HTTPException(response, response.reason)  # Possibly jank. It should be fine.191        # TODO: Webhook implementation192        await ctx.send(193            f"Campaign link created: <{SHORT_URL}{link_id}>\n"194            f"Your campaign ID is: `{link_id}`\n"195            f"Make sure your DMs are open. You will receive notifications there.",196            ephemeral=True197        )198    @command()199    async def all_campaigns(self, ctx: CustomContext) -> None:200        """Lists all the redirector campaigns you are listening to."""201        await ctx.defer(ephemeral=True)202        all_campaigns: list[tuple[str, datetime]] = []203        async for document in get_collection(self.bot).find({"listening_user": ctx.author.id}):204            all_campaigns.append((document["_id"], document["created_at"]))205        source: CampaignSource = CampaignSource(all_campaigns)206        await ViewMenuPages(source).start(ctx, ephemeral=True)207    @command()208    async def summarize_campaign(209            self,210            ctx: CustomContext,211            link_id: str = Option(name="campaign_id", description="The campaign ID to summarize."),212    ) -> None:213        """Summarizes a redirector campaign."""214        await ctx.defer(ephemeral=True)215        all_hits: list[LoggedEvent] = []216        document: Document = await get_listen_doc(get_collection(self.bot), link_id)217        async with self.client_session.get(f"{API_URL}hits/{document['_id']}") as response:218            data: list[dict[str, Any]] = await response.json()219            for entry in data:220                timestamp: datetime = datetime.fromisoformat(entry["timestamp"])221                all_hits.append(222                    LoggedEvent(223                        entry["link_id"],224                        entry["redirected_to"],225                        timestamp,226                        entry["remote"],227                        entry.get("user_agent"),228                        document=document,229                    )230                )231        source: HitSource = HitSource(all_hits, link_id)232        await ViewMenuPages(source).start(ctx, ephemeral=True)233def setup(bot: BOT_TYPES) -> None:234    """Loads the Redirector cog."""...

CWSRequests.py

Source:CWSRequests.py

1#!/usr/bin/env python32# Contest Management System - http://cms-dev.github.io/3# Copyright Â© 2010-2012 Giovanni Mascellani <mascellani@poisson.phc.unipi.it>4# Copyright Â© 2010-2018 Stefano Maggiolo <s.maggiolo@gmail.com>5# Copyright Â© 2010-2012 Matteo Boscariol <boscarim@hotmail.com>6# Copyright Â© 2014 Artem Iglikov <artem.iglikov@gmail.com>7# Copyright Â© 2016 Luca Wehrstedt <luca.wehrstedt@gmail.com>8# Copyright Â© 2017 Luca Chiodini <luca@chiodini.org>9#10# This program is free software: you can redistribute it and/or modify11# it under the terms of the GNU Affero General Public License as12# published by the Free Software Foundation, either version 3 of the13# License, or (at your option) any later version.14#15# This program is distributed in the hope that it will be useful,16# but WITHOUT ANY WARRANTY; without even the implied warranty of17# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the18# GNU Affero General Public License for more details.19#20# You should have received a copy of the GNU Affero General Public License21# along with this program.  If not, see <http://www.gnu.org/licenses/>.22import logging23import os24import random25import re26import tempfile27from urllib.parse import parse_qs, urlsplit28from cms import config29from cms.grading.languagemanager import filename_to_language30from cmscommon.crypto import decrypt_number31from cmstestsuite.web import GenericRequest, LoginRequest32logger = logging.getLogger(__name__)33class CWSLoginRequest(LoginRequest):34    def test_success(self):35        if not LoginRequest.test_success(self):36            return False37        if self.redirected_to.rstrip("/") != self.base_url.rstrip("/"):38            return False39        return True40class HomepageRequest(GenericRequest):41    """Load the main page of CWS.42    """43    def __init__(self, browser, username, loggedin, base_url=None):44        GenericRequest.__init__(self, browser, base_url)45        self.url = self.base_url46        self.username = username47        self.loggedin = loggedin48    def describe(self):49        return "check the main page"50    def test_success(self):51        if not GenericRequest.test_success(self):52            return False53        username_re = re.compile(self.username)54        if self.loggedin:55            if username_re.search(self.res_data) is None:56                return False57        else:58            if username_re.search(self.res_data) is not None:59                return False60        return True61class TaskRequest(GenericRequest):62    """Load a task page in CWS.63    """64    def __init__(self, browser, task_id, base_url=None):65        GenericRequest.__init__(self, browser, base_url)66        self.url = "%s/tasks/%s/description" % (self.base_url, task_id)67        self.task_id = task_id68    def describe(self):69        return "load page for task %s (%s)" % (self.task_id, self.url)70class TaskStatementRequest(GenericRequest):71    """Load a task statement in CWS.72    """73    def __init__(self, browser, task_id, language_code, base_url=None):74        GenericRequest.__init__(self, browser, base_url)75        self.url = "%s/tasks/%s/statements/%s" % (self.base_url,76                                                  task_id, language_code)77        self.task_id = task_id78    def describe(self):79        return "load statement for task %s (%s)" % (self.task_id, self.url)80    def specific_info(self):81        return '\nNO DATA DUMP FOR TASK STATEMENTS\n'82class SubmitRequest(GenericRequest):83    """Submit a solution in CWS.84    """85    def __init__(self, browser, task, submission_format,86                 filenames, language=None, base_url=None):87        GenericRequest.__init__(self, browser, base_url)88        self.url = "%s/tasks/%s/submit" % (self.base_url, task[1])89        self.task = task90        self.submission_format = submission_format91        self.filenames = filenames92        self.data = {}93        # If not passed, try to recover the language from the filenames.94        if language is None:95            for filename in filenames:96                lang = filename_to_language(filename)97                if lang is not None:98                    language = lang.name99                    break100        # Only send the language in the request if not None.101        if language is not None:102            self.data = {"language": language}103    def _prepare(self):104        GenericRequest._prepare(self)105        self.files = list(zip(self.submission_format, self.filenames))106    def describe(self):107        return "submit sources %s for task %s (ID %d) %s" % \108            (repr(self.filenames), self.task[1], self.task[0], self.url)109    def specific_info(self):110        return 'Task: %s (ID %d)\nFile: %s\n' % \111            (self.task[1], self.task[0], repr(self.filenames)) + \112            GenericRequest.specific_info(self)113    def test_success(self):114        if not GenericRequest.test_success(self):115            return False116        return self.get_submission_id() is not None117    def get_submission_id(self):118        # Only valid after self.execute()119        # Parse submission ID out of redirect.120        if self.redirected_to is None:121            return None122        query = parse_qs(urlsplit(self.redirected_to).query)123        if "submission_id" not in query or len(query["submission_id"]) != 1:124            logger.warning("Redirected to an unexpected page: `%s'",125                           self.redirected_to)126            return None127        try:128            submission_id = decrypt_number(query["submission_id"][0],129                                           config.secret_key)130        except Exception:131            logger.warning("Unable to decrypt submission id from page: `%s'",132                           self.redirected_to)133            return None134        return submission_id135class SubmitUserTestRequest(GenericRequest):136    """Submit a user test in CWS."""137    def __init__(self, browser, task, submission_format,138                 filenames, language=None, base_url=None):139        GenericRequest.__init__(self, browser, base_url)140        self.url = "%s/tasks/%s/test" % (self.base_url, task[1])141        self.task = task142        self.submission_format = submission_format143        self.filenames = filenames144        self.data = {}145        # If not passed, try to recover the language from the filenames.146        if language is None:147            for filename in filenames:148                lang = filename_to_language(filename)149                if lang is not None:150                    language = lang.name151                    break152        # Only send the language in the request if not None.153        if language is not None:154            self.data = {"language": language}155    def _prepare(self):156        GenericRequest._prepare(self)157        # Let's generate an arbitrary input file.158        # TODO: delete this file once we're done with it.159        _, temp_filename = tempfile.mkstemp()160        self.files = \161            list(zip(self.submission_format, self.filenames)) + \162            [("input", temp_filename)]163    def describe(self):164        return "submit user test %s for task %s (ID %d) %s" % \165            (repr(self.filenames), self.task[1], self.task[0], self.url)166    def specific_info(self):167        return 'Task: %s (ID %d)\nFile: %s\n' % \168            (self.task[1], self.task[0], repr(self.filenames)) + \169            GenericRequest.specific_info(self)170    def test_success(self):171        if not GenericRequest.test_success(self):172            return False173        return self.get_user_test_id() is not None174    def get_user_test_id(self):175        # Only valid after self.execute()176        # Parse submission ID out of redirect.177        if self.redirected_to is None:178            return None179        query = parse_qs(urlsplit(self.redirected_to).query)180        if "user_test_id" not in query or len(query["user_test_id"]) != 1:181            logger.warning("Redirected to an unexpected page: `%s'",182                           self.redirected_to)183            return None184        try:185            user_test_id = decrypt_number(query["user_test_id"][0],186                                          config.secret_key)187        except Exception:188            logger.warning("Unable to decrypt user test id from page: `%s'",189                           self.redirected_to)190            return None191        return user_test_id192class TokenRequest(GenericRequest):193    """Release test a submission.194    """195    def __init__(self, browser, task, submission_num, base_url=None):196        GenericRequest.__init__(self, browser, base_url)197        self.url = "%s/tasks/%s/submissions/%s/token" % (self.base_url,198                                                         task[1],199                                                         submission_num)200        self.task = task201        self.submission_num = submission_num202        self.data = {}203    def describe(self):204        return "release test the %s-th submission for task %s (ID %d)" % \205            (self.submission_num, self.task[1], self.task[0])206    def specific_info(self):207        return 'Task: %s (ID %d)\nSubmission: %s\n' % \208            (self.task[1], self.task[0], self.submission_num) + \209            GenericRequest.specific_info(self)210class SubmitRandomRequest(GenericRequest):211    """Submit a solution in CWS.212    """213    def __init__(self, browser, task, base_url=None,214                 submissions_path=None):215        GenericRequest.__init__(self, browser, base_url)216        self.url = "%s/tasks/%s/submit" % (self.base_url, task[1])217        self.task = task218        self.submissions_path = submissions_path219        self.data = {}220    def _prepare(self):221        """Select a random solution and prepare it for submission.222        If task/ is the task directory, it might contain files (only223        if the submission format is with a single file) and224        directory. If it contains a file, it is assumed that it is the225        only element in the submission format, and is the basename226        without extension of the file. If it is a directory, all files227        inside are assumed to be part of the submission format with228        their basenames without extension.229        """230        GenericRequest._prepare(self)231        # Select a random directory or file inside the task directory.232        task_path = os.path.join(self.submissions_path, self.task[1])233        sources = os.listdir(task_path)234        source = random.choice(sources)235        lang = filename_to_language(source)236        if lang is not None:237            self.data["language"] = lang.name238        self.source_path = os.path.join(task_path, source)239        # Compose the submission format240        self.files = []241        if os.path.isdir(self.source_path):242            submission_formats = os.listdir(self.source_path)243            self.files = [('%s.%%l' % (os.path.splitext(sf)[0]),244                           os.path.join(self.source_path, sf))245                          for sf in submission_formats]246        else:247            submission_format = os.path.splitext(source)[0]248            self.files = [('%s.%%l' % (submission_format), self.source_path)]249    def describe(self):250        return "submit source %s for task %s (ID %d) %s" % \251            (self.source_path, self.task[1], self.task[0], self.url)252    def specific_info(self):253        return 'Task: %s (ID %d)\nFile: %s\n' % \254            (self.task[1], self.task[0], self.source_path) + \...

reject_redirects_test.py

Source:reject_redirects_test.py

1# hammertime: A high-volume http fetch library2# Copyright (C) 2016-  Delve Labs inc.3#4# This program is free software; you can redistribute it and/or5# modify it under the terms of the GNU General Public License6# as published by the Free Software Foundation; either version 27# of the License.8#9# This program is distributed in the hope that it will be useful,10# but WITHOUT ANY WARRANTY; without even the implied warranty of11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the12# GNU General Public License for more details.13#14# You should have received a copy of the GNU General Public License15# along with this program; if not, write to the Free Software16# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.17from unittest import TestCase18from unittest.mock import MagicMock, patch19from aiohttp.test_utils import make_mocked_coro20from hammertime.http import Entry, StaticResponse21from hammertime.ruleset import RejectRequest22from tests.fixtures import async_test23from hammertime.rules import RejectCatchAllRedirect, RedirectLimiter24from hammertime.kb import KnowledgeBase25class TestRejectCatchAllRedirect(TestCase):26    def setUp(self):27        self.heuristic = RejectCatchAllRedirect()28        self.host = "http://example.com"29        self.fake_engine = MagicMock()30        self.heuristic.set_engine(self.fake_engine)31        self.heuristic.child_heuristics = MagicMock()32    @async_test()33    async def test_after_headers_request_random_filename_in_same_path_as_initial_request_if_response_is_redirect(self):34        self.fake_engine.perform_high_priority = make_mocked_coro(return_value=MagicMock())35        entry0 = self.create_redirected_request("/admin/restricted-resource.php", redirected_to="/admin/login.php")36        entry1 = self.create_redirected_request("/junkpath", redirected_to="/index.html")37        entry2 = self.create_redirected_request("/path/to/admin/resource", redirected_to="/path/to/admin/login.php")38        with patch("hammertime.rules.redirects.uuid4", MagicMock(return_value="uuid")):39            await self.heuristic.after_headers(entry0)40            await self.heuristic.after_headers(entry1)41            await self.heuristic.after_headers(entry2)42            self.assertRequested(self.host + "/admin/uuid", self.fake_engine.perform_high_priority, order=0)43            self.assertRequested(self.host + "/uuid", self.fake_engine.perform_high_priority, order=1)44            self.assertRequested(self.host + "/path/to/admin/uuid", self.fake_engine.perform_high_priority, order=2)45    @async_test()46    async def test_after_headers_doesnt_request_random_filename_if_response_is_not_redirect(self):47        self.fake_engine.perform_high_priority = make_mocked_coro()48        entry = Entry.create("http://example.com/junkpath", response=StaticResponse(404, {}, b"Not found"))49        await self.heuristic.after_headers(entry)50        self.fake_engine.perform_high_priority.assert_not_called()51    @async_test()52    async def test_after_headers_reject_request_if_request_for_random_file_has_same_redirect_as_initial_request(self):53        initial_request = self.create_redirected_request("/admin/resource.php", redirected_to="/admin/login.php")54        returned_entry = self.create_redirected_request("/admin/uuid", redirected_to="/admin/login.php")55        self.fake_engine.perform_high_priority = make_mocked_coro(return_value=returned_entry)56        with self.assertRaises(RejectRequest):57            await self.heuristic.after_headers(initial_request)58    @async_test()59    async def test_after_headers_transform_relative_location_to_absolute_location(self):60        initial_request = self.create_redirected_request("/admin/resource.php", redirected_to="/admin/login.php",61                                                         relative=True)62        returned_entry = self.create_redirected_request("/admin/uuid", redirected_to="/admin/login.php")63        self.fake_engine.perform_high_priority = make_mocked_coro(return_value=returned_entry)64        with self.assertRaises(RejectRequest):65            await self.heuristic.after_headers(initial_request)66    @async_test()67    async def test_ignore_when_path_is_present_in_redirect(self):68        initial_request = self.create_redirected_request("/admin/a.php", redirected_to="/404.php?path=/admin/a.php")69        returned_entry = self.create_redirected_request("/admin/uuid", redirected_to="/404.php?path=/admin/uuid")70        self.fake_engine.perform_high_priority = make_mocked_coro(return_value=returned_entry)71        with self.assertRaises(RejectRequest):72            with patch("hammertime.rules.redirects.uuid4", MagicMock(return_value="uuid")):73                await self.heuristic.after_headers(initial_request)74    @async_test()75    async def test_before_request_accept_request_if_random_file_not_redirected_to_same_path_as_initial_request(self):76        initial_request = self.create_redirected_request("/admin/resource.php", redirected_to="/admin/login.php")77        returned_entry = self.create_redirected_request("/admin/uuid", redirected_to="/catchAll.html")78        self.fake_engine.perform_high_priority = make_mocked_coro(return_value=returned_entry)79        await self.heuristic.after_headers(initial_request)80    @async_test()81    async def test_after_headers_add_request_response_to_knowledge_base(self):82        first_host = "http://example1.com"83        returned_entry = self.create_redirected_request("/anything", relative=True, redirected_to="/login.php")84        self.fake_engine.perform_high_priority = make_mocked_coro(return_value=returned_entry)85        entry0 = self.create_redirected_request("/admin/restricted-resource.php", host=first_host,86                                                redirected_to="/login.php")87        entry1 = self.create_redirected_request("/junkpath", host=first_host, redirected_to="/login.php")88        entry2 = self.create_redirected_request("/path/to/admin/resource", host=first_host, redirected_to="/login.php")89        second_host = "http://example2.com"90        entry3 = self.create_redirected_request("/admin/restricted-resource.php", host=second_host,91                                                redirected_to="/login.php")92        entry4 = self.create_redirected_request("/junkpath/", host=second_host, redirected_to="/login.php")93        entry5 = self.create_redirected_request("/path/to/admin/resource", host=second_host, redirected_to="/login.php")94        kb = KnowledgeBase()95        self.heuristic.set_kb(kb)96        with patch("hammertime.rules.redirects.uuid4", MagicMock(return_value="uuid")):97            await self.ignore_reject_request(self.heuristic.after_headers, entry0)98            await self.ignore_reject_request(self.heuristic.after_headers, entry1)99            await self.ignore_reject_request(self.heuristic.after_headers, entry2)100            await self.ignore_reject_request(self.heuristic.after_headers, entry3)101            await self.ignore_reject_request(self.heuristic.after_headers, entry4)102            await self.ignore_reject_request(self.heuristic.after_headers, entry5)103            self.assertEqual(kb.redirects["%s/admin/" % first_host], "%s/login.php" % first_host)104            self.assertEqual(kb.redirects["%s/" % first_host], "%s/login.php" % first_host)105            self.assertEqual(kb.redirects["%s/path/to/admin/" % first_host], "%s/login.php" % first_host)106            self.assertEqual(kb.redirects["%s/admin/" % second_host], "%s/login.php" % second_host)107            self.assertEqual(kb.redirects["%s/junkpath/" % second_host], "%s/login.php" % second_host)108            self.assertEqual(kb.redirects["%s/path/to/admin/" % second_host], "%s/login.php" % second_host)109    @async_test()110    async def test_after_headers_doesnt_request_random_filename_if_redirect_is_already_in_knowledge_base(self):111        self.fake_engine.perform_high_priority = make_mocked_coro()112        kb = KnowledgeBase()113        self.heuristic.set_kb(kb)114        kb.redirects["%s/wp-admin/" % self.host] = "%s/somewhere" % self.host115        entry = self.create_redirected_request("/wp-admin/admin.php", redirected_to="/wp-login.php")116        await self.heuristic.after_headers(entry)117        self.fake_engine.perform_high_priority.assert_not_called()118    def create_redirected_request(self, path, *, host=None, redirected_to, relative=False):119        host = host or self.host120        if relative:121            headers = {"location": redirected_to}122        else:123            headers = {"location": host + redirected_to}124        response = StaticResponse(code=302, headers=headers, content=b"content")125        return Entry.create(host + path, response=response)126    def assertRequested(self, url, request_method, order=None):127        requested = False128        if order is not None:129            call_list = [request_method.call_args_list[order]]130        else:131            call_list = request_method.call_args_list132        for call in call_list:133            args, kwargs = call134            entry = args[0]135            if entry.request.url == url:136                requested = True137                break138        self.assertTrue(requested)139    async def ignore_reject_request(self, coro, arg):140        try:141            await coro(arg)142        except RejectRequest:143            pass144def redirect(url, redirect_to):145    return Entry.create(url, response=StaticResponse(302, headers={146        "location": redirect_to,147    }))148class RedirectLimiterTest(TestCase):149    @async_test()150    async def test_obvious_redirect_to_404(self):151        limiter = RedirectLimiter(sequence_matching=False)152        with self.assertRaises(RejectRequest):153            entry = redirect("http://example.com/foobar", redirect_to="http://example.com/404.php")154            await limiter.after_headers(entry)155    @async_test()156    async def test_full_matches_only(self):157        limiter = RedirectLimiter(sequence_matching=False)158        await limiter.after_headers(redirect("http://example.com/foobar", redirect_to="http://example.com/foobar/1404"))159        await limiter.after_headers(redirect("http://example.com/foobar", redirect_to="http://example.com/foobar/4041"))160    @async_test()161    async def test_obvious_redirect_to_not_found(self):162        limiter = RedirectLimiter(sequence_matching=False)163        with self.assertRaises(RejectRequest):164            entry = redirect("http://example.com/foobar", redirect_to="http://example.com/not-found")165            await limiter.after_headers(entry)166    @async_test()167    async def test_reject_wildely_different_paths(self):168        limiter = RedirectLimiter()169        with self.assertRaises(RejectRequest):170            await limiter.after_headers(redirect("http://example.com/foobar",171                                                 redirect_to="http://example.com/hello-world"))172    @async_test()173    async def test_accept_similar_paths(self):174        limiter = RedirectLimiter()175        await limiter.after_headers(redirect("http://example.com/foobar", redirect_to="http://example.com/foobar/"))176    @async_test()177    async def test_with_relative_paths(self):178        limiter = RedirectLimiter()179        await limiter.after_headers(redirect("http://example.com/foobar", redirect_to="foobar/"))...

Playwright tutorial

LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.

Chapters:

What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.