fix: legacy region get_xxx_host and add more cases

Author: lihsai0

qiniu/region.py

@@ -17,7 +17,7 @@ def decorator(func):
         @functools.wraps(func)
         def wrapper(self, *args, **kwargs):
             if hasattr(self, key) and getattr(self, key):
-                return self.rs_host
+                return getattr(self, key)
             if is_customized_default('default_' + key):
                 return get_default('default_' + key)
             return func(self, *args, **kwargs)
@@ -196,6 +196,19 @@ def get_bucket_hosts(self, ak, bucket, home_dir=None, force=False):
         return bucket_hosts
 
     def get_bucket_hosts_to_cache(self, key, home_dir):
+        """
+        .. deprecated::
+            The cache has been replaced by CachedRegionsProvider
+
+        Parameters
+        ----------
+        key: str
+        home_dir: str
+
+        Returns
+        -------
+        dict
+        """
         ret = {}
         if len(self.host_cache) == 0:
             self.host_cache_from_file(home_dir)
@@ -209,11 +222,29 @@ def get_bucket_hosts_to_cache(self, key, home_dir):
         return ret
 
     def set_bucket_hosts_to_cache(self, key, val, home_dir):
+        """
+        .. deprecated::
+            The cache has been replaced by CachedRegionsProvider
+
+        Parameters
+        ----------
+        key: str
+        val: dict
+        home_dir: str
+        """
         self.host_cache[key] = val
         self.host_cache_to_file(home_dir)
         return
 
     def host_cache_from_file(self, home_dir):
+        """
+        .. deprecated::
+            The cache has been replaced by CachedRegionsProvider
+
+        Parameters
+        ----------
+        home_dir: str
+        """
         if home_dir is not None:
             self.home_dir = home_dir
         path = self.host_cache_file_path()
@@ -229,9 +260,26 @@ def host_cache_from_file(self, home_dir):
         return
 
     def host_cache_file_path(self):
+        """
+        .. deprecated::
+            The cache has been replaced by CachedRegionsProvider
+
+        Returns
+        -------
+        str
+        """
         return os.path.join(self.home_dir, ".qiniu_pythonsdk_hostscache.json")
 
     def host_cache_to_file(self, home_dir):
+        """
+        .. deprecated::
+            The cache has been replaced by CachedRegionsProvider
+
+        Parameters
+        ----------
+        home_dir: str
+
+        """
         path = self.host_cache_file_path()
         with open(path, 'w') as f:
             json.dump(self.host_cache, f)

tests/cases/test_services/test_storage/test_uploader.py

@@ -15,6 +15,7 @@
 from qiniu.http.endpoint import Endpoint
 from qiniu.http.region import ServiceName
 from qiniu.services.storage.uploader import _form_put
+from qiniu.services.storage.uploaders.abc import UploaderBase
 
 KB = 1024
 MB = 1024 * KB
@@ -885,3 +886,24 @@ def test_upload_acc_fallback_src_by_acc_unavailable(
         )
 
         assert ret['key'] == key, resp
+
+    def test_uploader_base_compatible(self, qn_auth, bucket_name):
+        if is_py2:
+            class MockUploader(UploaderBase):
+                def upload(
+                    self,
+                    **kwargs
+                ):
+                    pass
+            uploader = MockUploader(
+                bucket_name=bucket_name,
+                auth=qn_auth
+            )
+        else:
+            uploader = UploaderBase(
+                bucket_name=bucket_name,
+                auth=qn_auth
+            )
+
+        up_hosts = uploader._get_up_hosts()
+        assert len(up_hosts) > 0

tests/cases/test_zone/test_lagacy_region.py

@@ -1,31 +1,78 @@
+import pytest
+
 from qiniu.http.region import Region, ServiceName
 from qiniu.region import LegacyRegion
