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

How to use get_feed method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_feeds.py

Source:test_feeds.py

...15        StubUserDataProvider.clear()16        self.controller = UserController.from_settings()17        self.controller.create_user(username='a')18        self.controller.create_user(username='b')19    def get_feed(self, actor, page=None, page_size=None):20        data = {21            'actor': actor,22        }23        if page is not None:24            data['page'] = str(page)25        if page_size is not None:26            data['page_size'] = str(page_size)27        return self.client.get(self.url, data=data)28    def test_empty(self):29        response = self.get_feed(actor='a')30        self.assertResponseSuccess(response)31        self.assertResponseDataEqual(response, [])32    def test_nonexistent_actor(self):33        response = self.get_feed(actor='z')34        self.assertResponseSuccess(response)35        self.assertResponseDataEqual(response, [])36    def test_one_action(self):37        self.create_action(actor='a', target='b')38        response = self.get_feed(actor='a')39        self.assertResponseSuccess(response)40        data = response.data['data']41        self.assertEqual(len(data), 1)42        action_data = data[0]43        self.assertEqual(action_data['actor'], 'a')44        self.assertEqual(action_data['target'], 'b')45    def test_other_user_action(self):46        self.create_action(actor='a', target='b')47        response = self.get_feed(actor='b')48        self.assertResponseSuccess(response)49        self.assertEqual(len(response.data['data']), 0)50    def test_first_page(self):51        self.create_action(actor='a', target='b')52        response = self.get_feed(actor='a', page=0)53        self.assertResponseSuccess(response)54        self.assertEqual(len(response.data['data']), 1)55    def test_large_page(self):56        self.create_action(actor='a', target='b')57        response = self.get_feed(actor='a', page=777)58        self.assertResponseSuccess(response)59        self.assertEqual(len(response.data['data']), 0)60    def test_multiple_pages(self):61        page_size = 1062        for _ in range(2 * page_size + 1):63            self.create_action(actor='a', target='b')64        response_page0 = self.get_feed(actor='a', page=0, page_size=page_size)65        response_page1 = self.get_feed(actor='a', page=1, page_size=page_size)66        response_page2 = self.get_feed(actor='a', page=2, page_size=page_size)67        response_page3 = self.get_feed(actor='a', page=3, page_size=page_size)68        self.assertResponseSuccess(response_page0)69        self.assertResponseSuccess(response_page1)70        self.assertResponseSuccess(response_page2)71        self.assertResponseSuccess(response_page3)72        self.assertEqual(len(response_page0.data['data']), page_size)73        self.assertEqual(len(response_page1.data['data']), page_size)74        self.assertEqual(len(response_page2.data['data']), 1)75class FriendsFeedTestCase(CreateActionTestCaseMixin, DemoAPITestCase):76    @property77    def url(self):78        return reverse('api:friends-feed')79    def setUp(self):80        StubActionDataProvider.clear()81        StubUserDataProvider.clear()82        self.controller = UserController.from_settings()83        self.controller.create_user(username='a')84        self.controller.create_user(username='b')85        # b follows a.86        self.controller.follow(followee='a', follower='b')87    def get_feed(self, actor, page=None, page_size=None):88        data = {89            'actor': actor,90        }91        if page is not None:92            data['page'] = str(page)93        if page_size is not None:94            data['page_size'] = str(page_size)95        return self.client.get(self.url, data=data)96    def test_empty(self):97        response = self.get_feed(actor='a')98        self.assertResponseSuccess(response)99        self.assertResponseDataEqual(response, [])100    def test_nonexistent_actor(self):101        response = self.get_feed(actor='z')102        self.assertResponseSuccess(response)103        self.assertResponseDataEqual(response, [])104    def test_one_action(self):105        self.create_action(actor='b', target='a')106        response = self.get_feed(actor='a')107        self.assertResponseSuccess(response)108        self.assertEqual(len(response.data['data']), 0)109    def test_one_action_reverse(self):110        self.create_action(actor='a', target='b')111        response = self.get_feed(actor='b')112        self.assertResponseSuccess(response)113        data = response.data['data']114        self.assertEqual(len(data), 1)115        action_data = data[0]116        self.assertEqual(action_data['actor'], 'a')117        self.assertEqual(action_data['target'], 'b')118    def test_self_action(self):119        self.create_action(actor='a', target='b')120        response = self.get_feed(actor='a')121        self.assertResponseSuccess(response)122        self.assertEqual(len(response.data['data']), 0)123    def test_first_page(self):124        self.create_action(actor='a', target='b')125        response = self.get_feed(actor='b', page=0)126        self.assertResponseSuccess(response)127        self.assertEqual(len(response.data['data']), 1)128    def test_large_page(self):129        self.create_action(actor='a', target='b')130        response = self.get_feed(actor='b', page=777)131        self.assertResponseSuccess(response)132        self.assertEqual(len(response.data['data']), 0)133    def test_multiple_pages(self):134        page_size = 10135        for _ in range(2 * page_size + 1):136            self.create_action(actor='a', target='b')137        response_page0 = self.get_feed(actor='b', page=0, page_size=page_size)138        response_page1 = self.get_feed(actor='b', page=1, page_size=page_size)139        response_page2 = self.get_feed(actor='b', page=2, page_size=page_size)140        response_page3 = self.get_feed(actor='b', page=3, page_size=page_size)141        self.assertResponseSuccess(response_page0)142        self.assertResponseSuccess(response_page1)143        self.assertResponseSuccess(response_page2)144        self.assertResponseSuccess(response_page3)145        self.assertEqual(len(response_page0.data['data']), page_size)146        self.assertEqual(len(response_page1.data['data']), page_size)147        self.assertEqual(len(response_page2.data['data']), 1)148        self.assertEqual(len(response_page3.data['data']), 0)...

