~edoardo-serra/storm/select-for-update-support

« back to all changes in this revision

Viewing changes to tests/store/base.py

  • Committer: Edoardo Serra
  • Date: 2010-09-08 18:56:43 UTC
  • Revision ID: eserra@barbera-20100908185643-z4lnntw6ry8425dr
Unit tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
569
569
        self.assertEqual(True, result.is_empty())
570
570
        self.assertNotIn("ORDER BY", stream.getvalue())
571
571
 
 
572
    def test_is_empty_strips_for_update(self):
 
573
        """
 
574
        L{ResultSet.is_empty} strips the C{FOR UPDATE} clause, if one is
 
575
        present, since it's not supported in subqueries and thre is no reason
 
576
        to acquire a row-level lock while testing if a ResultSet is empty
 
577
        """
 
578
        stream = StringIO()
 
579
        self.addCleanup(debug, False)
 
580
        debug(True, stream)
 
581
 
 
582
        result = self.store.find(Foo, Foo.id == 300).for_update()
 
583
        result.order_by(Foo.id)
 
584
        self.assertEqual(True, result.is_empty())
 
585
        self.assertNotIn("FOR UPDATE", stream.getvalue())
 
586
 
572
587
    def test_is_empty_with_composed_key(self):
573
588
        result = self.store.find(Link, foo_id=300, bar_id=3000)
574
589
        self.assertEquals(result.is_empty(), True)
592
607
                          (30, "Title 10"),
593
608
                         ])
594
609
 
 
610
    def test_find_for_update(self):
 
611
        stream = StringIO()
 
612
        self.addCleanup(debug, False)
 
613
        debug(True, stream)
 
614
 
 
615
        result = self.store.find(Foo).for_update()
 
616
        lst = [(foo.id, foo.title) for foo in result]
 
617
        lst.sort()
 
618
        self.assertEquals(lst, [
 
619
                          (10, "Title 30"),
 
620
                          (20, "Title 20"),
 
621
                          (30, "Title 10"),
 
622
                         ])
 
623
        if self.__class__.__name__.startswith("SQLite"):
 
624
            return
 
625
        self.assertIn("FOR UPDATE", stream.getvalue())
 
626
 
595
627
    def test_find_from_cache(self):
596
628
        foo = self.store.get(Foo, 10)
597
629
        self.assertTrue(self.store.find(Foo, id=10).one() is foo)
940
972
    def test_find_count(self):
941
973
        self.assertEquals(self.store.find(Foo).count(), 3)
942
974
 
 
975
    def test_find_count_for_update(self):
 
976
        result = self.store.find(Foo)
 
977
        result.for_update()
 
978
        self.assertRaises(FeatureError, result.count, 3)
 
979
 
943
980
    def test_find_count_after_slice(self):
