Produits Entr'ouvert » w.c.s.

# -*- coding: utf-8 -*-

import json

import pytest

import hashlib

    user.roles = [role.id]

    user.store()

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

    resp = app.post(formdata.get_url() + 'jump/trigger/XXX', status=302)

    resp = app.post(formdata.get_url() + 'jump/trigger/XXX', status=302)

    resp = app.post(formdata.get_url() + 'jump/trigger/XXX', status=302)

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

def test_display_message(pub):

    user = create_user(pub)

    formdata1 = data_class()

    formdata1.status = 'wf-st1'

    formdata1.just_created()

    formdata1.jump_status('st1') # will add another evolution entry

    formdata1.evolution[0].time = datetime.datetime(2015, 1, 1, 0, 0, 0).timetuple()

    formdata1.evolution[1].time = datetime.datetime(2015, 1, 2, 0, 0, 0).timetuple()

    formdata2 = data_class()

    formdata2.status = 'wf-st1'

    formdata2.just_created()

    formdata2.jump_status('st1') # will add another evolution entry

    formdata2.evolution[0].time = datetime.datetime(2015, 1, 3, 0, 0, 0).timetuple()

    formdata2.evolution[1].time = datetime.datetime(2015, 1, 4, 0, 0, 0).timetuple()

                if form.get_widget('date').parse():

                    date = form.get_widget('date').parse()

                    date = time.strptime(date, misc.date_format())

                self.fd = StringIO()

                t = tarfile.open('wcs.tar.gz', 'w:gz', fileobj=self.fd)

    who = None

    status = None

    time = None

    comment = None

    parts = None

    def get_json_export_dict(self, user, anonymise=False):

        data = {

            'time': self.time,

        }

        if self.status:

            data['status'] = self.status[3:]

            previous_status = self.pop_previous_marked_status()

            assert previous_status, 'failed to compute previous status'

            status_id = previous_status.id

        if not self.evolution:

            self.evolution = []

        self.evolution.append(evo)

        self.store()

    def get_url(self, backoffice = False):

    def get_last_update_time(self):

        if hasattr(self, '_last_update_time'):

            return self._last_update_time

            return self.evolution[-1].time

        else:

            return self.receipt_time

                                    who varchar,

                                    status varchar,

                                    time timestamp,

                                    comment text,

                                    parts bytea,

                                    formdata_id integer REFERENCES %s (id) ON DELETE CASCADE)''' % (

    for field in (existing_fields - needed_fields):

        cur.execute('''ALTER TABLE %s DROP COLUMN %s CASCADE''' % (table_name, field))

    if rebuild_views or len(existing_fields - needed_fields):

        # views may have been dropped when dropping columns, so we recreate

        # them even if not asked to.

            self._evolution = []

            return self._evolution

        conn, cur = get_connection_and_cursor()

                            WHERE formdata_id = %%(id)s

                         ORDER BY id''' % self._table_name

        cur.execute(sql_statement, {'id': self.id})

    @classmethod

    def _row2evo(cls, row, formdata):

        o = wcs.formdata.Evolution(formdata)

        if o.time:

            o.time = o.time.timetuple()

        return o

    def set_evolution(self, value):

        if not object_dict:

            return

        conn, cur = get_connection_and_cursor()

                             FROM %s_evolutions''' % cls._table_name

        sql_statement += ''' WHERE formdata_id IN %(object_ids)s ORDER BY id'''

        cur.execute(sql_statement, {'object_ids': tuple(object_dict.keys())})

            row = cur.fetchone()

            if row is None:

                break

            formdata = object_dict.get(formdata_id)

            if not formdata:

                continue

                    sql_dict.update({'id': evo._sql_id})

                    sql_statement = '''UPDATE %s_evolutions SET

                                        time = %%(time)s,

                                        status = %%(status)s,

                                        comment = %%(comment)s,

                                        parts = %%(parts)s

                else:

                    sql_statement = '''INSERT INTO %s_evolutions (

                                               id, who, status,

                                               formdata_id)

                                        VALUES (DEFAULT, %%(who)s, %%(status)s,

                                                %%(parts)s, %%(formdata_id)s)

                                     RETURNING id''' % self._table_name

                sql_dict.update({

                        'who': evo.who,

                        'status': evo.status,

                        'time': datetime.datetime.fromtimestamp(time.mktime(evo.time)),

                        'comment': evo.comment,

                        'formdata_id': self.id,

                })

    return result

def migrate_global_views(conn, cur):

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

        # 19: add geolocation to views

        # 20: remove user hash stuff

        # 22: rebuild views

        migrate_views(conn, cur)

    if sql_level < 21:

        # 3: introduction of _structured for user fields

        if self.timeout:

            timeout = int(self.compute(self.timeout))

            if last:

                diff = time.time() - time.mktime(last)

                must_jump = (diff > timeout) and must_jump

            anchor_status = self.anchor_status_first or formdata.status

            for evolution in formdata.evolution:

                if evolution.status == anchor_status:

                    break

        elif self.anchor == 'latest-arrival':

            anchor_status = self.anchor_status_latest or formdata.status

            for evolution in reversed(formdata.evolution):

                if evolution.status == anchor_status:

                    break

        elif self.anchor == 'python':

            variables = get_publisher().substitutions.get_context_variables()