~launchpad-committers/storm/lp

« back to all changes in this revision

Viewing changes to storm/tests/databases/mysql.py

  • Committer: Colin Watson
  • Date: 2023-07-06 10:51:34 UTC
  • mfrom: (386.34.138 storm)
  • Revision ID: cjwatson@canonical.com-20230706105134-rzpb4opv8nfpg56r
Merge Storm 0.26 release.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
#
 
2
# Copyright (c) 2006, 2007 Canonical
 
3
#
 
4
# Written by Gustavo Niemeyer <gustavo@niemeyer.net>
 
5
#
 
6
# This file is part of Storm Object Relational Mapper.
 
7
#
 
8
# Storm is free software; you can redistribute it and/or modify
 
9
# it under the terms of the GNU Lesser General Public License as
 
10
# published by the Free Software Foundation; either version 2.1 of
 
11
# the License, or (at your option) any later version.
 
12
#
 
13
# Storm is distributed in the hope that it will be useful,
 
14
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
15
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
16
# GNU Lesser General Public License for more details.
 
17
#
 
18
# You should have received a copy of the GNU Lesser General Public License
 
19
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
20
#
 
21
from __future__ import print_function
 
22
 
 
23
import os
 
24
 
 
25
from six.moves.urllib.parse import urlunsplit
 
26
 
 
27
from storm.databases.mysql import MySQL
 
28
from storm.database import create_database
 
29
from storm.expr import Column, Insert
 
30
from storm.uri import URI
 
31
from storm.variables import IntVariable, UnicodeVariable
 
32
 
 
33
from storm.tests.databases.base import (
 
34
    DatabaseTest, DatabaseDisconnectionTest, UnsupportedDatabaseTest)
 
35
from storm.tests.databases.proxy import ProxyTCPServer
 
36
from storm.tests.helper import TestHelper
 
37
 
 
38
 
 
39
def create_proxy_and_uri(uri):
 
40
    """Create a TCP proxy to a Unix-domain database identified by `uri`."""
 
41
    proxy = ProxyTCPServer(uri.options["unix_socket"])
 
42
    proxy_host, proxy_port = proxy.server_address
 
43
    proxy_uri = URI(urlunsplit(
 
44
        ("mysql", "%s:%s" % (proxy_host, proxy_port), "/storm_test",
 
45
         "", "")))
 
46
    return proxy, proxy_uri
 
47
 
 
48
 
 
49
class MySQLTest(DatabaseTest, TestHelper):
 
50
 
 
51
    supports_microseconds = False
 
52
 
 
53
    def is_supported(self):
 
54
        return bool(os.environ.get("STORM_MYSQL_URI"))
 
55
 
 
56
    def create_database(self):
 
57
        self.database = create_database(os.environ["STORM_MYSQL_URI"])
 
58
 
 
59
    def create_tables(self):
 
60
        self.connection.execute("CREATE TABLE number "
 
61
                                "(one INTEGER, two INTEGER, three INTEGER)")
 
62
        self.connection.execute("CREATE TABLE test "
 
63
                                "(id INT AUTO_INCREMENT PRIMARY KEY,"
 
64
                                " title VARCHAR(50)) ENGINE=InnoDB")
 
65
        self.connection.execute("CREATE TABLE datetime_test "
 
66
                                "(id INT AUTO_INCREMENT PRIMARY KEY,"
 
67
                                " dt TIMESTAMP, d DATE, t TIME, td TEXT) "
 
68
                                "ENGINE=InnoDB")
 
69
        self.connection.execute("CREATE TABLE bin_test "
 
70
                                "(id INT AUTO_INCREMENT PRIMARY KEY,"
 
71
                                " b BLOB) ENGINE=InnoDB")
 
72
 
 
73
    def test_wb_create_database(self):
 
74
        database = create_database("mysql://un:pw@ht:12/db?unix_socket=us")
 
75
        self.assertTrue(isinstance(database, MySQL))
 
76
        for key, value in [("db", "db"), ("host", "ht"), ("port", 12),
 
77
                           ("user", "un"), ("passwd", "pw"),
 
78
                           ("unix_socket", "us")]:
 
79
            self.assertEqual(database._connect_kwargs.get(key), value)
 
80
 
 
81
    def test_charset_defaults_to_utf8mb3(self):
 
82
        result = self.connection.execute("SELECT @@character_set_client")
 
