Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use set_auth method in tempest

Best Python code snippet using tempest_python

remote.py

Source:remote.py

...81        proto = project_cfg[CONFIG_PROTOCOL]82        host = project_cfg[CONFIG_HOST]83        self._project = ProjectService(host, version)84        self._project.base_protocol = proto85        self._project.set_auth(self._token_project)86    def _init_metadata_service(self, version):87        """88        Method to initialize the Metadata Service from the config data89        Args:90            version (string): Version of Boss API to use.91        Returns:92            None93        Raises:94            (KeyError): if given invalid version.95        """96        metadata_cfg = self._load_config_section(CONFIG_METADATA_SECTION)97        self._token_metadata = metadata_cfg[CONFIG_TOKEN]98        proto = metadata_cfg[CONFIG_PROTOCOL]99        host = metadata_cfg[CONFIG_HOST]100        self._metadata = MetadataService(host, version)101        self._metadata.base_protocol = proto102        self._metadata.set_auth(self._token_metadata)103    def _init_volume_service(self, version):104        """105        Method to initialize the Volume Service from the config data106        Args:107            version (string): Version of Boss API to use.108        Returns:109            None110        Raises:111            (KeyError): if given invalid version.112        """113        volume_cfg = self._load_config_section(CONFIG_VOLUME_SECTION)114        self._token_volume = volume_cfg[CONFIG_TOKEN]115        proto = volume_cfg[CONFIG_PROTOCOL]116        host = volume_cfg[CONFIG_HOST]117        self._volume = VolumeService(host, version)118        self._volume.base_protocol = proto119        self._volume.set_auth(self._token_volume)120    def _load_config_section(self, section_name):121        """122        Method to load the specific Service section from the config file if it123        exists, or fall back to the default124        Args:125            section_name (str): The desired service section name126        Returns:127            (dict): the section parameters128        """129        if self._config.has_section(section_name):130            # Load specific section131            section = dict(self._config.items(section_name))132        elif self._config.has_section("Default"):133            # Load Default section134            section = dict(self._config.items("Default"))135        else:136            raise KeyError((137                "'{}' was not found in the configuration file and no default " +138                "configuration was provided."139            ).format(section_name))140        # Make sure section is valid141        if "protocol" in section and "host" in section and "token" in section:142            return section143        else:144            raise KeyError(145                "Missing values in configuration data. " +146                "Must contain: protocol, host, token"147            )148    @property149    def token_project(self):150        """151        Returns the current token152        """153        return self._token_project154    @token_project.setter155    def token_project(self, value):156        self._token_project = value157        self.project_service.set_auth(self._token_project)158    @property159    def token_metadata(self):160        """161        Returns metadata for the current token162        """163        return self._token_metadata164    @token_metadata.setter165    def token_metadata(self, value):166        self._token_metadata = value167        self.metadata_service.set_auth(self._token_metadata)168    @property169    def token_volume(self):170        """171        Get the current token volume172        """173        return self._token_volume174    @token_volume.setter175    def token_volume(self, value):176        self._token_volume = value177        self.volume_service.set_auth(self._token_volume)178    def list_groups(self, filtr=None):179        """180        Get the groups the logged in user is a member of.181        Optionally filter by 'member' or 'maintainer'.182        Args:183            filtr (optional[string|None]): ['member'|'maintainer']184        Returns:185            (list[string]): List of group names.186        Raises:187            requests.HTTPError on failure.188        """189        self.project_service.set_auth(self._token_project)190        return self.project_service.list_groups(filtr)191    def get_group(self, name, user_name=None):192        """193        Get information on the given group or whether or not a user is a member194        of the group.195        Args:196            name (string): Name of group to query.197            user_name (optional[string]): Supply None if not interested in198            determining if user is a member of the given group.199        Returns:200            (mixed): Dictionary if getting group information or bool if a user201                name is supplied.202        Raises:203            requests.HTTPError on failure.204        """205        self.project_service.set_auth(self._token_project)206        return self.project_service.get_group(name, user_name)207    def create_group(self, name):208        """209        Create a new group.210        Args:211            name (string): Name of the group to create.212        Returns:213            (bool): True on success.214        Raises:215            requests.HTTPError on failure.216        """217        self.project_service.set_auth(self._token_project)218        return self.project_service.create_group(name)219    def delete_group(self, name):220        """221        Delete given group.222        Args:223            name (string): Name of group.224        Returns:225            (bool): True on success.226        Raises:227            requests.HTTPError on failure.228        """229        self.project_service.set_auth(self._token_project)230        return self.project_service.delete_group(name)231    def list_group_members(self, name):232        """233        Get the members of a group.234        Args:235            name (string): Name of group to query.236        Returns:237            (list[string]): List of member names.238        Raises:239            requests.HTTPError on failure.240        """241        self.project_service.set_auth(self._token_project)242        return self.project_service.list_group_members(name)243    def add_group_member(self, grp_name, user):244        """245        Add the given user to the named group.246        Both group and user must already exist for this to succeed.247        Args:248            name (string): Name of group.249            user_name (string): User to add to group.250        Raises:251            requests.HTTPError on failure.252        """253        self.project_service.set_auth(self._token_project)254        self.project_service.add_group_member(grp_name, user)255    def delete_group_member(self, grp_name, user):256        """257        Delete the given user to the named group.258        Both group and user must already exist for this to succeed.259        Args:260            name (string): Name of group.261            user_name (string): User to delete from the group.262        Raises:263            requests.HTTPError on failure.264        """265        self.project_service.set_auth(self._token_project)266        self.project_service.delete_group_member(grp_name, user)267    def get_is_group_member(self, grp_name, user):268        """269        Check if the given user is a member of the named group.270        Note that a group maintainer is not considered a member unless the271        user is also explicitly added as a member.272        Args:273            name (string): Name of group.274            user_name (string): User of interest.275        Returns:276            (bool): False if user not a member.277        """278        self.project_service.set_auth(self._token_project)279        return self.project_service.get_is_group_member(grp_name, user)280    def list_group_maintainers(self, name):281        """282        Get the maintainers of a group.283        Args:284            name (string): Name of group to query.285        Returns:286            (list[string]): List of maintainer names.287        Raises:288            requests.HTTPError on failure.289        """290        self.project_service.set_auth(self._token_project)291        return self.project_service.list_group_maintainers(name)292    def add_group_maintainer(self, name, user):293        """294        Add the given user to the named group.295        Both group and user must already exist for this to succeed.296        Args:297            name (string): Name of group.298            user (string): User to add to group.299        Raises:300            requests.HTTPError on failure.301        """302        self.project_service.set_auth(self._token_project)303        self.project_service.add_group_maintainer(name, user)304    def delete_group_maintainer(self, grp_name, user):305        """306        Delete the given user to the named group.307        Both group and user must already exist for this to succeed.308        Args:309            name (string): Name of group.310            user (string): User to add to group.311        Raises:312            requests.HTTPError on failure.313        """314        self.project_service.set_auth(self._token_project)315        self.project_service.delete_group_maintainer(grp_name, user)316    def get_is_group_maintainer(self, grp_name, user):317        """318        Check if the given user is a member of the named group.319        Args:320            name (string): Name of group.321            user (string): User of interest.322        Returns:323            (bool): False if user not a member.324        """325        self.project_service.set_auth(self._token_project)326        return self.project_service.get_is_group_maintainer(grp_name, user)327    def list_permissions(self, group_name=None, resource=None):328        """329        List permission sets associated filtering by group and/or resource.330        Args:331            group_name (string): Name of group.332            resource (intern.resource.boss.Resource): Identifies which data333                model object to operate on.334        Returns:335            (list): List of permissions.336        Raises:337            requests.HTTPError on failure.338        """339        self.project_service.set_auth(self._token_project)340        return self.project_service.list_permissions(group_name, resource)341    def get_permissions(self, grp_name, resource):342        """343        Get permissions associated the group has with the given resource.344        Args:345            grp_name (string): Name of group.346            resource (intern.resource.boss.Resource): Identifies which data347                model object to operate on.348        Returns:349            (list): List of permissions.350        Raises:351            requests.HTTPError on failure.352        """353        self.project_service.set_auth(self._token_project)354        return self.project_service.get_permissions(grp_name, resource)355    def add_permissions(self, grp_name, resource, permissions):356        """357        Add additional permissions for the group associated with the resource.358        Args:359            grp_name (string): Name of group.360            resource (intern.resource.boss.Resource): Identifies which data361                model object to operate on.362            permissions (list): List of permissions to add to the given resource363        Raises:364            requests.HTTPError on failure.365        """366        self.project_service.set_auth(self._token_project)367        self.project_service.add_permissions(grp_name, resource, permissions)368    def update_permissions(self, grp_name, resource, permissions):369        """370        Update permissions for the group associated with the given resource.371        Args:372            grp_name (string): Name of group.373            resource (intern.resource.boss.Resource): Identifies which data374                model object to operate on375            permissions (list): List of permissions to add to the given resource376        Raises:377            requests.HTTPError on failure.378        """379        self.project_service.set_auth(self._token_project)380        self.project_service.update_permissions(grp_name, resource, permissions)381    def delete_permissions(self, grp_name, resource):382        """383        Removes permissions from the group for the given resource.384        Args:385            grp_name (string): Name of group.386            resource (intern.resource.boss.Resource): Identifies which data387                model object to operate on.388        Raises:389            requests.HTTPError on failure.390        """391        self.project_service.set_auth(self._token_project)392        self.project_service.delete_permissions(grp_name, resource)393    def get_user_roles(self, user):394        """395        Get roles associated with the given user.396        Args:397            user (string): User name.398        Returns:399            (list): List of roles that user has.400        Raises:401            requests.HTTPError on failure.402        """403        self.project_service.set_auth(self._token_project)404        return self.project_service.get_user_roles(user)405    def add_user_role(self, user, role):406        """407        Add role to given user.408        Args:409            user (string): User name.410            role (string): Role to assign.411        Raises:412            requests.HTTPError on failure.413        """414        self.project_service.set_auth(self._token_project)415        self.project_service.add_user_role(user, role)416    def delete_user_role(self, user, role):417        """418        Remove role from given user.419        Args:420            user (string): User name.421            role (string): Role to remove.422        Raises:423            requests.HTTPError on failure.424        """425        self.project_service.set_auth(self._token_project)426        self.project_service.delete_user_role(user, role)427    def get_user(self, user):428        """429        Get user's data (first and last name, email, etc).430        Args:431            user (string): User name.432        Returns:433            (dictionary): User's data encoded in a dictionary.434        Raises:435            requests.HTTPError on failure.436        """437        self.project_service.set_auth(self._token_project)438        return self.project_service.get_user(user)439    def get_user_groups(self, user):440        """441        Get user's group memberships.442        Args:443            user (string): User name.444        Returns:445            (list): User's groups.446        Raises:447            requests.HTTPError on failure.448        """449        self.project_service.set_auth(self._token_project)450        return self.project_service.get_user_groups(user)451    def add_user(452            self, user,453            first_name=None, last_name=None,454            email=None, password=None455        ):456        """457        Add a new user.458        Args:459            user (string): User name.460            first_name (optional[string]): User's first name.  Defaults to None.461            last_name (optional[string]): User's last name.  Defaults to None.462            email: (optional[string]): User's email address.  Defaults to None.463            password: (optional[string]): User's password.  Defaults to None.464        Raises:465            requests.HTTPError on failure.466        """467        self.project_service.set_auth(self._token_project)468        self.project_service.add_user(469            user, first_name, last_name, email, password)470    def delete_user(self, user):471        """472        Delete the given user.473        Args:474            user (string): User name.475        Raises:476            requests.HTTPError on failure.477        """478        self.project_service.set_auth(self._token_project)479        self.project_service.delete_user(user)480    def _list_resource(self, resource):481        """482        List all instances of the given resource type.483        Use the specific list_<resource>() methods instead:484            list_collections()485            list_experiments()486            list_channels()487            list_coordinate_frames()488        Args:489            resource (intern.resource.boss.BossResource): resource.name may be490                an empty string.491        Returns:492            (list)493        Raises:494            requests.HTTPError on failure.495        """496        self.project_service.set_auth(self._token_project)497        return super(BossRemote, self).list_project(resource=resource)498    def list_collections(self):499        """500        List all collections.501        Returns:502            (list)503        Raises:504            requests.HTTPError on failure.505        """506        coll = CollectionResource(name='')507        return self._list_resource(coll)508    def list_experiments(self, collection_name):509        """510        List all experiments that belong to a collection.511        Args:512            collection_name (string): Name of the parent collection.513        Returns:514            (list)515        Raises:516            requests.HTTPError on failure.517        """518        exp = ExperimentResource(519            name='', collection_name=collection_name, coord_frame='foo')520        return self._list_resource(exp)521    def list_channels(self, collection_name, experiment_name):522        """523        List all channels belonging to the named experiment that is part524        of the named collection.525        Args:526            collection_name (string): Name of the parent collection.527            experiment_name (string): Name of the parent experiment.528        Returns:529            (list)530        Raises:531            requests.HTTPError on failure.532        """533        dont_care = 'image'534        chan = ChannelResource(535            name='', collection_name=collection_name,536            experiment_name=experiment_name, type=dont_care)537        return self._list_resource(chan)538    def list_coordinate_frames(self):539        """540        List all coordinate_frames.541        Returns:542            (list)543        Raises:544            requests.HTTPError on failure.545        """546        cf = CoordinateFrameResource(name='')547        return self._list_resource(cf)548    def get_channel(self, chan_name, coll_name, exp_name):549        """550        Helper that gets a fully initialized ChannelResource for an *existing* channel.551        Args:552            chan_name (str): Name of channel.553            coll_name (str): Name of channel's collection.554            exp_name (str): Name of channel's experiment.555        Returns:556            (intern.resource.boss.ChannelResource)557        """558        chan = ChannelResource(chan_name, coll_name, exp_name)559        return self.get_project(chan)560    def create_project(self, resource):561        """562        Create the entity described by the given resource.563        Args:564            resource (intern.resource.boss.BossResource)565        Returns:566            (intern.resource.boss.BossResource): Returns resource of type567                requested on success.568        Raises:569            requests.HTTPError on failure.570        """571        self.project_service.set_auth(self._token_project)572        return self.project_service.create(resource)573    def get_project(self, resource):574        """575        Get attributes of the data model object named by the given resource.576        Args:577            resource (intern.resource.boss.BossResource): resource.name as well578                as any parents must be identified to succeed.579        Returns:580            (intern.resource.boss.BossResource): Returns resource of type581                requested on success.582        Raises:583            requests.HTTPError on failure.584        """585        self.project_service.set_auth(self._token_project)586        return self.project_service.get(resource)587    def update_project(self, resource_name, resource):588        """589        Updates an entity in the data model using the given resource.590        Args:591            resource_name (string): Current name of the resource (in case the592                resource is getting its name changed).593            resource (intern.resource.boss.BossResource): New attributes for594                the resource.595        Returns:596            (intern.resource.boss.BossResource): Returns updated resource of597                given type on success.598        Raises:599            requests.HTTPError on failure.600        """601        self.project_service.set_auth(self._token_project)602        return self.project_service.update(resource_name, resource)603    def delete_project(self, resource):604        """605        Deletes the entity described by the given resource.606        Args:607            resource (intern.resource.boss.BossResource)608        Raises:609            requests.HTTPError on a failure.610        """611        self.project_service.set_auth(self._token_project)612        self.project_service.delete(resource)613    def list_metadata(self, resource):614        """615        List all keys associated with the given resource.616        Args:617            resource (intern.resource.boss.BossResource)618        Returns:619            (list)620        Raises:621            requests.HTTPError on a failure.622        """623        self.metadata_service.set_auth(self._token_metadata)624        return self.metadata_service.list(resource)625    def create_metadata(self, resource, keys_vals):626        """627        Associates new key-value pairs with the given resource.628        Will attempt to add all key-value pairs even if some fail.629        Args:630            resource (intern.resource.boss.BossResource)631            keys_vals (dictionary): Collection of key-value pairs to assign to632                given resource.633        Raises:634            HTTPErrorList on failure.635        """636        self.metadata_service.set_auth(self._token_metadata)637        self.metadata_service.create(resource, keys_vals)638    def get_metadata(self, resource, keys):639        """640        Gets the values for given keys associated with the given resource.641        Args:642            resource (intern.resource.boss.BossResource)643            keys (list)644        Returns:645            (dictionary)646        Raises:647            HTTPErrorList on failure.648        """649        self.metadata_service.set_auth(self._token_metadata)650        return self.metadata_service.get(resource, keys)651    def update_metadata(self, resource, keys_vals):652        """653        Updates key-value pairs with the given resource.654        Will attempt to update all key-value pairs even if some fail.655        Keys must already exist.656        Args:657            resource (intern.resource.boss.BossResource)658            keys_vals (dictionary): Collection of key-value pairs to update on659                the given resource.660        Raises:661            HTTPErrorList on failure.662        """663        self.metadata_service.set_auth(self._token_metadata)664        self.metadata_service.update(resource, keys_vals)665    def delete_metadata(self, resource, keys):666        """667        Deletes the given key-value pairs associated with the given resource.668        Will attempt to delete all key-value pairs even if some fail.669        Args:670            resource (intern.resource.boss.BossResource)671            keys (list)672        Raises:673            HTTPErrorList on failure.674        """675        self.metadata_service.set_auth(self._token_metadata)676        self.metadata_service.delete(resource, keys)677    def parse_bossURI(self, uri): # type: (str) -> Resource678        """679        Parse a bossDB URI and handle malform errors.680        Arguments:681            uri (str): URI of the form bossdb://<collection>/<experiment>/<channel>682        Returns:683            Resource684        """685        t = uri.split("://")[1].split("/")686        if len(t) == 3:687            return self.get_channel(t[2], t[0], t[1])688        raise ValueError("Cannot parse URI " + uri + ".")689    def get_cutout(self, resource, resolution, x_range, y_range, z_range, time_range=None, id_list=[], no_cache=None, access_mode=CacheMode.no_cache, parallel: bool = True, **kwargs):...

test_app.py

Source:test_app.py

...13database_path = "sqlite:///{}".format(os.path.join(project_dir, database_name))14assistant_token = os.getenv('ASSISTANT_JWT')15director_token = os.getenv('DIRECTOR_JWT')16producer_token = os.getenv('PRODUCER_JWT')17def set_auth(role):18    if role == 'assistant':19        return {20            "Content-Type": "application/json",21            "Authorization": "Bearer {}".format(assistant_token)22            }23    elif role == 'director':24        return {25            "Content-Type": "application/json",26            'Authorization': 'Bearer {}'.format(director_token)27            }28    elif role == 'producer':29        return {30            "Content-Type": "application/json",31            'Authorization': 'Bearer {}'.format(producer_token)32            }33class testCase(unittest.TestCase):34    def setUp(self):35        app.config['TESTING'] = True36        app.config['WTF_CSRF_ENABLED'] = False37        app.config['DEBUG'] = False38        app.config['SQLALCHEMY_DATABASE_URI'] = database_path39        self.app = app.test_client()40        db.drop_all()41        db.create_all()42    def tearDown(self):43        """Executed after reach test"""44        pass45    def test_get_movies_assistant(self):46        res = self.app.get('/movies', headers=set_auth('assistant'))47        self.assertEqual(res.status_code, 200)48    def test_get_movies_director(self):49        res = self.app.get('/movies', headers=set_auth('director'))50        self.assertEqual(res.status_code, 200)51    def test_get_movies_producer(self):52        res = self.app.get('/movies', headers=set_auth('producer'))53        self.assertEqual(res.status_code, 200)54    def test_get_movies_fail(self):55        res = self.app.get('/movies', headers=set_auth(''))56        self.assertEqual(res.status_code, 401)57    def test_post_movie_assistant(self):58        data = {59            "title": "test_title"60        }61        res = self.app.post('/movies', json=data,62                            headers=set_auth('assistant'))63        self.assertEqual(res.status_code, 401)64    def test_post_movie_director(self):65        data = {66            "title": "test_title"67        }68        res = self.app.post('/movies', json=data, headers=set_auth('director'))69        self.assertEqual(res.status_code, 401)70    def test_post_movie_producer(self):71        data = {72            "title": "test_title"73        }74        res = self.app.post('/movies', json=data, headers=set_auth('producer'))75        self.assertEqual(res.status_code, 200)76        self.assertEqual(res.get_json()['success'], True)77    def test_post_movie_fail(self):78        res = self.app.post('/movies', json={}, headers=set_auth('producer'))79        self.assertEqual(res.status_code, 422)80        self.assertEqual(res.get_json()['success'], False)81    def test_patch_movie_assistant(self):82        data = {83            "title": "test_title"84        }85        res = self.app.post('/movies', json=data, headers=set_auth('producer'))86        updated_data = {87            "title": "updated_title"88        }89        res = self.app.patch('/movies/1', json=updated_data,90                             headers=set_auth('assistant'))91        self.assertEqual(res.status_code, 401)92    def test_patch_movie_director(self):93        data = {94            "title": "test_title"95        }96        res = self.app.post('/movies', json=data, headers=set_auth('producer'))97        updated_data = {98            "title": "updated_title"99        }100        res = self.app.patch('/movies/1', json=updated_data,101                             headers=set_auth('director'))102        self.assertEqual(res.status_code, 200)103    def test_patch_movie_producer(self):104        data = {105            "title": "test_title"106        }107        res = self.app.post('/movies', json=data, headers=set_auth('producer'))108        updated_data = {109            "title": "updated_title"110        }111        res = self.app.patch('/movies/1', json=updated_data,112                             headers=set_auth('producer'))113        self.assertEqual(res.status_code, 200)114        self.assertEqual(res.get_json()['success'], True)115    def test_patch_movie_fail(self):116        data = {117            "title": "test_title"118        }119        res = self.app.post('/movies', json=data, headers=set_auth('producer'))120        res = self.app.patch('/movies/100000', json={},121                             headers=set_auth('producer'))122        self.assertEqual(res.status_code, 404)123        self.assertEqual(res.get_json()['success'], False)124    def test_delete_movie_fail(self):125        res = self.app.delete('/movies/1000000', headers=set_auth('producer'))126        self.assertEqual(res.status_code, 404)127    def test_delete_movie_assistant(self):128        data = {129            "title": "test_title"130        }131        res = self.app.post('/movies', json=data, headers=set_auth('producer'))132        res = self.app.delete('/movies/1', headers=set_auth('assistant'))133        self.assertEqual(res.status_code, 401)134    def test_delete_movie_director(self):135        data = {136            "title": "test_title"137        }138        res = self.app.post('/movies', json=data, headers=set_auth('producer'))139        res = self.app.delete('/movies/1', headers=set_auth('director'))140        self.assertEqual(res.status_code, 401)141    def test_delete_movie_producer(self):142        data = {143            "title": "test_title"144        }145        res = self.app.post('/movies', json=data, headers=set_auth('producer'))146        res = self.app.delete('/movies/1', headers=set_auth('producer'))147        self.assertEqual(res.status_code, 200)148    def test_delete_movies_fail(self):149        res = self.app.delete('/movies/1', headers=set_auth(''))150        self.assertEqual(res.status_code, 401)151    # Actor Tests152    def test_get_actors_assistant(self):153        res = self.app.get('/actors', headers=set_auth('assistant'))154        self.assertEqual(res.status_code, 200)155    def test_get_actors_director(self):156        res = self.app.get('/actors', headers=set_auth('director'))157        self.assertEqual(res.status_code, 200)158    def test_get_actors_producer(self):159        res = self.app.get('/actors', headers=set_auth('producer'))160        self.assertEqual(res.status_code, 200)161    def test_get_actor_fail(self):162        res = self.app.get('/actors', headers=set_auth(''))163        self.assertEqual(res.status_code, 401)164    def test_post_actor_assistant(self):165        data = {166            "name": "updated_name",167            "gender": "female",168            "age": 30169        }170        res = self.app.post('/actors', json=data,171                            headers=set_auth('assistant'))172        self.assertEqual(res.status_code, 401)173    def test_post_actor_director(self):174        data = {175            "name": "updated_name",176            "gender": "female",177            "age": 30178        }179        res = self.app.post('/actors', json=data, headers=set_auth('director'))180        self.assertEqual(res.status_code, 200)181    def test_post_actor_producer(self):182        data = {183            "name": "updated_name",184            "gender": "female",185            "age": 30186        }187        res = self.app.post('/actors', json=data, headers=set_auth('producer'))188        self.assertEqual(res.status_code, 200)189        self.assertEqual(res.get_json()['success'], True)190    def test_post_actors_fail(self):191        res = self.app.post('/actors', json={}, headers=set_auth('producer'))192        self.assertEqual(res.status_code, 422)193        self.assertEqual(res.get_json()['success'], False)194    def test_patch_actor_assistant(self):195        data = {196            "name": "test_name",197            "gender": "female",198            "age": 30199        }200        res = self.app.post('/actors', json=data, headers=set_auth('producer'))201        updated_data = {202            "name": "updated_name",203            "gender": "female",204            "age": 35205        }206        res = self.app.patch('/actors/1', json=updated_data,207                             headers=set_auth('assistant'))208        self.assertEqual(res.status_code, 401)209    def test_patch_actor_director(self):210        data = {211            "name": "test_name",212            "gender": "female",213            "age": 30214        }215        res = self.app.post('/actors', json=data, headers=set_auth('director'))216        updated_data = {217            "name": "updated_name",218            "gender": "female",219            "age": 35220        }221        res = self.app.patch('/actors/1', json=updated_data,222                             headers=set_auth('director'))223        self.assertEqual(res.status_code, 200)224        self.assertEqual(res.get_json()['success'], True)225    def test_patch_actor_producer(self):226        data = {227            "name": "test_name",228            "gender": "female",229            "age": 30230        }231        res = self.app.post('/actors', json=data, headers=set_auth('producer'))232        updated_data = {233            "name": "updated_name",234            "gender": "female",235            "age": 35236        }237        res = self.app.patch('/actors/1', json=updated_data,238                             headers=set_auth('producer'))239        self.assertEqual(res.status_code, 200)240        self.assertEqual(res.get_json()['success'], True)241    def test_patch_actor_fail(self):242        data = {243            "name": "test_name",244            "gender": "female",245            "age": 30246        }247        res = self.app.post('/actors', json=data, headers=set_auth('producer'))248        res = self.app.patch('/actors/100000', json={},249                             headers=set_auth('producer'))250        self.assertEqual(res.status_code, 404)251        self.assertEqual(res.get_json()['success'], False)252    def test_delete_actor_assistant(self):253        data = {254            "name": "test_name",255            "gender": "female",256            "age": 30257        }258        res = self.app.post('/actors', json=data,259                            headers=set_auth('assistant'))260        res = self.app.delete('/actors/1', headers=set_auth('assistant'))261        self.assertEqual(res.status_code, 401)262    def test_delete_actor_director(self):263        data = {264            "name": "test_name",265            "gender": "female",266            "age": 30267        }268        res = self.app.post('/actors', json=data, headers=set_auth('director'))269        res = self.app.delete('/actors/1', headers=set_auth('director'))270        self.assertEqual(res.status_code, 200)271    def test_delete_actor_producer(self):272        data = {273            "name": "test_name",274            "gender": "female",275            "age": 30276        }277        res = self.app.post('/actors', json=data, headers=set_auth('producer'))278        res = self.app.delete('/actors/1', headers=set_auth('producer'))279        self.assertEqual(res.status_code, 200)280    def test_delete_actor_fail(self):281        logging.info("Intentionally Missing Auth Header (see below)")282        res = self.app.delete('/actors/1', headers=set_auth(''))283        self.assertEqual(res.status_code, 401)284if __name__ == '__main__':...