AI_Sparapalloni (1).py

Source:AI_Sparapalloni (1).py

1import requests2import time3# import numpy as np4ADAFRUIT_IO_USERNAME = "Eronion"5ADAFRUIT_IO_KEY = "aio_CtXf43ogyYT4rlEb5OQ5nuVywe1y"6class AiSparapalloni:7    #  Loop -->In questo loop infinito definisco come raccogliere i dati dai feed8    #          e come rispondere ad essi con comandi inviati ad altri feed9    def loop(self):10        # infinite loop for serial managing11        self.Inizializzazione()12        lasttime = time.time()13        while True:14            ts = time.time()15            if ts-lasttime > 0.5:  # ogni mezzo secondo16                val_status = self.Get_feed('status')17                if val_status == 1:18                    #  print("Status 1")19                    feedname_program = 'program'  # Questo sarÃ  il mio feed per l'ON e OFF20                    headers_program = {'X-AIO-Key': ADAFRUIT_IO_KEY}21                    url_program_previous = 'https://io.adafruit.com/api/v2/{}/feeds/{}/data/previous'.format(ADAFRUIT_IO_USERNAME, feedname_program)22                    print(url_program_previous)23                    myGET_program_previous = requests.get(url_program_previous, headers=headers_program)24                    responseJsonBody_previous = myGET_program_previous.json()25                    str_val_program_previous = responseJsonBody_previous.get('value', None)26                    val_program_previous = int(str_val_program_previous)27                    val_program_last = self.Get_feed('program')28                    print(str_val_program_previous + "<->" + str(val_program_last))29                    if val_program_last != val_program_previous:30                        if val_program_last == 3:31                            self.Tiro_in_loop()32                        elif val_program_last == 4:33                            self.Tiro_ad_obbiettivo()34                        elif val_program_last == 5:35                            self.Adattiva()36                lasttime = time.time()37    def Tiro_in_loop(self):38        print('tiro a loop')39        self.Post_feed('status', 1)40        while self.Get_feed('program') == 3 and self.Get_feed('status') == 1:41            print("rimango in loop")42        self.Post_feed('status', 0)43        print("Chiudo tiro a loop")44    def Tiro_ad_obbiettivo(self):45        print('Tiro ad obbiettivo')46        obbiettivo = self.Get_feed('obbiettivo')47        serie_completate = 048        self.Post_feed('status', 1)49        while self.Get_feed('program') == 4:50            tirati2 = self.Get_feed('tirati2')51            tirati3 = self.Get_feed('tirati3')52            if tirati2 == 10:53                realizzati2 = self.Get_feed('realizzati2')54                if realizzati2 >= obbiettivo:55                    serie_completate = serie_completate + 156                    self.Post_feed('serie', serie_completate)57                self.Azzera_statistiche()58            elif tirati3 ==10:59                realizzati3 = self.Get_feed('realizzati3')60                if realizzati3 >= obbiettivo:61                    serie_completate = serie_completate + 162                    self.Post_feed('serie', serie_completate)63                self.Azzera_statistiche()64        self.Post_feed('status', 0)65    def Adattiva(self): # DA DEFINIRE66        print('adattiva')67        livello = self.Get_feed('livello')68        serie_completate = 069        serie_di_fila = 070        self.Post_feed('status', 1)71        while self.Get_feed('program') == 5:72            tirati2 = self.Get_feed('tirati2')73            tirati3 = self.Get_feed('tirati3')74            if tirati2 == 10:75                realizzati2 = self.Get_feed('realizzati2')76                if realizzati2 >= livello:77                    serie_di_fila = serie_di_fila + 178                    serie_completate = serie_completate + 179                    self.Post_feed('serie', serie_completate)80                    if serie_di_fila >= 2:81                        livello=livello+182                        serie_di_fila = 083                else:84                    serie_di_fila = 085                self.Azzera_statistiche()86            elif tirati3 == 10:87                realizzati3 = self.Get_feed('realizzati3')88                if realizzati3 >= livello:89                    serie_di_fila = serie_di_fila + 190                    serie_completate = serie_completate + 191                    self.Post_feed('serie', serie_completate)92                    if serie_di_fila >= 2:93                        livello = livello + 194                        self.Post_feed('livello', livello)95                        serie_di_fila = 096                else:97                    serie_di_fila = 098        self.Post_feed('status', 0)99    def Inizializzazione(self):100        self.Post_feed('program', 0)101        self.Post_feed('tirati2', 0)102        self.Post_feed('tirati3', 0)103        self.Post_feed('realizzati2', 0)104        self.Post_feed('realizzati3', 0)105    def Azzera_statistiche(self):106        self.Post_feed('tirati2', 0)107        self.Post_feed('tirati3', 0)108        self.Post_feed('realizzati2', 0)109        self.Post_feed('realizzati3', 0)110    def Get_feed(self, feedname):111        headers = {'X-AIO-Key': ADAFRUIT_IO_KEY}112        url = 'https://io.adafruit.com/api/v2/{}/feeds/{}/data/last'.format(ADAFRUIT_IO_USERNAME, feedname)113        myGET = requests.get(url, headers=headers)114        responseJsonBody = myGET.json()115        str_val = responseJsonBody.get('value', None)116        time.sleep(60)117        return int(str_val)118    def Post_feed(self, feedname, valore):119        mypostdata = {'value': valore}120        headers = {'X-AIO-Key': ADAFRUIT_IO_KEY}121        url = 'https://io.adafruit.com/api/v2/{}/feeds/{}/data'.format(ADAFRUIT_IO_USERNAME, feedname)122        myPOST = requests.post(url, data=mypostdata, headers=headers)123        # print(myPOST.json())124        time.sleep(60)125        #  ACHTUNG!!!  get dal feed ti torna una stringa126if __name__ == '__main__':127    ai = AiSparapalloni()...

