mirror of
https://github.com/umap-project/umap.git
synced 2025-04-28 19:42:36 +02:00
chore: add basic tests for S3 storage
This commit is contained in:
parent
a04624c4c8
commit
81fa31f50b
5 changed files with 172 additions and 4 deletions
|
@ -61,6 +61,7 @@ test = [
|
|||
"pytest-playwright==0.6.2",
|
||||
"pytest-rerunfailures==15.0",
|
||||
"pytest-xdist>=3.5.0,<4",
|
||||
"moto[s3]==5.0.21"
|
||||
]
|
||||
docker = [
|
||||
"uwsgi==2.0.28",
|
||||
|
|
24
umap/migrations/0025_alter_datalayer_geojson.py
Normal file
24
umap/migrations/0025_alter_datalayer_geojson.py
Normal file
|
@ -0,0 +1,24 @@
|
|||
# Generated by Django 5.1.2 on 2024-11-29 15:45
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
import umap.models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
dependencies = [
|
||||
("umap", "0024_alter_map_share_status"),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AlterField(
|
||||
model_name="datalayer",
|
||||
name="geojson",
|
||||
field=models.FileField(
|
||||
blank=True,
|
||||
null=True,
|
||||
storage=umap.models.set_storage,
|
||||
upload_to=umap.models.upload_to,
|
||||
),
|
||||
),
|
||||
]
|
|
@ -426,7 +426,7 @@ class Pictogram(NamedModel):
|
|||
|
||||
attribution = models.CharField(max_length=300)
|
||||
category = models.CharField(max_length=300, null=True, blank=True)
|
||||
pictogram = models.FileField(upload_to="pictogram", storage=storages["default"])
|
||||
pictogram = models.FileField(upload_to="pictogram")
|
||||
|
||||
@property
|
||||
def json(self):
|
||||
|
@ -445,6 +445,10 @@ def upload_to(instance, filename):
|
|||
return instance.geojson.storage.make_filename(instance)
|
||||
|
||||
|
||||
def set_storage():
|
||||
return storages["data"]
|
||||
|
||||
|
||||
class DataLayer(NamedModel):
|
||||
"""
|
||||
Layer to store Features in.
|
||||
|
@ -470,7 +474,7 @@ class DataLayer(NamedModel):
|
|||
map = models.ForeignKey(Map, on_delete=models.CASCADE)
|
||||
description = models.TextField(blank=True, null=True, verbose_name=_("description"))
|
||||
geojson = models.FileField(
|
||||
upload_to=upload_to, blank=True, null=True, storage=storages["data"]
|
||||
upload_to=upload_to, blank=True, null=True, storage=set_storage
|
||||
)
|
||||
display_on_load = models.BooleanField(
|
||||
default=False,
|
||||
|
@ -519,7 +523,7 @@ class DataLayer(NamedModel):
|
|||
new.pk = uuid.uuid4()
|
||||
if map_inst:
|
||||
new.map = map_inst
|
||||
new.geojson = File(new.geojson.file.file)
|
||||
new.geojson = File(new.geojson.file.file, name="tmpname")
|
||||
new.save()
|
||||
return new
|
||||
|
||||
|
|
|
@ -7,6 +7,7 @@ from pathlib import Path
|
|||
from botocore.exceptions import ClientError
|
||||
from django.conf import settings
|
||||
from django.contrib.staticfiles.storage import ManifestStaticFilesStorage
|
||||
from django.core.files.base import File
|
||||
from django.core.files.storage import FileSystemStorage
|
||||
from rcssmin import cssmin
|
||||
from rjsmin import jsmin
|
||||
|
@ -112,7 +113,10 @@ class UmapS3(S3Storage):
|
|||
pass
|
||||
|
||||
def onDatalayerDelete(self, instance):
|
||||
pass
|
||||
return self.connection.meta.client.delete_object(
|
||||
Bucket=self.bucket_name,
|
||||
Key=instance.geojson.name,
|
||||
)
|
||||
|
||||
|
||||
class UmapFileSystem(FileSystemStorage):
|
||||
|
|
135
umap/tests/test_datalayer_s3.py
Normal file
135
umap/tests/test_datalayer_s3.py
Normal file
|
@ -0,0 +1,135 @@
|
|||
import json
|
||||
import os
|
||||
|
||||
import boto3
|
||||
import pytest
|
||||
from botocore.errorfactory import ClientError
|
||||
from django.core.files.base import ContentFile
|
||||
from django.core.files.storage import storages
|
||||
from moto import mock_aws
|
||||
|
||||
from umap.models import DataLayer
|
||||
|
||||
from .base import DataLayerFactory
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
def patch_storage():
|
||||
"""Mocked AWS Credentials for moto."""
|
||||
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
|
||||
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
|
||||
os.environ["AWS_SECURITY_TOKEN"] = "testing"
|
||||
os.environ["AWS_SESSION_TOKEN"] = "testing"
|
||||
os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
|
||||
before = DataLayer.geojson.field.storage
|
||||
|
||||
DataLayer.geojson.field.storage = storages.create_storage(
|
||||
{
|
||||
"BACKEND": "umap.storage.UmapS3",
|
||||
"OPTIONS": {
|
||||
"access_key": "testing",
|
||||
"secret_key": "testing",
|
||||
"bucket_name": "umap",
|
||||
"region_name": "us-east-1",
|
||||
},
|
||||
}
|
||||
)
|
||||
yield
|
||||
DataLayer.geojson.field.storage = before
|
||||
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
def mocked_aws():
|
||||
"""
|
||||
Mock all AWS interactions
|
||||
Requires you to create your own boto3 clients
|
||||
"""
|
||||
with mock_aws():
|
||||
client = boto3.client("s3", region_name="us-east-1")
|
||||
client.create_bucket(Bucket="umap")
|
||||
client.put_bucket_versioning(
|
||||
Bucket="umap", VersioningConfiguration={"Status": "Enabled"}
|
||||
)
|
||||
yield
|
||||
|
||||
|
||||
def test_can_create_datalayer(map, datalayer):
|
||||
other = DataLayerFactory(map=map)
|
||||
assert datalayer.geojson.name == f"{datalayer.pk}.geojson"
|
||||
assert other.geojson.name == f"{other.pk}.geojson"
|
||||
|
||||
|
||||
def test_clone_should_return_new_instance(map, datalayer):
|
||||
clone = datalayer.clone()
|
||||
assert datalayer.pk != clone.pk
|
||||
assert datalayer.name == clone.name
|
||||
assert datalayer.map == clone.map
|
||||
assert datalayer.geojson != clone.geojson
|
||||
assert datalayer.geojson.name != clone.geojson.name
|
||||
assert clone.geojson.name == f"{clone.pk}.geojson"
|
||||
|
||||
|
||||
def test_update_should_add_version(map, datalayer):
|
||||
assert len(datalayer.versions) == 1
|
||||
datalayer.geojson = ContentFile("{}", "foo.json")
|
||||
datalayer.save()
|
||||
assert len(datalayer.versions) == 2
|
||||
|
||||
|
||||
def test_get_version(map, datalayer):
|
||||
assert len(datalayer.versions) == 1
|
||||
datalayer.geojson = ContentFile('{"foo": "bar"}', "foo.json")
|
||||
datalayer.save()
|
||||
assert len(datalayer.versions) == 2
|
||||
latest = datalayer.versions[0]["ref"]
|
||||
version = datalayer.get_version(latest)
|
||||
assert json.loads(version) == {"foo": "bar"}
|
||||
older = datalayer.versions[1]["ref"]
|
||||
version = datalayer.get_version(older)
|
||||
assert json.loads(version) == {
|
||||
"_umap_options": {
|
||||
"browsable": True,
|
||||
"displayOnLoad": True,
|
||||
"name": "test datalayer",
|
||||
},
|
||||
"features": [
|
||||
{
|
||||
"geometry": {
|
||||
"coordinates": [
|
||||
14.68896484375,
|
||||
48.55297816440071,
|
||||
],
|
||||
"type": "Point",
|
||||
},
|
||||
"properties": {
|
||||
"_umap_options": {
|
||||
"color": "DarkCyan",
|
||||
"iconClass": "Ball",
|
||||
},
|
||||
"description": "Da place anonymous again 755",
|
||||
"name": "Here",
|
||||
},
|
||||
"type": "Feature",
|
||||
},
|
||||
],
|
||||
"type": "FeatureCollection",
|
||||
}
|
||||
|
||||
latest = datalayer.reference_version
|
||||
version = datalayer.get_version(latest)
|
||||
assert json.loads(version) == {"foo": "bar"}
|
||||
|
||||
|
||||
def test_delete_datalayer_should_delete_all_versions(datalayer):
|
||||
# create a new version
|
||||
datalayer.geojson = ContentFile('{"foo": "bar"}', "foo.json")
|
||||
datalayer.save()
|
||||
s3_key = datalayer.geojson.name
|
||||
datalayer.delete()
|
||||
with pytest.raises(ClientError):
|
||||
datalayer.geojson.storage.connection.meta.client.get_object(
|
||||
Bucket="umap",
|
||||
Key=s3_key,
|
||||
)
|
Loading…
Reference in a new issue