84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.cache import cache
|
|
import django_webtest
|
|
import psycopg2
|
|
from psycopg2 import sql
|
|
import pytest
|
|
|
|
from passerelle_reunion_unicite.models import UnicityReunionConnector
|
|
from passerelle.base.models import ApiUser, AccessRight
|
|
|
|
|
|
DB_NAME = 'wcs-reunion-unicite-test-%s' % os.environ.get("BRANCH_NAME", "").replace('/', '-')[:63]
|
|
TABLE_NAME = 'test_table'
|
|
|
|
|
|
@pytest.fixture
|
|
def wcsdb():
|
|
connection = psycopg2.connect(user='', password='', host='', port='', database='')
|
|
connection.set_session(autocommit=True)
|
|
cursor = connection.cursor()
|
|
cursor.execute(sql.SQL('DROP DATABASE IF EXISTS {}').format(sql.Identifier(DB_NAME)))
|
|
cursor.execute(sql.SQL('CREATE DATABASE {}').format(sql.Identifier(DB_NAME)))
|
|
new_conn = psycopg2.connect(user='', password='', host='', port='', database=DB_NAME)
|
|
new_conn.set_session(autocommit=True)
|
|
new_cursor = new_conn.cursor()
|
|
create_table = '''
|
|
CREATE TABLE {} (
|
|
id char(5) CONSTRAINT xxxx PRIMARY KEY,
|
|
f_siren varchar(40) NOT NULL,
|
|
f_iban varchar(40) NOT NULL,
|
|
status varchar(40) NOT NULL,
|
|
annee varchar(40) NOT NULL
|
|
)
|
|
'''
|
|
new_cursor.execute(sql.SQL(create_table).format(sql.Identifier(TABLE_NAME)))
|
|
yield new_cursor
|
|
new_conn.close()
|
|
cursor.execute(sql.SQL('DROP DATABASE {}').format(sql.Identifier(DB_NAME)))
|
|
connection.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def app(request):
|
|
wtm = django_webtest.WebTestMixin()
|
|
wtm._patch_settings()
|
|
cache.clear()
|
|
yield django_webtest.DjangoTestApp()
|
|
wtm._unpatch_settings()
|
|
|
|
|
|
@pytest.fixture
|
|
def connector(db):
|
|
connector = UnicityReunionConnector.objects.create(
|
|
slug='test', db_name=DB_NAME)
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(connector)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api,
|
|
resource_type=obj_type, resource_pk=connector.pk)
|
|
|
|
|
|
def test_unicite_siren_annee(app, connector, wcsdb):
|
|
wcsdb.execute(
|
|
sql.SQL('INSERT INTO {} VALUES (%s, %s, %s, %s, %s)').format(sql.Identifier(TABLE_NAME)),
|
|
['1', '1', '1', '1', '1']
|
|
)
|
|
resp = app.get('/passerelle-reunion-unicite/test/unicite_siren_annee?siren=1')
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == ['1']
|
|
|
|
|
|
def test_unicite_iban_annee(app, connector, wcsdb):
|
|
wcsdb.execute(
|
|
sql.SQL('INSERT INTO {} VALUES (%s, %s, %s, %s, %s)').format(sql.Identifier(TABLE_NAME)),
|
|
['1', '1', '1', '1', '1']
|
|
)
|
|
resp = app.get('/passerelle-reunion-unicite/test/unicite_iban_annee?iban=1')
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == ['1']
|