test_get_feed.py

Source:test_get_feed.py

...17# along with this program.  If not, see <http://www.gnu.org/licenses/>.18from gvm.errors import InvalidArgumentType, RequiredArgument19from gvm.protocols.gmpv208 import FeedType20class GmpGetFeedTestMixin:21    def test_get_feed(self):22        """23        Test basic get_feed calls with only resource_type except special24        cases for audit, policy, scan_config and task.25        """26        self.gmp.get_feed(FeedType.NVT)27        self.connection.send.has_been_called_with('<get_feeds type="NVT"/>')28        self.gmp.get_feed(FeedType.CERT)29        self.connection.send.has_been_called_with('<get_feeds type="CERT"/>')30        self.gmp.get_feed(FeedType.SCAP)31        self.connection.send.has_been_called_with('<get_feeds type="SCAP"/>')32        self.gmp.get_feed(FeedType.GVMD_DATA)33        self.connection.send.has_been_called_with(34            '<get_feeds type="GVMD_DATA"/>'35        )36    def test_get_feed_missing_type(self):37        """38        Test get_feed calls with missing resource_type39        """40        with self.assertRaises(RequiredArgument):41            self.gmp.get_feed(feed_type=None)42        with self.assertRaises(RequiredArgument):43            self.gmp.get_feed(feed_type="")44        with self.assertRaises(RequiredArgument):45            self.gmp.get_feed("")46    def test_get_feed_invalid_type(self):47        """48        Test get_feed calls with invalid resource_type49        """50        with self.assertRaises(InvalidArgumentType):...

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.