botlists.py

Source:botlists.py

...100class DiscordLists(commands.Cog):101    def __init__(self, bot):102        self.bot = bot103        self.api = discordlists.Client(self.bot)104        self.api.set_auth("discord.bots.gg", bot.config.DBGG_TOKEN)105        self.api.set_auth("discord.boats", bot.config.DBOATS_TOKEN)106        self.api.set_auth("voidbots.net", bot.config.GLENN_TOKEN)107        self.api.set_auth("mythicalbots.xyz", bot.config.MYTH_TOKEN)108        self.api.set_auth("botsfordiscord.com", bot.config.BFD_TOKEN)109        self.api.set_auth("botlist.space", bot.config.BOTSPACE_TOKEN)110        self.api.set_auth("discordbots.co", bot.config.DISCORD_BOTS_TOKEN)111        self.api.set_auth('bladebotlist.xyz', bot.config.BBL_TOKEN)112        self.api.set_auth('blist.xyz', bot.config.BLIST_TOKEN)113        self.api.set_auth('space-bot-list.xyz', bot.config.SBL_TOKEN)114        self.api.set_auth('infinitybotlist.com', bot.config.INFINITY_TOKEN)115        self.api.start_loop()116        self.help_icon = ''117        self.big_icon = ''118    @commands.command(hidden=True)119    async def post(self, ctx):120        """121        Manually posts guild count using discordlists.py (BotBlock)122        """123        try:124            result = await self.api.post_count()125        except Exception as e:126            try:127                await ctx.send("Request failed: `{}`".format(e))128                return...

