420 lines
15 KiB
Python
420 lines
15 KiB
Python
import datetime
|
|
import re
|
|
|
|
from webtest import Upload
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.files import File
|
|
from django.utils.six import StringIO
|
|
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.text
|
|
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.text
|
|
assert 'Geographic information system' in resp.text
|
|
resp = resp.click('Base Adresse Web Service')
|
|
resp.forms[0]['title'] = 'Test Connector'
|
|
resp.forms[0]['slug'] = '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.text
|
|
|
|
resp = app.get('/manage/', status=200)
|
|
assert 'Test Connector' in resp.text
|
|
|
|
def test_add_connector_unique_slug(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Add Connector')
|
|
resp = resp.click('Base Adresse Web Service')
|
|
resp.forms[0]['title'] = 'Test Connector'
|
|
resp.forms[0]['slug'] = 'test-connector'
|
|
resp.forms[0]['description'] = 'Connector for a simple test'
|
|
resp.forms[0]['service_url'] = 'https://api-adresse.data.gouv.fr/'
|
|
resp1 = resp.forms[0].submit()
|
|
assert resp1.status_int == 302
|
|
|
|
resp2 = resp.forms[0].submit()
|
|
assert 'There were errors processing your form.' in resp2.text
|
|
assert 'this Identifier already exists.' in resp2.text
|
|
resp.forms[0]['slug'] = 'foo'
|
|
resp2 = resp.forms[0].submit()
|
|
assert resp2.status_int == 302
|
|
|
|
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.text):
|
|
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.text
|
|
|
|
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(b'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.text
|
|
|
|
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'
|
|
|
|
# Notification delays
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('availability check')
|
|
assert resp.form['notification_delays'].value == '0'
|
|
resp.form['notification_delays'] = '0,5,100'
|
|
resp = resp.form.submit().follow()
|
|
assert not resp.pyquery('.messages .warning')
|
|
assert csv.availability_parameters.notification_delays == '0,5,100'
|
|
|
|
resp = app.get(csv.get_absolute_url())
|
|
resp = resp.click('availability check')
|
|
resp.form['notification_delays'] = 'x'
|
|
resp = resp.form.submit()
|
|
assert len(resp.pyquery('#id_notification_delays_p .error'))
|
|
|
|
def test_jobs(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')
|
|
|
|
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 not 'jobs' in resp.text
|
|
|
|
csv.add_job('sample_job')
|
|
resp = app.get(csv.get_absolute_url())
|
|
assert 'jobs' in resp.text
|
|
assert 'sample_job' in resp.text
|
|
|
|
resp = resp.click('full page', index=1)
|
|
assert resp.text.count('<tr data-pk') == 1
|
|
assert resp.text.count('sample_job') == 1
|
|
|
|
resp.form['q'] = 'sample'
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<tr data-pk') == 1
|
|
|
|
resp.form['q'] = 'blah'
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<tr data-pk') == 0
|
|
|
|
resp.form['q'] = datetime.date.today().strftime('%d/%m/%Y')
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<tr data-pk') == 1
|
|
|
|
resp.form['q'] = datetime.date.today().strftime('%d/%m/2010')
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<tr data-pk') == 0
|
|
|
|
resp.form['q'] = ''
|
|
resp = resp.form.submit()
|
|
assert resp.text.count('<tr data-pk') == 1
|
|
job_pk = re.findall(r'data-pk="(.*)"', resp.text)[0]
|
|
base_url = re.findall(r'data-job-base-url="(.*)"', resp.text)[0]
|
|
resp = app.get(base_url + job_pk + '/')
|
|
resp = app.get(base_url + '12345' + '/', status=404)
|
|
|
|
|
|
def test_manager_import_export(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')
|
|
csv2 = CsvDataSource.objects.create(csv_file=File(data, 't.csv'),
|
|
columns_keynames='id, text', slug='test2', title='a title', description='a description')
|
|
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,
|
|
)
|
|
|
|
# export site
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
resp = resp.click('Export')
|
|
assert resp.headers['content-type'] == 'application/json'
|
|
site_export = resp.text
|
|
|
|
# invalid json
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Import')
|
|
resp.form['site_json'] = Upload('export.json', b'garbage', 'application/json')
|
|
resp = resp.form.submit()
|
|
assert 'File is not in the expected JSON format.' in resp.text
|
|
|
|
# empty json
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Import')
|
|
resp.form['site_json'] = Upload('export.json', b'{}', 'application/json')
|
|
resp = resp.form.submit().follow()
|
|
assert CsvDataSource.objects.count() == 2
|
|
|
|
# import site
|
|
CsvDataSource.objects.all().delete()
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Import')
|
|
resp.form['site_json'] = Upload('export.json', site_export.encode('utf-8'), 'application/json')
|
|
resp = resp.form.submit().follow()
|
|
assert CsvDataSource.objects.count() == 2
|
|
|
|
# export connector
|
|
resp = app.get('/%s/%s/' % (csv.get_connector_slug(), csv.slug), status=200)
|
|
resp = resp.click('Export')
|
|
assert resp.headers['content-type'] == 'application/json'
|
|
connector_export = resp.text
|
|
|
|
# import connector
|
|
csv.delete()
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Import')
|
|
resp.form['site_json'] = Upload('export.json', connector_export.encode('utf-8'),
|
|
'application/json')
|
|
resp = resp.form.submit().follow()
|
|
assert CsvDataSource.objects.count() == 2
|
|
assert CsvDataSource.objects.filter(slug='test').exists()
|
|
|
|
# import users
|
|
ApiUser.objects.all().delete()
|
|
AccessRight.objects.all().delete()
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Import')
|
|
resp.form['site_json'] = Upload('export.json', site_export.encode('utf-8'), 'application/json')
|
|
resp = resp.form.submit().follow()
|
|
assert not ApiUser.objects.exists()
|
|
assert not AccessRight.objects.exists()
|
|
resp = resp.click('Import')
|
|
resp.form['import_users'] = True
|
|
resp.form['site_json'] = Upload('export.json', site_export.encode('utf-8'), 'application/json')
|
|
resp = resp.form.submit().follow()
|
|
assert ApiUser.objects.filter(username='public').exists()
|
|
assert AccessRight.objects.filter(codename='can_access').exists()
|