Removed old upload API that conflicts with urllib3

Author: giohappy

geonode/upload/api/tests_old.py

@@ -16,38 +16,14 @@
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 #
 #########################################################################
-
-from geonode.base.models import ResourceBase
-from geonode.geoserver.helpers import gs_catalog
 import os
-import shutil
 import logging
-import tempfile
-from io import IOBase
-from urllib.request import urljoin
-
-from django.conf import settings
 
 from django.urls import reverse
-from django.contrib.auth import authenticate, get_user_model
-from django.test.utils import override_settings
-
-from requests_toolbelt.multipart.encoder import MultipartEncoder
+from django.contrib.auth import get_user_model
 
 from rest_framework.test import APITestCase
 
-from seleniumrequests import Firefox
-
-# from selenium.common import exceptions
-# from selenium.webdriver.common.by import By
-from selenium.webdriver import FirefoxOptions
-from selenium.webdriver.support.wait import WebDriverWait
-from selenium.webdriver.support import expected_conditions as EC
-from selenium.webdriver.firefox.firefox_binary import FirefoxBinary
-
-from webdriver_manager.firefox import GeckoDriverManager
-
-from geonode.tests.base import GeoNodeLiveTestSupport
 from geonode.geoserver.helpers import ogc_server_settings
 from geonode.upload.models import UploadSizeLimit, UploadParallelismLimit
 
@@ -60,180 +36,6 @@
 logger = logging.getLogger("importer")
 
 
-@override_settings(
-    DEBUG=True,
-    ALLOWED_HOSTS=["*"],
-    SITEURL=LIVE_SERVER_URL,
-    CSRF_COOKIE_SECURE=False,
-    CSRF_COOKIE_HTTPONLY=False,
-    CORS_ORIGIN_ALLOW_ALL=True,
-    SESSION_COOKIE_SECURE=False,
-    DEFAULT_MAX_PARALLEL_UPLOADS_PER_USER=5,
-)
-class UploadApiTests(GeoNodeLiveTestSupport, APITestCase):
-    port = 0
-
-    @classmethod
-    def setUpClass(cls):
-        super().setUpClass()
-
-        try:
-            """Instantiate selenium driver instance"""
-            binary = FirefoxBinary("/usr/bin/firefox")
-            opts = FirefoxOptions()
-            opts.add_argument("--headless")
-            executable_path = GeckoDriverManager().install()
-            cls.selenium = Firefox(firefox_binary=binary, firefox_options=opts, executable_path=executable_path)
-            cls.selenium.implicitly_wait(10)
-        except Exception as e:
-            logger.error(e)
-
-    @classmethod
-    def tearDownClass(cls):
-        """Quit selenium driver instance"""
-        try:
-            cls.selenium.quit()
-        except Exception as e:
-            logger.debug(e)
-        super().tearDownClass()
-
-    def setUp(self):
-        super().setUp()
-        self.temp_folder = tempfile.mkdtemp(dir=CURRENT_LOCATION)
-        self.session_id = None
-        self.csrf_token = None
-
-    def tearDown(self):
-        shutil.rmtree(self.temp_folder, ignore_errors=True)
-        return super().tearDown()
-
-    def set_session_cookies(self, url=None):
-        # selenium will set cookie domain based on current page domain
-        self.selenium.get(url or f"{self.live_server_url}/")
-        self.csrf_token = self.selenium.get_cookie("csrftoken")["value"]
-        self.session_id = self.selenium.get_cookie(settings.SESSION_COOKIE_NAME)["value"]
-        self.selenium.add_cookie(
-            {"name": settings.SESSION_COOKIE_NAME, "value": self.session_id, "secure": False, "path": "/"}
-        )
-        self.selenium.add_cookie({"name": "csrftoken", "value": self.csrf_token, "secure": False, "path": "/"})
-
-    def click_button(self, label):
-        selector = f"//button[contains(., '{label}')]"
-        self.selenium.find_element_by_xpath(selector).click()
-
-    def do_login(self, username="admin", password="admin"):
-        """Method to login the GeoNode site"""
-        assert authenticate(username=username, password=password)
-        self.assertTrue(self.client.login(username=username, password=password))  # Native django test client
-
-        url = urljoin(settings.SITEURL, f"{reverse('account_login')}?next=/layers")
-        self.set_session_cookies(url)
-        self.selenium.save_screenshot(os.path.join(self.temp_folder, "login.png"))
-
-        title = self.selenium.title
-        current_url = self.selenium.current_url
-        logger.debug(f" ---- title: {title} / current_url: {current_url}")
-
-        username_input = self.selenium.find_element_by_xpath('//input[@id="id_login"][@type="text"]')
-        username_input.send_keys(username)
-        password_input = self.selenium.find_element_by_xpath('//input[@id="id_password"][@type="password"]')
-        password_input.send_keys(password)
-
-        self.selenium.save_screenshot(os.path.join(self.temp_folder, "login-set_fields.png"))
-        self.click_button("Sign In")
-        self.selenium.save_screenshot(os.path.join(self.temp_folder, "login-sign_in.png"))
-
-        title = self.selenium.title
-        current_url = self.selenium.current_url
-        logger.debug(f" ---- title: {title} / current_url: {current_url}")
-
-        # Wait until the response is received
-        WebDriverWait(self.selenium, 10).until(EC.title_contains("Explore Layers"))
-        self.set_session_cookies(url)
-
-    def do_logout(self):
-        url = urljoin(settings.SITEURL, f"{reverse('account_logout')}")
-        self.selenium.get(url)
-        self.click_button("Log out")
-
-    def do_upload_step(self, step=None):
-        step = urljoin(settings.SITEURL, reverse("data_upload", args=[step] if step else []))
-        return step
-
-    def live_upload_file(self, _file):
-        """function that uploads a file, or a collection of files, to
-        the GeoNode"""
-        spatial_files = ("dbf_file", "shx_file", "prj_file")
-        base, ext = os.path.splitext(_file)
-        params = {
-            # make public since wms client doesn't do authentication
-            "csrfmiddlewaretoken": self.csrf_token,
-            "permissions": '{ "users": {"AnonymousUser": ["view_resourcebase"]} , "groups":{}}',
-            "time": "false",
-            "charset": "UTF-8",
-        }
-        cookies = {settings.SESSION_COOKIE_NAME: self.session_id, "csrftoken": self.csrf_token}
-        headers = {
-            "X-CSRFToken": self.csrf_token,
-            "X-Requested-With": "XMLHttpRequest",
-            "Set-Cookie": f"csrftoken={self.csrf_token}; sessionid={self.session_id}",
-        }
-        url = self.do_upload_step()
-        logger.debug(f" ---- UPLOAD URL: {url} / cookies: {cookies} / headers: {headers}")
-
-        # deal with shapefiles
-        if ext.lower() == ".shp":
-            for spatial_file in spatial_files:
-                ext, _ = spatial_file.split("_")
-                file_path = f"{base}.{ext}"
-                # sometimes a shapefile is missing an extra file,
-                # allow for that
-                if os.path.exists(file_path):
-                    params[spatial_file] = open(file_path, "rb")
-
-        with open(_file, "rb") as base_file:
-            params["base_file"] = base_file
-            for name, value in params.items():
-                if isinstance(value, IOBase):
-                    params[name] = (os.path.basename(value.name), value)
-
-            # refresh to exchange cookies with the server.
-            self.selenium.refresh()
-            self.selenium.get(url)
-            self.selenium.save_screenshot(os.path.join(self.temp_folder, "upload-page.png"))
-            logger.debug(f" ------------ UPLOAD FORM: {params}")
-            encoder = MultipartEncoder(fields=params)
-            headers["Content-Type"] = encoder.content_type
-            response = self.selenium.request("POST", url, data=encoder, headers=headers)
-
-        # Closes the files
-        for spatial_file in spatial_files:
-            if isinstance(params.get(spatial_file), IOBase):
-                params[spatial_file].close()
-
-        try:
-            logger.error(f" -- response: {response.status_code} / {response.json()}")
-            return response, response.json()
-        except ValueError:
-            logger.exception(ValueError(f"probably not json, status {response.status_code} / {response.content}"))
-            return response, response.content
-
-    def _cleanup_layer(self, layer_name):
-        # removing the layer from geonode
-        x = ResourceBase.objects.filter(alternate__icontains=layer_name)
-        if x.exists():
-            for el in x.iterator():
-                el.delete()
-        # removing the layer from geoserver
-        dataset = gs_catalog.get_layer(layer_name)
-        if dataset:
-            gs_catalog.delete(dataset, purge="all", recurse=True)
-        # removing the layer from geoserver
-        store = gs_catalog.get_store(layer_name, workspace="geonode")
-        if store:
-            gs_catalog.delete(store, purge="all", recurse=True)
-
-
 class UploadSizeLimitTests(APITestCase):
     fixtures = [
         "group_test_data.json",