test_unread_comments_counters.py

Source:test_unread_comments_counters.py

1import pytest2from . import factories3from art17 import models4from .conftest import create_user5@pytest.fixture6def setup_common(app):7    factories.EtcDataSpeciesRegionFactory(8        assesment_speciesname="Canis lupus",9        speciesname="Canis lupus",10        group="Mammals",11    )12    factories.SpeciesManualAssessmentFactory(region="ALP")13    factories.SpeciesManualAssessmentFactory(region="ALP", user_id="conclusion_user")14    factories.EtcDicBiogeoregFactory()15    factories.EtcDicHdHabitat()16    factories.HabitattypesManualAssessmentsFactory(region="ALP")17    factories.HabitattypesManualAssessmentsFactory(18        region="ALP", user_id="conclusion_user"19    )20    models.db.session.commit()21@pytest.fixture22def setup(app):23    factories.CommentFactory(region="ALP")24    factories.CommentFactory(id=2, author_id="user2", region="ALP")25    factories.CommentFactory(26        id=3, author_id="user3", region="ALP", user_id="conclusion_user"27    )28    factories.WikiFactory(region_code="ALP")29    factories.WikiChangeFactory()30    factories.WikiCommentFactory()31    factories.HabitatCommentFactory(region="ALP")32    factories.HabitatCommentFactory(id=2, author_id="user2", region="ALP")33    factories.HabitatCommentFactory(34        id=3, author_id="user3", region="ALP", user_id="conclusion_user"35    )36    factories.WikiFactory(37        id=2,38        region_code="ALP",39        habitatcode=1110,40        assesment_speciesname=None,41    )42    factories.WikiChangeFactory(43        id=2,44        wiki_id=2,45        body="The conservation status in the marine Baltic region",46    )47    factories.WikiCommentFactory(id=2, wiki_id=2)48    models.db.session.commit()49@pytest.fixture50def setup_deleted(app):51    factories.CommentFactory(id=4, author_id="user4", region="ALP", deleted=1)52    factories.CommentFactory(53        id=5, author_id="user4", region="ALP", deleted=1, user_id="conclusion_user"54    )55    factories.WikiCommentFactory(id=3, author_id="user3", deleted=1)56    factories.HabitatCommentFactory(id=4, author_id="user4", region="ALP", deleted=1)57    factories.HabitatCommentFactory(58        id=5, author_id="user4", region="ALP", deleted=1, user_id="conclusion_user"59    )60    factories.WikiCommentFactory(id=4, wiki_id=2, author_id="user4", deleted=1)61@pytest.mark.xfail62@pytest.mark.parametrize(63    "user, path, args",64    [65        (66            "someuser",67            "/species/progress/",68            {"period": "1", "group": "Mammals", "conclusion": "population"},69        ),70        (71            "someuser",72            "/habitat/progress/",73            {"period": "1", "group": "coastal habitats", "conclusion": "area"},74        ),75    ],76)77def test_unread_conclusion_comments(78    app, client, set_auth, setup_common, setup, user, path, args79):80    create_user(user, app)81    set_auth.update({"user_id": user})82    resp = client.get(path, args)83    assert resp.status_code == 20084    assert "Unread comments for my conclusions: 2" in resp.text85    assert "Unread comments for all conclusions: 3" in resp.text86    assert "Unread comments for data sheet info: 1" in resp.text87@pytest.mark.xfail88@pytest.mark.parametrize(89    "path, args",90    [91        (92            "/species/progress/",93            {"period": "1", "group": "Mammals", "conclusion": "population"},94        ),95        (96            "/habitat/progress/",97            {"period": "1", "group": "coastal habitats", "conclusion": "area"},98        ),99    ],100)101def test_unread_conclusion_comments_anonymous_user(102    app, client, set_auth, setup, path, args103):104    resp = client.get(path, args)105    assert resp.status_code == 200106    assert "Unread comments for my conclusions:" not in resp.text107    assert "Unread comments for all conclusions:" not in resp.text108    assert "Unread comments for data sheet info:" not in resp.text109@pytest.mark.xfail110@pytest.mark.parametrize(111    "user, path, args",112    [113        (114            "someuser",115            "/species/progress/",116            {"period": "1", "group": "Mammals", "conclusion": "population"},117        ),118        (119            "someuser",120            "/habitat/progress/",121            {"period": "1", "group": "coastal habitats", "conclusion": "area"},122        ),123    ],124)125def test_unread_conclusion_comments_zero_comments(126    app, client, set_auth, setup_common, user, path, args127):128    create_user(user, app)129    set_auth.update({"user_id": user})130    resp = client.get(path, args)131    assert resp.status_code == 200132    assert "Unread comments for my conclusions: 0" in resp.text133    assert "Unread comments for all conclusions: 0" in resp.text134    assert "Unread comments for data sheet info: 0" in resp.text135@pytest.mark.xfail136@pytest.mark.parametrize(137    "user, path, args",138    [139        (140            "someuser",141            "/species/progress/",142            {"period": "1", "group": "Mammals", "conclusion": "population"},143        ),144        (145            "someuser",146            "/habitat/progress/",147            {"period": "1", "group": "coastal habitats", "conclusion": "area"},148        ),149    ],150)151def test_unread_conclusion_comments_deleted_comments(152    app, client, set_auth, setup_common, setup, setup_deleted, user, path, args153):154    create_user(user, app)155    set_auth.update({"user_id": user})156    resp = client.get(path, args)157    assert resp.status_code == 200158    assert "Unread comments for my conclusions: 2" in resp.text159    assert "Unread comments for all conclusions: 3" in resp.text160    assert "Unread comments for data sheet info: 1" in resp.text161@pytest.mark.xfail162@pytest.mark.parametrize(163    "username, path, args, comment_cls, wiki_id, read, unread_comments",164    [165        (166            "someuser",167            "/species/progress/",168            {"period": "1", "group": "Mammals", "conclusion": "population"},169            factories.CommentFactory,170            1,171            True,172            (2, 3, 1),173        ),174        (175            "someuser",176            "/habitat/progress/",177            {"period": "1", "group": "coastal habitats", "conclusion": "area"},178            factories.HabitatCommentFactory,179            2,180            True,181            (2, 3, 1),182        ),183        (184            "someuser",185            "/species/progress/",186            {"period": "1", "group": "Mammals", "conclusion": "population"},187            factories.CommentFactory,188            1,189            False,190            (3, 5, 2),191        ),192        (193            "someuser",194            "/habitat/progress/",195            {"period": "1", "group": "coastal habitats", "conclusion": "area"},196            factories.HabitatCommentFactory,197            2,198            False,199            (3, 5, 2),200        ),201    ],202)203def test_unread_conclusion_comments_read_comments(204    app,205    client,206    set_auth,207    setup_common,208    setup,209    username,210    path,211    args,212    comment_cls,213    wiki_id,214    read,215    unread_comments,216):217    user = create_user(username, app)218    set_auth.update({"user_id": username})219    comment = comment_cls(id=10, author_id="user10", region="ALP")220    comment_other_conclusion = comment_cls(221        id=11, author_id="user11", region="ALP", user_id="conclusion_user"222    )223    wiki_comment = factories.WikiCommentFactory(224        id=10, author_id="user10", wiki_id=wiki_id225    )226    models.db.session.commit()227    if read:228        comment.readers.append(user)229        comment_other_conclusion.readers.append(user)230        wiki_comment.readers.append(user)231        models.db.session.commit()232    resp = client.get(path, args)233    assert resp.status_code == 200234    mine, all, dsi = unread_comments235    assert "Unread comments for my conclusions: {0}".format(mine) in resp.text236    assert "Unread comments for all conclusions: {0}".format(all) in resp.text...

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.