How to use _create_rule method in localstack

Best Python code snippet using localstack_python

test_rules_api.py

Source:test_rules_api.py

...87 body.update({'active': active})88 if priority is not None:89 body.update({'priority': priority})90 return body91 def _create_rule(self, name, org_id=None, pool_id="", owner_id="",92 conditions=None, active=None, priority=None,93 set_allowed=True):94 if not org_id:95 org_id = self.org_id96 if not conditions:97 conditions = [98 {"type": "name_starts_with", "meta_info": "name_starts_with"}99 ]100 if pool_id == "":101 pool_id = self.org_pool_id102 if owner_id == "":103 owner_id = self.employee['id']104 if set_allowed:105 _, employee = self.client.employee_get(owner_id)106 self.set_allowed_pair(employee['auth_user_id'], pool_id)107 rule_body = self._prepare_rule_body(108 name=name,109 pool_id=pool_id,110 owner_id=owner_id,111 conditions=conditions,112 active=active,113 priority=priority,114 )115 code, rule = self.client.rules_create(org_id, rule_body)116 self.assertEqual(code, 201, rule)117 return rule118 def _test_params(self, call, params, *args, required_parameters=None,119 entities_should_exist=None):120 data = params.copy()121 unexpected_param_name = "unexpected_param"122 data[unexpected_param_name] = self.gen_id()123 code, response = call(*args, data)124 self.assertEqual(code, 400)125 self.assertTrue('Unexpected parameters' in response['error']['reason'])126 self.assertTrue(unexpected_param_name in response['error']['reason'])127 if required_parameters:128 for param in required_parameters:129 data = params.copy()130 data.pop(param)131 code, response = call(*args, data)132 self.assertEqual(code, 400)133 self.assertEqual(response['error']['reason'],134 '%s is not provided' % param)135 if entities_should_exist:136 for param in entities_should_exist:137 data = params.copy()138 id = str(self.gen_id())139 data[param] = id140 code, response = call(*args, data)141 self.assertEqual(code, 400)142 self.assertTrue('%s %s doesn\'t exist' % (param, id)143 in response['error']['reason'])144 self.assertTrue(response['error']['error_code'], 'OE0212')145class TestRuleApi(TestRulesApiBase):146 @patch(AUTHORIZE_ACTION_METHOD)147 def test_create_rule(self, p_authorize):148 """149 Basic create rule request flow.150 Steps:151 - 1. Check initial count of rules. Should be 0.152 - 2. Create FullPair rule.153 - 3. Verify response code is 201.154 - 4. Verify count rules is 1.155 - 12. Create rule in inactive state.156 - verify state157 """158 p_authorize.return_value = True159 rules = self._get_rules()160 self.assertEqual(len(rules), 0)161 full_rule = self._create_rule('Test rule')162 self.assertEqual(self.p_activities_publish.call_count, 1)163 self.assertEqual(full_rule['pool_name'], self.organization['name'])164 self.assertEqual(full_rule['pool_purpose'], 'business_unit')165 self.assertEqual(full_rule['owner_name'], self.employee['name'])166 self.assertEqual(full_rule['creator_name'], self.employee['name'])167 self.assertEqual(full_rule['priority'], 1)168 rules = self._get_rules()169 self.assertEqual(len(rules), 1)170 self.assertEqual(rules[0]['id'], full_rule['id'])171 inactive = self._create_rule("Inactive rule",172 active=False)173 self.assertEqual(self.p_activities_publish.call_count, 2)174 code, rule = self.client.rule_get(inactive['id'])175 self.assertEqual(code, 200)176 self.assertEqual(rule['active'], False)177 rules = self._get_rules()178 self.assertEqual(len(rules), 2)179 code, rules = self.client.rules_list(self.org_id,180 valid_rules_only=True)181 self.assertEqual(code, 200)182 all_rules = rules['rules']183 for rule in all_rules:184 self.assertTrue(rule['id'] != inactive['id'])185 self.assertIn('priority', rule)186 @patch(AUTHORIZE_ACTION_METHOD)187 def test_create_rule_with_priority(self, p_authorize):188 """189 - 1. Check initial count of rules. Should be 0.190 - 2. Create 2 rules191 - 3. Create rules with priority 2192 - 4. Get list of rules and check priorities193 """194 p_authorize.return_value = True195 rules = self._get_rules()196 self.assertEqual(len(rules), 0)197 rule1 = self._create_rule('rule1')198 rule3 = self._create_rule('rule3')199 self.assertEqual(self.p_activities_publish.call_count, 2)200 rules = self._get_rules()201 self.assertEqual(len(rules), 2)202 rule2 = self._create_rule('rule2', priority=2)203 self.assertEqual(self.p_activities_publish.call_count, 3)204 self.assertEqual(rule2['priority'], 2)205 rule_body = self._prepare_rule_body(206 name='rule4',207 pool_id=self.org_pool_id,208 owner_id=self.employee['id'],209 conditions=[{"type": "name_starts_with", "meta_info": "name_starts_with"}],210 active=None,211 priority=5,212 )213 code, resp = self.client.rules_create(self.org_id, rule_body)214 self.assertEqual(code, 400)215 self.verify_error_code(resp, 'OE0217')216 rule_body.update({'priority': 0})217 code, resp = self.client.rules_create(self.org_id, rule_body)218 self.assertEqual(code, 400)219 self.verify_error_code(resp, 'OE0224')220 rules = self._get_rules()221 self.assertEqual(len(rules), 3)222 for priority, rule in enumerate(rules, start=1):223 self.assertEqual(rule['priority'], priority)224 def test_params_create_rule(self):225 """226 Test required parameters for creation rule.227 Steps:228 - 1. Verify requests without param:229 - name230 - conditions231 - 2. Verify requests with param, but entity doesn't exist:232 - owner_id233 - pool_id234 - 3. Verify request with conditions as empty array235 - err_code should be 400236 - localized code should be E0216237 - 3. Verify request without conditions in body238 - err_code should be 400239 - localized code should be E0216240 - 4. Try to create rule without owner or pool.241 - localized code should be E0216242 - 5. Try to create rule with unsupported param in condition243 - localized code should be E0212244 - 6. Try to create rule with unsupported condition type245 - localized code should be E0430246 - 7. Try to create rule with absent meta in condition247 - localized code should be E0430248 - 8. Try to create rule with absent type in condition249 - localized code should be E0430250 - 9. Try to create rule with pool in another org251 - localized code should be E0002252 - 10. Try to create rule with owner in another org253 - localized code should be E0002254 - 11. Try to create rule with invalid condition format255 - localized code should be E0344256 """257 rule_body = self._prepare_rule_body(258 name="Test rule",259 pool_id=self.org_pool_id,260 owner_id=self.employee['id'],261 conditions=[262 {"type": "name_starts_with", "meta_info": "test_"}263 ]264 )265 self._test_params(266 self.client.rules_create, rule_body, self.org_id,267 required_parameters=['name', 'conditions'],268 entities_should_exist=['owner_id', 'pool_id']269 )270 rule_body_copy = rule_body.copy()271 rule_body_copy['name'] = None272 code, resp = self.client.rules_create(self.org_id, rule_body_copy)273 self.assertEqual(code, 400)274 self.verify_error_code(resp, 'OE0216')275 rule_body_copy = rule_body.copy()276 rule_body_copy['conditions'] = []277 code, resp = self.client.rules_create(self.org_id, rule_body_copy)278 self.assertEqual(code, 400)279 self.verify_error_code(resp, 'OE0216')280 rule_body_copy = rule_body.copy()281 rule_body_copy.pop('conditions')282 code, resp = self.client.rules_create(self.org_id, rule_body_copy)283 self.assertEqual(code, 400)284 self.verify_error_code(resp, 'OE0216')285 rule_body_copy = rule_body.copy()286 rule_body_copy['conditions'] = None287 code, resp = self.client.rules_create(self.org_id, rule_body_copy)288 self.assertEqual(code, 400)289 self.verify_error_code(resp, 'OE0216')290 rule_body_copy = rule_body.copy()291 rule_body_copy['pool_id'] = None292 rule_body_copy['owner_id'] = None293 code, resp = self.client.rules_create(self.org_id, rule_body_copy)294 self.assertEqual(code, 400)295 self.verify_error_code(resp, 'OE0216')296 rule_body_copy = rule_body.copy()297 rule_body_copy.pop('pool_id')298 rule_body_copy.pop('owner_id')299 code, resp = self.client.rules_create(self.org_id, rule_body_copy)300 self.assertEqual(code, 400)301 self.verify_error_code(resp, 'OE0216')302 rule_body_copy = copy.deepcopy(rule_body)303 rule_body_copy['conditions'][0]['unsupported_param'] = 'test_val'304 code, resp = self.client.rules_create(self.org_id, rule_body_copy)305 self.assertEqual(code, 400)306 self.verify_error_code(resp, 'OE0212')307 rule_body_copy = copy.deepcopy(rule_body)308 rule_body_copy['conditions'][0]['type'] = 'UNSUPPORTED_TYPE'309 code, resp = self.client.rules_create(self.org_id, rule_body_copy)310 self.assertEqual(code, 400)311 self.verify_error_code(resp, 'OE0430')312 required_condition_fields = ['type', 'meta_info']313 for param in required_condition_fields:314 rule_body_copy = copy.deepcopy(rule_body)315 rule_body_copy['conditions'][0].pop(param)316 code, resp = self.client.rules_create(self.org_id,317 rule_body_copy)318 self.assertEqual(code, 400)319 self.verify_error_code(resp, 'OE0216')320 rule_body_copy = copy.deepcopy(rule_body)321 rule_body_copy['conditions'][0]['unsupported_key'] = None322 code, resp = self.client.rules_create(self.org_id,323 rule_body_copy)324 self.assertEqual(code, 400)325 self.verify_error_code(resp, 'OE0212')326 rule_body_copy = rule_body.copy()327 rule_body_copy['owner_id'] = self.org2_employee['id']328 code, resp = self.client.rules_create(self.org_id, rule_body_copy)329 self.assertEqual(code, 400)330 self.verify_error_code(resp, 'OE0005')331 rule_body_copy = rule_body.copy()332 rule_body_copy['pool_id'] = self.org2_pool_id333 code, resp = self.client.rules_create(self.org_id, rule_body_copy)334 self.assertEqual(code, 400)335 self.verify_error_code(resp, 'OE0005')336 rule_body_copy = rule_body.copy()337 rule_body_copy['conditions'] = ["123", "123"]338 code, resp = self.client.rules_create(self.org_id, rule_body_copy)339 self.assertEqual(code, 400)340 self.verify_error_code(resp, 'OE0344')341 @patch(AUTHORIZE_ACTION_METHOD)342 def test_list_rules(self, p_authorize):343 """344 Test list rules API with query params345 Steps:346 - 1. Create rules:347 - rule1: full pair rule for pool_org and user1 as owner348 - rule2: full pair rule for pool_org and user2 as owner349 - rule3: full pair rule for sub_pool1 and user2 as owner350 - rule4: full pair rule for sub_pool1 and user1 as owner351 - 2. Request list rules. Should return all 4 rules352 - 3. Request list rules with owner as user1. Should return:353 - rule1, rule4354 - 4. Request list rules with pool as org_pool. Should return:355 - rule1, rule2356 - 5. Request list rules with org_pool and user1 as owner.357 Should return:358 - rule1359 -9. Request list rules with invalid owner_id.360 Should return:361 - empty array.362 """363 def check_list_rules(expected_rules, **kwargs):364 expected_ids = set([r['id'] for r in expected_rules])365 rules = self._get_rules(**kwargs)366 self.assertEqual(len(rules), len(expected_rules))367 for rule in rules:368 self.assertTrue(rule['id'] in expected_ids)369 p_authorize.return_value = True370 rules = self._get_rules()371 self.assertEqual(len(rules), 0)372 rule1 = self._create_rule("rule1")373 rule2 = self._create_rule("rule2", owner_id=self.employee2['id'])374 rule3 = self._create_rule(375 "rule3", pool_id=self.sub_pools[0]['id'],376 owner_id=self.employee2['id'])377 rule4 = self._create_rule(378 "rule4", pool_id=self.sub_pools[0]['id'],379 owner_id=self.employee['id'])380 rules = self._get_rules()381 self.assertEqual(len(rules), 4)382 check_list_rules([rule1, rule4], owner_id=self.employee['id'])383 check_list_rules([rule1, rule2], pool_id=self.org_pool_id)384 check_list_rules([rule1], owner_id=self.employee['id'],385 pool_id=self.org_pool_id)386 check_list_rules([], owner_id=self.gen_id())387 @patch(AUTHORIZE_ACTION_METHOD)388 def test_create_rule_with_existing_name(self, p_authorize):389 """390 Test for trying crete rule with name that is already exist391 Steps:392 - 1. Check initial count of rules. Should be 0.393 - 2. Create FullPair rule with name TestRule.394 - 3. Verify response code is 201.395 - 4. Verify count rules is 1.396 - 5. Create FullPair rule with the same name.397 - verify localized code is E0431398 - 6. Verify count rules is still 1.399 - 5. Create rules with the same name.400 - verify localized code is E0431401 - 6. Verify count rules is still 1.402 """403 p_authorize.return_value = True404 rules = self._get_rules()405 self.assertEqual(len(rules), 0)406 rule_body = self._prepare_rule_body(407 name="TestRule",408 pool_id=self.org_pool_id,409 owner_id=self.employee['id'],410 conditions=[411 {"type": "name_starts_with", "meta_info": "test_"}412 ]413 )414 code, full_rule = self.client.rules_create(self.org_id, rule_body)415 self.assertEqual(code, 201)416 rules = self._get_rules()417 self.assertEqual(len(rules), 1)418 code, resp = self.client.rules_create(self.org_id, rule_body)419 self.assertEqual(code, 409)420 self.verify_error_code(resp, 'OE0149')421 rules = self._get_rules()422 self.assertEqual(len(rules), 1)423 p_authorize.return_value = False424 rule_body_copy = rule_body.copy()425 rule_body_copy.pop('pool_id')426 code, resp = self.client.rules_create(self.org_id, rule_body_copy)427 self.assertEqual(code, 409)428 self.verify_error_code(resp, 'OE0149')429 rules = self._get_rules()430 self.assertEqual(len(rules), 1)431 rule_body_copy = rule_body.copy()432 rule_body_copy.pop('owner_id')433 code, resp = self.client.rules_create(self.org_id, rule_body_copy)434 self.assertEqual(code, 409)435 self.verify_error_code(resp, 'OE0149')436 rules = self._get_rules()437 self.assertEqual(len(rules), 1)438 @patch(AUTHORIZE_ACTION_METHOD)439 def test_create_full_rule_with_invalid_target_pair(self, p_authorize):440 """441 Test creation rule with invalid target pair.442 Steps:443 - 1. Check initial count of rules. Should be 0.444 - 2. Create FullPair rule with invalid target pair.445 - 3. Verify response code is 403.446 - 4. Verify localized code is E0379447 - 4. Verify count rules is 0.448 """449 p_authorize.return_value = False450 rules = self._get_rules()451 self.assertEqual(len(rules), 0)452 rule_body = self._prepare_rule_body(453 name="TestRule",454 pool_id=self.org_pool_id,455 owner_id=self.employee['id'],456 conditions=[457 {"type": "name_starts_with", "meta_info": "test_"}458 ]459 )460 code, resp = self.client.rules_create(self.org_id, rule_body)461 self.assertEqual(code, 403)462 self.verify_error_code(resp, "OE0379")463 rules = self._get_rules()464 self.assertEqual(len(rules), 0)465 @patch(AUTHORIZE_ACTION_METHOD)466 def test_create_rule_with_invalid_tag_is_meta(self, p_authorize):467 """468 Test creation rule with invalid tag is meta condition469 Steps:470 - 1. Try to create rule with tags_is conditions with all possible471 invalid meta_info parameter472 - 2. Verify all tries are failed473 - 3. Verify no new rule created474 """475 p_authorize.return_value = True476 rules = self._get_rules()477 self.assertEqual(len(rules), 0)478 valid_rule_body = self._prepare_rule_body(479 name="TestRule",480 pool_id=self.org_pool_id,481 owner_id=self.employee['id'],482 conditions=[483 {"type": "name_starts_with", "meta_info": "test_"}484 ]485 )486 def check_response(meta_info, error_code):487 conditions = [488 {"type": "tag_is", "meta_info": meta_info}489 ]490 valid_rule_body['conditions'] = conditions491 code, resp = self.client.rules_create(self.org_id,492 valid_rule_body)493 self.assertEqual(code, 400)494 self.verify_error_code(resp, error_code)495 check_response(None, "OE0216")496 check_response(123, "OE0214")497 check_response({}, "OE0214")498 check_response("{}", "OE0216")499 check_response("123", "OE0344")500 check_response("\"123\"", "OE0344")501 check_response("{\"123\"}", "OE0219")502 check_response("{\"key\": \"key\"}", "OE0216")503 check_response("{\"key\": 123}", "OE0214")504 check_response("{\"value\": \"value\"}", "OE0216")505 check_response("{\"key\": \"key\", \"value\": 123}", "OE0214")506 check_response("{\"key\": \"k\", \"value\": \"v\", \"new\": \"value\"}",507 "OE0212")508 check_response("{\"key\": \"key\", \"unexpected\": \"value\"}", "OE0216")509 rules = self._get_rules()510 self.assertEqual(len(rules), 0)511 @patch(AUTHORIZE_ACTION_METHOD)512 def test_update_rule(self, p_authorize):513 """514 Basic test for rule update515 Steps:516 - 1. Check initial count of rules. Should be 0.517 - 2. Create FullPair rule with 2 conditions.518 - 3. Update rule:519 - update name520 - update active521 - update pool_id522 - update owner_id523 - delete one condition524 - update one existing condition525 - add new condition526 - 4. Verify all changes with GET rule API527 - 5. Verify extra fields in PATCH response528 """529 final_conditions_after_update = set()530 p_authorize.return_value = True531 rules = self._get_rules()532 self.assertEqual(len(rules), 0)533 conditions = [534 {"type": "name_starts_with", "meta_info": "NAME_STARTS_WITH"},535 {"type": "name_contains", "meta_info": "NAME_CONTAINS"},536 ]537 full_rule = self._create_rule("TestRule", conditions=conditions)538 rules = self._get_rules()539 self.assertEqual(len(rules), 1)540 code, original_rule = self.client.rule_get(full_rule['id'])541 self.assertEqual(code, 200)542 modified_rule_body = copy.deepcopy(original_rule)543 modified_rule_body['name'] = 'new_name'544 modified_rule_body['active'] = False545 modified_rule_body['owner_id'] = self.employee2['id']546 modified_rule_body['pool_id'] = self.sub_pools[0]['id']547 modified_rule_body['conditions'][0]['type'] = 'name_ends_with'548 final_conditions_after_update.add('name_ends_with')549 modified_rule_body['conditions'][0]['meta_info'] = 'NAME_ENDS_WITH'550 modified_rule_body['conditions'] = modified_rule_body['conditions'][:-1]551 modified_rule_body['conditions'].append(552 {"type": "name_is", "meta_info": "NAME_IS"})553 final_conditions_after_update.add('name_is')554 code, patched_response = self.client.rule_update(original_rule['id'],555 modified_rule_body)556 self.assertEqual(code, 200)557 self.assertEqual(self.p_activities_publish.call_count, 2)558 code, modified_rule = self.client.rule_get(full_rule['id'])559 self.assertEqual(code, 200)560 self.assertEqual(modified_rule['name'], 'new_name')561 self.assertEqual(modified_rule['active'], False)562 self.assertEqual(modified_rule['pool_id'],563 self.sub_pools[0]['id'])564 self.assertEqual(modified_rule['owner_id'], self.employee2['id'])565 self.assertEqual(len(modified_rule['conditions']), 2)566 for condition in modified_rule['conditions']:567 self.assertTrue(condition['type'] in final_conditions_after_update)568 self.assertTrue(569 condition['meta_info'].lower() in final_conditions_after_update)570 self.assertEqual(patched_response['creator_name'],571 self.employee['name'])572 self.assertEqual(patched_response['owner_name'], self.employee2['name'])573 self.assertEqual(patched_response['pool_purpose'], "budget")574 self.assertEqual(patched_response['pool_name'],575 self.sub_pools[0]['name'])576 for field in ["creator_name", "pool_name",577 "owner_name", "pool_purpose"]:578 self.assertTrue(field in patched_response.keys())579 @patch(AUTHORIZE_ACTION_METHOD)580 def test_update_rule_with_existing_name(self, p_authorize):581 """582 Basic test for rule update583 Steps:584 - 1. Check initial count of rules. Should be 0.585 - 2. Create two rules with names rule0 and rule1586 - 3. Create rule with name rule2587 - 4. Create rule with name rule3588 - 5. Try to update rule0 with names rule1, rule2, rule3.589 - all updates should be failed.590 - verify localized code is E0431591 - 6. Check that rule0 has the same name592 - 7. Create rule for org2 with name rule_new_org.593 - 8. Update rule0 with name rule_new_org. Should be succeeded.594 """595 p_authorize.return_value = True596 rules = self._get_rules()597 self.assertEqual(len(rules), 0)598 rule0 = self._create_rule("rule0")599 rule1 = self._create_rule("rule1")600 rule2 = self._create_rule("rule2")601 rule3 = self._create_rule("rule3")602 for name in [rule1['name'], rule2['name'], rule3['name']]:603 rule0['name'] = name604 code, resp = self.client.rule_update(rule0['id'], rule0)605 self.assertEqual(code, 409)606 self.verify_error_code(resp, 'OE0149')607 code, rule = self.client.rule_get(rule0['id'])608 self.assertEqual(code, 200)609 self.assertEqual(rule['name'], "rule0")610 new_rule_name = "rule_new_org"611 with self.switch_user(self.org2_user_id):612 self._create_rule(new_rule_name, org_id=self.org2_id,613 owner_id=self.org2_employee['id'],614 pool_id=self.org2_pool_id)615 rule0['name'] = new_rule_name616 code, resp = self.client.rule_update(rule0['id'],617 rule0)618 # 5 rules created and one updated619 self.assertEqual(self.p_activities_publish.call_count, 6)620 self.assertEqual(code, 200)621 code, rule = self.client.rule_get(rule0['id'])622 self.assertEqual(code, 200)623 self.assertEqual(rule['name'], new_rule_name)624 @patch(AUTHORIZE_ACTION_METHOD)625 def test_update_rule_with_invalid_tag_is_meta(self, p_authorize):626 """627 Test update rule with invalid tag is meta condition628 Steps:629 - 1. Create valid rule with tag_is condition.630 - 2. Try to update rule with tags_is conditions with all possible631 invalid meta_info parameter632 - 3. Verify all tries are failed633 - 4. Verify no new rule created634 """635 p_authorize.return_value = True636 rules = self._get_rules()637 self.assertEqual(len(rules), 0)638 conditions = [639 {640 "type": "tag_is",641 "meta_info": "{\"key\":\"key\",\"value\":\"value\"}"642 }643 ]644 rule1 = self._create_rule('rule1', conditions=conditions)645 rules = self._get_rules()646 self.assertEqual(len(rules), 1)647 valid_rule_body = self._prepare_rule_body(648 name="TestRule",649 pool_id=self.org_pool_id,650 owner_id=self.employee['id'],651 conditions=[652 {"type": "name_starts_with", "meta_info": "test_"}653 ]654 )655 def check_response(meta_info, error_code):656 conditions = [657 {"type": "tag_is", "meta_info": meta_info}658 ]659 valid_rule_body['conditions'] = conditions660 code, resp = self.client.rule_update(rule1['id'],661 valid_rule_body)662 self.assertEqual(code, 400)663 self.verify_error_code(resp, error_code)664 check_response(None, "OE0216")665 check_response(123, "OE0214")666 check_response({}, "OE0214")667 check_response("{}", "OE0216")668 check_response("123", "OE0344")669 check_response("\"123\"", "OE0344")670 check_response("{\"123\"}", "OE0219")671 check_response("{\"key\": \"key\"}", "OE0216")672 check_response("{\"key\": 123}", "OE0214")673 check_response("{\"value\": \"value\"}", "OE0216")674 check_response("{\"key\": \"key\", \"value\": 123}", "OE0214")675 check_response("{\"key\": \"k\", \"value\": \"v\", \"new\": \"value\"}",676 "OE0212")677 check_response("{\"key\": \"key\", \"unexpected\": \"value\"}", "OE0216")678 rules = self._get_rules()679 self.assertEqual(len(rules), 1)680 @patch(AUTHORIZE_ACTION_METHOD)681 def test_update_rule_name_is_null(self, p_authorize):682 """683 Testing update rule name with None value684 Steps:685 - 1. Check initial count of rules. Should be 0.686 - 2. Create two FullPair rules with name rule0.687 - 3. Try to update rule0 with name is None688 - update should be failed.689 - verify localized code is E0216690 - 4. Try to update rule0 with name is integer value691 - update should be failed.692 - verify localized code is E0214693 """694 p_authorize.return_value = True695 rules = self._get_rules()696 self.assertEqual(len(rules), 0)697 full_rule = self._create_rule("rule0")698 full_rule['name'] = None699 code, resp = self.client.rule_update(full_rule['id'],700 full_rule)701 self.assertEqual(code, 400)702 self.verify_error_code(resp, 'OE0216')703 full_rule['name'] = 0704 code, resp = self.client.rule_update(full_rule['id'],705 full_rule)706 self.assertEqual(code, 400)707 self.verify_error_code(resp, 'OE0214')708 @patch(AUTHORIZE_ACTION_METHOD)709 def test_get_update_rule_invalid_id(self, p_authorize):710 """711 Test get_update_rule for invalid rule id712 Steps:713 - 1. Get rule by invalid id. Verify714 - code is 404715 - localized code is E0002716 - 2. Update rule by invalid id. Verify717 - code is 404718 - localized code is E0002719 - 3. Create rule for org2720 - 4. Try to update rule in org1 for rule id from org 2721 """722 p_authorize.return_value = True723 code, resp = self.client.rule_get(self.gen_id())724 self.assertEqual(code, 404)725 self.verify_error_code(resp, "OE0002")726 code, resp = self.client.rule_update(self.gen_id(), {})727 self.assertEqual(code, 404)728 self.verify_error_code(resp, "OE0002")729 # TODO uncomment after OSB-412 is done730 # with self.switch_user(self.org2_user_id):731 # org2_rule = self._create_rule(732 # 'new_rule_name',733 # org_id=self.org2_id,734 # owner_id=self.org2_employee['id'],735 # budget_id=self.org2_budget_id)736 # org2_rule_id = org2_rule['id']737 # code, resp = self.client.rule_get(org2_rule_id)738 # self.assertEqual(code, 404)739 # self.verify_error_code(resp, "OE0002")740 # code, resp = self.client.rule_update(org2_rule_id, {})741 # self.assertEqual(code, 404)742 # self.verify_error_code(resp, "OE0002")743 @patch(AUTHORIZE_ACTION_METHOD)744 def test_update_with_change_rule_priority(self, p_authorize):745 """746 Test update rule with changing rule priority747 Steps:748 - 2. Create 3 rules749 - 3. Update rule3 and set priority 1750 - 5. Verify new rule order751 """752 p_authorize.return_value = True753 rules = self._get_rules()754 self.assertEqual(len(rules), 0)755 rule1 = self._create_rule('rule1')756 rule2 = self._create_rule('rule2')757 rule3 = self._create_rule('rule3')758 rules = self._get_rules()759 self.assertEqual(len(rules), 3)760 original_order = [r['id'] for r in rules]761 rule2['priority'] = 0762 code, resp = self.client.rule_update(rule2['id'], rule2)763 self.assertEqual(code, 400)764 self.verify_error_code(resp, 'OE0224')765 rule2['priority'] = 5766 code, resp = self.client.rule_update(rule2['id'], rule2)767 self.assertEqual(code, 400)768 self.verify_error_code(resp, 'OE0217')769 rule1['priority'] = 1770 code, _ = self.client.rule_update(rule1['id'], rule1)771 self.assertEqual(code, 200)772 rules = self._get_rules()773 self.assertEqual(len(rules), 3)774 expected_new_order = [original_order[2], original_order[0],775 original_order[1]]776 new_order = [r['id'] for r in rules]777 self.assertEqual(expected_new_order, new_order)778 @patch(AUTHORIZE_ACTION_METHOD)779 def test_update_without_pool_and_owner(self, p_authorize):780 """781 Test update rule without owner and pool specified782 Steps:783 - 1. Create full pair rule784 - 2. Update rule remove owner and pool785 - verify it's failed786 - verify localized code787 - 3. Update rule (remove owner)788 - 4. Update rule (remove pool)789 - verify it's failed790 - verify localized code791 """792 p_authorize.return_value = True793 rule = self._create_rule('test rule')794 rule_tmp = copy.deepcopy(rule)795 rule_tmp['pool_id'] = None796 rule_tmp['owner_id'] = None797 code, resp = self.client.rule_update(rule['id'], rule_tmp)798 self.assertEqual(code, 400)799 self.verify_error_code(resp, "OE0216")800 rule_tmp = copy.deepcopy(rule)801 rule_tmp['owner_id'] = None802 code, resp = self.client.rule_update(rule['id'], rule_tmp)803 self.assertEqual(code, 400)804 self.verify_error_code(resp, "OE0216")805 rule_tmp['pool_id'] = None806 code, resp = self.client.rule_update(rule['id'], rule_tmp)807 self.assertEqual(code, 400)808 self.verify_error_code(resp, "OE0216")809 @patch(AUTHORIZE_ACTION_METHOD)810 def test_update_remove_all_conditions(self, p_authorize):811 """812 Test update rule without conditions813 Steps:814 - 1. Create full pair rule815 - 2. Update rule without conditions(empty array)816 - verify it's failed817 - verify localized code is E0216818 """819 p_authorize.return_value = True820 rule = self._create_rule('test rule')821 rule['conditions'] = []822 code, resp = self.client.rule_update(rule['id'], rule)823 self.assertEqual(code, 400)824 self.verify_error_code(resp, "OE0216")825 @patch(AUTHORIZE_ACTION_METHOD)826 def test_update_invalid_conditions(self, p_authorize):827 """828 Test update rule with invalid conditions829 Steps:830 - 1. Create full pair rule831 - 2. Update rule with832 - not supported type in condition(E0430)833 - not supported key in condition(E0212)834 - with condition type is None(E0430)835 - with condition meta is None(E0216)836 - verify all requests are failed837 """838 p_authorize.return_value = True839 rule = self._create_rule('test rule')840 tmp_rule = copy.deepcopy(rule)841 tmp_rule['conditions'][0]['type'] = "unsupported_type"842 code, resp = self.client.rule_update(rule['id'], tmp_rule)843 self.assertEqual(code, 400)844 self.verify_error_code(resp, "OE0430")845 tmp_rule = copy.deepcopy(rule)846 tmp_rule['conditions'][0]['unsupported_key'] = "dummy"847 code, resp = self.client.rule_update(rule['id'], tmp_rule)848 self.assertEqual(code, 400)849 self.verify_error_code(resp, "OE0212")850 tmp_rule = copy.deepcopy(rule)851 tmp_rule['conditions'][0]["type"] = None852 code, resp = self.client.rule_update(rule['id'], tmp_rule)853 self.assertEqual(code, 400)854 self.verify_error_code(resp, "OE0430")855 tmp_rule = copy.deepcopy(rule)856 tmp_rule['conditions'][0]["meta_info"] = None857 code, resp = self.client.rule_update(rule['id'], tmp_rule)858 self.assertEqual(code, 400)859 self.verify_error_code(resp, "OE0216")860 @patch(AUTHORIZE_ACTION_METHOD)861 def test_update_invalid_target_pair(self, p_authorize):862 """863 Test update rule without owner and pool specified864 Steps:865 - 1. Create full pair rule866 - 2. Update rule with invalid target pair867 - verify it's failed868 - verify localized code is E0379869 """870 p_authorize.return_value = True871 rule = self._create_rule('test rule')872 p_authorize.return_value = False873 code, resp = self.client.rule_update(rule['id'], rule)874 self.assertEqual(code, 403)875 self.verify_error_code(resp, "OE0379")876 @patch(AUTHORIZE_ACTION_METHOD)877 def test_update_active_is_none(self, p_authorize):878 """879 Test update rule with invalid active field880 Steps:881 - 1. Create full pair rule882 - 2. Update rule with active is None883 - verify it's failed884 - verify localized code is E0212885 """886 p_authorize.return_value = True887 rule = self._create_rule('test rule')888 rule['active'] = None889 code, resp = self.client.rule_update(rule['id'], rule)890 self.assertEqual(code, 400)891 self.verify_error_code(resp, "OE0216")892 @patch(AUTHORIZE_ACTION_METHOD)893 def test_update_invalid_pool_or_owner_id(self, p_authorize):894 """895 Test update rule with specifying invalid owner id.896 Steps:897 - 1. Create full pair rule898 - 2. Update rule with random UUID for pool_id and owner_id899 - verify it's failed900 - verify localized code is E0002901 -3. Update rule with pool_id and owner_id from another org.902 """903 p_authorize.return_value = True904 rule = self._create_rule('test rule')905 for param in ['pool_id', 'owner_id']:906 rule_tmp = copy.deepcopy(rule)907 rule_tmp[param] = self.gen_id()908 code, resp = self.client.rule_update(rule['id'], rule_tmp)909 self.assertEqual(code, 400)910 self.verify_error_code(resp, "OE0005")911 rule_tmp = copy.deepcopy(rule)912 rule_tmp['pool_id'] = self.org2_pool_id913 code, resp = self.client.rule_update(rule['id'], rule_tmp)914 self.assertEqual(code, 400)915 self.verify_error_code(resp, "OE0005")916 rule_tmp = copy.deepcopy(rule)917 rule_tmp['owner_id'] = self.org2_employee['id']918 code, resp = self.client.rule_update(rule['id'], rule_tmp)919 self.assertEqual(code, 400)920 self.verify_error_code(resp, "OE0005")921 @patch(AUTHORIZE_ACTION_METHOD)922 def test_test_patch_rule_fields(self, p_authorize):923 """924 Test update rule with field by filed925 Steps:926 - 1. Create full pair rule927 - 2. Update all fields in rule one by one using only this key in body928 - 3. Verify all fields are updated929 """930 p_authorize.return_value = True931 rule = self._create_rule('test rule')932 new_rule = {933 'name': 'new_name',934 'pool_id': self.sub_pools[0]['id'],935 'owner_id': self.employee2['id'],936 'conditions':937 [{"type": "name_ends_with", "meta_info": "name_ends_with"}]938 }939 for key in new_rule.keys():940 body = {key: new_rule[key]}941 code, resp = self.client.rule_update(rule['id'], body)942 self.assertEqual(code, 200)943 code, rule = self.client.rule_get(rule['id'])944 self.assertEqual(code, 200)945 if key == 'conditions':946 for cond in rule[key]:947 cond.pop('id')948 self.assertEqual(rule[key], new_rule[key])949 @patch(AUTHORIZE_ACTION_METHOD)950 def test_delete_rule(self, p_authorize):951 """952 Test delete rule953 Steps:954 - 1. Create rule rule1955 - 2. Create rule rule2956 - 3. Create rule rule3957 - 4. Verify count of rules, should be 3958 - 5. Delete rules one by one and check count -1959 - 6. Try to delete rule that is doesn't exist960 """961 p_authorize.return_value = True962 rules = self._get_rules()963 self.assertEqual(len(rules), 0)964 rule1 = self._create_rule('rule1')965 rule2 = self._create_rule('rule2')966 rule3 = self._create_rule('rule3')967 rules = self._get_rules()968 self.assertEqual(len(rules), 3)969 code, _ = self.client.rule_delete(rule1['id'])970 self.assertEqual(code, 204)971 self.assertEqual(self.p_activities_publish.call_count, 4)972 rules = self._get_rules()973 self.assertEqual(len(rules), 2)974 code, _ = self.client.rule_delete(rule2['id'])975 self.assertEqual(code, 204)976 self.assertEqual(self.p_activities_publish.call_count, 5)977 rules = self._get_rules()978 self.assertEqual(len(rules), 1)979 code, _ = self.client.rule_delete(rule3['id'])980 self.assertEqual(code, 204)981 self.assertEqual(self.p_activities_publish.call_count, 6)982 rules = self._get_rules()983 self.assertEqual(len(rules), 0)984 code, resp = self.client.rule_delete(self.gen_id())985 self.assertEqual(code, 404)986 self.verify_error_code(resp, 'OE0002')987 @patch(AUTHORIZE_ACTION_METHOD)988 def test_delete_rule_priority_change(self, p_authorize):989 """990 Test delete rule991 Steps:992 - 1. Create 3 rules993 - 2. Verify count of rules, should be 3994 - 3. Delete middle value995 - 4. Check priority order updated and have no gaps996 """997 p_authorize.return_value = True998 rules = self._get_rules()999 self.assertEqual(len(rules), 0)1000 rule1 = self._create_rule('rule1')1001 rule2 = self._create_rule('rule2')1002 rule3 = self._create_rule('rule3')1003 rules = self._get_rules()1004 self.assertEqual(len(rules), 3)1005 code, _ = self.client.rule_delete(rule2['id'])1006 self.assertEqual(code, 204)1007 rules = self._get_rules()1008 self.assertEqual(len(rules), 2)1009 for i, rule in enumerate(rules, start=1):1010 self.assertEqual(rule['priority'], i)1011class TestApplyRuleApi(TestRulesApiBase):1012 def setUp(self, version='v2'):1013 super().setUp(version)1014 patch('rest_api_server.controllers.cloud_account.'1015 'CloudAccountController._configure_report').start()1016 patch('rest_api_server.handlers.v1.base.BaseAuthHandler.'1017 '_get_user_info', return_value=self.user).start()1018 aws_cloud_acc = {1019 'name': 'my cloud_acc',1020 'type': 'aws_cnr',1021 'config': {1022 'access_key_id': 'key',1023 'secret_access_key': 'secret',1024 'config_scheme': 'create_report'1025 }1026 }1027 _, self.cloud_acc = self.create_cloud_account(1028 self.org_id, aws_cloud_acc, auth_user_id=self.user_id)1029 _, cloud_rules = self.client.rules_list(self.org_id)1030 self.set_allowed_pair(self.user_id,1031 cloud_rules['rules'][0]['pool_id'])1032 self.cloud_acc_id = self.cloud_acc['id']1033 def _create_resources(self, names, cloud_resource_ids=[]):1034 valid_body = {'resources': []}1035 for i, name in enumerate(names):1036 cloud_resource_id = 'res_id_%s' % self.gen_id()1037 if i < len(cloud_resource_ids):1038 cloud_resource_id = cloud_resource_ids[i]1039 valid_body['resources'].append(1040 {1041 'cloud_resource_id': cloud_resource_id,1042 'name': name,1043 'resource_type': 'VM',1044 }1045 )1046 code, result = self.cloud_resource_create_bulk(1047 self.cloud_acc_id, valid_body, return_resources=True)1048 self.assertEqual(code, 200)1049 return result['resources']1050 def _verify_assignments(self, resources, expected_map):1051 self.assertEqual(len(resources), len(expected_map.keys()))1052 for resource in resources:1053 self.assertEqual(1054 resource.get('employee_id'),1055 expected_map[resource['name']]['employee_id'])1056 self.assertEqual(1057 resource.get('pool_id'),1058 expected_map[resource['name']]['pool_id'])1059 @patch(AUTHORIZE_ACTION_METHOD)1060 @patch(RULE_APPLY_CONDITION_CLOUD_IS)1061 def test_apply_rules_all_supported_conditions(self, p_rule_apply,1062 p_authorize):1063 """1064 Test checks all supported conditions in one rule1065 Steps:1066 - 1. Create full pair rule rule1 with:1067 - name_is1068 - name_starts_with1069 - name_ends_with1070 - name_contains1071 - 2. Create resources:1072 - one matched all conditions1073 - one doesn't match name_is1074 - 3. Verify the first resource is assigned, the second one is not.1075 """1076 p_authorize.return_value = True1077 p_rule_apply.return_value = False1078 conditions = [1079 {"type": "name_starts_with", "meta_info": "starts_"},1080 {"type": "name_ends_with", "meta_info": "_ends"},1081 {"type": "name_contains", "meta_info": "contains"},1082 {"type": "name_is", "meta_info": "starts_contains_ends"},1083 ]1084 self._create_rule('test_rule', conditions=conditions,1085 pool_id=self.sub_pools_ids[0],1086 owner_id=self.employee2['id'])1087 resources = self._create_resources(1088 ['starts_contains_ends', 'starts_contains_new_ends']1089 )1090 expected_map = {1091 'starts_contains_ends': {1092 'pool_id': self.sub_pools_ids[0],1093 'employee_id': self.employee2['id']1094 },1095 'starts_contains_new_ends': {1096 'pool_id': self.org_pool_id,1097 'employee_id': self.employee['id']1098 }1099 }1100 self._verify_assignments(resources, expected_map)1101 @patch(AUTHORIZE_ACTION_METHOD)1102 @patch(RULE_APPLY_CONDITION_CLOUD_IS)1103 def test_apply_rules_multiple_rules_multiple_resources(self, p_rule_apply,1104 p_authorize):1105 """1106 Test checks applying rules. Multiple rules, multiple resources1107 Steps:1108 - 1. Create multiple rules rule1, rule2, rule31109 - 2. Create resources:1110 - two resources matched rule11111 - two resources matched rule21112 - two resources matched rule31113 - two doesn't match any rules1114 - 3. Verify the resource assignments.1115 """1116 p_authorize.return_value = True1117 p_rule_apply.return_value = False1118 conditions = [1119 {"type": "name_starts_with", "meta_info": "starts_"},1120 ]1121 rule1 = self._create_rule('rule1', conditions=conditions,1122 pool_id=self.sub_pools_ids[0],1123 owner_id=self.employee['id'])1124 conditions = [1125 {"type": "name_ends_with", "meta_info": "_ends"},1126 ]1127 rule2 = self._create_rule('rule2', conditions=conditions,1128 pool_id=self.sub_pools_ids[1],1129 owner_id=self.employee2['id'])1130 conditions = [1131 {"type": "name_contains", "meta_info": "contains"},1132 ]1133 rule3 = self._create_rule('rule3', conditions=conditions,1134 pool_id=self.sub_pools_ids[2],1135 owner_id=self.employee3['id'])1136 res_map = {1137 'starts_name_1': 'cloud_resource_id_1',1138 'name_2': 'starts_/res_2/id',1139 'name_3_ends': 'cloud_resource_id_3',1140 'name_4': 'res_4/_ends',1141 'name_contains_5': 'cloud_resource_id_5',1142 'name_6': 'res_id/contains_6',1143 'name_7': 'res_7_id',1144 'name_8': 'res_8_id'1145 }1146 resources = self._create_resources(1147 names=list(res_map.keys()),1148 cloud_resource_ids=list(res_map.values())1149 )1150 expected_map = {1151 'starts_name_1': {1152 'pool_id': self.sub_pools_ids[0],1153 'employee_id': self.employee['id']1154 },1155 'name_2': {1156 'pool_id': self.sub_pools_ids[0],1157 'employee_id': self.employee['id']1158 },1159 'name_3_ends': {1160 'pool_id': self.sub_pools_ids[1],1161 'employee_id': self.employee2['id']1162 },1163 'name_4': {1164 'pool_id': self.sub_pools_ids[1],1165 'employee_id': self.employee2['id']1166 },1167 'name_contains_5': {1168 'pool_id': self.sub_pools_ids[2],1169 'employee_id': self.employee3['id']1170 },1171 'name_6': {1172 'pool_id': self.sub_pools_ids[2],1173 'employee_id': self.employee3['id']1174 },1175 'name_7': {1176 'pool_id': self.org_pool_id,1177 'employee_id': self.employee['id']1178 },1179 'name_8': {1180 'pool_id': self.org_pool_id,1181 'employee_id': self.employee['id']1182 },1183 }1184 self._verify_assignments(resources, expected_map)1185 applied_rules_map = {1186 'starts_name_1': [1187 {'id': rule1['id'], 'name': rule1['name'],1188 'pool_id': rule1['pool_id']}1189 ],1190 'name_2': [1191 {'id': rule1['id'], 'name': rule1['name'],1192 'pool_id': rule1['pool_id']}1193 ],1194 'name_3_ends': [1195 {'id': rule2['id'], 'name': rule2['name'],1196 'pool_id': rule2['pool_id']}1197 ],1198 'name_4': [1199 {'id': rule2['id'], 'name': rule2['name'],1200 'pool_id': rule2['pool_id']}1201 ],1202 'name_contains_5': [1203 {'id': rule3['id'], 'name': rule3['name'],1204 'pool_id': rule3['pool_id']}1205 ],1206 'name_6': [1207 {'id': rule3['id'], 'name': rule3['name'],1208 'pool_id': rule3['pool_id']}1209 ],1210 'name_7': None,1211 'name_8': None1212 }1213 for r in resources:1214 name = r.get('name')1215 applied_rules = applied_rules_map[name]1216 self.assertEqual(applied_rules, r.get('applied_rules'))1217 if applied_rules:1218 self.assertEqual(1219 r.get('pool_id'), applied_rules[0]['pool_id'])1220 @patch(AUTHORIZE_ACTION_METHOD)1221 def test_apply_rules_rules_priority(self, p_authorize):1222 """1223 Test checks applying rules by priority.1224 Steps:1225 - 1. Create rule1 and rule2 with conflicted conditions1226 - 2. Create 4 resources matched both rules1227 - 3. Verify the resource assignments.1228 """1229 p_authorize.return_value = True1230 conditions = [1231 {"type": "name_starts_with", "meta_info": "rule"},1232 ]1233 self._create_rule('rule1', conditions=conditions,1234 pool_id=self.sub_pools_ids[0],1235 owner_id=self.employee2['id'])1236 self._create_rule('rule2', conditions=conditions,1237 pool_id=self.sub_pools_ids[1],1238 owner_id=self.employee3['id'])1239 resources = self._create_resources(1240 [1241 'rule1',1242 'rule2',1243 'rule3',1244 'rule4',1245 ]1246 )1247 expected_map = {1248 'rule1': {1249 'pool_id': self.sub_pools_ids[1],1250 'employee_id': self.employee3['id']1251 },1252 'rule2': {1253 'pool_id': self.sub_pools_ids[1],1254 'employee_id': self.employee3['id']1255 },1256 'rule3': {1257 'pool_id': self.sub_pools_ids[1],1258 'employee_id': self.employee3['id']1259 },1260 'rule4': {1261 'pool_id': self.sub_pools_ids[1],1262 'employee_id': self.employee3['id']1263 }1264 }1265 self._verify_assignments(resources, expected_map)1266 @patch(AUTHORIZE_ACTION_METHOD)1267 @patch(RULE_APPLY_CONDITION_CLOUD_IS)1268 def test_apply_rules_no_rules(self, p_rule_apply, p_authorize):1269 """1270 Test checks creation resources in case of no rules are available in org1271 Steps:1272 - 1. Create 4 resources.1273 - 2. Verify root assignments.1274 - 3. Create 4 rules1275 - 4. Create resource that doesn't match any rules.1276 - 5. Verify root assignments.1277 """1278 p_authorize.return_value = True1279 p_rule_apply.return_value = False1280 resources = self._create_resources(1281 [1282 'rule1',1283 'rule2',1284 'rule3',1285 'rule4',1286 ]1287 )1288 expected_map = {1289 'rule1': {1290 'pool_id': self.org_pool_id,1291 'employee_id': self.employee['id']1292 },1293 'rule2': {1294 'pool_id': self.org_pool_id,1295 'employee_id': self.employee['id']1296 },1297 'rule3': {1298 'pool_id': self.org_pool_id,1299 'employee_id': self.employee['id']1300 },1301 'rule4': {1302 'pool_id': self.org_pool_id,1303 'employee_id': self.employee['id']1304 }1305 }1306 self._verify_assignments(resources, expected_map)1307 conditions = [1308 {"type": "name_starts_with", "meta_info": "rule"},1309 ]1310 self._create_rule('rule1', conditions=conditions,1311 pool_id=self.sub_pools_ids[0],1312 owner_id=self.employee2['id'])1313 self._create_rule('rule2', conditions=conditions,1314 pool_id=self.sub_pools_ids[1],1315 owner_id=self.employee3['id'])1316 resources = self._create_resources(1317 [1318 '_rule1',1319 '_rule2',1320 '_rule3',1321 '_rule4',1322 ]1323 )1324 expected_map = {1325 '_rule1': {1326 'pool_id': self.org_pool_id,1327 'employee_id': self.employee['id']1328 },1329 '_rule2': {1330 'pool_id': self.org_pool_id,1331 'employee_id': self.employee['id']1332 },1333 '_rule3': {1334 'pool_id': self.org_pool_id,1335 'employee_id': self.employee['id']1336 },1337 '_rule4': {1338 'pool_id': self.org_pool_id,1339 'employee_id': self.employee['id']1340 }1341 }1342 self._verify_assignments(resources, expected_map)1343 @patch(AUTHORIZE_ACTION_METHOD)1344 @patch(RULE_APPLY_CONDITION_CLOUD_IS)1345 def test_apply_rules_disabled_rules(self, p_rule_apply, p_authorize):1346 """1347 Test checks creation resources in case disabled rules1348 Steps:1349 - 1. Create 2 rules rule1 and rule2.1350 - 2. Create resource matched both rules1351 - 3. Verify rule1 is applied1352 - 4. Disable rule11353 - 5. Create resource matched both rules1354 - 3. Verify rule2 is applied1355 - 4. Disable rule21356 - 5. Create resource matched both rules1357 - 3. Verify no rules are applied1358 """1359 p_authorize.return_value = True1360 p_rule_apply.return_value = False1361 conditions = [1362 {"type": "name_starts_with", "meta_info": "rule"},1363 ]1364 rule1 = self._create_rule('rule1', conditions=conditions,1365 pool_id=self.sub_pools_ids[0],1366 owner_id=self.employee2['id'])1367 rule2 = self._create_rule('rule2', conditions=conditions,1368 pool_id=self.sub_pools_ids[1],1369 owner_id=self.employee3['id'])1370 resources = self._create_resources(1371 [1372 'rule2'1373 ]1374 )1375 expected_map = {1376 'rule2': {1377 'pool_id': self.sub_pools_ids[1],1378 'employee_id': self.employee3['id']1379 }1380 }1381 self._verify_assignments(resources, expected_map)1382 code, resp = self.client.rule_update(rule2['id'], {'active': False})1383 self.assertEqual(code, 200)1384 resources = self._create_resources(1385 [1386 'rule1'1387 ]1388 )1389 expected_map = {1390 'rule1': {1391 'pool_id': self.sub_pools_ids[0],1392 'employee_id': self.employee2['id']1393 }1394 }1395 self._verify_assignments(resources, expected_map)1396 code, resp = self.client.rule_update(rule1['id'], {'active': False})1397 self.assertEqual(code, 200)1398 resources = self._create_resources(1399 [1400 'rule3'1401 ]1402 )1403 expected_map = {1404 'rule3': {1405 'pool_id': self.org_pool_id,1406 'employee_id': self.employee['id']1407 }1408 }1409 self._verify_assignments(resources, expected_map)1410 @patch(AUTHORIZE_ACTION_METHOD)1411 def test_apply_rules_complex(self, p_authorize):1412 """1413 Test checks creation resources in case complex rules1414 Steps:1415 - 1. Create multiple rules with multiple condition types.1416 - 2. Create resource one resource match only one rule.1417 - 3. Verify assignments.1418 """1419 p_authorize.return_value = True1420 conditions = [1421 {"type": "name_starts_with", "meta_info": "qa"},1422 {"type": "name_ends_with", "meta_info": "prod"},1423 {"type": "name_contains", "meta_info": "resource"},1424 ]1425 self._create_rule('full_pair_rule1', conditions=conditions,1426 pool_id=self.sub_pools_ids[0],1427 owner_id=self.employee2['id'])1428 conditions = [1429 {"type": "name_is", "meta_info": "special_name"},1430 ]1431 self._create_rule('full_pair_rule2', conditions=conditions,1432 pool_id=self.sub_pools_ids[1],)1433 resources = self._create_resources(1434 [1435 'qa_resource_prod',1436 'qa_resource1_prod',1437 'special_name',1438 ]1439 )1440 expected_map = {1441 'qa_resource_prod': {1442 'pool_id': self.sub_pools_ids[0],1443 'employee_id': self.employee2['id']1444 },1445 'qa_resource1_prod': {1446 'pool_id': self.sub_pools_ids[0],1447 'employee_id': self.employee2['id']1448 },1449 'special_name': {1450 'pool_id': self.sub_pools_ids[1],1451 'employee_id': self.employee['id']1452 },1453 }1454 self._verify_assignments(resources, expected_map)1455 @patch(AUTHORIZE_ACTION_METHOD)1456 def test_apply_rules_tag_condition(self, p_authorize):1457 """1458 Test checks creation resources in case of tag condition in rule1459 Steps:1460 - 1. Create rule with tag_is condition rule11461 - 2. Create resources with tags matched rule11462 - 3. Verify assignment.1463 """1464 p_authorize.return_value = True1465 conditions = [1466 {1467 "type": "tag_is",1468 "meta_info": "{\"key\":\"key\",\"value\":\"value\"}"1469 }1470 ]1471 self._create_rule('rule1', conditions=conditions)1472 valid_body = {'resources': []}1473 valid_body['resources'].append(1474 {1475 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1476 'name': 'resource',1477 'resource_type': 'VM',1478 'tags': {1479 "key": "value",1480 "key1": "value1",1481 "key2": "value2",1482 }1483 }1484 )1485 valid_body['resources'].append(1486 {1487 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1488 'name': 'resource1',1489 'resource_type': 'VM',1490 'tags': {1491 "key1": "value1",1492 "key": "value",1493 "key2": "value2",1494 }1495 }1496 )1497 valid_body['resources'].append(1498 {1499 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1500 'name': 'resource2',1501 'resource_type': 'VM',1502 'tags': {1503 "key1": "value1",1504 "key2": "value2",1505 "key": "value"1506 }1507 }1508 )1509 code, result = self.cloud_resource_create_bulk(1510 self.cloud_acc_id, valid_body, return_resources=True)1511 self.assertEqual(code, 200)1512 expected_map = {1513 'resource': {1514 'pool_id': self.org_pool_id,1515 'employee_id': self.employee['id']1516 },1517 'resource1': {1518 'pool_id': self.org_pool_id,1519 'employee_id': self.employee['id']1520 },1521 'resource2': {1522 'pool_id': self.org_pool_id,1523 'employee_id': self.employee['id']1524 },1525 }1526 self._verify_assignments(result['resources'], expected_map)1527 @patch(AUTHORIZE_ACTION_METHOD)1528 def test_apply_rules_tag_exists_condition(self, p_authorize):1529 """1530 Test checks creation resources in case of tag exists condition in rule1531 Steps:1532 - 1. Create rule with tag_exists condition rule11533 - 2. Create resources with tags matched rule11534 - 3. Verify assignment.1535 """1536 p_authorize.return_value = True1537 conditions = [1538 {"type": "tag_exists", "meta_info": "key"}1539 ]1540 self._create_rule('rule1', conditions=conditions)1541 valid_body = {'resources': []}1542 valid_body['resources'].append(1543 {1544 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1545 'name': 'resource',1546 'resource_type': 'VM',1547 'tags': {1548 "key": "value",1549 "key1": "value1",1550 "key2": "value2",1551 }1552 }1553 )1554 valid_body['resources'].append(1555 {1556 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1557 'name': 'resource1',1558 'resource_type': 'VM',1559 'tags': {1560 "key1": "value1",1561 "key": "value",1562 "key2": "value2",1563 }1564 }1565 )1566 valid_body['resources'].append(1567 {1568 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1569 'name': 'resource2',1570 'resource_type': 'VM',1571 'tags': {1572 "key1": "value1",1573 "key2": "value2",1574 "key": "value"1575 }1576 }1577 )1578 valid_body['resources'].append(1579 {1580 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1581 'name': 'resource3',1582 'resource_type': 'VM',1583 'tags': {1584 "key1": "value1",1585 "key2": "value2",1586 "key": "value3"1587 }1588 }1589 )1590 code, result = self.cloud_resource_create_bulk(1591 self.cloud_acc_id, valid_body, return_resources=True)1592 self.assertEqual(code, 200)1593 expected_map = {1594 'resource': {1595 'pool_id': self.org_pool_id,1596 'employee_id': self.employee['id']1597 },1598 'resource1': {1599 'pool_id': self.org_pool_id,1600 'employee_id': self.employee['id']1601 },1602 'resource2': {1603 'pool_id': self.org_pool_id,1604 'employee_id': self.employee['id']1605 },1606 'resource3': {1607 'pool_id': self.org_pool_id,1608 'employee_id': self.employee['id']1609 },1610 }1611 self._verify_assignments(result['resources'], expected_map)1612 @patch(AUTHORIZE_ACTION_METHOD)1613 def test_apply_rules_environment_resources(self, p_authorize):1614 p_authorize.return_value = True1615 name = 'some_name'1616 conditions = [1617 {"type": "name_is", "meta_info": name}1618 ]1619 self._create_rule('rule1', conditions=conditions)1620 code, res = self.environment_resource_create(1621 self.org_id, {'name': name, 'resource_type': 'type'})1622 self.assertEqual(code, 201)1623 expected_map = {1624 name: {1625 'pool_id': self.org_pool_id,1626 'employee_id': self.employee['id']1627 }1628 }1629 self._verify_assignments([res], expected_map)1630 @patch(AUTHORIZE_ACTION_METHOD)1631 def test_apply_rules_cluster(self, p_authorize):1632 p_authorize.return_value = True1633 conditions = [1634 {"type": "tag_exists", "meta_info": "key"}1635 ]1636 self._create_rule('rule1', conditions=conditions)1637 valid_body = {'resources': []}1638 valid_body['resources'].extend([1639 {1640 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1641 'name': 'resource',1642 'resource_type': 'VM',1643 'tags': {1644 "key": "value",1645 }1646 },1647 {1648 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1649 'name': 'resource1',1650 'resource_type': 'VM',1651 'tags': {1652 "key": "value",1653 }1654 }])1655 code, ct = self.client.cluster_type_create(1656 self.org_id, {'name': 'some', 'tag_key': 'key'})1657 self.assertEqual(code, 201)1658 code, result = self.cloud_resource_create_bulk(1659 self.cloud_acc_id, valid_body, return_resources=True,1660 behavior='update_existing')1661 self.assertEqual(code, 200)1662 expected_map = {1663 'resource': {1664 'pool_id': self.org_pool_id,1665 'employee_id': self.employee['id']1666 },1667 'resource1': {1668 'pool_id': self.org_pool_id,1669 'employee_id': self.employee['id']1670 }1671 }1672 cluster = next(self.resources_collection.find(1673 {'cluster_type_id': ct['id']}))1674 self._verify_assignments(result['resources'], expected_map)1675 self.assertEqual(cluster.get('pool_id'), self.org_pool_id)1676 @patch(AUTHORIZE_ACTION_METHOD)1677 def test_apply_rules_tag_value_starts_condition(self, p_authorize):1678 """1679 Test checks creation resources in case of tag value starts with1680 condition in rule1681 Steps:1682 - 1. Create rule with tag_value_starts_with condition rule11683 - 2. Create resources with tags matched rule11684 - 3. Verify assignment.1685 """1686 p_authorize.return_value = True1687 conditions = [1688 {1689 "type": "tag_value_starts_with",1690 "meta_info": "{\"key\":\"key\",\"value\":\"value\"}"1691 }1692 ]1693 self._create_rule('rule1', conditions=conditions)1694 valid_body = {'resources': []}1695 valid_body['resources'].append(1696 {1697 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1698 'name': 'resource',1699 'resource_type': 'VM',1700 'tags': {1701 "key": "value",1702 "key1": "value1",1703 "key2": "value2",1704 }1705 }1706 )1707 valid_body['resources'].append(1708 {1709 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1710 'name': 'resource1',1711 'resource_type': 'VM',1712 'tags': {1713 "key1": "value1",1714 "key": "value value",1715 "key2": "value2",1716 }1717 }1718 )1719 valid_body['resources'].append(1720 {1721 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1722 'name': 'resource2',1723 'resource_type': 'VM',1724 'tags': {1725 "key1": "value1",1726 "key2": "value2",1727 "key": "value vbivalue"1728 }1729 }1730 )1731 valid_body['resources'].append(1732 {1733 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1734 'name': 'resource3',1735 'resource_type': 'VM',1736 'tags': {1737 "key": "valueue",1738 "key1": "value1",1739 "key2": "value2",1740 }1741 }1742 )1743 code, result = self.cloud_resource_create_bulk(1744 self.cloud_acc_id, valid_body, return_resources=True)1745 self.assertEqual(code, 200)1746 expected_map = {1747 'resource': {1748 'pool_id': self.org_pool_id,1749 'employee_id': self.employee['id']1750 },1751 'resource1': {1752 'pool_id': self.org_pool_id,1753 'employee_id': self.employee['id']1754 },1755 'resource2': {1756 'pool_id': self.org_pool_id,1757 'employee_id': self.employee['id']1758 },1759 'resource3': {1760 'pool_id': self.org_pool_id,1761 'employee_id': self.employee['id']1762 },1763 }1764 self._verify_assignments(result['resources'], expected_map)1765 @patch(AUTHORIZE_ACTION_METHOD)1766 @patch(RULE_APPLY_CONDITION_CLOUD_IS)1767 def test_apply_rules_resource_type(self, p_rule_apply, p_authorize):1768 """1769 Test checks creation resources in case of res type condition in rule1770 Steps:1771 - 1. Create rule with resource_type_is condition rule11772 - 2. Create resource with type matched rule1 and two more resource1773 which are not1774 - 3. Verify assignment.1775 """1776 p_authorize.return_value = True1777 p_rule_apply.return_value = False1778 conditions = [1779 {"type": "resource_type_is", "meta_info": "VM"}1780 ]1781 self._create_rule('rule1', conditions=conditions)1782 valid_body = {'resources': []}1783 valid_body['resources'].append(1784 {1785 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1786 'name': 'resource',1787 'resource_type': 'VM'1788 }1789 )1790 valid_body['resources'].append(1791 {1792 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1793 'name': 'resource1',1794 'resource_type': 'VM1'1795 }1796 )1797 valid_body['resources'].append(1798 {1799 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1800 'name': 'resource2',1801 'resource_type': 'vm'1802 }1803 )1804 code, result = self.cloud_resource_create_bulk(1805 self.cloud_acc_id, valid_body, return_resources=True)1806 self.assertEqual(code, 200)1807 expected_map = {1808 'resource': {1809 'pool_id': self.org_pool_id,1810 'employee_id': self.employee['id']1811 },1812 'resource1': {1813 'pool_id': self.org_pool_id,1814 'employee_id': self.employee['id']1815 },1816 'resource2': {1817 'pool_id': self.org_pool_id,1818 'employee_id': self.employee['id']1819 },1820 }1821 self._verify_assignments(result['resources'], expected_map)1822 @patch(AUTHORIZE_ACTION_METHOD)1823 @patch(RULE_APPLY_CONDITION_CLOUD_IS)1824 def test_apply_rules_region(self, p_rule_apply, p_authorize):1825 """1826 Test checks creation resources in case of res region condition in rule1827 Steps:1828 - 1. Create rule with region_is condition rule11829 - 2. Create resource with type matched rule1 and two more resource1830 which are not1831 - 3. Verify assignment.1832 """1833 p_authorize.return_value = True1834 p_rule_apply.return_value = False1835 conditions = [1836 {"type": "region_is", "meta_info": "test_region"}1837 ]1838 self._create_rule('rule1', conditions=conditions,1839 pool_id=self.sub_pools_ids[0],1840 owner_id=self.employee2['id'])1841 valid_body = {'resources': []}1842 valid_body['resources'].append(1843 {1844 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1845 'name': 'resource',1846 'region': 'test_region1',1847 'resource_type': 'VM'1848 }1849 )1850 # matched resource1851 valid_body['resources'].append(1852 {1853 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1854 'name': 'resource1',1855 'region': 'test_region',1856 'resource_type': 'VM1'1857 }1858 )1859 valid_body['resources'].append(1860 {1861 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1862 'name': 'resource2',1863 'region': 'test_region3',1864 'resource_type': 'vm'1865 }1866 )1867 code, result = self.cloud_resource_create_bulk(1868 self.cloud_acc_id, valid_body, return_resources=True)1869 self.assertEqual(code, 200)1870 expected_map = {1871 'resource': {1872 'pool_id': self.org_pool_id,1873 'employee_id': self.employee['id']1874 },1875 'resource1': {1876 'pool_id': self.sub_pools_ids[0],1877 'employee_id': self.employee2['id']1878 },1879 'resource2': {1880 'pool_id': self.org_pool_id,1881 'employee_id': self.employee['id']1882 },1883 }1884 self._verify_assignments(result['resources'], expected_map)1885 @patch(AUTHORIZE_ACTION_METHOD)1886 def test_apply_rules_cloud_is(self, p_authorize):1887 """1888 Test checks creation resources in case of cloud condition in rule1889 Steps:1890 - 1. Create rule with cloud_is condition rule11891 - 2. Create resources for right cloud1892 - 3. Verify assignment.1893 - 4. Create resources for wrong cloud1894 - 5. Verify assignment.1895 - 6. Verify that get rules returns related cloud acc entities1896 """1897 p_authorize.return_value = True1898 auth_user_id_1 = self.gen_id()1899 _, employee1 = self.client.employee_create(1900 self.org_id, {'name': 'name1', 'auth_user_id': auth_user_id_1})1901 conditions = [1902 {"type": "cloud_is", "meta_info": self.cloud_acc_id}1903 ]1904 self._create_rule('rule1', conditions=conditions,1905 pool_id=self.sub_pools_ids[0],1906 owner_id=employee1['id'])1907 valid_body = {'resources': []}1908 valid_body['resources'].append(1909 {1910 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1911 'name': 'resource',1912 'resource_type': 'VM'1913 }1914 )1915 valid_body['resources'].append(1916 {1917 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1918 'name': 'resource1',1919 'resource_type': 'VM1'1920 }1921 )1922 valid_body['resources'].append(1923 {1924 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1925 'name': 'resource2',1926 'resource_type': 'vm'1927 }1928 )1929 code, result = self.cloud_resource_create_bulk(1930 self.cloud_acc_id, valid_body, return_resources=True)1931 self.assertEqual(code, 200)1932 expected_map = {1933 'resource': {1934 'pool_id': self.sub_pools_ids[0],1935 'employee_id': employee1['id']1936 },1937 'resource1': {1938 'pool_id': self.sub_pools_ids[0],1939 'employee_id': employee1['id']1940 },1941 'resource2': {1942 'pool_id': self.sub_pools_ids[0],1943 'employee_id': employee1['id']1944 },1945 }1946 self._verify_assignments(result['resources'], expected_map)1947 aws_cloud_acc1 = {1948 'name': 'my cloud_acc1',1949 'type': 'aws_cnr',1950 'config': {1951 'access_key_id': 'key',1952 'secret_access_key': 'secret',1953 'config_scheme': 'create_report'1954 }1955 }1956 _, cloud_acc1 = self.create_cloud_account(1957 self.org_id, aws_cloud_acc1, auth_user_id=auth_user_id_1)1958 cloud_acc_id = cloud_acc1['id']1959 valid_body = {'resources': []}1960 valid_body['resources'].append(1961 {1962 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1963 'name': 'resource',1964 'resource_type': 'VM'1965 }1966 )1967 valid_body['resources'].append(1968 {1969 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1970 'name': 'resource1',1971 'resource_type': 'VM1'1972 }1973 )1974 valid_body['resources'].append(1975 {1976 'cloud_resource_id': 'res_id_%s' % self.gen_id(),1977 'name': 'resource2',1978 'resource_type': 'vm'1979 }1980 )1981 code, result = self.cloud_resource_create_bulk(1982 cloud_acc_id, valid_body, return_resources=True)1983 self.assertEqual(code, 200)1984 expected_map = {1985 'resource': {1986 'pool_id': self.org_pool_id,1987 'employee_id': self.employee['id']1988 },1989 'resource1': {1990 'pool_id': self.org_pool_id,1991 'employee_id': self.employee['id']1992 },1993 'resource2': {1994 'pool_id': self.org_pool_id,1995 'employee_id': self.employee['id']1996 },1997 }1998 self._verify_assignments(result['resources'], expected_map)1999 code, rules = self.client.rules_list(2000 organization_id=self.org_id)2001 self.assertEqual(code, 200)2002 self.assertIn(self.cloud_acc_id, rules['entities'])2003 @patch(AUTHORIZE_ACTION_METHOD)2004 def test_apply_rules_with_deleted_condition(self, p_authorize):2005 """2006 Test checks that deleted conditions don't apply on filtering.2007 Steps:2008 - 1. Create rule with one condition.2009 - 2. Modify rule. Add one new condition, remove existing one.2010 - 3. Create resource matches to the new condition,2011 but doesn't match to the deleted one.2012 - 3. Verify assignments.2013 """2014 p_authorize.return_value = True2015 conditions = [2016 {"type": "name_starts_with", "meta_info": "qa"}2017 ]2018 rule = self._create_rule('full_pair_rule1', conditions=conditions)2019 conditions = [2020 {"type": "name_starts_with", "meta_info": "prod"}2021 ]2022 code, patched_response = self.client.rule_update(2023 rule['id'],2024 {'conditions': conditions}2025 )2026 resources = self._create_resources(2027 [2028 'prod_resource_prod',2029 ]2030 )2031 expected_map = {2032 'prod_resource_prod': {2033 'pool_id': self.org_pool_id,2034 'employee_id': self.employee['id']2035 }2036 }2037 self._verify_assignments(resources, expected_map)2038 @patch(AUTHORIZE_ACTION_METHOD)2039 def test_rule_prioritize(self, p_authorize):2040 p_authorize.return_value = True2041 for rule_name in ['rule1', 'rule2', 'rule3', 'rule4']:2042 conditions = [{"type": "name_starts_with", "meta_info": rule_name}]2043 self._create_rule(rule_name, conditions=conditions)2044 code, rules = self.client.rules_list(self.org_id)2045 self.assertEqual(code, 200)2046 initial_rule_order = [r['id'] for r in rules['rules']]2047 # prioritize third rule2048 expected_new_order = [initial_rule_order[2], initial_rule_order[0],2049 initial_rule_order[1], initial_rule_order[3],2050 initial_rule_order[4]]2051 code, response = self.client.rule_prioritize(initial_rule_order[2])2052 self.assertEqual(code, 200)2053 new_order = [r['id'] for r in response['rules']]2054 self.assertEqual(expected_new_order, new_order)2055 code, list_response = self.client.rules_list(self.org_id)2056 self.assertEqual(code, 200)2057 self.assertEqual(list_response, response)2058 @patch(AUTHORIZE_ACTION_METHOD)2059 def test_rule_prioritize_first(self, p_authorize):2060 p_authorize.return_value = True2061 for rule_name in ['rule1', 'rule2', 'rule3', 'rule4']:2062 conditions = [{"type": "name_starts_with", "meta_info": rule_name}]2063 self._create_rule(rule_name, conditions=conditions)2064 code, rules = self.client.rules_list(self.org_id)2065 self.assertEqual(code, 200)2066 initial_rule_order = [r['id'] for r in rules['rules']]2067 code, response = self.client.rule_prioritize(initial_rule_order[0])2068 self.assertEqual(code, 200)2069 new_order = [r['id'] for r in response['rules']]2070 self.assertEqual(initial_rule_order, new_order)2071 @patch(AUTHORIZE_ACTION_METHOD)2072 def test_rule_promote(self, p_authorize):2073 p_authorize.return_value = True2074 for rule_name in ['rule1', 'rule2', 'rule3', 'rule4']:2075 conditions = [{"type": "name_starts_with", "meta_info": rule_name}]2076 self._create_rule(rule_name, conditions=conditions)2077 code, rules = self.client.rules_list(self.org_id)2078 self.assertEqual(code, 200)2079 initial_rule_order = [r['id'] for r in rules['rules']]2080 # promote fourth rule2081 expected_new_order = [initial_rule_order[0], initial_rule_order[1],2082 initial_rule_order[3], initial_rule_order[2], initial_rule_order[4]]2083 code, resp = self.client.rule_promote(initial_rule_order[0])2084 self.assertEqual(code, 200)2085 code, response = self.client.rule_promote(initial_rule_order[3])2086 self.assertEqual(code, 200)2087 new_order = [r['id'] for r in response['rules']]2088 self.assertEqual(expected_new_order, new_order)2089 @patch(AUTHORIZE_ACTION_METHOD)2090 def test_rule_demote(self, p_authorize):2091 p_authorize.return_value = True2092 for rule_name in ['rule1', 'rule2', 'rule3', 'rule4']:2093 conditions = [{"type": "name_starts_with", "meta_info": rule_name}]2094 self._create_rule(rule_name, conditions=conditions)2095 code, rules = self.client.rules_list(self.org_id)2096 self.assertEqual(code, 200)2097 initial_rule_order = [r['id'] for r in rules['rules']]2098 # demote first rule2099 expected_new_order = [initial_rule_order[1], initial_rule_order[0],2100 initial_rule_order[2], initial_rule_order[4],2101 initial_rule_order[3]]2102 code, resp = self.client.rule_demote(initial_rule_order[3])2103 self.assertEqual(code, 200)2104 code, response = self.client.rule_demote(initial_rule_order[0])2105 self.assertEqual(code, 200)2106 new_order = [r['id'] for r in response['rules']]2107 self.assertEqual(expected_new_order, new_order)2108 @patch(AUTHORIZE_ACTION_METHOD)2109 def test_rule_deprioritize(self, p_authorize):2110 p_authorize.return_value = True2111 for rule_name in ['rule1', 'rule2', 'rule3', 'rule4']:2112 conditions = [{"type": "name_starts_with", "meta_info": rule_name}]2113 self._create_rule(rule_name, conditions=conditions)2114 code, rules = self.client.rules_list(self.org_id)2115 self.assertEqual(code, 200)2116 initial_rule_order = [r['id'] for r in rules['rules']]2117 # deprioritize second rule2118 expected_new_order = [initial_rule_order[0], initial_rule_order[2],2119 initial_rule_order[3], initial_rule_order[4],2120 initial_rule_order[1]]2121 code, response = self.client.rule_deprioritize(initial_rule_order[1])2122 self.assertEqual(code, 200)2123 new_order = [r['id'] for r in response['rules']]2124 self.assertEqual(expected_new_order, new_order)2125 @patch(AUTHORIZE_ACTION_METHOD)2126 @patch(RULE_APPLY_CONDITION_CLOUD_IS)2127 def test_invalid_rule_deactivated(self, p_rule_apply, p_authorize):2128 p_authorize.return_value = True2129 p_rule_apply.return_value = False2130 m_activities_publish = patch(2131 'rest_api_server.controllers.base.'2132 'BaseController.publish_activities_task').start()2133 conditions = [2134 {"type": "name_starts_with", "meta_info": "prod"}2135 ]2136 rule = self._create_rule('rule1', conditions=conditions,2137 pool_id=self.sub_pools_ids[2],2138 owner_id=self.employee3['id'],2139 set_allowed=False)2140 resources = self._create_resources(['prod_resource_prod'])2141 code, rule = self.client.rule_get(rule['id'])2142 self.assertEqual(code, 200)2143 self.assertFalse(rule['active'])2144 m_activities_publish.assert_has_calls([2145 call(rule['organization_id'], rule['id'], 'rule',2146 'rule_deactivated', ANY, 'rule.rule_deactivated',2147 add_token=True)])2148class TestRulesApplyApi(TestRulesApiBase):2149 def setUp(self, version='v2'):2150 super().setUp()2151 self._mock_auth_user(self.user_id)2152 patch('rest_api_server.handlers.v1.base.BaseAuthHandler._get_user_info',2153 return_value=self.user).start()2154 patch(AUTHORIZE_ACTION_METHOD, return_value=True).start()2155 patch('rest_api_server.controllers.cloud_account.'2156 'CloudAccountController._configure_report').start()2157 aws_cloud_acc = {2158 'name': 'my cloud_acc',2159 'type': 'aws_cnr',2160 'config': {2161 'access_key_id': 'key',2162 'secret_access_key': 'secret',2163 'config_scheme': 'create_report'2164 }2165 }2166 _, self.cloud_acc = self.create_cloud_account(2167 self.org_id, aws_cloud_acc, auth_user_id=self.user_id)2168 _, cloud_acc_rule = self.client.rules_list(self.org_id)2169 self.cloud_acc_rule = cloud_acc_rule['rules'][0]2170 self.cloud_acc_id = self.cloud_acc['id']2171 def _create_resources(self, names, pool_id=None, employee_id=None,2172 tags=None):2173 valid_body = {'resources': []}2174 for name in names:2175 resource_dict = {2176 'cloud_resource_id': 'res_id_%s' % self.gen_id(),2177 'name': name,2178 'resource_type': 'VM',2179 }2180 if not pool_id:2181 pool_id = self.org_pool_id2182 if not employee_id:2183 employee_id = self.employee['id']2184 resource_dict['pool_id'] = pool_id2185 resource_dict['employee_id'] = employee_id2186 if tags:2187 resource_dict['tags'] = tags2188 valid_body['resources'].append(resource_dict)2189 code, result = self.cloud_resource_create_bulk(2190 self.cloud_acc_id, valid_body, return_resources=True)2191 self.assertEqual(code, 200)2192 return result['resources']2193 def test_rules_apply_api_invalid_params(self):2194 code, resp = self.client.rules_apply(self.org_id, pool_id=1)2195 self.assertEqual(code, 400)2196 self.verify_error_code(resp, 'OE0214')2197 code, resp = self.client.rules_apply(self.org_id, pool_id=' ')2198 self.assertEqual(code, 400)2199 self.verify_error_code(resp, 'OE0416')2200 code, resp = self.client.rules_apply(self.org_id, pool_id='')2201 self.assertEqual(code, 400)2202 self.verify_error_code(resp, 'OE0215')2203 code, resp = self.client.rules_apply(self.org_id, pool_id='a' * 500)2204 self.assertEqual(code, 400)2205 self.verify_error_code(resp, 'OE0215')2206 code, resp = self.client.rules_apply(2207 self.org_id, pool_id=self.org_pool_id, include_children='a')2208 self.assertEqual(code, 400)2209 self.verify_error_code(resp, 'OE0226')2210 body_unexpected = {'pool_id': self.org_pool_id, 'aa': 'bb'}2211 code, resp = self.client.post(self.client.rules_apply_url(self.org_id),2212 body_unexpected)2213 self.assertEqual(code, 400)2214 self.verify_error_code(resp, 'OE0212')2215 def verify_assignment(self, resource_id, pool_id, employee_id):2216 code, resource = self.client.cloud_resource_get(resource_id)2217 self.assertEqual(code, 200)2218 self.assertEqual(resource['pool_id'], pool_id)2219 self.assertEqual(resource['employee_id'], employee_id)2220 def test_rules_apply_api_include_children(self):2221 sub_pool_id = self.sub_pools[0]['id']2222 code, sub_sub_pool = self.client.pool_create(2223 self.org_id,2224 {'name': "sub_sub_pool",2225 'parent_id': sub_pool_id})2226 self.assertEqual(code, 201)2227 sub_sub_pool_id = sub_sub_pool['id']2228 # root pool2229 resources2 = self._create_resources(2230 ['my_2', 'm3'], pool_id=self.org_pool_id,2231 employee_id=self.employee['id'])2232 # sub pool2233 resources3 = self._create_resources(2234 ['my_3', 'm4'], pool_id=sub_pool_id,2235 employee_id=self.employee['id'])2236 # sub sub pool2237 resources4 = self._create_resources(2238 ['my_4', 'm5'], pool_id=sub_sub_pool_id,2239 employee_id=self.employee['id'])2240 conditions = [2241 {"type": "name_starts_with", "meta_info": "my_"}2242 ]2243 self._create_rule('rule1', conditions=conditions, set_allowed=True)2244 # only processing resources from root2245 code, resp = self.client.rules_apply(self.org_id,2246 pool_id=self.org_pool_id)2247 self.assertEqual(code, 201)2248 self.assertEqual(resp['processed'], 2)2249 self.assertEqual(resp['updated_assignments'], 1)2250 self.verify_assignment(resources2[0]['id'], self.org_pool_id,2251 self.employee['id'])2252 self.verify_assignment(2253 resources2[1]['id'], self.cloud_acc_rule['pool_id'],2254 self.employee['id'])2255 # processing resources from sub and sub sub2256 code, resp = self.client.rules_apply(2257 self.org_id, pool_id=sub_pool_id, include_children=True)2258 self.assertEqual(code, 201)2259 self.assertEqual(resp['processed'], 4)2260 self.assertEqual(resp['updated_assignments'], 4)2261 self.verify_assignment(resources3[0]['id'], self.org_pool_id,2262 self.employee['id'])2263 self.verify_assignment(2264 resources3[1]['id'], self.cloud_acc_rule['pool_id'],2265 self.employee['id'])2266 self.verify_assignment(resources4[0]['id'], self.org_pool_id,2267 self.employee['id'])2268 self.verify_assignment(2269 resources4[1]['id'], self.cloud_acc_rule['pool_id'],2270 self.employee['id'])2271 def test_reapply_deleted_rule(self):2272 self._create_resources(['my_1'])2273 rule = self._create_rule('rule1', conditions=[2274 {"type": "name_starts_with", "meta_info": "m"}2275 ], set_allowed=True)2276 code, _ = self.client.rules_apply(self.org_id,2277 pool_id=self.org_pool_id)2278 self.assertEqual(code, 201)2279 self.client.rule_delete(rule['id'])2280 rule2 = self._create_rule('rule2', conditions=[2281 {"type": "name_starts_with", "meta_info": "my_"}2282 ], pool_id=self.sub_pools_ids[0], set_allowed=True)2283 m_activities_publish = patch(2284 'rest_api_server.controllers.base.'2285 'BaseController.publish_activities_task').start()2286 code, _ = self.client.rules_apply(2287 self.org_id, pool_id=self.org_pool_id)2288 self.assertEqual(code, 201)2289 m_activities_publish.assert_has_calls([2290 call(self.org_id, self.org_id, 'organization',2291 'rules_processing_started', ANY,2292 'organization.rules_processing_started', add_token=True),2293 call(self.org_id, rule2['id'], 'rule',2294 'rule_applied', ANY, 'rule.rule_applied', add_token=True),2295 call(self.org_id, self.org_id, 'organization',2296 'rules_processing_completed', ANY,2297 'organization.rules_processing_completed', add_token=True),2298 ])2299 def test_rules_apply_api_clusters(self):2300 m_activities_publish = patch(2301 'rest_api_server.controllers.base.'2302 'BaseController.publish_activities_task').start()2303 code, c_type = self.client.cluster_type_create(2304 self.org_id, {'name': 'c_type', 'tag_key': 'tn'})2305 self.assertEqual(code, 201)2306 resources = self._create_resources(['my_1', 'm2'], tags={'tn': 'tv'})2307 conditions = [2308 {"type": "tag_exists", "meta_info": "tn"}2309 ]2310 rule = self._create_rule('rule1', conditions=conditions,2311 set_allowed=True)2312 code, resp = self.client.rules_apply(self.org_id, self.org_pool_id)2313 m_activities_publish.assert_has_calls([2314 call(self.org_id, self.org_id, 'organization',2315 'rules_processing_started', ANY,2316 'organization.rules_processing_started', add_token=True),2317 call(self.org_id, rule['id'], 'rule',2318 'rule_applied', ANY, 'rule.rule_applied', add_token=True),2319 call(self.org_id, self.org_id, 'organization',2320 'rules_processing_completed', ANY,2321 'organization.rules_processing_completed', add_token=True),2322 ])2323 self.assertEqual(code, 201)2324 self.assertEqual(resp['processed'], 1)...

