source: OpenRLabs-Git/deploy/rlabs-docker/web2py-rlabs/gluon/packages/dal/tests/base.py

main
Last change on this file was 42bd667, checked in by David Fuertes <dfuertes@…>, 4 years ago

Historial Limpio
  • Property mode set to 100755
File size: 7.3 KB
Line 
1# -*- coding: utf-8 -*-
2
3from ._compat import unittest
4from ._adapt import DEFAULT_URI, drop, IS_MSSQL, IS_IMAP, IS_GAE, IS_TERADATA, IS_ORACLE
5from pydal import DAL, Field
6from pydal._compat import PY2
7
8
9@unittest.skipIf(IS_IMAP, "Reference not Null unsupported on IMAP")
10@unittest.skipIf(IS_ORACLE, "Reference Not Null unsupported on Oracle")
11class TestReferenceNOTNULL(unittest.TestCase):
12    # 1:N not null
13
14    def testRun(self):
15        for ref, bigint in [("reference", False), ("big-reference", True)]:
16            db = DAL(DEFAULT_URI, check_reserved=["all"], bigint_id=bigint)
17            if bigint and "big-id" not in db._adapter.types:
18                continue
19            db.define_table("tt", Field("vv"))
20            db.define_table(
21                "ttt", Field("vv"), Field("tt_id", "%s tt" % ref, notnull=True)
22            )
23            self.assertRaises(Exception, db.ttt.insert, vv="pydal")
24            # The following is mandatory for backends as PG to close the aborted transaction
25            db.commit()
26            drop(db.ttt)
27            drop(db.tt)
28            db.close()
29
30
31@unittest.skipIf(IS_IMAP, "Reference Unique unsupported on IMAP")
32@unittest.skipIf(IS_GAE, "Reference Unique unsupported on GAE")
33@unittest.skipIf(IS_ORACLE, "Reference Unique unsupported on Oracle")
34class TestReferenceUNIQUE(unittest.TestCase):
35    # 1:1 relation
36
37    def testRun(self):
38        for ref, bigint in [("reference", False), ("big-reference", True)]:
39            db = DAL(DEFAULT_URI, check_reserved=["all"], bigint_id=bigint)
40            if bigint and "big-id" not in db._adapter.types:
41                continue
42            db.define_table("tt", Field("vv"))
43            db.define_table(
44                "ttt",
45                Field("vv"),
46                Field("tt_id", "%s tt" % ref, unique=True),
47                Field("tt_uq", "integer", unique=True),
48            )
49            id_1 = db.tt.insert(vv="pydal")
50            id_2 = db.tt.insert(vv="pydal")
51            # Null tt_id
52            db.ttt.insert(vv="pydal", tt_uq=1)
53            # first insert is OK
54            db.ttt.insert(tt_id=id_1, tt_uq=2)
55            self.assertRaises(Exception, db.ttt.insert, tt_id=id_1, tt_uq=3)
56            self.assertRaises(Exception, db.ttt.insert, tt_id=id_2, tt_uq=2)
57            # The following is mandatory for backends as PG to close the aborted transaction
58            db.commit()
59            drop(db.ttt)
60            drop(db.tt)
61            db.close()
62
63
64@unittest.skipIf(IS_IMAP, "Reference Unique not Null unsupported on IMAP")
65@unittest.skipIf(IS_GAE, "Reference Unique not Null unsupported on GAE")
66@unittest.skipIf(IS_ORACLE, "Reference Unique not Null unsupported on Oracle")
67class TestReferenceUNIQUENotNull(unittest.TestCase):
68    # 1:1 relation not null
69
70    def testRun(self):
71        for ref, bigint in [("reference", False), ("big-reference", True)]:
72            db = DAL(DEFAULT_URI, check_reserved=["all"], bigint_id=bigint)
73            if bigint and "big-id" not in db._adapter.types:
74                continue
75            db.define_table("tt", Field("vv"))
76            db.define_table(
77                "ttt",
78                Field("vv"),
79                Field("tt_id", "%s tt" % ref, unique=True, notnull=True),
80            )
81            self.assertRaises(Exception, db.ttt.insert, vv="pydal")
82            db.commit()
83            id_i = db.tt.insert(vv="pydal")
84            # first insert is OK
85            db.ttt.insert(tt_id=id_i)
86            self.assertRaises(Exception, db.ttt.insert, tt_id=id_i)
87            # The following is mandatory for backends as PG to close the aborted transaction
88            db.commit()
89            drop(db.ttt)
90            drop(db.tt)
91            db.close()
92
93
94@unittest.skipIf(IS_IMAP, "Skip unicode on IMAP")
95@unittest.skipIf(IS_MSSQL and not PY2, "Skip unicode on py3 and MSSQL")
96class TestUnicode(unittest.TestCase):
97    def testRun(self):
98        db = DAL(DEFAULT_URI, check_reserved=["all"])
99        db.define_table("tt", Field("vv"))
100        vv = "ἀγοραζε"
101        id_i = db.tt.insert(vv=vv)
102        row = db(db.tt.id == id_i).select().first()
103        self.assertEqual(row.vv, vv)
104        db.commit()
105        drop(db.tt)
106        db.close()
107
108
109class TestParseDateTime(unittest.TestCase):
110    def testRun(self):
111        db = DAL(DEFAULT_URI, check_reserved=["all"])
112
113        #: skip for adapters that use drivers for datetime parsing
114        if db._adapter.parser.registered.get("datetime") is None:
115            return
116
117        parse = lambda v: db._adapter.parser.parse(v, "datetime", "datetime")
118
119        dt = parse("2015-09-04t12:33:36.223245")
120        self.assertEqual(dt.microsecond, 223245)
121        self.assertEqual(dt.hour, 12)
122
123        dt = parse("2015-09-04t12:33:36.223245Z")
124        self.assertEqual(dt.microsecond, 223245)
125        self.assertEqual(dt.hour, 12)
126
127        dt = parse("2015-09-04t12:33:36.223245-2:0")
128        self.assertEqual(dt.microsecond, 223245)
129        self.assertEqual(dt.hour, 10)
130
131        dt = parse("2015-09-04t12:33:36+1:0")
132        self.assertEqual(dt.microsecond, 0)
133        self.assertEqual(dt.hour, 13)
134
135        dt = parse("2015-09-04t12:33:36.123")
136        self.assertEqual(dt.microsecond, 123000)
137
138        dt = parse("2015-09-04t12:33:36.00123")
139        self.assertEqual(dt.microsecond, 1230)
140
141        dt = parse("2015-09-04t12:33:36.1234567890")
142        self.assertEqual(dt.microsecond, 123456)
143        db.close()
144
145
146@unittest.skipIf(IS_IMAP, "chained join unsupported on IMAP")
147@unittest.skipIf(IS_TERADATA, "chained join unsupported on TERADATA")
148class TestChainedJoinUNIQUE(unittest.TestCase):
149    # 1:1 relation
150
151    def testRun(self):
152        db = DAL(DEFAULT_URI, check_reserved=["all"])
153        db.define_table("aa", Field("name"))
154        db.define_table("bb", Field("aa", "reference aa"), Field("name"))
155        for k in ("x", "y", "z"):
156            i = db.aa.insert(name=k)
157            for j in ("u", "v", "w"):
158                db.bb.insert(aa=i, name=k + j)
159        db.commit()
160        rows = db(db.aa).select()
161        rows.join(db.bb.aa, fields=[db.bb.name], orderby=[db.bb.name])
162        self.assertEqual(rows[0].bb[0].name, "xu")
163        self.assertEqual(rows[0].bb[1].name, "xv")
164        self.assertEqual(rows[0].bb[2].name, "xw")
165        self.assertEqual(rows[1].bb[0].name, "yu")
166        self.assertEqual(rows[1].bb[1].name, "yv")
167        self.assertEqual(rows[1].bb[2].name, "yw")
168        self.assertEqual(rows[2].bb[0].name, "zu")
169        self.assertEqual(rows[2].bb[1].name, "zv")
170        self.assertEqual(rows[2].bb[2].name, "zw")
171
172        rows = db(db.bb).select()
173        rows.join(db.aa.id, fields=[db.aa.name])
174
175        self.assertEqual(rows[0].aa.name, "x")
176        self.assertEqual(rows[1].aa.name, "x")
177        self.assertEqual(rows[2].aa.name, "x")
178        self.assertEqual(rows[3].aa.name, "y")
179        self.assertEqual(rows[4].aa.name, "y")
180        self.assertEqual(rows[5].aa.name, "y")
181        self.assertEqual(rows[6].aa.name, "z")
182        self.assertEqual(rows[7].aa.name, "z")
183        self.assertEqual(rows[8].aa.name, "z")
184
185        rows_json = rows.as_json()
186        drop(db.bb)
187        drop(db.aa)
188        db.close()
189
190
191class TestNullAdapter(unittest.TestCase):
192    # Test that NullAdapter can define tables
193
194    def testRun(self):
195        db = DAL(None)
196        db.define_table("no_table", Field("aa"))
197        self.assertIsInstance(db.no_table.aa, Field)
198        self.assertIsInstance(db.no_table["aa"], Field)
199        db.close()
Note: See TracBrowser for help on using the repository browser.