83
        self.assertEqual(result.get_one(), ("utf8mb3",))
 
84
 
 
85
    def test_charset_option(self):
 
86
        uri = URI(os.environ["STORM_MYSQL_URI"])
 
87
        uri.options["charset"] = "ascii"
 
88
        database = create_database(uri)
 
89
        connection = database.connect()
 
90
        result = connection.execute("SELECT @@character_set_client")
 
91
        self.assertEqual(result.get_one(), ("ascii",))
 
92
 
 
93
    def test_get_insert_identity(self):
 
94
        # Primary keys are filled in during execute() for MySQL
 
95
        pass
 
96
 
 
97
    def test_get_insert_identity_composed(self):
 
98
        # Primary keys are filled in during execute() for MySQL
 
99
        pass
 
100
 
 
101
    def test_execute_insert_auto_increment_primary_key(self):
 
102
        id_column = Column("id", "test")
 
103
        id_variable = IntVariable()
 
104
        title_column = Column("title", "test")
 
105
        title_variable = UnicodeVariable(u"testing")
 
106
 
 
107
        # This is not part of the table.  It is just used to show that
 
108
        # only one primary key variable is set from the insert ID.
 
109
        dummy_column = Column("dummy", "test")
 
110
        dummy_variable = IntVariable()
 
111
 
 
112
        insert = Insert({title_column: title_variable},
 
113
                        primary_columns=(id_column, dummy_column),
 
114
                        primary_variables=(id_variable, dummy_variable))
 
115
        self.connection.execute(insert)
 
116
        self.assertTrue(id_variable.is_defined())
 
117
        self.assertFalse(dummy_variable.is_defined())
 
118
 
 
119
        # The newly inserted row should have the maximum id value for
 
120
        # the table.
 
121
        result = self.connection.execute("SELECT MAX(id) FROM test")
 
122
        self.assertEqual(result.get_one()[0], id_variable.get())
 
123
 
 
124
    def test_mysql_specific_reserved_words(self):
 
125
        reserved_words = """
 
126
            accessible analyze asensitive before bigint binary blob call
 
127
            change condition current_user database databases day_hour
 
128
            day_microsecond day_minute day_second delayed deterministic
 
129
            distinctrow div dual each elseif enclosed escaped exit explain
 
130
            float4 float8 force fulltext high_priority hour_microsecond
 
131
            hour_minute hour_second if ignore index infile inout int1 int2
 
132
            int3 int4 int8 iterate keys kill leave limit linear lines load
 
133
            localtime localtimestamp lock long longblob longtext loop
 
134
            low_priority master_ssl_verify_server_cert mediumblob mediumint
 
135
            mediumtext middleint minute_microsecond minute_second mod modifies
 
136
            no_write_to_binlog optimize optionally out outfile purge range
 
137
            read_write reads regexp release rename repeat replace require
 
138
            return rlike schemas second_microsecond sensitive separator show
 
139
            spatial specific sql_big_result sql_calc_found_rows
 
140
            sql_small_result sqlexception sqlwarning ssl starting
 
141
            straight_join terminated tinyblob tinyint tinytext trigger undo
 
142
            unlock unsigned use utc_date utc_time utc_timestamp varbinary
 
143
            varcharacter while xor year_month zerofill
 
144
            """.split()
 
145
        for word in reserved_words:
 
146
            self.assertTrue(self.connection.compile.is_reserved_word(word),
 
147
                            "Word missing: %s" % (word,))
 
148
 
 
149
 
 
150
class MySQLUnsupportedTest(UnsupportedDatabaseTest, TestHelper):
 
151
 
 
152
    dbapi_module_names = ["MySQLdb"]
 
153
    db_module_name = "mysql"
 
154
 
 
155
 
 
156
class MySQLDisconnectionTest(DatabaseDisconnectionTest, TestHelper):
 
157
 
 
158
    environment_variable = "STORM_MYSQL_URI"
 
159
    host_environment_variable = "STORM_MYSQL_HOST_URI"
 
160
    default_port = 3306
 
161
 
 
162
    def create_proxy(self, uri):
 
163
        """See `DatabaseDisconnectionMixin.create_proxy`."""
 
164
        if "unix_socket" in uri.options:
 
165
            return create_proxy_and_uri(uri)[0]
 
166
        else:
 
167
            return super(MySQLDisconnectionTest, self).create_proxy(uri)