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

How to use _get_stream method in Airtest

Best Python code snippet using Airtest

tests.py

Source:tests.py

...48        self._sticker(self.user, sticker_type_id=self._epic_sticker.type_id, sender=sender)49        self.user.kv.update()50        for notification in self.user.redis.notifications.get():51            self.api_post('/api/notification/acknowledge', {'nkey': notification['nkey']}, user=self.user)52    def _get_stream(self):53        return '<html><body>' + self.api_post('/api/activity/activity_stream', user=self.user) + '</body></html>'54    def test_sticker(self):55        resp = self._get_stream()56        self.assertNumCssMatches(0, resp, '.sticker_activity')57        self._sticker(self.user)58        resp = self._get_stream()59        self.assertNumCssMatches(1, resp, '.sticker_activity')60    def test_many_stickers(self):61        resp = self._get_stream()62        self.assertNumCssMatches(0, resp, '.sticker_activity')63        COUNT = 564        for _ in xrange(5):65            self._sticker(self.user)66        resp = self._get_stream()67        self.assertNumCssMatches(COUNT, resp, '.sticker_activity')68    def test_sticker_thumbnail(self):69        resp = self._get_stream()70        self.assertNumCssMatches(0, resp, '.thumbnail')71        self._sticker(self.user)72        resp = self._get_stream()73        self.assertNumCssMatches(1, resp, '.thumbnail')74    def test_epic_sticker(self):75        resp = self._get_stream()76        self.assertNumCssMatches(0, resp, '.epic_sticker_activity')77        self._give_epic_sticker()78        resp = self._get_stream()79        self.assertNumCssMatches(1, resp, '.epic_sticker_activity')80    def test_markup_is_unescaped(self):81        self._sticker(self.user)82        resp = self._get_stream()83        self.assertNumCssMatches(1, resp, '.sticker_container')84    def test_level_up(self):85        resp = self._get_stream()86        self.assertNumCssMatches(0, resp, '.level_up_activity')87        self.user.redis.user_kv.hincrby('sticker_inbox', STICKER_SCHEDULE[0])88        self.api_post('/api/user/level_up', {}, user=self.user)89        resp = self._get_stream()90        self.assertNumCssMatches(1, resp, '.level_up_activity')91        (message,) = self.css_select(resp, '.level_up_activity')92        self.assertTrue(str(STICKER_REWARDS[0]) in message.xpath('string()'))93    def test_remix(self):94        resp = self._get_stream()95        self.assertNumCssMatches(0, resp, '.remix_activity')96        original_content = create_content()97        remix_content = create_content()98        remix_content.remix_of = original_content99        remix_content.save()100        op = self.post_comment(reply_content=create_content().id)101        original = self.post_comment(parent_comment=op.id, reply_content=original_content.id, user=self.user)102        remixer = create_user()103        remix = self.post_comment(parent_comment=op.id, reply_content=remix_content.id)104        resp = self._get_stream()105        self.assertNumCssMatches(1, resp, '.remix_activity')106    def test_reply_to_your_thread(self):107        def check(count):108            resp = self._get_stream()109            self.assertNumCssMatches(count, resp, '.thread_reply_activity')110        check(0)111        op = self.post_comment(reply_content=create_content().id, user=self.user)112        check(0)113        for n in xrange(1, 2):114            reply = self.post_comment(parent_comment=op.id, reply_content=create_content().id)115            check(n)116    def test_at_reply(self):117        def check(count):118            resp = self._get_stream()119            self.assertNumCssMatches(count, resp, '.reply_activity')120        check(0)121        op = self.post_comment(reply_content=create_content().id, user=self.user)122        check(0)123        reply = self.post_comment(parent_comment=op.id, reply_content=create_content().id)124        check(0)125        at_reply = self.post_comment(parent_comment=op.id, reply_content=create_content().id, replied_comment=op.id)...

tee.py

Source:tee.py

