Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use read_event method in keyboard

Best Python code snippet using keyboard

queue_buffer.py

Source:queue_buffer.py

1import numpy as np2from threading import Event3from utils.circular_buffer import CircularBuffer4class QueueBuffer(object):5    """6    Queue using Circular Buffer Implementation7    Allows AT MOST 1 reader and 1 writer threads -> NO MULTI READERS OR WRITERS8    """9    def __init__(self, shape: tuple = (0, 0), buffer=None):10        self.buffer = CircularBuffer(shape, buffer=buffer)11        assert self.capacity % 2 == 0, "buffer size must be divisible by 2"12        self.read_idx = 013        self.write_idx = 014        self.read_event = Event()15        self.write_event = Event()16    @property17    def capacity(self):18        return self.buffer.buf_size19    def empty(self):20        return self.read_idx == self.write_idx21    def size(self):22        return self.write_idx - self.read_idx23    def full(self):24        return self.size() == self.capacity25    def put(self, data: np.ndarray, length=None, put_incrementally=False) -> int:26        if length is None:27            length = np.shape(data)[0]28        if not put_incrementally:29            self.read_event.clear()30            while self.write_idx + length - self.read_idx > self.capacity:  # we must wait31                self.read_event.wait()32                self.read_event.clear()33            self.buffer.put(self.write_idx % self.capacity, data, length=length)34            self.write_idx += length35            self.write_event.set()36        else:37            self.read_event.clear()38            if self.write_idx + length - self.read_idx <= self.capacity:  # we can put everything in39                self.buffer.put(self.write_idx % self.capacity, data, length=length)40                self.write_idx += length41                self.write_event.set()42            else:43                remaining = length44                while remaining > 0:45                    # get available space46                    avail = self.capacity - self.size()47                    # clear event48                    self.read_event.clear()49                    # wait for space and loop again if necessary50                    if avail == 0:51                        self.read_event.wait()52                        continue53                    # don't add to much54                    if avail > remaining:55                        avail = remaining56                    # fill the available space57                    self.buffer.put(self.write_idx % self.capacity,58                                    data[-remaining:], length=avail)  # pylint: disable=invalid-unary-operand-type59                    # update write_idx and write_event60                    self.write_idx += avail61                    self.write_event.set()62                    # update remaining63                    remaining -= avail64                    # wait if necessary:65                    if remaining > 0:66                        self.read_event.wait()67        return True68    def put_nowait(self, data: np.ndarray, length=None):69        if length is None:70            length = np.shape(data)[0]71        if self.write_idx + length - self.read_idx > self.capacity:72            return False73        else:74            self.buffer.put(self.write_idx % self.capacity, data, length=length)75            self.write_idx += length76            self.write_event.set()77            return True78    # def put_force(self, data: np.ndarray, length=None):79    #     if length is None:80    #         length = np.shape(data)[0]81    #     # put no matter what82    #     self.buffer.put(self.write_idx % self.capacity, data, length=length)83    #     self.write_idx += length84    #     if self.write_idx - self.read_idx > self.capacity:  # bump up read idx85    #         self.read_idx += length86    #     self.write_event.set()87    #     return True88    def get_into(self, output: np.ndarray, length=None) -> int:89        if length is None:90            length = np.shape(output)[0]91        self.read_event.clear()92        while self.read_idx + length > self.write_idx:93            self.read_event.wait()94            self.read_event.clear()95        self.buffer.get_into(self.read_idx % self.capacity, output, length=length)96        self.read_idx += length97        self.read_event.set()98        return True99    def get_into_nowait(self, output: np.ndarray, length=None) -> int:100        if length is None:101            length = np.shape(output)[0]102        if self.read_idx + length > self.write_idx:103            return False104        else:105            self.buffer.get_into(self.read_idx % self.capacity, output, length=length)106            self.read_idx += length107            self.read_event.set()108            return True109    def get(self, length):110        self.read_event.clear()111        while self.read_idx + length > self.write_idx:112            self.read_event.wait()113            self.read_event.clear()114        output = self.buffer.get(self.read_idx % self.capacity, length)115        self.read_idx += length116        self.read_event.set()117        return output118    def get_nowait(self, length):119        if self.read_idx + length > self.write_idx:120            return None121        else:122            output = self.buffer.get(self.read_idx % self.capacity, length)123            self.read_idx += length124            self.read_event.set()125            return output126    # def get_into_force(self, output: np.ndarray, length=None) -> int:127    #     if length is None:128    #         length = np.shape(output)[0]129    #     # get no matter what130    #     self.buffer.get_into(self.read_idx % self.capacity, output, length=length)131    #     self.read_idx += length132    #     if self.read_idx > self.write_idx:  # let's force it -> bump up write idx133    #         self.write_idx = self.read_idx134    #     self.read_event.set()135    #     return True136    # def get_force(self, length):137    #     # get no matter what138    #     output = self.buffer.get(self.read_idx % self.capacity, length)139    #     self.read_idx += length140    #     if self.read_idx > self.write_idx:  # let's force it -> bump up write idx141    #         self.write_idx = self.read_idx142    #     self.read_event.set()...

test_access_rights.py

Source:test_access_rights.py

...16            'start': datetime(2020, 2, 2, 8, 0),17            'stop': datetime(2020, 2, 2, 18, 0),18            'user_id': user.id,19        }, **values))20    def read_event(self, user, events, field):21        data = events.with_user(user).read([field])22        if len(events) == 1:23            return data[0][field]24        mapped_data = {record['id']: record for record in data}25        # Keep the same order26        return [mapped_data[eid][field] for eid in events.ids]27    def test_private_read_name(self):28        event = self.create_event(29            self.john,30            privacy='private',31            name='my private event',32        )33        self.assertEqual(self.read_event(self.john, event, 'name'), 'my private event', "Owner should be able to read the event")34        self.assertEqual(self.read_event(self.raoul, event, 'name'), 'Busy', "Private value should be obfuscated")35        with self.assertRaises(AccessError):36            self.read_event(self.portal, event, 'name')37    def test_private_other_field(self):38        event = self.create_event(39            self.john,40            privacy='private',41            location='in the Sky',42        )43        self.assertEqual(self.read_event(self.john, event, 'location'), 'in the Sky', "Owner should be able to read the event")44        self.assertEqual(self.read_event(self.raoul, event, 'location'), False, "Private value should be obfuscated")45        with self.assertRaises(AccessError):46            self.read_event(self.portal, event, 'location')47    def test_private_and_public(self):48        private = self.create_event(49            self.john,50            privacy='private',51            location='in the Sky',52        )53        public = self.create_event(54            self.john,55            privacy='public',56            location='In Hell',57        )58        [private_location, public_location] = self.read_event(self.raoul, private + public, 'location')59        self.assertEqual(private_location, False, "Private value should be obfuscated")60        self.assertEqual(public_location, 'In Hell', "Public value should not be obfuscated")61    def test_read_group_public(self):62        event = self.create_event(self.john)63        data = self.env['calendar.event'].with_user(self.raoul).read_group([('id', '=', event.id)], fields=['start'], groupby='start')64        self.assertTrue(data, "It should be able to read group")65        data = self.env['calendar.event'].with_user(self.raoul).read_group([('id', '=', event.id)], fields=['name'],66                                                                           groupby='name')67        self.assertTrue(data, "It should be able to read group")68    def test_read_group_private(self):69        event = self.create_event(self.john, privacy='private')70        result = self.env['calendar.event'].with_user(self.raoul).read_group([('id', '=', event.id)], fields=['name'], groupby='name')71        self.assertFalse(result, "Private events should not be fetched")72    def test_read_group_agg(self):73        event = self.create_event(self.john)74        data = self.env['calendar.event'].with_user(self.raoul).read_group([('id', '=', event.id)], fields=['start'], groupby='start:week')75        self.assertTrue(data, "It should be able to read group")76    def test_read_group_list(self):77        event = self.create_event(self.john)78        data = self.env['calendar.event'].with_user(self.raoul).read_group([('id', '=', event.id)], fields=['start'], groupby=['start'])79        self.assertTrue(data, "It should be able to read group")80    def test_private_attendee(self):81        event = self.create_event(82            self.john,83            privacy='private',84            location='in the Sky',85        )86        partners = (self.john|self.raoul).mapped('partner_id')87        event.write({'partner_ids': [(6, 0, partners.ids)]})88        self.assertEqual(self.read_event(self.raoul, event, 'location'), 'in the Sky',89                         "Owner should be able to read the event")90        with self.assertRaises(AccessError):...

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.