mirror of https://github.com/zulip/zulip.git
test_uploads: Extract 'use_s3_backend' to test helpers.
This commit is contained in:
parent
269cd3b8bf
commit
31ea495842
|
@ -1,11 +1,13 @@
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
from typing import (cast, Any, Callable, Dict, Generator, Iterable, Iterator, List, Mapping,
|
from typing import (
|
||||||
Optional, Set, Sized, Tuple, Union, IO, Text)
|
cast, Any, Callable, Dict, Generator, Iterable, Iterator, List, Mapping,
|
||||||
|
Optional, Set, Sized, Tuple, Union, IO, Text, TypeVar
|
||||||
|
)
|
||||||
|
|
||||||
from django.core import signing
|
from django.core import signing
|
||||||
from django.core.urlresolvers import LocaleRegexURLResolver
|
from django.core.urlresolvers import LocaleRegexURLResolver
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
from django.test import TestCase
|
from django.test import TestCase, override_settings
|
||||||
from django.test.client import (
|
from django.test.client import (
|
||||||
BOUNDARY, MULTIPART_CONTENT, encode_multipart,
|
BOUNDARY, MULTIPART_CONTENT, encode_multipart,
|
||||||
)
|
)
|
||||||
|
@ -13,6 +15,8 @@ from django.template import loader
|
||||||
from django.http import HttpResponse, HttpResponseRedirect
|
from django.http import HttpResponse, HttpResponseRedirect
|
||||||
from django.db.utils import IntegrityError
|
from django.db.utils import IntegrityError
|
||||||
|
|
||||||
|
import zerver.lib.upload
|
||||||
|
from zerver.lib.upload import S3UploadBackend, LocalUploadBackend
|
||||||
from zerver.lib.avatar import avatar_url
|
from zerver.lib.avatar import avatar_url
|
||||||
from zerver.lib.cache import get_cache_backend
|
from zerver.lib.cache import get_cache_backend
|
||||||
from zerver.lib.initial_password import initial_password
|
from zerver.lib.initial_password import initial_password
|
||||||
|
@ -56,6 +60,7 @@ import unittest
|
||||||
from six.moves import urllib
|
from six.moves import urllib
|
||||||
from six import binary_type
|
from six import binary_type
|
||||||
from zerver.lib.str_utils import NonBinaryStr
|
from zerver.lib.str_utils import NonBinaryStr
|
||||||
|
from moto import mock_s3_deprecated
|
||||||
|
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
import fakeldap
|
import fakeldap
|
||||||
|
@ -481,3 +486,18 @@ def load_subdomain_token(response):
|
||||||
assert isinstance(response, HttpResponseRedirect)
|
assert isinstance(response, HttpResponseRedirect)
|
||||||
token = response.url.rsplit('/', 1)[1]
|
token = response.url.rsplit('/', 1)[1]
|
||||||
return signing.loads(token, salt='zerver.views.auth.log_into_subdomain')
|
return signing.loads(token, salt='zerver.views.auth.log_into_subdomain')
|
||||||
|
|
||||||
|
FuncT = TypeVar('FuncT', bound=Callable[..., None])
|
||||||
|
|
||||||
|
def use_s3_backend(method):
|
||||||
|
# type: (FuncT) -> FuncT
|
||||||
|
@mock_s3_deprecated
|
||||||
|
@override_settings(LOCAL_UPLOADS_DIR=None)
|
||||||
|
def new_method(*args, **kwargs):
|
||||||
|
# type: (*Any, **Any) -> Any
|
||||||
|
zerver.lib.upload.upload_backend = S3UploadBackend()
|
||||||
|
try:
|
||||||
|
return method(*args, **kwargs)
|
||||||
|
finally:
|
||||||
|
zerver.lib.upload.upload_backend = LocalUploadBackend()
|
||||||
|
return new_method
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
from django.test import TestCase, override_settings
|
from django.test import TestCase
|
||||||
from unittest import skip
|
from unittest import skip
|
||||||
|
|
||||||
from zerver.lib.avatar import (
|
from zerver.lib.avatar import (
|
||||||
|
@ -14,6 +14,7 @@ from zerver.lib.test_helpers import (
|
||||||
avatar_disk_path,
|
avatar_disk_path,
|
||||||
get_test_image_file,
|
get_test_image_file,
|
||||||
POSTRequestMock,
|
POSTRequestMock,
|
||||||
|
use_s3_backend,
|
||||||
)
|
)
|
||||||
from zerver.lib.test_runner import slow
|
from zerver.lib.test_runner import slow
|
||||||
from zerver.lib.upload import sanitize_name, S3UploadBackend, \
|
from zerver.lib.upload import sanitize_name, S3UploadBackend, \
|
||||||
|
@ -47,9 +48,7 @@ import base64
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
from django.utils.timezone import now as timezone_now
|
from django.utils.timezone import now as timezone_now
|
||||||
|
|
||||||
from moto import mock_s3_deprecated
|
from typing import Any, Callable, Text
|
||||||
|
|
||||||
from typing import Any, Callable, TypeVar, Text
|
|
||||||
|
|
||||||
def destroy_uploads():
|
def destroy_uploads():
|
||||||
# type: () -> None
|
# type: () -> None
|
||||||
|
@ -972,21 +971,6 @@ class LocalStorageTest(UploadSerializeMixin, ZulipTestCase):
|
||||||
# type: () -> None
|
# type: () -> None
|
||||||
destroy_uploads()
|
destroy_uploads()
|
||||||
|
|
||||||
FuncT = TypeVar('FuncT', bound=Callable[..., None])
|
|
||||||
|
|
||||||
def use_s3_backend(method):
|
|
||||||
# type: (FuncT) -> FuncT
|
|
||||||
@mock_s3_deprecated
|
|
||||||
@override_settings(LOCAL_UPLOADS_DIR=None)
|
|
||||||
def new_method(*args, **kwargs):
|
|
||||||
# type: (*Any, **Any) -> Any
|
|
||||||
zerver.lib.upload.upload_backend = S3UploadBackend()
|
|
||||||
try:
|
|
||||||
return method(*args, **kwargs)
|
|
||||||
finally:
|
|
||||||
zerver.lib.upload.upload_backend = LocalUploadBackend()
|
|
||||||
return new_method
|
|
||||||
|
|
||||||
class S3Test(ZulipTestCase):
|
class S3Test(ZulipTestCase):
|
||||||
|
|
||||||
@use_s3_backend
|
@use_s3_backend
|
||||||
|
|
Loading…
Reference in New Issue