From 8c73893b9184732f1f6700d3f9d7db1c2ca012fe Mon Sep 17 00:00:00 2001
From: Serghei Mihai <smihai@entrouvert.com>
Date: Mon, 25 Nov 2019 17:16:50 +0100
Subject: [PATCH] feeder: preseve categories and form names order (#36930)

---
 tests/test_wcs.py  |  29 ++++++++++--
 wcs_olap/feeder.py | 114 ++++++++++++++++++++++++++++++++++++++-------
 2 files changed, 122 insertions(+), 21 deletions(-)

diff --git a/tests/test_wcs.py b/tests/test_wcs.py
index a636dce..7a94fb1 100644
--- a/tests/test_wcs.py
+++ b/tests/test_wcs.py
@@ -10,6 +10,8 @@ import mock
 
 import utils
 
+from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
+
 
 def test_wcs_fixture(wcs, postgres_db, tmpdir, olap_cmd, caplog):
     olap_cmd()
@@ -18,6 +20,7 @@ def test_wcs_fixture(wcs, postgres_db, tmpdir, olap_cmd, caplog):
         ('agent', 'id'),
         ('agent', 'label'),
         ('category', 'id'),
+        ('category', 'ref'),
         ('category', 'label'),
         ('channel', 'id'),
         ('channel', 'label'),
@@ -67,6 +70,7 @@ def test_wcs_fixture(wcs, postgres_db, tmpdir, olap_cmd, caplog):
         ('formdata_demande_field_itemOpen', 'id'),
         ('formdata_demande_field_itemOpen', 'label'),
         ('formdef', 'id'),
+        ('formdef', 'ref'),
         ('formdef', 'category_id'),
         ('formdef', 'label'),
         ('hour', 'id'),
@@ -97,14 +101,14 @@ def test_wcs_fixture(wcs, postgres_db, tmpdir, olap_cmd, caplog):
             assert json_schema == expected_json_schema
 
 
-def test_requests_exception(wcs, postgres_db, tmpdir, olap_cmd, caplog):
+def test_requests_exception(wcs, tmpdir, olap_cmd, caplog):
     with mock.patch('requests.get', side_effect=requests.RequestException('wat!')):
         with pytest.raises(SystemExit):
             olap_cmd(no_log_errors=False)
     assert 'wat!' in caplog.text
 
 
-def test_requests_not_ok(wcs, postgres_db, tmpdir, olap_cmd, caplog):
+def test_requests_not_ok(wcs, tmpdir, olap_cmd, caplog):
     with mock.patch('requests.get') as mocked_get:
         mocked_get.return_value.ok = False
         mocked_get.return_value.status_code = 401
@@ -114,7 +118,7 @@ def test_requests_not_ok(wcs, postgres_db, tmpdir, olap_cmd, caplog):
     assert 'invalid signature' in caplog.text
 
 
-def test_requests_not_json(wcs, postgres_db, tmpdir, olap_cmd, caplog):
+def test_requests_not_json(wcs, tmpdir, olap_cmd, caplog):
     with mock.patch('requests.get') as mocked_get:
         mocked_get.return_value.ok = True
         mocked_get.return_value.json.side_effect = ValueError('invalid JSON')
@@ -197,3 +201,22 @@ formdata.store()
             formdata = c.fetchone()
             assert formdata[0] == bazouka_id
             assert formdata[1] == open_new_id
+
+
+def test_create_reference_column(postgres_db, olap_cmd):
+
+    olap_cmd()
+    conn = postgres_db.conn()
+    with postgres_db.conn() as conn:
+        with conn.cursor() as c:
+            conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
+            c.execute('ALTER TABLE olap.category DROP COLUMN ref')
+            c.execute('ALTER TABLE olap.formdef DROP COLUMN ref')
+
+    olap_cmd()
+
+    with postgres_db.conn() as conn:
+        with conn.cursor() as c:
+            c.execute('SELECT * FROM olap.category')
+            for line in c.fetchall():
+                assert len(line) == 3
diff --git a/wcs_olap/feeder.py b/wcs_olap/feeder.py
index 9cf58a1..d4148c3 100644
--- a/wcs_olap/feeder.py
+++ b/wcs_olap/feeder.py
@@ -25,7 +25,7 @@ def quote(name):
 
 
 def slugify(s):
-    return s.replace('-', '_').replace(' ', '_')
+    return s.replace('-', '_').replace(' ', '_').lower()
 
 
 class Context(object):
@@ -368,6 +368,10 @@ CREATE TABLE public.dates AS (SELECT
         self.ex(query, vars=(name,))
         return self.cur.fetchone()[0]
 
+    def update_table_sequence_number(self, name):
+        self.ex("""SELECT setval(pg_get_serial_sequence('{name}', 'id'),
+        (SELECT GREATEST(1, MAX(id)) FROM {name}))""", ctx={'name': quote(name)})
+
     def create_labeled_table_serial(self, name, comment):
         self.create_table(
             name, [['id', 'serial primary key'], ['label', 'varchar']], comment=comment)
@@ -378,9 +382,69 @@ CREATE TABLE public.dates AS (SELECT
                 'INSERT INTO {schema_temp}.{name} SELECT * FROM {schema}.{name}',
                 ctx={'name': quote(name)}
             )
-            # Update sequence
-            self.ex("""SELECT setval(pg_get_serial_sequence('{name}', 'id'),
-            (SELECT GREATEST(1, MAX(id)) FROM {name}))""", ctx={'name': quote(name)})
+            self.update_table_sequence_number(name)
+
+    def create_referenced_table(self, name, fields, comment):
+        # add primary key and reference fields
+        new_fields = [['id', 'serial primary key'], ['ref', 'varchar UNIQUE']] + fields
+        self.create_table(name, new_fields, comment=comment)
+
+        if self.prev_table_exists(name):
+            # verify if ref exists in table schema
+            self.ex("""SELECT column_name FROM information_schema.columns
+            WHERE table_name=%s AND table_schema='{schema}'""", vars=(name,))
+            columns = [c[0] for c in self.cur.fetchall()]
+            if 'ref' in columns:
+                self.ex('INSERT INTO {schema_temp}.{name} SELECT * FROM {schema}.{name}', ctx={'name': quote(name)})
+            else:
+                fields_labels =  ', '.join([f[0] for f in fields])
+                self.ex('SELECT {fields} FROM {schema}.{name}', ctx={'name': quote(name),
+                                                                     'fields': fields_labels})
+                new_values = []
+                for line in self.cur.fetchall():
+                    result = (slugify(line[-1]),) + line
+                    new_values.append(result)
+                if new_values:
+                    columns_number = len(line) + 1
+                    columns_values = ', '.join(['%s' for x in xrange(columns_number)])
+                    tmpl = ', '.join(['(DEFAULT, %s)' % columns_values] * len(new_values))
+                    query = 'INSERT INTO {schema_temp}.{name} VALUES %s' % tmpl
+                    query = self.ex(query, ctx={'name': quote(name)},
+                                    vars=list(itertools.chain(*new_values)))
+
+
+    def do_referenced_data(self, name, data, result_column, update_column='label'):
+        to_insert = []
+
+        for item in data:
+            ref = item[0]
+            self.ex(
+                'SELECT ref, {column} FROM {name} WHERE ref = %s',
+                ctx={'name': quote(name), 'column': quote(update_column)},
+                vars=(ref,))
+            if self.cur.fetchall():
+                for item in self.cur.fetchall():
+                    self.ex('UPDATE {name} SET {column}=%s WHERE ref=%s',
+                            ctx={'name': quote(name), 'column': quote(update_column)},
+                            vars=[item[1], ref])
+            else:
+                to_insert.append(item)
+
+        if to_insert:
+            columns_values = ', '.join(['%s' for x in xrange(len(item))])
+            tmpl = ', '.join(['(DEFAULT, %s)' % columns_values] * len(data))
+
+            query = 'INSERT INTO {name} VALUES %s' % tmpl
+            self.ex(query, ctx={'name': quote(name)}, # 'column': quote(update_column)},
+                    vars=list(itertools.chain(*to_insert)))
+
+        result = {}
+        self.ex('SELECT id, {column} FROM {name}', ctx={'name': quote(name),
+                                                        'column': result_column})
+        for _id, column  in self.cur.fetchall():
+            result[column] = _id
+        return result
+
 
     def create_labeled_table(self, name, labels, comment=None):
         self.create_table(
@@ -446,6 +510,30 @@ CREATE TABLE public.dates AS (SELECT
     def add_dim(self, **kwargs):
         self.dimensions.append(self.tpl(kwargs))
 
+    def do_category_table(self):
+        fields = [['label', 'varchar']]
+        table_name = self.ctx['category_table']
+        self.create_referenced_table(table_name, fields, 'catégorie')
+        categories_data = [(c.id, c.name) for c in self.categories]
+        tmp_cat_map = self.do_referenced_data(table_name, categories_data, 'label')
+        self.update_table_sequence_number(table_name)
+        # remap categories ids to ids in the table
+        return dict((c.id, tmp_cat_map[c.name]) for c in self.categories)
+
+    def do_formdef_table(self):
+        categories_mapping = self.do_category_table()
+
+        formdef_fields = [['category_id', 'integer REFERENCES {category_table} (id)'],
+                          ['label', 'varchar']
+        ]
+        table_name = self.ctx['form_table']
+        self.create_referenced_table(table_name, formdef_fields, 'types de formulaire')
+
+        formdefs_data = [(form.slug, categories_mapping.get(form.schema.category_id),
+                          form.schema.name) for form in self.formdefs if form.count]
+        self.formdefs_mapping = self.do_referenced_data(table_name, formdefs_data, 'ref')
+        self.update_table_sequence_number(table_name)
+
     def do_base_table(self):
         # channels
         self.create_labeled_table('channel', [[c[0], c[2]] for c in self.channels],
@@ -457,20 +545,15 @@ CREATE TABLE public.dates AS (SELECT
         self.role_mapping = dict(
             (role.id, tmp_role_map[role.name]) for role in self.roles)
 
-        # categories
-        tmp_cat_map = self.create_labeled_table(
-            'category', enumerate(c.name for c in self.categories), comment='catégorie')
-        self.categories_mapping = dict((c.id, tmp_cat_map[c.name]) for c in self.categories)
+        # forms
+        self.do_formdef_table()
 
         self.create_labeled_table('hour', zip(range(0, 24), map(str, range(0, 24))),
                                   comment='heures')
 
         self.create_labeled_table('status', self.status,
                                   comment='statuts simplifiés')
-        self.ex('CREATE TABLE {form_table} (id serial PRIMARY KEY,'
-                ' category_id integer REFERENCES {category_table} (id),'
-                ' label varchar)')
-        self.ex('COMMENT ON TABLE {form_table} IS %s', vars=('types de formulaire',))
+
         # agents
         self.create_labeled_table_serial('agent', comment='agents')
 
@@ -606,11 +689,6 @@ class WcsFormdefFeeder(object):
         self.status_mapping = dict((s.id, tmp_status_map[s.name]) for s in statuses)
 
     def do_data_table(self):
-        self.ex('INSERT INTO {form_table} (category_id, label) VALUES (%s, %s) RETURNING (id)',
-                vars=[self.categories_mapping.get(self.formdef.schema.category_id),
-                      self.formdef.schema.name])
-        self.formdef_sql_id = self.cur.fetchone()[0]
-
         columns = OrderedDict()
         columns['status_id'] = {'sql_col_name': 'status_id', 'sql_col_def': 'smallint REFERENCES {status_table} (id)'}
 
@@ -766,7 +844,7 @@ class WcsFormdefFeeder(object):
             if channel == 'web' and data.submission.backoffice:
                 channel = 'backoffice'
             row = {
-                'formdef_id': self.formdef_sql_id,
+                'formdef_id': self.formdefs_mapping[self.formdef.slug],
                 'receipt_time': data.receipt_time,
                 'hour_id': data.receipt_time.hour,
                 'channel_id': self.channel_to_id[channel],
-- 
2.25.0

