267 lines
8.0 KiB
Python
267 lines
8.0 KiB
Python
from io import BytesIO
|
|
from urllib.parse import parse_qsl, urlparse
|
|
|
|
import django_webtest
|
|
import pytest
|
|
from django.contrib.auth.models import User
|
|
from django.core.cache import cache
|
|
from django.core.files import File
|
|
from django.core.management import call_command
|
|
from django.db import connection
|
|
from django.db.migrations.executor import MigrationExecutor
|
|
from httmock import HTTMock, remember_called, response, urlmatch
|
|
|
|
from tests.utils import make_resource
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def media(settings, tmpdir):
|
|
settings.MEDIA_ROOT = str(tmpdir.mkdir('media'))
|
|
|
|
|
|
@pytest.fixture
|
|
def app(request):
|
|
wtm = django_webtest.WebTestMixin()
|
|
wtm._patch_settings()
|
|
request.addfinalizer(wtm._unpatch_settings)
|
|
cache.clear()
|
|
return django_webtest.DjangoTestApp()
|
|
|
|
|
|
@urlmatch(netloc='^api-adresse.data.gouv.fr$', path='^/search/$')
|
|
@remember_called
|
|
def api_adresse_data_gouv_fr_search(url, request):
|
|
return response(
|
|
200,
|
|
{
|
|
'limit': 1,
|
|
'attribution': 'BAN',
|
|
'version': 'draft',
|
|
'licence': 'ODbL 1.0',
|
|
'query': 'plop',
|
|
'type': 'FeatureCollection',
|
|
'features': [
|
|
{
|
|
'geometry': {'type': 'Point', 'coordinates': [-0.593775, 47.474633]},
|
|
'properties': {
|
|
'citycode': '49007',
|
|
'name': 'Rue Roger Halope',
|
|
'id': '49007_6950_be54bd',
|
|
'city': 'Angers',
|
|
'context': '49, Maine-et-Loire, Pays de la Loire',
|
|
'score': 0.14097272727272728,
|
|
'label': 'Rue Roger Halope 49000 Angers',
|
|
'postcode': '49000',
|
|
'type': 'street',
|
|
},
|
|
'type': 'Feature',
|
|
}
|
|
],
|
|
},
|
|
request=request,
|
|
)
|
|
|
|
|
|
@urlmatch(netloc='^api-adresse.data.gouv.fr$', path='^/reverse/$')
|
|
@remember_called
|
|
def api_adresse_data_gouv_fr_reverse(url, request):
|
|
return response(
|
|
200,
|
|
{
|
|
'limit': 1,
|
|
'attribution': 'BAN',
|
|
'version': 'draft',
|
|
'licence': 'ODbL 1.0',
|
|
'type': 'FeatureCollection',
|
|
'features': [
|
|
{
|
|
'geometry': {'type': 'Point', 'coordinates': [-0.593775, 47.474633]},
|
|
'properties': {
|
|
'citycode': '49007',
|
|
'name': 'Rue Roger Halope',
|
|
'id': '49007_6950_be54bd',
|
|
'city': 'Angers',
|
|
'distance': 0,
|
|
'context': '49, Maine-et-Loire, Pays de la Loire',
|
|
'score': 1.0,
|
|
'label': 'Rue Roger Halope 49000 Angers',
|
|
'postcode': '49000',
|
|
'type': 'street',
|
|
},
|
|
'type': 'Feature',
|
|
}
|
|
],
|
|
},
|
|
request=request,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_api_adresse_data_gouv_fr_search():
|
|
with HTTMock(api_adresse_data_gouv_fr_search):
|
|
yield api_adresse_data_gouv_fr_search
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_api_adresse_data_gouv_fr_reverse():
|
|
with HTTMock(api_adresse_data_gouv_fr_reverse):
|
|
yield api_adresse_data_gouv_fr_reverse
|
|
|
|
|
|
@pytest.fixture
|
|
def endpoint_dummy_cache(monkeypatch):
|
|
from django.core.cache import caches
|
|
|
|
import passerelle.views
|
|
|
|
monkeypatch.setattr(passerelle.views, 'cache', caches['dummy'])
|
|
|
|
|
|
@urlmatch()
|
|
def internal_server_error(url, request):
|
|
return response(500, 'Internal server error')
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_500():
|
|
with HTTMock(internal_server_error):
|
|
yield None
|
|
|
|
|
|
@pytest.fixture
|
|
def dummy_csv_datasource(db):
|
|
from passerelle.apps.csvdatasource.models import CsvDataSource, Query
|
|
|
|
data = b'''id,label
|
|
1,a
|
|
2,b
|
|
3,c'''
|
|
obj = make_resource(
|
|
CsvDataSource,
|
|
slug='dummy-slug',
|
|
title='Dummy Title',
|
|
description='dummy description',
|
|
csv_file=File(BytesIO(data), 'dummy.csv'),
|
|
)
|
|
|
|
Query.objects.create(
|
|
resource=obj,
|
|
slug='dummy-query',
|
|
structure='array',
|
|
label='Dummy Query',
|
|
description='dummy query description',
|
|
projections='id:int(id)\ntext:label',
|
|
)
|
|
|
|
return obj
|
|
|
|
|
|
@pytest.fixture
|
|
def relax_openssl(tmpdir):
|
|
"""OpenSSL default configuration has been really strict for some years,
|
|
this fixture set a temporary really permisive ciphers list."""
|
|
import os
|
|
|
|
openssl_cnf_path = tmpdir / 'openssl.cnf'
|
|
with openssl_cnf_path.open('w') as fd:
|
|
fd.write(
|
|
'''
|
|
[default_conf]
|
|
ssl_conf = ssl_sect
|
|
|
|
[ssl_sect]
|
|
system_default = system_default_sect
|
|
|
|
[system_default_sect]
|
|
CipherString = ALL'''
|
|
)
|
|
old_value = os.environ.get('OPENSSL_CONF', None)
|
|
try:
|
|
os.environ['OPENSSL_CONF'] = str(openssl_cnf_path)
|
|
yield
|
|
finally:
|
|
if old_value is None:
|
|
del os.environ['OPENSSL_CONF']
|
|
else:
|
|
os.environ['OPENSSL_CONF'] = old_value
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_cache():
|
|
cache.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def simple_user(db):
|
|
return User.objects.create_user('user', password='user')
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_user(db):
|
|
return User.objects.create_superuser('admin', email=None, password='admin')
|
|
|
|
|
|
@pytest.fixture
|
|
def httpbin():
|
|
@urlmatch(netloc=r'^httpbin\.org$', path=r'^/cookies/set$')
|
|
def httpbin_cookies_set(url, request):
|
|
headers = []
|
|
cookies = {}
|
|
for key, value in parse_qsl(urlparse(request.url).query):
|
|
cookies[key] = value
|
|
headers.append(('Set-Cookie', f'{key}={value}; Path=/'))
|
|
headers.append(('Location', '/cookies'))
|
|
return response(200, b'', headers=headers, request=request)
|
|
|
|
with HTTMock(httpbin_cookies_set) as fixture:
|
|
fixture.url = 'http://httpbin.org'
|
|
yield fixture
|
|
|
|
|
|
@pytest.fixture()
|
|
def migration(request, transactional_db):
|
|
# see https://gist.github.com/asfaltboy/b3e6f9b5d95af8ba2cc46f2ba6eae5e2
|
|
"""
|
|
This fixture returns a helper object to test Django data migrations.
|
|
The fixture returns an object with two methods;
|
|
- `before` to initialize db to the state before the migration under test
|
|
- `after` to execute the migration and bring db to the state after the migration
|
|
The methods return `old_apps` and `new_apps` respectively; these can
|
|
be used to initiate the ORM models as in the migrations themselves.
|
|
For example:
|
|
def test_foo_set_to_bar(migration):
|
|
old_apps = migration.before([('my_app', '0001_inital')])
|
|
Foo = old_apps.get_model('my_app', 'foo')
|
|
Foo.objects.create(bar=False)
|
|
assert Foo.objects.count() == 1
|
|
assert Foo.objects.filter(bar=False).count() == Foo.objects.count()
|
|
# executing migration
|
|
new_apps = migration.apply([('my_app', '0002_set_foo_bar')])
|
|
Foo = new_apps.get_model('my_app', 'foo')
|
|
|
|
assert Foo.objects.filter(bar=False).count() == 0
|
|
assert Foo.objects.filter(bar=True).count() == Foo.objects.count()
|
|
Based on: https://gist.github.com/blueyed/4fb0a807104551f103e6
|
|
"""
|
|
|
|
class Migrator:
|
|
def before(self, targets, at_end=True):
|
|
"""Specify app and starting migration names as in:
|
|
before([('app', '0001_before')]) => app/migrations/0001_before.py
|
|
"""
|
|
executor = MigrationExecutor(connection)
|
|
executor.migrate(targets)
|
|
executor.loader.build_graph()
|
|
return executor._create_project_state(with_applied_migrations=True).apps
|
|
|
|
def apply(self, targets):
|
|
"""Migrate forwards to the "targets" migration"""
|
|
executor = MigrationExecutor(connection)
|
|
executor.migrate(targets)
|
|
executor.loader.build_graph()
|
|
return executor._create_project_state(with_applied_migrations=True).apps
|
|
|
|
yield Migrator()
|
|
|
|
call_command('migrate', verbosity=0)
|