944
981
        """
945
982
        When we slice a ResultSet obtained after a set operation (like union),
1007
1044
    def test_find_max(self):
1008
1045
        self.assertEquals(self.store.find(Foo).max(Foo.id), 30)
1009
1046
 
 
1047
    def test_find_max_for_update(self):
 
1048
        result = self.store.find(Foo)
 
1049
        result.for_update()
 
1050
        self.assertRaises(FeatureError, result.max, Foo.id)
 
1051
 
1010
1052
    def test_find_max_expr(self):
1011
1053
        self.assertEquals(self.store.find(Foo).max(Foo.id + 1), 31)
1012
1054
 
1028
1070
    def test_find_min(self):
1029
1071
        self.assertEquals(self.store.find(Foo).min(Foo.id), 10)
1030
1072
 
 
1073
    def test_find_min_for_update(self):
 
1074
        result = self.store.find(Foo)
 
1075
        result.for_update()
 
1076
        self.assertRaises(FeatureError, result.min, Foo.id)
 
1077
 
1031
1078
    def test_find_min_expr(self):
1032
1079
        self.assertEquals(self.store.find(Foo).min(Foo.id - 1), 9)
1033
1080
 
1049
1096
    def test_find_avg(self):
1050
1097
        self.assertEquals(self.store.find(Foo).avg(Foo.id), 20)
1051
1098
 
 
1099
    def test_find_avg_for_update(self):
 
1100
        result = self.store.find(Foo)
 
1101
        result.for_update()
 
1102
        self.assertRaises(FeatureError, result.avg, Foo.id)
 
1103
 
1052
1104
    def test_find_avg_expr(self):
1053
1105
        self.assertEquals(self.store.find(Foo).avg(Foo.id + 10), 30)
1054
1106
 
1062
1114
    def test_find_sum(self):
1063
1115
        self.assertEquals(self.store.find(Foo).sum(Foo.id), 60)
1064
1116
 
 
1117
    def test_find_sum_for_update(self):
 
1118
        result = self.store.find(Foo)
 
1119
        result.for_update()
 
1120
        self.assertRaises(FeatureError, result.sum, Foo.id)
 
1121
 
1065
1122
    def test_find_sum_expr(self):
1066
1123
        self.assertEquals(self.store.find(Foo).sum(Foo.id * 2), 120)
1067
1124
 
1546
1603
        result = list(result)
1547
1604
        self.assertEquals(result, [(2L, 2L), (2L, 2L), (2L, 3L), (3L, 6L)])
1548
1605
 
 
1606
    def test_find_group_by_for_update(self):
 
1607
        result = self.store.find((Count(FooValue.id), Sum(FooValue.value1)))
 
1608
        result.group_by(FooValue.value2)
 
1609
        self.assertRaises(FeatureError, result.for_update)
 
1610
 
1549
1611
    def test_find_group_by_table(self):
1550
1612
        result = self.store.find(
1551
1613
            (Sum(FooValue.value2), Foo), Foo.id == FooValue.foo_id)
2271
2333
        self.assertEquals(foo.id, 20)
2272
2334
        self.assertEquals(foo.title, "Title 20")
2273
2335
 
 
2336
    def test_sub_select_for_update(self):
 
2337
        foo = self.store.find(Foo, Foo.id == Select(SQL("20"))
 
2338
                             ).for_update().one()
 
2339
        self.assertTrue(foo)
 
2340
        self.assertEquals(foo.id, 20)
 
2341
        self.assertEquals(foo.title, "Title 20")
 
2342
 
2274
2343
    def test_cache_has_improper_object(self):
2275
2344
        foo = self.store.get(Foo, 20)
2276
2345
        self.store.remove(foo)
5472
5541
                          (30, "Title 10"),
5473
5542
                         ])
5474
5543
 
 
5544
    def test_result_union_for_update(self):
 
5545
        result1 = self.store.find(Foo, id=30).for_update()
 
5546
        result2 = self.store.find(Foo, id=10)
 
5547
        self.assertRaises(FeatureError, result1.union, result2)
 
5548
        self.assertRaises(FeatureError, result2.union, result1)
 
5549
 
5475
5550
    def test_result_union_duplicated(self):
5476
5551
        result1 = self.store.find(Foo, id=30)
5477
5552
        result2 = self.store.find(Foo, id=30)
5544
5619
                          (30, "Title 10"),
5545
5620
                         ])
5546
5621
 
 
5622
    def test_result_difference_for_update(self):
 
5623
        result1 = self.store.find(Foo).for_update()
 
5624
        result2 = self.store.find(Foo, id=10)
 
5625
        self.assertRaises(FeatureError, result1.difference, result2)
 
5626
        self.assertRaises(FeatureError, result2.difference, result1)
 
5627
 
5547
5628
    def test_result_difference_with_empty(self):
5548
5629
        if self.__class__.__name__.startswith("MySQL"):
5549
5630
            return
5605
5686
                          (30, "Title 10"),
5606
5687
                         ])
5607
5688
 
 
5689
    def test_result_intersection_for_update(self):
 
5690
        result1 = self.store.find(Foo).for_update()
 
5691
        result2 = self.store.find(Foo, Foo.id.is_in((10, 30)))
 
5692
        self.assertRaises(FeatureError, result1.intersection, result2)
 
5693
        self.assertRaises(FeatureError, result2.intersection, result1)
 
5694
 
5608
5695
    def test_result_intersection_with_empty(self):
5609
5696
        if self.__class__.__name__.startswith("MySQL"):
5610
5697
            return
6048
6135
        self.assertEquals(self.result.order_by(Foo.title), self.result)
6049
6136
        self.assertEquals(self.empty.order_by(Foo.title), self.empty)
6050
6137
 
 
6138
    def test_for_update(self):
 
6139
        self.assertEquals(self.result.for_update(), self.result)
 
6140
        self.assertEquals(self.empty.for_update(), self.empty)
 
6141
 
6051
6142
    def test_remove(self):
6052
6143
        self.assertEquals(self.result.remove(), 0)
6053
6144
        self.assertEquals(self.empty.remove(), 0)