How to use put_permission method in localstack

Best Python code snippet using localstack_python

cloudwatch_events.py

Source:cloudwatch_events.py

...15            None16        """17        self.logger = logging.getLogger(__name__)18        self.cwe_client = boto3.client("events", config=boto3_config)19    def put_permission(self, principal, event_bus_name):20        """Puts permission on CloudWatch event bus. Running it permits the specified\21            AWS account or AWS organization to put events to your account's default event bus22        Args:23            principal (string): 12-digit Amazon Web Services account ID or the Organization\24                 Arn that you are permitting to put events to specified event bus25            event_bus_name (string):  name of the event bus associated with the rule26        Returns:27            None: put_permission api response28        Raises:29            EventBridge.Client.exceptions.ResourceNotFoundException: \30                thrown when AWS resource not found31            ClientError: general exception provided by an AWS service to your Boto3 client's request32        """33        log_message = {34            "METHOD": "put_permission",35            "MESSAGE": f"putting permission on CloudWatch Event bus for principal:\36                 {principal}, event_bus_name: {event_bus_name}",37        }38        self.logger.debug(str(log_message))39        try:40            if re.match("(arn:aws:organizations:).*", principal):41                org_id = principal.split("/")[-1]42                condition = {43                    "Type": "StringEquals",44                    "Key": "aws:PrincipalOrgID",45                    "Value": org_id,46                }47                # Once we specify a condition with an AWS organization ID, the recommendation48                # is we use "*" as the value or Principal to grant permission to all the accounts49                #  in the named organization.50                self.cwe_client.put_permission(51                    Action="events:PutEvents",52                    Principal="*",53                    StatementId=org_id,54                    Condition=condition,55                    EventBusName=event_bus_name,56                )57            else:58                self.cwe_client.put_permission(59                    Action="events:PutEvents",60                    Principal=principal,61                    StatementId=principal,62                    EventBusName=event_bus_name,63                )64            return None  # the API response always returns None65        except (66            self.cwe_client.exceptions.ResourceNotFoundException,67            ClientError,68        ) as err:69            log_message["EXCEPTION"] = str(err)70            self.logger.error(str(log_message))71            raise72    def remove_permission(self, principal, event_bus_name):...

test_file_acl.py

Source:test_file_acl.py

...71    assert acl.get_perm_of('user', userA) == '---'72def test_apply_to_file(module_file, userA):73    os.system('setfacl -m u:' + userA + ':--- ' + module_file)74    acl = FileACL(file=module_file)75    acl.put_permission('user', userA, '+rwx')76    assert acl.get_perm_of('user', userA) == 'rwx'77    # Creating a new FileACL object makes sure that permissions78    # are actually written to Inode and are not just stored inside79    # the FileACL object80    acl = FileACL(file=module_file)81    assert acl.get_perm_of('user', userA) == 'rwx'82@pytest.mark.parametrize('instruction, expected_permset', test_parameters)83def test_write_user(module_file, userA, instruction, expected_permset):84    acl = FileACL(file=module_file)85    acl.put_permission('user', userA, instruction)86    assert acl.get_perm_of('user', userA) == expected_permset87####88## group-specific tests89####90def test_read_group(module_file, groupA):91    os.system('setfacl -m g:' + groupA + ':rwx ' + module_file)92    acl = FileACL(file=module_file)93    assert acl.get_perm_of('group', groupA) == 'rwx'94    os.system('setfacl -m g:' + groupA + ':--- ' + module_file)95    acl = FileACL(file=module_file)96    assert acl.get_perm_of('group', groupA) == '---'97@pytest.mark.parametrize('instruction, expected_permset', test_parameters)98def test_write_group(module_file, groupA, instruction, expected_permset):99    acl = FileACL(file=module_file)100    acl.put_permission('group', groupA , instruction)101    assert acl.get_perm_of('group', groupA) == expected_permset102####103## test user-group dynamic104####105def test_group_membership(fresh_file, userA, groupA):106    '''Tests that a user has access to a file (only) through group membership.107    108    This test assumes the user has no individual acl entry.'''109    acl = FileACL(file=fresh_file)110    acl.put_permission('group', groupA , '=r')111    assert acl.get_perm_of('group', groupA) == 'r--'112    acl.put_permission('other', None , '=')113    ...

entries.py

Source:entries.py

1"""2    marvin.views.entries3    ~~~~~~~~~~~~~~~~~~~~4    CRUD endpoints for stream entries.5"""6# pylint: disable=no-self-use7from .. import db8from ..models import Entry, EntryForm, Stream9from ..permissions import login_required10from flask.ext.restful import Resource11from flask.ext.principal import Permission, UserNeed12class EntryDetailView(Resource):13    """ RUD interface to entries. """14    def get(self, entry_id):15        """ Get the entry with the given ID. """16        entry = Entry.query.get(entry_id)17        return {18            'entry': entry.to_json(),19        }20    @login_required21    def put(self, entry_id):22        """ Update the entry with the given ID. """23        entry = Entry.query.get_or_404(entry_id)24        put_permission = Permission(UserNeed(entry.stream.creator_id))25        if put_permission.can():26            form = EntryForm(obj=entry)27            if form.validate_on_submit():28                form.populate_obj(entry)29                return {30                    'msg': 'Entry updated.',31                    'entry': entry.to_json(),32                }33            return {34                'msg': 'Some attributes did not pass validation.',35                'errors': form.errors,36            }, 40037        else:38            return {39                'msg': "Only the stream creator can edit it's entries.",40            }, 40341    @login_required42    def delete(self, entry_id):43        """ Delete the entry with the given ID. """44        entry = Entry.query.get(entry_id)45        delete_permission = Permission(UserNeed(entry.stream.creator_id))46        if delete_permission.can():47            db.session.delete(entry)48            return {'msg': 'Entry deleted.'}49        else:50            return {51                'msg': 'Only the stream creator can delete entries.',52            }, 40353class CreateEntryView(Resource):54    """ Create interface to entries. """55    @login_required56    def post(self, stream_id):57        """ Create new entry. """58        stream = Stream.query.get_or_404(stream_id)59        add_entry_to_stream_permission = Permission(UserNeed(stream.creator_id))60        if add_entry_to_stream_permission.can():61            form = EntryForm()62            if form.validate_on_submit():63                entry = Entry()64                form.populate_obj(entry)65                entry.stream = stream66                db.session.add(entry)67                db.session.commit()68                return {69                    'msg': 'Entry created.',70                    'entry': entry.to_json(),71                }, 20172            return {73                'msg': 'Some attributes did not pass validation.',74                'errors': form.errors,75            }, 40076        else:77            return {78                'msg': 'Only the creator can add entries to streams',...

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.