tests: run against postgresql (#67383)

tests/test_admin_pages.py

@@ -20,7 +20,7 @@ from wcs.workflows import Workflow
 from wcs.formdef import FormDef
 from wcs import fields
 
-from utilities import get_app, login, create_temporary_pub
+from utilities import get_app, login, create_temporary_pub, clean_temporary_pub
 
 
 def setup_module(module):
@@ -36,16 +36,20 @@ def setup_module(module):
     pub.write_cfg()
 
 
+def teardown_module(module):
+    clean_temporary_pub()
+
+
 def create_superuser():
     global user1
-    if pub.user_class.has_key('admin'):
-        user1 = pub.user_class.get('admin')
+    if pub.user_class.get_users_with_name_identifier('admin'):
+        user1 = pub.user_class.get_users_with_name_identifier('admin')[0]
         user1.is_admin = True
         user1.roles = []
         return
     user1 = pub.user_class(name='admin')
-    user1.id = 'admin'
     user1.is_admin = True
+    user1.name_identifiers = ['admin']
     user1.roles = []
     user1.store()
 

tests/test_backoffice_pages.py

@@ -10,12 +10,11 @@ from wcs.qommon.ident.password_accounts import PasswordAccount
 from wcs.qommon.http_request import HTTPRequest
 from wcs.qommon.template import get_current_theme
 from wcs.categories import Category
-from wcs.roles import Role
 from wcs.workflows import Workflow
 from wcs.formdef import FormDef
 from wcs import fields
 
-from utilities import get_app, login, create_temporary_pub
+from utilities import get_app, login, create_temporary_pub, clean_temporary_pub
 
 
 def setup_module(module):
@@ -31,11 +30,15 @@ def setup_module(module):
     pub.write_cfg()
 
 
+def teardown_module(module):
+    clean_temporary_pub()
+
+
 def create_agent():
-    if pub.user_class.has_key('agent'):
-        return
+    if pub.user_class.get_users_with_name_identifier('agent'):
+        return pub.user_class.get_users_with_name_identifier('agent')[0]
     user1 = pub.user_class(name='agent')
-    user1.id = 'admin'
+    user1.name_identifiers = ['agent']
     user1.is_admin = False
     user1.store()
 
@@ -51,8 +54,8 @@ def create_agent():
 
 
 def create_role():
-    Role.wipe()
-    role = Role(name='foobar')
+    pub.role_class.wipe()
+    role = pub.role_class(name='foobar')
     role.allows_backoffice_access = True
     role.store()
     return role

tests/test_user_pages.py

@@ -10,12 +10,11 @@ from wcs.qommon.ident.password_accounts import PasswordAccount
 from wcs.qommon.http_request import HTTPRequest
 from wcs.qommon.template import get_current_theme
 from wcs.categories import Category
-from wcs.roles import Role
 from wcs.workflows import Workflow
 from wcs.formdef import FormDef
 from wcs import fields
 
-from utilities import get_app, login, create_temporary_pub
+from utilities import get_app, login, create_temporary_pub, clean_temporary_pub
 
 
 def setup_module(module):
@@ -31,11 +30,15 @@ def setup_module(module):
     pub.write_cfg()
 
 
+def teardown_module(module):
+    clean_temporary_pub()
+
+
 def create_user():
-    if pub.user_class.has_key('user'):
-        return
+    if pub.user_class.get_users_with_name_identifier('user'):
+        return pub.user_class.get_users_with_name_identifier('user')[0]
     user1 = pub.user_class(name='user')
-    user1.id = 'admin'
+    user1.name_identifiers = ['user']
     user1.is_admin = False
     user1.store()
 

tests/utilities.py

@@ -1,7 +1,11 @@
 import configparser
 import os
+import random
+import shutil
 import tempfile
 
+import psycopg2
+
 from webtest import TestApp
 from quixote import cleanup, get_publisher
 from django.conf import settings
@@ -10,7 +14,7 @@ from wcs.qommon.publisher import set_publisher_class
 import wcs
 import wcs.wsgi
 from wcs.qommon.http_request import HTTPRequest
-from wcs import publisher, compat
+from wcs import publisher, compat, sql
 
 import wcs.middleware
 
@@ -31,12 +35,69 @@ def create_temporary_pub():
     compat.CompatWcsPublisher.configure(config)
     compat.CompatWcsPublisher.init_publisher_class()
     pub = compat.CompatWcsPublisher.create_publisher()
-    # allow saving the user
     pub.app_dir = os.path.join(APP_DIR, 'example.net')
     os.mkdir(pub.app_dir)
+
+    # set classes
+    pub.user_class = sql.SqlUser
+    pub.role_class = sql.Role
+    pub.token_class = sql.Token
+    pub.tracking_code_class = sql.TrackingCode
+    pub.session_class = sql.Session
+    pub.custom_view_class = sql.CustomView
+    pub.snapshot_class = sql.Snapshot
+    pub.loggederror_class = sql.LoggedError
+
+    conn = psycopg2.connect(user=os.environ['USER'], dbname='postgres')
+    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
+    i = 0
+    while True:
+        dbname = 'wcstests%d' % random.randint(0, 100000)
+        try:
+            cur = conn.cursor()
+            cur.execute('CREATE DATABASE %s' % dbname)
+            break
+        except psycopg2.Error:
+            if i < 5:
+                i += 1
+                continue
+            raise
+        finally:
+            cur.close()
+
+    pub.cfg['postgresql'] = {'database': dbname, 'user': os.environ['USER']}
+    pub.write_cfg()
+
+    sql.do_user_table()
+    sql.do_role_table()
+    sql.do_tokens_table()
+    sql.do_tracking_code_table()
+    sql.do_session_table()
+    sql.do_transient_data_table()
+    sql.do_custom_views_table()
+    sql.do_snapshots_table()
+    sql.do_loggederrors_table()
+    sql.do_meta_table()
+    sql.init_global_table()
+    conn.close()
+
     return pub
 
 
+def clean_temporary_pub():
+    pub = get_publisher()
+    conn = psycopg2.connect(user=os.environ['USER'], dbname='postgres')
+    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
+    try:
+        cur = conn.cursor()
+        cur.execute('DROP DATABASE %s' % pub.cfg['postgresql']['database'])
+        cur.close()
+    except psycopg2.Error as e:
+        print(e)
+    shutil.rmtree(pub.app_dir)
+    pub.cleanup()
+
+
 def get_app(pub, https=False):
     extra_environ = {'HTTP_HOST': 'example.net', 'REMOTE_ADDR': '127.0.0.1'}
     if https: