Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- def test_policy_rule_evaluation(self):
- try:
- # create policy alice in PE1
- self.assertEqual(self.pe1.post(
- suffix='policies', json={'name': 'alice'}).status_code,
- 201)
- # add rule to PE1
- j = {'rule': 'p(x) :- q(x)', 'name': 'rule0'}
- self.assertEqual(self.pe1.post(
- suffix='policies/alice/rules', json=j).status_code, 201)
- # add data to PE1
- j = {'rule': 'q(1)', 'name': 'rule1'}
- self.assertEqual(self.pe1.post(
- suffix='policies/alice/rules', json=j).status_code, 201)
- # add data to PE2
- j = {'rule': 'q(2)', 'name': 'rule2'}
- self.assertEqual(self.pe2.post(
- suffix='policies/alice/rules', json=j).status_code, 201)
- # eval on PE1
- helper.retry_check_function_return_value_table(
- lambda: [x['data'] for x in
- self.pe1.get('policies/alice/tables/p/rows').json()[
- 'results']],
- [[1], [2]])
- # eval on PE2
- helper.retry_check_function_return_value_table(
- lambda: [x['data'] for x in
- self.pe2.get('policies/alice/tables/p/rows').json()[
- 'results']],
- [[1], [2]])
- except Exception:
- LOG.error('PE1 process output:\n%s' % self.read_output_file(self.outfiles[1]))
- LOG.error('PE2 process output:\n%s' % self.read_output_file(self.outfiles[2]))
- raise
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement