689 class MigrationCommandsComputedTC(MigrationTC): |
689 class MigrationCommandsComputedTC(MigrationTC): |
690 """ Unit tests for computed relations and attributes |
690 """ Unit tests for computed relations and attributes |
691 """ |
691 """ |
692 appid = 'datacomputed' |
692 appid = 'datacomputed' |
693 |
693 |
|
694 def setUp(self): |
|
695 MigrationTC.setUp(self) |
|
696 # ensure vregistry is reloaded, needed by generated hooks for computed |
|
697 # attributes |
|
698 self.repo.vreg.set_schema(self.repo.schema) |
|
699 |
694 def test_computed_relation_add_relation_definition(self): |
700 def test_computed_relation_add_relation_definition(self): |
695 self.assertNotIn('works_for', self.schema) |
701 self.assertNotIn('works_for', self.schema) |
696 with self.mh() as (cnx, mh): |
702 with self.mh() as (cnx, mh): |
697 with self.assertRaises(ExecutionError) as exc: |
703 with self.assertRaises(ExecutionError) as exc: |
698 mh.cmd_add_relation_definition('Employee', 'works_for', |
704 mh.cmd_add_relation_definition('Employee', 'works_for', |
753 ('Company', 'whatever', 'Person')) |
759 ('Company', 'whatever', 'Person')) |
754 self.assertEqual(str(exc.exception), |
760 self.assertEqual(str(exc.exception), |
755 'Cannot synchronize a relation definition for a computed ' |
761 'Cannot synchronize a relation definition for a computed ' |
756 'relation (whatever)') |
762 'relation (whatever)') |
757 |
763 |
|
764 # computed attributes migration ############################################ |
|
765 |
|
766 def setup_add_score(self): |
|
767 with self.admin_access.client_cnx() as cnx: |
|
768 assert not cnx.execute('Company X') |
|
769 c = cnx.create_entity('Company') |
|
770 e1 = cnx.create_entity('Employee', reverse_employees=c) |
|
771 n1 = cnx.create_entity('Note', note=2, concerns=e1) |
|
772 e2 = cnx.create_entity('Employee', reverse_employees=c) |
|
773 n2 = cnx.create_entity('Note', note=4, concerns=e2) |
|
774 cnx.commit() |
|
775 |
|
776 def assert_score_initialized(self, mh): |
|
777 self.assertEqual(self.schema['score'].rdefs['Company', 'Float'].formula, |
|
778 'Any AVG(NN) WHERE X employees E, N concerns E, N note NN') |
|
779 fields = self.table_schema(mh, '%sCompany' % SQL_PREFIX) |
|
780 self.assertEqual(fields['%sscore' % SQL_PREFIX], 'float') |
|
781 self.assertEqual([[3.0]], |
|
782 mh.rqlexec('Any CS WHERE C score CS, C is Company').rows) |
|
783 |
|
784 def test_computed_attribute_add_relation_type(self): |
|
785 self.assertNotIn('score', self.schema) |
|
786 self.setup_add_score() |
|
787 with self.mh() as (cnx, mh): |
|
788 mh.cmd_add_relation_type('score') |
|
789 self.assertIn('score', self.schema) |
|
790 self.assertEqual(self.schema['score'].objects(), ('Float',)) |
|
791 self.assertEqual(self.schema['score'].subjects(), ('Company',)) |
|
792 self.assert_score_initialized(mh) |
|
793 |
|
794 def test_computed_attribute_add_attribute(self): |
|
795 self.assertNotIn('score', self.schema) |
|
796 self.setup_add_score() |
|
797 with self.mh() as (cnx, mh): |
|
798 mh.cmd_add_attribute('Company', 'score') |
|
799 self.assertIn('score', self.schema) |
|
800 self.assert_score_initialized(mh) |
|
801 |
758 |
802 |
759 if __name__ == '__main__': |
803 if __name__ == '__main__': |
760 unittest_main() |
804 unittest_main() |