fuzzy_object.py

Source:fuzzy_object.py

...159 self.sim = None160 return161 if fuzzy_type == FuzzyType.BOTH:162 rules = []163 rules.append( self._create_rule("N", "N", "N") )164 rules.append( self._create_rule("N", "Z", "N") )165 rules.append( self._create_rule("N", "P", "Z") )166 rules.append( self._create_rule("Z", "N", "N") )167 rules.append( self._create_rule("Z", "Z", "Z") )168 rules.append( self._create_rule("Z", "P", "P") )169 rules.append( self._create_rule("P", "N", "Z") )170 rules.append( self._create_rule("P", "Z", "P") )171 rules.append( self._create_rule("P", "P", "P") )172# rules.append( self._create_rule("NB", "NB", "NB") )173# rules.append( self._create_rule("NB", "NS", "NB") )174# rules.append( self._create_rule("NB", "Z", "NS") )175# rules.append( self._create_rule("NB", "PS", "NS") )176# rules.append( self._create_rule("NB", "PB", "Z") )177#178# rules.append( self._create_rule("NS", "NB", "NB") )179# rules.append( self._create_rule("NS", "NS", "NS") )180# rules.append( self._create_rule("NS", "Z", "NS") )181# rules.append( self._create_rule("NS", "PS", "Z") )182# rules.append( self._create_rule("NS", "PB", "PS") )183#184# rules.append( self._create_rule("Z", "NB", "NS") )185# rules.append( self._create_rule("Z", "NS", "NS") )186# rules.append( self._create_rule("Z", "Z", "Z") )187# rules.append( self._create_rule("Z", "PS", "PS") )188# rules.append( self._create_rule("Z", "PB", "PS") )189#190# rules.append( self._create_rule("PS", "NB", "NS") )191# rules.append( self._create_rule("PS", "NS", "Z") )192# rules.append( self._create_rule("PS", "Z", "PS") )193# rules.append( self._create_rule("PS", "PS", "PS") )194# rules.append( self._create_rule("PS", "PB", "PB") )195#196# rules.append( self._create_rule("PB", "NB", "Z") )197# rules.append( self._create_rule("PB", "NS", "PS") )198# rules.append( self._create_rule("PB", "Z", "PS") )199# rules.append( self._create_rule("PB", "PS", "PB") )200# rules.append( self._create_rule("PB", "PB", "PB") )201 self.output_ctrl = ctrl.ControlSystem( rules )202 self.sim = ctrl.ControlSystemSimulation(self.output_ctrl)203# self._plots.request_plot( self.output_ctrl, "rules" )204 return205 if fuzzy_type == FuzzyType.ERR:206 rules = []207 rules.append( ctrl.Rule(self.ant_err["N"], self.con_output["N"]) )208 rules.append( ctrl.Rule(self.ant_err["Z"], self.con_output["Z"]) )209 rules.append( ctrl.Rule(self.ant_err["P"], self.con_output["P"]) )210# rules.append( ctrl.Rule(self.ant_err["NB"], self.con_output["NB"]) )211# rules.append( ctrl.Rule(self.ant_err["NS"], self.con_output["NS"]) )212# rules.append( ctrl.Rule(self.ant_err[ "Z"], self.con_output[ "Z"]) )213# rules.append( ctrl.Rule(self.ant_err["PS"], self.con_output["PS"]) )214# rules.append( ctrl.Rule(self.ant_err["PB"], self.con_output["PB"]) )215 self.output_ctrl = ctrl.ControlSystem( rules )216 self.sim = ctrl.ControlSystemSimulation(self.output_ctrl)217# self._plots.request_plot( self.output_ctrl, "rules" )218 return219 if fuzzy_type == FuzzyType.DERR:220 rules = []221 rules.append( ctrl.Rule(self.ant_derr["N"], self.con_output["N"]) )222 rules.append( ctrl.Rule(self.ant_derr["Z"], self.con_output["Z"]) )223 rules.append( ctrl.Rule(self.ant_derr["P"], self.con_output["P"]) )224# rules.append( ctrl.Rule(self.ant_derr["NB"], self.con_output["NB"]) )225# rules.append( ctrl.Rule(self.ant_derr["NS"], self.con_output["NS"]) )226# rules.append( ctrl.Rule(self.ant_derr[ "Z"], self.con_output[ "Z"]) )227# rules.append( ctrl.Rule(self.ant_derr["PS"], self.con_output["PS"]) )228# rules.append( ctrl.Rule(self.ant_derr["PB"], self.con_output["PB"]) )229 self.output_ctrl = ctrl.ControlSystem( rules )230 self.sim = ctrl.ControlSystemSimulation(self.output_ctrl)231# self._plots.request_plot( self.output_ctrl, "rules" )232 return233 self.sim = None234 raise ValueError('Invalid state: ' + fuzzy_type)235 def reset_state(self):236 self.error = Error()237 @synchronized238 def set_err_range(self, value):239 absval = abs(value)240 if absval < 0.1:241 self.ant_err = None242 return243 invert = False244 if value < 0.0:245 invert = True246 self.ant_err = ctrl.Antecedent(np.arange(-absval, absval, 1), 'err')247 self.ant_err.automf(names=["N", "Z", "P"], invert=invert)248 ##self.ant_err.automf(names=["NB", "NS", "Z", "PS", "PB"], invert=invert)249 self._plots.request_plot( self.ant_err, "err" )250 @synchronized251 def set_derr_range(self, value):252 absval = abs(value)253 if absval < 0.1:254 self.ant_derr = None255 return256 invert = False257 if value < 0.0:258 invert = True259 self.ant_derr = ctrl.Antecedent(np.arange(-absval, absval, 1), 'derr')260 self.ant_derr.automf(names=["N", "Z", "P"], invert=invert)261 ##self.ant_derr.automf(names=["NB", "NS", "Z", "PS", "PB"], invert=invert)262 self._plots.request_plot( self.ant_derr, "derr" )263 @synchronized264 def set_output_range(self, value):265 absval = abs(value)266 if absval < 0.1:267 self.con_output = None268 return269 invert = False270 if value < 0.0:271 invert = True272 self.con_output = ctrl.Consequent(np.arange(-absval, absval, 1), 'output')273 self.con_output.automf(names=["N", "Z", "P"], invert=invert)274 ##self.con_output.automf(names=["NB", "NS", "Z", "PS", "PB"], invert=invert)275 self._plots.request_plot( self.con_output, "output" )276 @synchronized277 def compute(self, valueInput):278 fuzzy_type = self.type()279 if fuzzy_type == FuzzyType.INVALID:280 self.error.calculate(valueInput)281 return 0282 err = self.error.calculate(valueInput)283 if fuzzy_type == FuzzyType.BOTH or fuzzy_type == FuzzyType.ERR:284 self.sim.input['err'] = err[0]285 if fuzzy_type == FuzzyType.BOTH or fuzzy_type == FuzzyType.DERR:286 self.sim.input['derr'] = err[1]287 self.sim.compute()288 self._plots.process_requests()289 output = self.sim.output['output']290 return output291 def close(self):292 self._plots.close_all()293 def _create_rule(self, err, derr, out):294 rule = ctrl.Rule(self.ant_err[err] | self.ant_derr[derr], self.con_output[out])295 return rule296class FuzzyObject():297 def __init__(self, controller_name, err_param=90, derr_param=30, output_param=20):298 self.controller_name = controller_name299 self.fuzzy = Fuzzy(err_param, derr_param, output_param)300 rospy.Subscriber("/self_balancer/" + self.controller_name + "/err_max", Float64, self._err_callback)301 rospy.Subscriber("/self_balancer/" + self.controller_name + "/derr_max", Float64, self._derr_callback)302 rospy.Subscriber("/self_balancer/" + self.controller_name + "/output_max", Float64, self._output_callback)303 self.output_pub = rospy.Publisher("/self_balancer/" + self.controller_name + "/output", Float64, queue_size=10)304 self.err_pub = rospy.Publisher("/self_balancer/" + self.controller_name + "/err", Float64, queue_size=10)305 self.derr_pub = rospy.Publisher("/self_balancer/" + self.controller_name + "/derr", Float64, queue_size=10)306 def reset_state(self):307 rospy.loginfo("resetting Fuzzy")...

