fix: legacy region get_xxx_host and add more cases

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
  }