848 lines
30 KiB
Python
848 lines
30 KiB
Python
import datetime
|
|
import json
|
|
import re
|
|
import uuid
|
|
from io import StringIO
|
|
|
|
import freezegun
|
|
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.files import File
|
|
from django.utils.encoding import force_bytes
|
|
from django.utils.timezone import now
|
|
from webtest import Upload
|
|
|
|
from passerelle.apps.csvdatasource.models import CsvDataSource, Query
|
|
from passerelle.apps.photon.models import Photon
|
|
from passerelle.base.models import AccessRight, ApiUser, Job, ResourceLog, ResourceStatus
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
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')
|
|
assert resp.html.find('input', {'name': 'title'}).attrs['data-slug-sync'] == 'slug'
|
|
|
|
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_edit_connector(app, admin_user):
|
|
Photon.objects.create(
|
|
slug='test-connector',
|
|
title='Test Connector',
|
|
description='Connector for a simple test',
|
|
service_url='https://example.org/',
|
|
)
|
|
app = login(app)
|
|
resp = app.get('/manage/photon/test-connector/edit', status=200)
|
|
assert 'Test Connector' in resp.text
|
|
|
|
resp.forms[0]['title'] = 'Connector updated'
|
|
resp.forms[0]['description'] = 'simple test'
|
|
resp.forms[0]['service_url'] = 'https://example.net/'
|
|
resp = resp.forms[0].submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith('/photon/test-connector/')
|
|
resp = resp.follow()
|
|
assert 'Connector updated' in resp.text
|
|
|
|
resp = app.get('/manage/photon/test-connector/edit', status=200)
|
|
resp.forms[0]['slug'] = 'test'
|
|
resp = resp.forms[0].submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith('/photon/test/')
|
|
|
|
|
|
def test_edit_connector_unique_slug(app, admin_user):
|
|
Photon.objects.create(
|
|
slug='test-connector',
|
|
title='Test Connector',
|
|
description='Connector for a simple test',
|
|
service_url='https://example.org/',
|
|
)
|
|
Photon.objects.create(
|
|
slug='other-connector',
|
|
title='Other Connector',
|
|
description='Connector for another test',
|
|
service_url='https://example.org/',
|
|
)
|
|
app = login(app)
|
|
resp = app.get('/manage/photon/test-connector/edit', status=200)
|
|
assert 'Test Connector' in resp.text
|
|
assert resp.html.find('input', {'name': 'title'}).attrs['data-slug-sync'] == 'slug'
|
|
|
|
resp.forms[0]['slug'] = 'other-connector'
|
|
resp = resp.forms[0].submit()
|
|
assert resp.status_int == 200
|
|
assert 'Photon with this Identifier already exists.' in resp.text
|
|
|
|
# if title and slug name are not in sync, disable prepopulate slug by JS
|
|
resp.forms[0]['slug'] = 'another-connector'
|
|
resp = resp.forms[0].submit()
|
|
resp = app.get('/manage/photon/another-connector/edit', status=200)
|
|
assert 'data-slug-sync' not in resp.html.find('input', {'name': 'title'}).attrs
|
|
|
|
|
|
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
|
|
|
|
resp = resp.click('Add API User')
|
|
resp.form['username'] = 'bar'
|
|
resp.form['fullname'] = 'Bar'
|
|
resp.form['keytype'] = 'SIGN'
|
|
resp.form['key'] = '123'
|
|
resp = resp.form.submit()
|
|
assert 'Signature Key length must be at least 256 bits.' 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'] = 'Wed, 29 Sep 2021 00:40:48 +0200'
|
|
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 = resp.click('Search for logs from the same call')
|
|
assert resp.text.count('<td class="timestamp">') == 2
|
|
|
|
resp = app.get(base_url + '12345' + '/', status=404)
|
|
|
|
resp = app.get('/manage/csvdatasource/test/logs/?log_id=%s' % log_pk)
|
|
assert 'log_target' in resp.context
|
|
|
|
# unknown id
|
|
resp = app.get('/manage/csvdatasource/test/logs/?log_id=0')
|
|
assert 'log_target' not in resp.context
|
|
# bad id
|
|
resp = app.get('/manage/csvdatasource/test/logs/?log_id=foo')
|
|
assert 'log_target' not in resp.context
|
|
|
|
resp = app.get('/manage/csvdatasource/test/logs/?log_level=')
|
|
assert resp.text.count('<td class="timestamp">') == 4
|
|
assert resp.text.count('level-warning') == 2
|
|
assert resp.text.count('level-info') == 2
|
|
resp = app.get('/manage/csvdatasource/test/logs/?log_level=INFO')
|
|
assert resp.text.count('<td class="timestamp">') == 2
|
|
assert resp.text.count('level-warning') == 0
|
|
assert resp.text.count('level-info') == 2
|
|
resp = app.get('/manage/csvdatasource/test/logs/?log_level=ERROR')
|
|
assert resp.text.count('<td class="timestamp">') == 0
|
|
assert resp.text.count('level-warning') == 0
|
|
assert resp.text.count('level-info') == 0
|
|
resp = app.get('/manage/csvdatasource/test/logs/?log_level=foobar')
|
|
assert resp.text.count('<td class="timestamp">') == 4
|
|
assert resp.text.count('level-warning') == 2
|
|
assert resp.text.count('level-info') == 2
|
|
|
|
with freezegun.freeze_time('2020-10-06 14:08:12'):
|
|
log1 = ResourceLog.objects.create(
|
|
appname=csv.get_connector_slug(), slug=csv.slug, levelno=50, message='hop'
|
|
)
|
|
log2 = ResourceLog.objects.create(
|
|
appname=csv.get_connector_slug(), slug=csv.slug, levelno=42, message='hop'
|
|
)
|
|
resp = app.get('/manage/csvdatasource/test/logs/%s/' % log1.pk)
|
|
assert 'title="Critical - Oct. 6, 2020 14:08:12"' in resp.text
|
|
resp = app.get('/manage/csvdatasource/test/logs/%s/' % log2.pk)
|
|
assert 'title="Level 42 - Oct. 6, 2020 14:08:12"' in resp.text
|
|
|
|
|
|
def test_logs_tab_pagination(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
|
|
|
|
for i in range(20):
|
|
app.get('/csvdatasource/test/query/foobar/?q=toto%s' % i)
|
|
|
|
resp = app.get(csv.get_absolute_url())
|
|
assert resp.pyquery('#panel-logs[hidden]')
|
|
assert resp.pyquery('#panel-endpoints:not([hidden])')
|
|
assert resp.pyquery('#table-logs tr').length == 10
|
|
assert resp.pyquery('#panel-logs .paginator a:first').text() == '2'
|
|
resp = resp.click('2')
|
|
assert resp.pyquery('#panel-logs:not([hidden])')
|
|
assert resp.pyquery('#panel-endpoints[hidden]')
|
|
assert resp.pyquery('#table-logs tr').length == 10
|
|
resp.pyquery('#panel-logs .paginator a:first').text()
|
|
assert resp.pyquery('#panel-logs .paginator a:first').text() == '1'
|
|
|
|
|
|
def test_logs_search(app, admin_user):
|
|
csv = CsvDataSource.objects.create(csv_file=File(StringIO('1;t\n'), 't.csv'), slug='t', title='t')
|
|
app = login(app)
|
|
|
|
transaction_id = str(uuid.uuid4())
|
|
looks_like_transaction_id = str(uuid.uuid4())
|
|
log1 = ResourceLog.objects.create(
|
|
appname=csv.get_connector_slug(),
|
|
slug=csv.slug,
|
|
levelno=1,
|
|
message='hop',
|
|
transaction_id=transaction_id,
|
|
extra={'transaction_id': transaction_id, 'foo': 'plop'},
|
|
)
|
|
log2 = ResourceLog.objects.create(
|
|
appname=csv.get_connector_slug(),
|
|
slug=csv.slug,
|
|
levelno=1,
|
|
message='plop',
|
|
transaction_id=transaction_id,
|
|
extra={'bar': 'hop'},
|
|
)
|
|
log3 = ResourceLog.objects.create(
|
|
appname=csv.get_connector_slug(), slug=csv.slug, levelno=1, message='foo', extra={'bar': 'hop'}
|
|
)
|
|
log4 = ResourceLog.objects.create(
|
|
appname=csv.get_connector_slug(), slug=csv.slug, levelno=1, message=looks_like_transaction_id
|
|
)
|
|
|
|
resp = app.get('/manage/csvdatasource/t/logs/')
|
|
assert list(resp.context['page_obj'].object_list) == [log4, log3, log2, log1]
|
|
|
|
resp.form['q'] = transaction_id
|
|
resp = resp.form.submit()
|
|
assert list(resp.context['page_obj'].object_list) == [log2, log1]
|
|
|
|
resp.form['q'] = looks_like_transaction_id
|
|
resp = resp.form.submit()
|
|
assert list(resp.context['page_obj'].object_list) == [log4]
|
|
|
|
resp.form['q'] = str(uuid.uuid4())
|
|
resp = resp.form.submit()
|
|
assert list(resp.context['page_obj'].object_list) == []
|
|
|
|
resp.form['q'] = 'not there'
|
|
resp = resp.form.submit()
|
|
assert list(resp.context['page_obj'].object_list) == []
|
|
|
|
resp.form['q'] = 'hop'
|
|
resp = resp.form.submit()
|
|
assert list(resp.context['page_obj'].object_list) == [log3, log2, log1]
|
|
|
|
resp.form['q'] = 'plop'
|
|
resp = resp.form.submit()
|
|
assert list(resp.context['page_obj'].object_list) == [log2, log1]
|
|
|
|
resp.form['q'] = 'foo'
|
|
resp = resp.form.submit()
|
|
assert list(resp.context['page_obj'].object_list) == [log3, log1]
|
|
|
|
|
|
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')
|
|
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 'jobs' not 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)
|
|
|
|
resp = app.get('/manage/csvdatasource/test/jobs/?job_id=%s' % job_pk)
|
|
assert 'job_target' in resp.context
|
|
|
|
# unknown id
|
|
resp = app.get('/manage/csvdatasource/test/jobs/?job_id=0')
|
|
assert 'job_target' not in resp.context
|
|
# bad id
|
|
resp = app.get('/manage/csvdatasource/test/jobs/?job_id=foo')
|
|
assert 'job_target' not in resp.context
|
|
|
|
|
|
def test_job_restart(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)
|
|
|
|
# unknown job
|
|
app.get('/manage/csvdatasource/test/jobs/0/restart/', status=404)
|
|
app.get('/manage/csvdatasource/test/jobs/foo/restart/', status=404)
|
|
|
|
# create a job, with wrong status
|
|
job = Job.objects.create(resource=csv, status='registered')
|
|
app.get('/manage/csvdatasource/test/jobs/%s/restart/' % job.pk, status=404)
|
|
|
|
job.status = 'failed'
|
|
job.done_timestamp = now()
|
|
job.status_details = {'error_summary': 'foo bar'}
|
|
job.save()
|
|
|
|
# unknown connector
|
|
app.get('/manage/csvdatasourc/test/jobs/%s/restart/' % job.pk, status=404)
|
|
app.get('/manage/csvdatasource/teste/jobs/%s/restart/' % job.pk, status=404)
|
|
|
|
# ok, restart job
|
|
resp = app.get('/manage/csvdatasource/test/jobs/%s/restart/' % job.pk, status=302)
|
|
new_job = Job.objects.latest('pk')
|
|
assert resp.location.endswith('/manage/csvdatasource/test/jobs/')
|
|
assert new_job.status == 'registered'
|
|
assert new_job.done_timestamp is None
|
|
assert new_job.status_details == {}
|
|
|
|
job.refresh_from_db()
|
|
assert job.status == 'restarted'
|
|
assert job.status_details == {'error_summary': 'foo bar', 'new_job_pk': new_job.pk}
|
|
|
|
resp = app.get('/manage/csvdatasource/test/jobs/%s/' % job.pk)
|
|
assert '/manage/csvdatasource/test/jobs/?job_id=%s' % new_job.pk in resp
|
|
|
|
|
|
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',
|
|
)
|
|
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,
|
|
)
|
|
|
|
# cannot export if not connected
|
|
resp = app.get('/%s/%s/' % (csv.get_connector_slug(), csv.slug), status=200)
|
|
assert 'Export' not in resp.html.find('ul', {'class': 'extra-actions-menu'}).text
|
|
|
|
# export site
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
with freezegun.freeze_time('2020-09-01'):
|
|
resp = resp.click('Export')
|
|
assert resp.headers['content-type'] == 'application/json'
|
|
assert resp.headers['content-disposition'] == 'attachment; filename="export_connectors_20200901.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
|
|
|
|
# unknown connectors
|
|
export = {
|
|
'resources': [
|
|
{'@type': 'passerelle-resource', 'resource_type': 'not_installed_app1.app_model'},
|
|
{'@type': 'passerelle-resource', 'resource_type': 'not_installed_app2.app_model'},
|
|
]
|
|
}
|
|
resp = app.get('/manage/', status=200)
|
|
resp = resp.click('Import')
|
|
resp.form['site_json'] = Upload('export.json', force_bytes(json.dumps(export)), 'application/json')
|
|
resp = resp.form.submit()
|
|
assert (
|
|
resp.html.find('ul', {'class': 'errorlist'}).li.text
|
|
== 'Unknown connectors: not_installed_app1, not_installed_app2'
|
|
)
|
|
|
|
# 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)
|
|
assert 'Export' in resp.html.find('ul', {'class': 'extra-actions-menu'}).text
|
|
with freezegun.freeze_time('2020-09-01'):
|
|
resp = resp.click('Export')
|
|
assert resp.headers['content-type'] == 'application/json'
|
|
assert resp.headers['content-disposition'] == 'attachment; filename="export_%s_%s_20200901.json"' % (
|
|
csv.get_connector_slug(),
|
|
csv.slug,
|
|
)
|
|
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()
|
|
|
|
|
|
def test_manager_open_access_information(app, admin_user):
|
|
photon = Photon.objects.create(slug='t')
|
|
|
|
app = login(app)
|
|
resp = app.get(photon.get_absolute_url())
|
|
assert resp.html.find('div', {'id': 'panel-security'}).div.text.strip() == 'Access is open.'
|
|
|
|
|
|
def test_manager_add_open_access_warning(app, admin_user):
|
|
csv = CsvDataSource.objects.create(csv_file=File(StringIO('1;t\n'), 't.csv'), slug='t', title='t')
|
|
private = ApiUser.objects.create(username='private', fullname='private', keytype='', key='xxx')
|
|
public = ApiUser.objects.create(username='public', fullname='private', keytype='', key='')
|
|
assert AccessRight.objects.count() == 0
|
|
|
|
# adding private api user works
|
|
app = login(app)
|
|
resp = app.get(csv.get_absolute_url())
|
|
assert 'Access is limited' in resp.html.find('div', {'id': 'panel-security'}).div.text.strip()
|
|
resp = resp.click('Add', href='/can_access/')
|
|
resp.form['apiuser'] = private.pk
|
|
resp = resp.form.submit().follow()
|
|
assert AccessRight.objects.count() == 1
|
|
|
|
# adding public user displays a warning
|
|
resp = resp.click('Add', href='/can_access/')
|
|
resp.form['apiuser'] = public.pk
|
|
resp = resp.form.submit()
|
|
assert AccessRight.objects.count() == 1
|
|
assert 'user has no security' in resp.text
|
|
|
|
resp = resp.form.submit()
|
|
assert AccessRight.objects.count() == 1
|
|
assert 'user has no security' in resp.text
|
|
|
|
# user has to check a box to procceed
|
|
resp.form['confirm_open_access'] = True
|
|
resp.form.submit().follow()
|
|
assert AccessRight.objects.count() == 2
|
|
|
|
|
|
def test_manager_open_access_flag(app, admin_user):
|
|
csv = CsvDataSource.objects.create(csv_file=File(StringIO('1;t\n'), 't.csv'), slug='t', title='t')
|
|
api = ApiUser.objects.create(username='test', fullname='test', keytype='', key='test')
|
|
|
|
app = login(app)
|
|
resp = app.get('/manage/', status=200)
|
|
assert len(resp.pyquery('li.connector')) == 1
|
|
assert 'open access' not in resp.text
|
|
|
|
obj_type = ContentType.objects.get_for_model(csv)
|
|
AccessRight.objects.create(codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=csv.pk)
|
|
|
|
resp = app.get('/manage/', status=200)
|
|
assert 'open access' not in resp.text
|
|
|
|
api.key = ''
|
|
api.save()
|
|
|
|
resp = app.get('/manage/', status=200)
|
|
assert 'open access' in resp.text
|
|
|
|
|
|
class TestRequestsSubstitutionsDisplay:
|
|
@pytest.fixture
|
|
def setup(self, settings):
|
|
settings.CONNECTORS_SETTINGS = {
|
|
'photon/t': {
|
|
'requests_substitutions': [
|
|
{
|
|
'search': 'abcd',
|
|
'replace': 'efgh',
|
|
}
|
|
]
|
|
}
|
|
}
|
|
photon = Photon.objects.create(slug='t')
|
|
return {'url': photon.get_absolute_url()}
|
|
|
|
def test_shown_when_logged(self, setup, app, admin_user):
|
|
app = login(app)
|
|
resp = app.get(setup['url'])
|
|
paragraph = str(resp.pyquery('#requests-substitutions'))
|
|
assert 'Requests substitutions' in paragraph
|
|
assert 'URL:' not in paragraph
|
|
assert 'Search: <code>abcd</code>' in paragraph
|
|
assert 'Replace: <code>efgh</code>' in paragraph
|
|
|
|
def test_absent_when_unlogged(self, setup, app):
|
|
resp = app.get(setup['url'])
|
|
paragraph = str(resp.pyquery('#requests-substitutions'))
|
|
assert 'Requests substitutions' not in paragraph
|
|
|
|
def test_url(self, setup, app, admin_user, settings):
|
|
settings.CONNECTORS_SETTINGS['photon/t']['requests_substitutions'][0]['url'] = 'https://example.com/'
|
|
app = login(app)
|
|
resp = app.get(setup['url'])
|
|
paragraph = str(resp.pyquery('#requests-substitutions'))
|
|
assert 'URL: https://example.com/' in paragraph
|