...75        in the '_downstream' attribute of the Tee object.76        """77        pass78    @staticmethod79    def _get_stream():80        """Get the stream 'pointer' this object is Teeing.81        For example, if this object was Teeing sys.stdout, it would82        return the current value of sys.stdout.83        """84        pass85    def __init__(self, pass_through=True):86        self.pass_through = pass_through87        self._downstream = self._set_stream(self)88    def __enter__(self):89        return self90    def __exit__(self, *args, **kwargs):91        self.reset()92    def reset(self):93        """Remove the current Tee object from the list of Tee objects.94        This removes the current object from the linked list of Tee95        objects associated with the stream returned by96        self._get_stream().97        Consider the case of several chained Tee objects (e.g. t1, t2,98        t3) where the original '<stdout>' open file is downstream from99        t1, t1 is downstream from t2, and t2 is downstream from t3:100        t3 => t2 => t1 => <open file '<stdout>', ...>101        ^102        sys.stdout103        sys.stdout points to t3. If we call t1.reset() it would be104        incorrect to set sys.stdout to the <open file ...> (t1's105        downstream). Instead sys.stdout must continue to point to t3106        and t2._downstream should be updated to point to the <open107        file ...>. This ensures that Tee objects may leave the chain108        as-needed without losing connections.109        """110        prev, cur = None, self._get_stream()111        if cur == self:112            self._set_stream(self._downstream)113            return114        while hasattr(cur, '_downstream'):115            prev, cur = cur, cur._downstream116            if cur == self:117                prev._downstream = cur._downstream118    def __getattr__(self, attr):119        return getattr(self._downstream, attr)120    def write(self, *args, **kwargs):121        if self.pass_through:122            self._downstream.write(*args, **kwargs)123    def flush(self, *args, **kwargs):124        if self.pass_through:...

part_definition.py

Source:part_definition.py

...54                # this is part is equal to whole upload source - so we use `get_content_sha1()`55                # and if sha1 is already given, we skip computing it again56                self._sha1 = self.upload_source.get_content_sha1()57            else:58                with self._get_stream() as stream:59                    self._sha1, _ = hex_sha1_of_unlimited_stream(stream)60        return self._sha161    def get_execution_step(self, execution_step_factory):62        return execution_step_factory.create_upload_execution_step(63            self._get_stream,64            stream_length=self.length,65            stream_sha1=self.get_sha1(),66        )67    def _get_stream(self):68        fp = self.upload_source.open()69        return wrap_with_range(70            fp, self.upload_source.get_content_length(), self.relative_offset, self.length71        )72class UploadSubpartsEmergePartDefinition(BaseEmergePartDefinition):73    def __init__(self, upload_subparts):74        self.upload_subparts = upload_subparts75        self._is_hashable = all(subpart.is_hashable() for subpart in upload_subparts)76        self._sha1 = None77    def __repr__(self):78        return '<{classname} upload_subparts={upload_subparts}>'.format(79            classname=self.__class__.__name__,80            upload_subparts=repr(self.upload_subparts),81        )82    def get_length(self):83        return sum(subpart.length for subpart in self.upload_subparts)84    def get_part_id(self):85        if self.is_hashable():86            return self.get_sha1()87        else:88            return tuple(subpart.get_subpart_id() for subpart in self.upload_subparts)89    def is_hashable(self):90        return self._is_hashable91    def get_sha1(self):92        if self._sha1 is None and self.is_hashable():93            with self._get_stream() as stream:94                self._sha1, _ = hex_sha1_of_unlimited_stream(stream)95        return self._sha196    def get_execution_step(self, execution_step_factory):97        return execution_step_factory.create_upload_execution_step(98            partial(self._get_stream, emerge_execution=execution_step_factory.emerge_execution),99            stream_length=self.get_length(),100            stream_sha1=self.get_sha1(),101        )102    def _get_stream(self, emerge_execution=None):103        return ChainedStream(104            [105                subpart.get_stream_opener(emerge_execution=emerge_execution)106                for subpart in self.upload_subparts107            ]108        )109class CopyEmergePartDefinition(BaseEmergePartDefinition):110    def __init__(self, copy_source, relative_offset, length):111        self.copy_source = copy_source112        self.relative_offset = relative_offset113        self.length = length114    def __repr__(self):115        return (116            '<{classname} copy_source={copy_source} relative_offset={relative_offset} '...

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.