How to use create_relation method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_context.py

Source:test_context.py Github

copy

Full Screen

...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])...

Full Screen

Full Screen

test_relation.py

Source:test_relation.py Github

copy

Full Screen

...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)...

Full Screen

Full Screen

GraphCreater.py

Source:GraphCreater.py Github

copy

Full Screen

...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()...

Full Screen

Full Screen

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:

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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful