251 lines
8.9 KiB
Python
251 lines
8.9 KiB
Python
import datetime
|
|
import re
|
|
from StringIO import StringIO
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.files import File
|
|
import pytest
|
|
|
|
from passerelle.base.models import ApiUser, AccessRight, ResourceLog, ResourceStatus
|
|
from passerelle.apps.csvdatasource.models import CsvDataSource, Query
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
@pytest.fixture
|
|
def simple_user():
|
|
try:
|
|
user = User.objects.get(username='user')
|
|
except User.DoesNotExist:
|
|
user = User.objects.create_user('user', password='user')
|
|
return user
|
|
|
|
@pytest.fixture
|
|
def admin_user():
|
|
try:
|
|
user = User.objects.get(username='admin')
|
|
except User.DoesNotExist:
|
|
user = User.objects.create_superuser('admin', email=None, password='admin')
|
|
return user
|
|
|
|
def login(app, username='admin', password='admin'):
|
|
login_page = app.get('/login/')
|
|
login_form = login_page.forms[0]
|
|
login_form['username'] = username
|
|
login_form['password'] = password
|
|
resp = login_form.submit()
|
|
assert resp.status_int == 302
|
|
return app
|
|
|
|
def test_homepage_redirect(app):
|
|
assert app.get('/', status=302).location.endswith('/manage/')
|
|
|
|
def test_unlogged_access(app):
|
|
# connect while not being logged in
|
|
assert app.get('/manage/', status=302).location.endswith('/login/?next=/manage/')
|
|
|
|
def test_simple_user_access(app, simple_user):
|
|
# connect while being logged as a simple user
|
|
app = login(app, username='user', password='user')
|
|
assert app.get('/manage/', status=403)
|
|
assert app.get('/manage/add', status=403)
|
|
assert app.get('/manage/ovh/add', status=403)
|
|
assert app.get('/manage/access/', status=403)
|
|
|
|
def test_access(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/', status=200)
|
|
assert 'Add Connector' in resp.body
|
|
assert app.get('/manage/access/', status=200)
|
|
|
|
def test_add_connector(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Add Connector')
|
|
assert 'Business Process Connectors' in resp.body
|
|
assert 'Geographic information system' in resp.body
|
|
resp = resp.click('Base Adresse Web Service')
|
|
resp.forms[0]['title'] = 'Test Connector'
|
|
resp.forms[0]['description'] = 'Connector for a simple test'
|
|
resp.forms[0]['service_url'] = 'https://api-adresse.data.gouv.fr/'
|
|
resp = resp.forms[0].submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith('/base-adresse/test-connector/')
|
|
resp = resp.follow()
|
|
assert 'Base Adresse Web Service - Test Connector' in resp.body
|
|
|
|
resp = app.get('/manage/', status=200)
|
|
assert 'Test Connector' in resp.body
|
|
|
|
def test_visit_connectors(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Add Connector')
|
|
for link in re.findall('href="(/manage.*add)"', resp.body):
|
|
resp = app.get(link, status=200)
|
|
|
|
def test_access_management(app, admin_user):
|
|
assert ApiUser.objects.count() == 0
|
|
app = login(app)
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Access Management')
|
|
resp = resp.click('Add API User')
|
|
resp.form['username'] = 'foo'
|
|
resp.form['fullname'] = 'Foo'
|
|
resp = resp.form.submit().follow()
|
|
assert ApiUser.objects.count() == 1
|
|
assert ApiUser.objects.get(username='foo').fullname == 'Foo'
|
|
|
|
resp = resp.click('Add API User')
|
|
resp.form['username'] = 'bar'
|
|
resp.form['fullname'] = 'Bar'
|
|
resp.form['keytype'] = 'API'
|
|
resp = resp.form.submit()
|
|
assert 'Key can not be empty' in resp.body
|
|
|
|
def test_menu_json(app, admin_user):
|
|
app.get('/manage/menu.json', status=302)
|
|
|
|
app = login(app)
|
|
resp = app.get('/manage/menu.json')
|
|
assert resp.headers['content-type'] == 'application/json'
|
|
assert resp.json[0]['label'] == 'Web Services'
|
|
|
|
resp = app.get('/manage/menu.json?callback=FooBar')
|
|
assert resp.headers['content-type'] == 'application/javascript'
|
|
assert resp.content.startswith('FooBar([{"')
|
|
|
|
def test_logs(app, admin_user):
|
|
data = StringIO('1;Foo\n2;Bar\n3;Baz')
|
|
csv = CsvDataSource.objects.create(csv_file=File(data, 't.csv'),
|
|
columns_keynames='id, text', slug='test', title='a title', description='a description')
|
|
|
|
query = Query(slug='fooba', resource=csv, structure='array')
|
|
query.projections = '\n'.join(['id:int(id)', 'text:text'])
|
|
query.save()
|
|
|
|
api = ApiUser.objects.create(username='public',
|
|
fullname='public',
|
|
description='access for all',
|
|
keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(csv)
|
|
AccessRight.objects.create(codename='can_access',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=csv.pk,
|
|
)
|
|
|
|
app = login(app)
|
|
resp = app.get(csv.get_absolute_url())
|
|
assert '<p>No records found</p>' in resp.body
|
|
|
|
app.get('/csvdatasource/test/query/foobar/')
|
|
resp = app.get(csv.get_absolute_url())
|
|
assert 'endpoint GET /csvdatasource/test/query/foobar/ ' in resp.text
|
|
|
|
app.get('/csvdatasource/test/query/foobar/?q=toto')
|
|
resp = app.get(csv.get_absolute_url())
|
|
assert 'endpoint GET /csvdatasource/test/query/foobar/?q=toto' in resp.text
|
|
|
|
resp = resp.click('full page')
|
|
assert resp.text.count('<td class="timestamp">') == 4
|
|
assert resp.text.count('Error occurred while processing request') == 2
|
|
|
|
resp.form['q'] = 'toto'
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<td class="timestamp">') == 1
|
|
|
|
resp.form['q'] = datetime.date.today().strftime('%d/%m/%Y')
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<td class="timestamp">') == 4
|
|
|
|
resp.form['q'] = datetime.date.today().strftime('%d/%m/2010')
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<td class="timestamp">') == 0
|
|
|
|
resp.form['q'] = ''
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<td class="timestamp">') == 4
|
|
log_pk = re.findall(r'data-pk="(.*)"', resp.text)[0]
|
|
base_url = re.findall(r'data-log-base-url="(.*)"', resp.text)[0]
|
|
resp = app.get(base_url + log_pk + '/')
|
|
resp = app.get(base_url + '12345' + '/', status=404)
|
|
|
|
def test_logging_parameters(app, admin_user):
|
|
data = StringIO('1;Foo\n2;Bar\n3;Baz')
|
|
csv = CsvDataSource.objects.create(csv_file=File(data, 't.csv'),
|
|
columns_keynames='id, text', slug='test', title='a title', description='a description')
|
|
app = login(app)
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('logging parameters')
|
|
resp.form['log_level'] = 'ERROR'
|
|
resp = resp.form.submit()
|
|
assert CsvDataSource.objects.get(id=csv.id).log_level == 'ERROR'
|
|
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('logging parameters')
|
|
resp.form['log_level'] = 'DEBUG'
|
|
resp = resp.form.submit()
|
|
assert CsvDataSource.objects.get(id=csv.id).log_level == 'DEBUG'
|
|
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('logging parameters')
|
|
resp.form['trace_emails'] = 'fred@localhost'
|
|
resp = resp.form.submit()
|
|
assert CsvDataSource.objects.get(id=csv.id).logging_parameters.trace_emails == 'fred@localhost'
|
|
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('logging parameters')
|
|
assert resp.form['trace_emails'].value == 'fred@localhost'
|
|
|
|
|
|
def test_availability_parameters(app, admin_user, monkeypatch):
|
|
data = StringIO('1;Foo\n2;Bar\n3;Baz')
|
|
csv = CsvDataSource.objects.create(
|
|
csv_file=File(data, 't.csv'),
|
|
columns_keynames='id, text', slug='test', title='a title', description='a description')
|
|
app = login(app)
|
|
resp = app.get(csv.get_absolute_url())
|
|
|
|
assert csv.availability_parameters.run_check
|
|
# csv connector has the default check_status which does nothing
|
|
# so availability check is hidden
|
|
assert 'availability check' not in resp.text
|
|
|
|
def check_status(*args, **kwargs):
|
|
return True
|
|
|
|
monkeypatch.setattr(CsvDataSource, 'check_status', check_status)
|
|
|
|
resp = app.get(csv.get_absolute_url())
|
|
assert 'availability check' in resp.text
|
|
|
|
resp = resp.click('availability check')
|
|
assert 'up' in resp.text
|
|
resp.form['run_check'] = False
|
|
resp = resp.form.submit()
|
|
# Connector status not changed, availability parameters changed
|
|
assert not csv.availability_parameters.run_check
|
|
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('availability check')
|
|
resp.form['run_check'] = True
|
|
resp = resp.form.submit()
|
|
|
|
# Connector down
|
|
resource_type = ContentType.objects.get_for_model(csv)
|
|
status = ResourceStatus(
|
|
resource_type=resource_type, resource_pk=csv.pk,
|
|
status='down', message='')
|
|
status.save()
|
|
assert csv.down()
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('availability check')
|
|
resp.form['run_check'] = False
|
|
resp = resp.form.submit()
|
|
# Connector is put back up
|
|
assert not csv.availability_parameters.run_check
|
|
assert not csv.down()
|
|
status = csv.get_availability_status()
|
|
assert status.status == 'up'
|