Produits Entr'ouvert » w.c.s.

    assert pub.cfg['misc']['default-position'] == {'lat': 1.234, 'lon': -1.234}

    assert data_class.get(resp.json['data']['id']).data['5'] == {'lat': 1.5, 'lon': 2.25}

    for map_value in ('1.5;2.25', {'lat': 1.5, 'lon': 2.25}):

        formdata.data['5'] = map_value

        payload = json.loads(

                    json.dumps(formdata.get_json_export_dict(),

                        cls=qommon.misc.JSONEncoder))

        signed_url = sign_url('http://example.net/api/formdefs/test/submit?orig=coucou', '1234')

        url = signed_url[len('http://example.net'):]

        resp = get_app(pub).post_json(url, payload)

        assert resp.json['err'] == 0

        new_formdata = formdef.data_class().get(resp.json['data']['id'])

        assert new_formdata.data['0'] == formdata.data['0']

        assert new_formdata.data['1'] == formdata.data['1']

        assert new_formdata.data['1_display'] == formdata.data['1_display']

        assert new_formdata.data['1_structured'] == formdata.data['1_structured']

        assert new_formdata.data['2'] == formdata.data['2']

        assert new_formdata.data['2_display'] == formdata.data['2_display']

        assert new_formdata.data['3'] == formdata.data['3']

        assert new_formdata.data['4'].get_content() == formdata.data['4'].get_content()

        assert new_formdata.data['5'] == {'lat': 1.5, 'lon': 2.25}

        assert new_formdata.data['bo1'] == formdata.data['bo1']

        assert not new_formdata.data.get('6')

        assert new_formdata.user_id is None

            "value\r\n" % (

                '%(lat)s;%(lon)s' % pub.get_default_position(),

                today))

    assert card1.data['1'] == {'lat': 48.81, 'lon': 2.37}

    assert fields.MapField().get_json_value({'lat': 42.2, 'lon': 10.2}) == {'lat': 42.2, 'lon': 10.2}

    assert fields.MapField().get_json_value({'lat': -42.2, 'lon': 10.2}) == {'lat': -42.2, 'lon': 10.2}

    assert fields.MapField().get_json_value(None) is None

    assert data.data == {'0': {'lat': 1.234, 'lon': -1.234}, '1': 'bla'}

def test_map_field_migration(pub):

    if pub.is_using_postgresql():

        pytest.skip('this is not relevant in SQL mode')

        return

    user = create_user(pub)

    formdef = create_formdef()

    formdef.data_class().wipe()

    formdef.fields = [fields.MapField(id='0', label='map', type='map')]

    formdef.store()

    formdata = formdef.data_class()()

    formdata.user_id = user.id

    formdata.data = {'0': '1;2'}  # legacy data type

    formdata.just_created()

    formdata.store()

    app = login(get_app(pub), username='foo', password='foo')

    resp = app.get(formdata.get_url())

    assert 'data-init-lat="1"' in resp

    assert 'data-init-lng="2"' in resp

    assert data.data == {'1': {'lat': 1.234, 'lon': -1.234}, '3': 'bar'}

        '7': {'lat': 2, 'lon': 4},  # map

            assert WorkflowStatusItem.compute('=form_var_map["lat"]', raises=True) == 2

            assert WorkflowStatusItem.compute('{{ form_var_map_lat }}', raises=True) == '2'

            assert WorkflowStatusItem.compute('{{ form_var_map_lon }}', raises=True) == '4'

    assert pub.get_default_position() == {'lat': 50.84, 'lon': 4.36}

@postgresql

