Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use create_relation method in dbt-osmosis
Best Python code snippet using dbt-osmosis_python
test_context.py
Source:test_context.py 
...229 context.create_document(1, 'python.pdf', 'pdf', '/tmp/python.pdf')230 context.create_document(2, 'rust.pdf', 'pdf', '/tmp/rust.pdf')231 context.create_document(3, 'lua.pdf', 'pdf', '/tmp/lua.pdf')232 context.create_tag(1, 'book')233 context.create_relation(1, 1)234 context.create_relation(2, 1)235 context.create_relation(3, 1)236 document_ids = context.find_document_ids([1])237 self.assertEqual(document_ids, [1, 2, 3])238 self.assertEqual(context.find_tag_ids([1]), [1])239 self.assertEqual(context.find_tag_ids([2]), [1])240 self.assertEqual(context.find_tag_ids([3]), [1])241 def test_create_relation_with_invalid_document(self):242 context = Context()243 context.create_tag(1, 'book')244 with self.assertRaises(ValueError):245 context.create_relation(1, 1)246 def test_create_relation_with_invalid_tag(self):247 context = Context()248 context.create_document(1, 'python.pdf', 'pdf', '/tmp/python.pdf')249 with self.assertRaises(ValueError):250 context.create_relation(1, 1)251 def test_remove_relations(self):252 context = Context()253 context.create_document(1, 'python.pdf', 'pdf', '/tmp/python.pdf')254 context.create_document(2, 'rust.pdf', 'pdf', '/tmp/rust.pdf')255 context.create_document(3, 'lua.pdf', 'pdf', '/tmp/lua.pdf')256 context.create_tag(1, 'book')257 context.create_relation(1, 1)258 context.create_relation(2, 1)259 context.create_relation(3, 1)260 self.assertEqual(context.count_relations(), 3)261 context.destroy_relation(2, 1)262 self.assertEqual(context.count_relations(), 2)263 context.destroy_relation(1, 1)264 self.assertEqual(context.count_relations(), 1)265 context.destroy_relation(3, 1)266 self.assertEqual(context.count_relations(), 0)267 def test_remove_missing_relation(self):268 context = Context()269 context.create_tag(1, 'book')270 context.create_document(1, 'python.pdf', 'pdf', '/tmp/python.pdf')271 with self.assertRaises(ValueError):272 context.destroy_relation(1, 1)273 def test_remove_relations_with_document(self):274 context = Context()275 context.create_document(1, 'python.pdf', 'pdf', '/tmp/python.pdf')276 context.create_document(2, 'rust.pdf', 'pdf', '/tmp/rust.pdf')277 context.create_document(3, 'lua.pdf', 'pdf', '/tmp/lua.pdf')278 context.create_tag(1, 'book')279 context.create_relation(1, 1)280 context.create_relation(2, 1)281 context.create_relation(3, 1)282 context.destroy_document(1)283 self.assertEqual(context.count_relations(), 2)284 context.destroy_document(3)285 self.assertEqual(context.count_relations(), 1)286 context.destroy_document(2)287 self.assertEqual(context.count_relations(), 0)288 def test_remove_relations_with_tag(self):289 context = Context()290 context.create_document(1, 'python.pdf', 'pdf', '/tmp/python.pdf')291 context.create_document(2, 'rust.pdf', 'pdf', '/tmp/rust.pdf')292 context.create_document(3, 'lua.pdf', 'pdf', '/tmp/lua.pdf')293 context.create_tag(1, 'book')294 context.create_relation(1, 1)295 context.create_relation(2, 1)296 context.create_relation(3, 1)297 context.destroy_tag(1)298 self.assertEqual(context.count_relations(), 0)299 def test_multiple_tags_and_documents(self):300 context = Context()301 context.create_document(1, 'python.pdf', 'pdf', '/tmp/python.pdf')302 context.create_document(2, 'tkinter.pdf', 'pdf', '/tmp/tkinter.pdf')303 context.create_document(3, 'lua.pdf', 'pdf', '/tmp/lua.pdf')304 context.create_document(4, 'index.py', 'py', '/tmp/index.py')305 context.create_tag(5, 'book')306 context.create_tag(6, 'python')307 context.create_relation(1, 5)308 context.create_relation(2, 5)309 context.create_relation(3, 5)310 context.create_relation(1, 6)311 context.create_relation(2, 6)312 context.create_relation(4, 6)313 self.assertEqual(context.count_relations(), 6)314 book_ids = context.find_document_ids([5])315 self.assertEqual(book_ids, [1, 2, 3])316 python_ids = context.find_document_ids([6])317 self.assertEqual(python_ids, [1, 2, 4])318 python_book_ids = context.find_document_ids([5, 6])319 self.assertEqual(python_book_ids, [1, 2])320 document_ids = context.find_document_ids([])321 self.assertEqual(document_ids, [1, 2, 3, 4])322 tag_ids = context.find_tag_ids([1, 2])323 self.assertEqual(tag_ids, [5, 6])324 tag_ids = context.find_tag_ids([1, 3])325 self.assertEqual(tag_ids, [5])326 tag_ids = context.find_tag_ids([3, 4])...
test_relation.py
Source:test_relation.py
...37 self.assertEqual(resp.status_code, status.HTTP_200_OK)38 self.assertEqual(resp.data['related'], [])39 @utils.store_samples('relation-list')40 def test_list_two_patch_relation(self):41 relation = create_relation()42 patches = create_patches(2, project=self.project, related=relation)43 # nobody44 resp = self.client.get(self.api_url(item=patches[0].pk))45 self.assertEqual(resp.status_code, status.HTTP_200_OK)46 self.assertIn('related', resp.data)47 self.assertEqual(len(resp.data['related']), 1)48 self.assertEqual(resp.data['related'][0]['id'], patches[1].pk)49 resp = self.client.get(self.api_url(item=patches[1].pk))50 self.assertEqual(resp.status_code, status.HTTP_200_OK)51 self.assertIn('related', resp.data)52 self.assertEqual(len(resp.data['related']), 1)53 self.assertEqual(resp.data['related'][0]['id'], patches[0].pk)54 @utils.store_samples('relation-create-forbidden')55 def test_create_two_patch_relation_nobody(self):56 patches = create_patches(2, project=self.project)57 resp = self.client.patch(58 self.api_url(item=patches[0].pk), {'related': [patches[1].pk]}59 )60 self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)61 def test_create_two_patch_relation_user(self):62 patches = create_patches(2, project=self.project)63 self.client.force_authenticate(user=self.normal_user)64 resp = self.client.patch(65 self.api_url(item=patches[0].pk), {'related': [patches[1].pk]}66 )67 self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)68 @utils.store_samples('relation-create-maintainer')69 def test_create_two_patch_relation_maintainer(self):70 patches = create_patches(2, project=self.project)71 self.client.force_authenticate(user=self.maintainer)72 resp = self.client.patch(73 self.api_url(item=patches[0].pk), {'related': [patches[1].pk]}74 )75 self.assertEqual(resp.status_code, status.HTTP_200_OK)76 # reload and verify77 patches = Patch.objects.all()78 self.assertIsNotNone(patches[0].related)79 self.assertIsNotNone(patches[1].related)80 self.assertEqual(patches[1].related, patches[0].related)81 def test_delete_two_patch_relation_nobody(self):82 relation = create_relation()83 patch = create_patches(2, project=self.project, related=relation)[0]84 self.assertEqual(PatchRelation.objects.count(), 1)85 resp = self.client.patch(self.api_url(item=patch.pk), {'related': []})86 self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)87 self.assertEqual(PatchRelation.objects.count(), 1)88 @utils.store_samples('relation-delete')89 def test_delete_two_patch_relation_maintainer(self):90 relation = create_relation()91 patch = create_patches(2, project=self.project, related=relation)[0]92 self.assertEqual(PatchRelation.objects.count(), 1)93 self.client.force_authenticate(user=self.maintainer)94 resp = self.client.patch(self.api_url(item=patch.pk), {'related': []})95 self.assertEqual(resp.status_code, status.HTTP_200_OK)96 self.assertEqual(PatchRelation.objects.count(), 0)97 self.assertEqual(98 Patch.objects.filter(related__isnull=False).exists(), False99 )100 def test_create_three_patch_relation(self):101 patches = create_patches(3, project=self.project)102 self.client.force_authenticate(user=self.maintainer)103 resp = self.client.patch(104 self.api_url(item=patches[0].pk),105 {'related': [patches[1].pk, patches[2].pk]},106 )107 self.assertEqual(resp.status_code, status.HTTP_200_OK)108 # reload and verify109 patches = Patch.objects.all()110 self.assertIsNotNone(patches[0].related)111 self.assertIsNotNone(patches[1].related)112 self.assertIsNotNone(patches[2].related)113 self.assertEqual(patches[0].related, patches[1].related)114 self.assertEqual(patches[1].related, patches[2].related)115 def test_delete_from_three_patch_relation(self):116 relation = create_relation()117 patch = create_patches(3, project=self.project, related=relation)[0]118 self.assertEqual(PatchRelation.objects.count(), 1)119 self.client.force_authenticate(user=self.maintainer)120 resp = self.client.patch(self.api_url(item=patch.pk), {'related': []})121 self.assertEqual(resp.status_code, status.HTTP_200_OK)122 self.assertIsNone(Patch.objects.get(id=patch.pk).related)123 self.assertEqual(PatchRelation.objects.count(), 1)124 self.assertEqual(PatchRelation.objects.first().patches.count(), 2)125 @utils.store_samples('relation-extend-through-new')126 def test_extend_relation_through_new(self):127 relation = create_relation()128 existing_patch_a = create_patches(129 2, project=self.project, related=relation130 )[0]131 new_patch = create_patch(project=self.project)132 self.client.force_authenticate(user=self.maintainer)133 resp = self.client.patch(134 self.api_url(item=new_patch.pk), {'related': [existing_patch_a.pk]}135 )136 self.assertEqual(resp.status_code, status.HTTP_200_OK)137 self.assertEqual(relation, Patch.objects.get(pk=new_patch.pk).related)138 self.assertEqual(relation.patches.count(), 3)139 def test_extend_relation_through_old(self):140 relation = create_relation()141 existing_patch_a = create_patches(142 2, project=self.project, related=relation143 )[0]144 new_patch = create_patch(project=self.project)145 # maintainer146 self.client.force_authenticate(user=self.maintainer)147 resp = self.client.patch(148 self.api_url(item=existing_patch_a.pk), {'related': [new_patch.pk]}149 )150 self.assertEqual(resp.status_code, status.HTTP_200_OK)151 self.assertEqual(relation, Patch.objects.get(id=new_patch.id).related)152 self.assertEqual(relation.patches.count(), 3)153 def test_extend_relation_through_new_two(self):154 relation = create_relation()155 existing_patch_a = create_patches(156 2, project=self.project, related=relation157 )[0]158 new_patch_a = create_patch(project=self.project)159 new_patch_b = create_patch(project=self.project)160 self.client.force_authenticate(user=self.maintainer)161 resp = self.client.patch(162 self.api_url(item=new_patch_a.pk),163 {'related': [existing_patch_a.pk, new_patch_b.pk]},164 )165 self.assertEqual(resp.status_code, status.HTTP_200_OK)166 self.assertEqual(167 relation, Patch.objects.get(id=new_patch_a.id).related168 )169 self.assertEqual(170 relation, Patch.objects.get(id=new_patch_b.id).related171 )172 self.assertEqual(relation.patches.count(), 4)173 @utils.store_samples('relation-extend-through-old')174 def test_extend_relation_through_old_two(self):175 relation = create_relation()176 existing_patch_a = create_patches(177 2, project=self.project, related=relation178 )[0]179 new_patch_a = create_patch(project=self.project)180 new_patch_b = create_patch(project=self.project)181 # maintainer182 self.client.force_authenticate(user=self.maintainer)183 resp = self.client.patch(184 self.api_url(item=existing_patch_a.pk),185 {'related': [new_patch_a.pk, new_patch_b.pk]},186 )187 self.assertEqual(resp.status_code, status.HTTP_200_OK)188 self.assertEqual(189 relation, Patch.objects.get(id=new_patch_a.id).related190 )191 self.assertEqual(192 relation, Patch.objects.get(id=new_patch_b.id).related193 )194 self.assertEqual(relation.patches.count(), 4)195 def test_remove_one_patch_from_relation_bad(self):196 relation = create_relation()197 patches = create_patches(3, project=self.project, related=relation)198 keep_patch_a = patches[1]199 keep_patch_b = patches[1]200 # this should do nothing - it is interpreted as201 # _adding_ keep_patch_b again which is a no-op.202 # maintainer203 self.client.force_authenticate(user=self.maintainer)204 resp = self.client.patch(205 self.api_url(item=keep_patch_a.pk), {'related': [keep_patch_b.pk]}206 )207 self.assertEqual(resp.status_code, status.HTTP_200_OK)208 self.assertEqual(relation.patches.count(), 3)209 def test_remove_one_patch_from_relation_good(self):210 relation = create_relation()211 target_patch = create_patches(212 3, project=self.project, related=relation213 )[0]214 # maintainer215 self.client.force_authenticate(user=self.maintainer)216 resp = self.client.patch(217 self.api_url(item=target_patch.pk), {'related': []}218 )219 self.assertEqual(resp.status_code, status.HTTP_200_OK)220 self.assertIsNone(Patch.objects.get(id=target_patch.id).related)221 self.assertEqual(relation.patches.count(), 2)222 @utils.store_samples('relation-forbid-moving-between-relations')223 def test_forbid_moving_patch_between_relations(self):224 """Test the break-before-make logic"""225 relation_a = create_relation()226 create_patches(2, project=self.project, related=relation_a)227 relation_b = create_relation()228 create_patches(2, project=self.project, related=relation_b)229 patch_a = relation_a.patches.first()230 patch_b = relation_b.patches.first()231 self.client.force_authenticate(user=self.maintainer)232 resp = self.client.patch(233 self.api_url(item=patch_a.pk), {'related': [patch_b.pk]}234 )235 self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT)236 resp = self.client.patch(237 self.api_url(item=patch_b.pk), {'related': [patch_a.pk]}238 )239 self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT)240 def test_cross_project_different_maintainers(self):241 patch_a = create_patch(project=self.project)...
GraphCreater.py
Source:GraphCreater.py
...29 % orgs_data['é¨éå称'][i]).data()[0]30 node = {'é¨éå
ç ': search_node['p.é¨éå
ç '],31 'ç´å±ä¸çº§': search_node['p.ç´å±ä¸çº§'], }32 # ç»å¶é¨éä¹é´çé¶å±å
³ç³»33 create_relation(self.graph, 'é¨é', 'é¨é', node['ç´å±ä¸çº§'], node['é¨éå
ç '])34 return None35 def initialize_stations(self):36 '''37 该å½æ°ç¨äºå°ç«æ¹éåå§å38 :return: None39 '''40 self.clear_graph()41 stations_list = os.listdir(self.stations_dir)42 for station in stations_list:43 self.__initialize_station(os.path.join(self.stations_dir, station))44 return None45 def __initialize_station(self, station_dir):46 try:47 station_info = pd.read_csv(os.path.join(station_dir, 'å°ç«.csv'), encoding='gbk')48 except FileNotFoundError:49 return None50 # çæå°ç«èç¹51 create_node(self.graph, 'å°ç«',52 {'name': station_info['å°ç«å称'][0],53 '代å·': station_info['å°ç«ä»£å·'][0],54 'æºæ¿é¢ç§¯': station_info['æºæ¿é¢ç§¯'][0]})55 # çæè·¯ç±ä¿¡æ¯56 self.__initialize_routes(station_dir, station_info['å°ç«å称'][0])57 self.__initialize_equips(station_dir, station_info['å°ç«å称'][0])58 return None59 def __initialize_routes(self, station_dir, station_name):60 '''61 该å½æ°ç¨äºçæè·¯ç±è¿æ¥é¨åçå¾è°±62 :param station_dir: å°ç«æ件夹63 :param station_name: å°ç«å称64 :return:65 '''66 try:67 stations_data = pd.read_csv(os.path.join(station_dir, 'è·¯ç±.csv'), encoding='gbk')68 except FileNotFoundError:69 return None70 # çæè·¯ç±èç¹71 for i in range(len(stations_data)):72 # 建ç«çº¿è·¯èç¹73 create_node(self.graph, '线路',74 {'name': stations_data['线路å称'][i],75 '纤è¯æ°é': stations_data['纤è¯æ°é'][i],76 'å¨ç¨çº¤è¯': stations_data['æ¿è½½ç³»ç»'][i],77 'ä¸å¯ç¨çº¤è¯': str(stations_data['ä¸å¯ç¨çº¤è¯'][i]).replace('ã', '/')})78 # å建è³å°ç«è¿æ¥79 # æ¬ç«¯å°ç«è³çº¿è·¯çè¿æ¥80 create_relation(self.graph, 'å°ç«', '线路', station_name, stations_data['线路å称'][i])81 # 对端å°ç«å°çº¿è·¯çè¿æ¥82 create_relation(self.graph, 'å°ç«', '线路', stations_data['éè¾¾æ¹å'][i], stations_data['线路å称'][i])83 # å°ç«å°å°ç«ä¹é´çè¿æ¥84 create_relation(self.graph, 'å°ç«', 'å°ç«', station_name, stations_data['éè¾¾æ¹å'][i])85 # 建ç«ç³»ç»èç¹86 systems = str(stations_data['æ¿è½½ç³»ç»'][i]).split('ã')87 for j in range(len(systems)):88 system = systems[j].split('[')[0]89 if str(system) == 'nan':90 break91 create_node(self.graph, 'ç³»ç»', {'name': system})92 create_relation(self.graph, 'å°ç«', 'ç³»ç»', station_name, system)93 create_relation(self.graph, '线路', 'ç³»ç»', stations_data['线路å称'][i], system)94 # 建ç«ä¸ç»§èç¹95 relays = str(stations_data['ä¸ç»§ç«'][i]).split('ã')96 for relay in relays:97 if relay == 'nan':98 break99 create_node(self.graph, 'ä¸ç»§', {'name': relay})100 create_relation(self.graph, '线路', 'ä¸ç»§', stations_data['线路å称'][i], relay)101 return None102 def __initialize_equips(self, station_dir, station_name):103 try:104 equips_data = pd.read_csv(os.path.join(station_dir, '设å¤.csv'), encoding='gbk')105 except FileNotFoundError:106 return None107 # çæ设å¤èç¹108 for i in range(len(equips_data)):109 create_node(self.graph, '设å¤',110 {'name': equips_data['设å¤ç±»å'][i],111 'å称': equips_data['设å¤å称'][i],112 'ç¼ç ': equips_data['设å¤ç¼ç '][i]},113 overwrite=False)114 # 建ç«å°ç«å°è®¾å¤çè¿æ¥115 create_relation(self.graph, 'å°ç«', '设å¤', station_name, equips_data['设å¤ç¼ç '][i], sub_key='ç¼ç ')116 # 建ç«è®¾å¤å°è®¾å¤çè¿æ¥117 for equ in str(equips_data['è¿æ¥è®¾å¤'][i]).split('ã'):118 if equ == '' or equ == 'nan':119 continue120 create_relation(self.graph, '设å¤', '设å¤', equips_data['设å¤ç¼ç '][i],121 int(equ), main_key='ç¼ç ', sub_key='ç¼ç ', newnode=False)122 # 建ç«è®¾å¤å°çº¿è·¯çè¿æ¥123 for route in equips_data['è¿æ¥è·¯ç±'][i].split('ã'):124 create_relation(self.graph, '设å¤', '线路', equips_data['设å¤ç¼ç '][i], route, main_key='ç¼ç ')125 def clear_graph(self):126 '''127 该å½æ°ç¨äºæ¸
空ç¥è¯å¾è°±ï¼æ
ç¨ï¼128 :return: None129 '''130 self.graph.run('match(n) detach delete n')131 return None132if __name__ == '__main__':133 g = GraphCreater()...
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?
