Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run dbt-osmosis automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