test_rules.py

Source:test_rules.py

2from tracpro.test.cases import TracProTest3from .. import rules4from . import factories5class TestGetCategory(TracProTest):6 def _create_rule(self, category):7 return {8 'category': category,9 'test': {},10 }11 def test_get_category_direct(self):12 """Return rule['category'] if it is not a dictionary."""13 rule = self._create_rule('cats')14 self.assertEqual(rules.get_category(rule), "cats")15 def test_get_category_base(self):16 """get_category returns the base category name, if available."""17 rule = self._create_rule({'base': 'base', 'eng': 'eng'})18 self.assertEqual(rules.get_category(rule), "base")19 def test_get_category_other(self):20 """get_category returns the first available translation."""21 rule = self._create_rule({'eng': 'eng'})22 self.assertEqual(rules.get_category(rule), "eng")23class TestGetAllCategories(TracProTest):24 def _create_rule(self, category):25 return {26 'category': {'base': category},27 'test': {},28 }29 def test_get_all_no_rules(self):30 question = factories.Question(rules=[])31 self.assertEqual(rules.get_all_categories(question), ["Other"])32 def test_get_all(self):33 question = factories.Question(rules=[34 self._create_rule('a'),35 self._create_rule('b'),36 ])37 self.assertEqual(38 rules.get_all_categories(question),39 ['a', 'b', 'Other'])40 def test_get_all__include_other(self):41 """Ensure that "Other" is at the end of the list."""42 question = factories.Question(rules=[43 self._create_rule('a'),44 self._create_rule('Other'),45 self._create_rule('b'),46 ])47 self.assertEqual(48 rules.get_all_categories(question),49 ['a', 'b', 'Other'])50 def test_get_all_with_answers(self):51 """Include Answer categories that aren't in the rules."""52 question = factories.Question(rules=[53 self._create_rule('a'),54 self._create_rule('b'),55 ])56 factories.Answer(question=question, category='apple')57 factories.Answer(question=question, category='a')58 self.assertEqual(59 rules.get_all_categories(question, answers=question.answers.all()),60 ['a', 'b', 'apple', 'Other'])61 def test_duplicates(self):62 question = factories.Question(rules=[63 self._create_rule('b'),64 self._create_rule('a'),65 self._create_rule('a'),66 self._create_rule('b'),67 ])68 self.assertEqual(69 rules.get_all_categories(question),70 ['b', 'a', 'Other'])71class CheckRuleTestBase(object):72 def call(self, test):73 val, kwargs = test74 return self.rule(val, **kwargs)75 def test_false(self):76 unexpected_successes = []77 for test in self.false_tests:78 if self.call(test) is not False:79 unexpected_successes.append(test)80 if unexpected_successes:...

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 localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

View sample repo View DocumentationSign up Free

›

_