def test_migration_43_map_data_type():

    formdef = FormDef()

    formdef.name = 'test map migration'

    formdef.fields = [

        fields.MapField(id='1', label='map', type='map'),

    ]

    formdef.store()

    formdata1 = formdef.data_class(mode='sql')()

    formdata1.just_created()

    formdata1.store()

    formdata2 = formdef.data_class(mode='sql')()

    formdata2.just_created()

    formdata2.store()

    formdata3 = formdef.data_class(mode='sql')()

    formdata3.just_created()

    formdata3.store()

    conn, cur = sql.get_connection_and_cursor()

    cur.execute('UPDATE wcs_meta SET value = 42 WHERE key = %s', ('sql_level',))

    conn.commit()

    cur.close()

    conn, cur = sql.get_connection_and_cursor()

    cur.execute('ALTER TABLE %s DROP COLUMN f1 CASCADE' % sql.get_formdef_table_name(formdef))

    cur.execute('ALTER TABLE %s ADD COLUMN f1 VARCHAR' % sql.get_formdef_table_name(formdef))

    cur.execute('UPDATE ' + sql.get_formdef_table_name(formdef) + ' SET f1 = %s WHERE id = %s', ('1;2', formdata1.id))

    cur.execute('UPDATE ' + sql.get_formdef_table_name(formdef) + ' SET f1 = %s WHERE id = %s', ('', formdata2.id))

    cur.execute('UPDATE ' + sql.get_formdef_table_name(formdef) + ' SET f1 = %s WHERE id = %s', (None, formdata3.id))

    conn.commit()

    cur.close()

    sql.migrate()

    conn, cur = sql.get_connection_and_cursor()

    assert migration_level(cur) >= 43

    conn.commit()

    cur.close()

    assert formdef.data_class(mode='sql').get(formdata1.id).data['1'] == {'lat': 1, 'lon': 2}

    assert formdef.data_class(mode='sql').get(formdata2.id).data['1'] is None

    assert formdef.data_class(mode='sql').get(formdata3.id).data['1'] is None

    assert widget.parse() == {'lat': 1.23, 'lon': 2.34}

    formdata.data = {'2': {'lat': 48.8337085, 'lon': 2.3233693}}

    formdata.data = {'2': {'lat': 48.8337085, 'lon': 2.3233693}}

    formdata.data = {'2': {'lat': 48.8337085, 'lon': 3.3233693}}

    formdata.data = {'2': {'lat': 48.8337085, 'lon': 4.3233693}}

                value = '%(lat)s;%(lon)s' % get_publisher().get_default_position()

                        value=geoloc_value,

        return (coords, False)

        if isinstance(value, str):  # compatiblity with old pickled data

            return indent + value

        return indent + '%(lat)s;%(lon)s' % value

        return {'lat': lat, 'lon': lon}

        return value

        if isinstance(value, str):

            # backward compatibility

            return self.convert_value_from_str(value)

        return value

        if isinstance(d.get(field.id), str) and field.convert_value_from_str:

        latlng_value = None

        if isinstance(value, str):  # legacy data type

            latlng_value = value

        elif value:

            latlng_value = '%s;%s' % (value['lat'], value['lon'])

        self.add(HiddenWidget, 'latlng', value=latlng_value)

    def point2str(self, value):

        if not value:

            return None

        return '%s;%s' % (value['lat'], value['lon'])

    def transfer_form_value(self, request):

        request.form[self.get_widget('latlng').name] = self.point2str(self.value)

        if isinstance(self.value, str):

        if isinstance(self.value, dict):

            return {'lat': self.value['lat'],

                    'lng': self.value['lon']}

            self.value = misc.normalize_geolocation({'lat': lat, 'lon': lon})

            default_position = self.get_site_option('default_position') or '50.84;4.36'

        if isinstance(default_position, str):

            default_position = {

                'lat': float(default_position.split(';')[0]),

                'lon': float(default_position.split(';')[1]),

            }

        default_position = self.get_default_position()

        attrs['data-def-lat'] = default_position['lat']

        attrs['data-def-lng'] = default_position['lon']

{% localize off %}

<input type="hidden" name="{{widget.name}}$latlng" {% if widget.value %}value="{{widget.value.lat}};{{widget.value.lon}}"{% endif %}>

{% endlocalize %}

    'map': 'jsonb',

SQL_LEVEL = 43

@guard_postgres

def migrate_map_data_type():

    conn, cur = get_connection_and_cursor()

    from wcs.formdef import FormDef

    from wcs.carddef import CardDef

    had_changes = False

    for formdef in FormDef.select() + CardDef.select():

        table_name = get_formdef_table_name(formdef)

        cur.execute('''SELECT column_name, udt_name FROM information_schema.columns

                        WHERE table_schema = 'public'

                          AND table_name = %s''', (table_name,))

        existing_fields = {x[0]: x[1] for x in cur.fetchall()}

        for field in formdef.get_all_fields():

            if field.type != 'map':

                continue

            database_field_id = get_field_id(field)

            if existing_fields.get(database_field_id) == 'jsonb':

                # already ok

                continue

            sql_type = SQL_TYPE_MAPPING.get(field.key, 'varchar')

            cur.execute('''ALTER TABLE %s ADD COLUMN %s %s''' % (

                                    table_name,

                                    'tmp_' + database_field_id,

                                    sql_type))

            cur.execute('''UPDATE %(table_name)s

                              SET tmp_%(column)s = jsonb_build_object(

                                        'lat', split_part(%(column)s, ';', 1)::float,

                                        'lon', split_part(%(column)s, ';', 2)::float)

                              WHERE %(column)s IS NOT NULL

                                AND %(column)s != '' ''' % {

                             'table_name': table_name,

                             'column': database_field_id

                        })

            cur.execute('''ALTER TABLE %s DROP COLUMN %s CASCADE''' % (

                table_name,

                database_field_id))

            cur.execute('''ALTER TABLE %s RENAME COLUMN tmp_%s TO %s''' % (

                table_name,

                database_field_id,

                database_field_id))

            had_changes = True

    if had_changes:

        # views have to be recreated

        migrate_views(conn, cur)

    conn.commit()

    cur.close()

    if sql_level < 43:

        # 43: migrate map field data type

        migrate_map_data_type()

    def __str__(self):

        # backward compatibility

        value = self._data.get(self._field.id)

        if not value:

            return ''

        return '%(lat)s;%(lon)s' % value