tests: add more real tests

tests/conftest.py

@@ -2,18 +2,20 @@ import os
 import glob
 import json
 from contextlib import closing
+import subprocess
+import tempfile
+import shutil
 
 import pytest
 
 import django_webtest
 
-import sqlparse
-
 import psycopg2
 
 
 from django.db import connection
 from django.contrib.auth.models import User
+from django.core.management import call_command
 
 
 @pytest.fixture
@@ -47,48 +49,85 @@ def admin(db):
 SCHEMA_PATHS = os.path.join(os.path.dirname(__file__), 'fixtures/')
 
 
-def load_schema(settings, tmpdir, schema):
+def load_schema_db(schema):
     import random
 
     database_name = 'db%s' % random.getrandbits(20)
+    tmpdir = tempfile.mkdtemp()
+    bijoe_schema_dir = os.path.join(tmpdir, 'schemas')
+    os.makedirs(bijoe_schema_dir)
+
     try:
         with closing(psycopg2.connect('')) as conn:
             conn.set_isolation_level(0)
             with conn.cursor() as cursor:
                 cursor.execute('CREATE DATABASE %s' % database_name)
 
-        schema_dir = tmpdir.mkdir('schemas')
-        SCHEMA_DIR = os.path.join(SCHEMA_PATHS, schema)
+        schema_dir = os.path.join(SCHEMA_PATHS, schema)
 
         # copy schemas and set pg_dsn
-        for schema_path in glob.glob(os.path.join(SCHEMA_DIR, '*.json')):
+        for schema_path in glob.glob(os.path.join(schema_dir, '*_schema.json')):
             with open(schema_path) as f:
                 schema = json.load(f)
                 schema['pg_dsn'] = 'dbname=%s' % database_name
-                new_schema_path = schema_dir.join(os.path.basename(schema_path))
-                new_schema_path.write(json.dumps(schema))
+                new_schema_path = os.path.join(bijoe_schema_dir, os.path.basename(schema_path))
+                with open(new_schema_path, 'w') as fp:
+                    json.dump(schema, fp)
 
-        settings.BIJOE_SCHEMAS = [str(schema_dir.join('*.json'))]
-        with closing(psycopg2.connect(database=database_name)) as conn:
-            conn.set_isolation_level(0)
-            with conn.cursor() as cursor:
-                for sql_path in sorted(glob.glob(os.path.join(SCHEMA_DIR, '*.sql'))):
-                    with open(sql_path) as sql_file:
-                        sql = sql_file.read()
-                        statements = sqlparse.split(sql)
-                        for statement in statements:
-                            if not statement.strip(';').strip():
-                                continue
-                            cursor.execute(statement.rstrip(';'))
-        yield
+        # load data
+        for sql_path in sorted(glob.glob(os.path.join(schema_dir, '*.sql'))):
+            process = subprocess.Popen(['psql', '--single-transaction', database_name, '-f', sql_path],
+                                       stdout=subprocess.PIPE,
+                                       stderr=subprocess.PIPE)
+            stdout, stderr = process.communicate()
+            return_code = process.returncode
+            assert return_code == 0, [stdout, stderr]
+
+        # load fixtures
+        fixtures = sorted(glob.glob(os.path.join(schema_dir, '*_fixture.json')))
+
+        d = {
+            'schema_dir': schema_dir,
+            'database_name': database_name,
+            'bijoe_schemas': [os.path.join(bijoe_schema_dir, '*_schema.json')],
+            'fixtures': fixtures
+        }
+        tables_path = os.path.join(schema_dir, 'tables.json')
+        if os.path.exists(tables_path):
+            with open(tables_path) as f:
+                d['tables'] = json.load(f)
+        yield d
     finally:
         with closing(psycopg2.connect('')) as conn:
             conn.set_isolation_level(0)
             with conn.cursor() as cursor:
                 cursor.execute('DROP DATABASE IF EXISTS %s' % database_name)
+        shutil.rmtree(tmpdir)
+
+
+@pytest.fixture(scope='module')
+def schema1_db():
+    for x in load_schema_db('schema1'):
+        yield x
 
 
 @pytest.fixture
-def schema1(db, settings, tmpdir):
-    for _ in load_schema(settings, tmpdir, 'schema1'):
-        yield _
+def schema1(db, schema1_db, settings):
+    settings.BIJOE_SCHEMAS = schema1_db['bijoe_schemas']
+    for json_fixture in schema1_db['fixtures']:
+        call_command('loaddata', json_fixture)
+    return schema1_db
+
+
+@pytest.fixture(scope='module')
+def schema2_db():
+    for x in load_schema_db('schema2'):
+        yield x
+
+
+@pytest.fixture
+def schema2(db, schema2_db, settings):
+    settings.BIJOE_SCHEMAS = schema2_db['bijoe_schemas']
+    for json_fixture in schema2_db['fixtures']:
+        call_command('loaddata', json_fixture)
+    return schema2_db

tests/fixtures/schema2/README

@@ -0,0 +1 @@
+exemple datas

tests/test_schema2.py

@@ -0,0 +1,34 @@
+import json
+import os
+
+import pytest
+
+from utils import login, get_table
+
+
+def pytest_generate_tests(metafunc):
+    if hasattr(metafunc, 'function'):
+        fcode = metafunc.function.func_code
+        if 'visualization' in fcode.co_varnames[:fcode.co_argcount]:
+            tables = json.load(
+                open(
+                    os.path.join(
+                        os.path.dirname(__file__),
+                        'fixtures',
+                        'schema2',
+                        'tables.json')))
+            metafunc.parametrize(['visualization'], [[x] for x in tables])
+
+
+@pytest.fixture(autouse=True)
+def freezetime(freezer):
+    freezer.move_to('2018-12-07 16:53:00')
+
+
+def test_simple(schema2, app, admin, visualization):
+    login(app, admin)
+    response = app.get('/')
+
+    visualization_page = response.click(lambda x: x == visualization)
+    assert 'big-msg-info' not in visualization_page
+    assert schema2['tables'][visualization] == get_table(visualization_page)

tox.ini

@@ -24,9 +24,9 @@ deps =
 	pytest-cov
 	pytest-random
 	pytest-django
+	pytest-freezegun
 	WebTest
 	django-webtest<1.9.3
-	sqlparse
 	pyquery
 commands =
         dj111: py.test {posargs: --junitxml=test_{envname}_results.xml --cov-report xml --cov-report html --cov=bijoe --random tests/}