-from qiniu.compat import json
+from qiniu.compat import json, is_py2
+
+
+@pytest.fixture
+def mocked_hosts():
+    mocked_hosts = {
+        ServiceName.UP: ['https://up.python-example.qiniu.com', 'https://up-2.python-example.qiniu.com'],
+        ServiceName.IO: ['https://io.python-example.qiniu.com'],
+        ServiceName.RS: ['https://rs.python-example.qiniu.com'],
+        ServiceName.RSF: ['https://rsf.python-example.qiniu.com'],
+        ServiceName.API: ['https://api.python-example.qiniu.com']
+    }
+    yield mocked_hosts
+
+
+@pytest.fixture
+def mock_legacy_region(mocked_hosts):
+    region = LegacyRegion(
+        up_host=mocked_hosts[ServiceName.UP][0],
+        up_host_backup=mocked_hosts[ServiceName.UP][1],
+        io_host=mocked_hosts[ServiceName.IO][0],
+        rs_host=mocked_hosts[ServiceName.RS][0],
+        rsf_host=mocked_hosts[ServiceName.RSF][0],
+        api_host=mocked_hosts[ServiceName.API][0]
+    )
+    yield region
 
 
 class TestLegacyRegion:
-    def test_compatible_with_http_region(self):
-        mocked_hosts = {
-            ServiceName.UP: ['https://up.python-example.qiniu.com', 'https://up-2.python-example.qiniu.com'],
-            ServiceName.IO: ['https://io.python-example.qiniu.com'],
-            ServiceName.RS: ['https://rs.python-example.qiniu.com'],
-            ServiceName.RSF: ['https://rsf.python-example.qiniu.com'],
-            ServiceName.API: ['https://api.python-example.qiniu.com']
-        }
+    def test_get_hosts_from_self(self, mocked_hosts, mock_legacy_region, qn_auth, bucket_name):
+        cases = [
+            # up will always query from the old version,
+            # which version implements the `get_up_host_*` method
+            (
+                mock_legacy_region.get_io_host(qn_auth.get_access_key(), None),
+                mocked_hosts[ServiceName.IO][0]
+            ),
+            (
+                mock_legacy_region.get_rs_host(qn_auth.get_access_key(), None),
+                mocked_hosts[ServiceName.RS][0]
+            ),
+            (
+                mock_legacy_region.get_rsf_host(qn_auth.get_access_key(), None),
+                mocked_hosts[ServiceName.RSF][0]
+            ),
+            (
+                mock_legacy_region.get_api_host(qn_auth.get_access_key(), None),
+                mocked_hosts[ServiceName.API][0]
+            )
+        ]
+        for actual, expect in cases:
+            assert actual == expect
+
+    def test_get_hosts_from_query(self, qn_auth, bucket_name):
+        up_token = qn_auth.upload_token(bucket_name)
+        region = LegacyRegion()
+        up_host = region.get_up_host_by_token(up_token, None)
+        up_host_backup = region.get_up_host_backup_by_token(up_token, None)
+        if is_py2:
+            up_host = up_host.encode()
+            up_host_backup = up_host_backup.encode()
+        assert type(up_host) is str and len(up_host) > 0
+        assert type(up_host_backup) is str and len(up_host_backup) > 0
+        assert up_host != up_host_backup
 
-        region = LegacyRegion(
-            up_host=mocked_hosts[ServiceName.UP][0],
-            up_host_backup=mocked_hosts[ServiceName.UP][1],
-            io_host=mocked_hosts[ServiceName.IO][0],
-            rs_host=mocked_hosts[ServiceName.RS][0],
-            rsf_host=mocked_hosts[ServiceName.RSF][0],
-            api_host=mocked_hosts[ServiceName.API][0]
-        )
-        assert isinstance(region, Region)
+    def test_compatible_with_http_region(self, mocked_hosts, mock_legacy_region):
+        assert isinstance(mock_legacy_region, Region)
         assert mocked_hosts == {
             k: [
                 e.get_value()
-                for e in region.services[k]
+                for e in mock_legacy_region.services[k]
             ]
             for k in mocked_hosts
         }