Produits Entr'ouvert » w.c.s.

from wcs.testdef import TestDef

def test_sql_testdef_unicity(pub):

    TestDef.do_table()

    testdef = TestDef()

    testdef.slug = 'test-1'

    testdef.object_type = 'formdef'

    testdef.object_id = '1'

    testdef.store()

    # same slug, different object_id

    testdef.id = None

    testdef.object_id = '2'

    testdef.store()

    # same slug, object_id and object_type

    testdef.id = None

    with pytest.raises(psycopg2.errors.UniqueViolation):

        testdef.store()

class TestDef(SqlMixin):

    _table_name = 'testdef'

    _table_static_fields = [

        ('id', 'serial'),

        ('name', 'varchar'),

        ('slug', 'varchar'),

        ('object_type', 'varchar'),

        ('object_id', 'varchar'),

        ('data', 'jsonb'),

    ]

    id = None

    @classmethod

    @guard_postgres

    def do_table(cls, conn=None, cur=None):

        conn, cur = get_connection_and_cursor()

        table_name = cls._table_name

        cur.execute(

            '''SELECT COUNT(*) FROM information_schema.tables

                        WHERE table_schema = 'public'

                          AND table_name = %s''',

            (table_name,),

        )

        if cur.fetchone()[0] == 0:

            cur.execute(

                '''CREATE TABLE %s (id SERIAL PRIMARY KEY,

                                            name varchar,

                                            slug varchar NOT NULL,

                                            object_type varchar NOT NULL,

                                            object_id varchar NOT NULL,

                                            data jsonb,

                                            UNIQUE(slug, object_type, object_id)

                                            )'''

                % table_name

            )

        conn.commit()

        cur.close()

    @guard_postgres

    def store(self):

        sql_dict = {x[0]: getattr(self, x[0], None) for x in self._table_static_fields if x[0] != 'id'}

        conn, cur = get_connection_and_cursor()

        column_names = list(sql_dict.keys())

        if not self.id:

            sql_statement = '''INSERT INTO %s (id, %s)

                               VALUES (DEFAULT, %s)

                               RETURNING id''' % (

                self._table_name,

                ', '.join(column_names),

                ', '.join(['%%(%s)s' % x for x in column_names]),

            )

            cur.execute(sql_statement, sql_dict)

            self.id = cur.fetchone()[0]

        else:

            sql_dict['id'] = self.id

            sql_statement = '''UPDATE %s SET %s WHERE id = %%(id)s RETURNING id''' % (

                self._table_name,

                ', '.join(['%s = %%(%s)s' % (x, x) for x in column_names]),

            )

            cur.execute(sql_statement, sql_dict)

        conn.commit()

        cur.close()

    @classmethod

    def _row2ob(cls, row, **kwargs):

        o = cls.__new__(cls)

        for attr, value in zip([x[0] for x in cls._table_static_fields], row):

            setattr(o, attr, value)

        return o

    @classmethod

    def get_data_fields(cls):

        return []

SQL_LEVEL = (71, 'add testdef table')

    if sql_level < 71:

        # 71: create testdef table

        TestDef.do_table()

# w.c.s. - web application for online forms

# Copyright (C) 2005-2022  Entr'ouvert

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, see <http://www.gnu.org/licenses/>.

from wcs import sql

class TestDef(sql.TestDef):

    _names = 'testdef'

    name = ''

    slug = None

    object_type = None  # (formdef, carddef, etc.)

    object_id = None

    data = None  # (json export of formdata, carddata, etc.)