from magento.magento_api import _sku_path from magento.magento_api import MagentoRestClient from magento.oauth_client import ( MagentoOAuthClient, _IMMEDIATE_RETRY_COUNT, _SERVER_DOWN_RECOVERY_WAITS, build_magento_oauth_kwargs, ) import warnings from urllib3.exceptions import InsecureRequestWarning class _FakeOAuth: def __init__(self, responses): self.responses = list(responses) self.calls = [] def _request(self, method, path, json=None, params=None): self.calls.append((method, path, json, params)) return self.responses.pop(0) def test_sku_path_double_encodes_slash_for_magento_oauth(): assert _sku_path("ACH-BBC36/39L/R") == "ACH-BBC36%252F39L%252FR" def test_sku_path_still_encodes_other_reserved_chars(): assert _sku_path("SKU A&B") == "SKU%20A%26B" def test_get_product_media_retries_single_encoded_slash_path(): oauth = _FakeOAuth( [ (404, {"message": "The product that was requested doesn't exist."}, "404"), ( 200, [{"id": 11, "file": "/a/b.jpg"}], None, ), ] ) client = MagentoRestClient(oauth) status, entries, err = client.get_product_media("ACH-BBC45/48L/R") assert status == 200 assert err is None assert entries == [{"id": 11, "file": "/a/b.jpg"}] assert oauth.calls[0][0:2] == ("GET", "products/ACH-BBC45%252F48L%252FR/media") assert oauth.calls[1][0:2] == ("GET", "products/ACH-BBC45%2F48L%2FR/media") def test_get_product_falls_back_to_search_for_slash_sku(monkeypatch): oauth = _FakeOAuth( [ (404, {"message": "missing"}, "404"), (404, {"message": "missing"}, "404"), ] ) client = MagentoRestClient(oauth) def _search(*, sku=None, page_size=20, entity_id=None): assert sku == "ACH-BCOR33L/R" return 200, [{"sku": "ACH-BCOR33L/R", "media_gallery_entries": [{"id": 7}]}], None monkeypatch.setattr(client, "search_products", _search) product = client.get_product("ACH-BCOR33L/R") assert product["sku"] == "ACH-BCOR33L/R" assert product["media_gallery_entries"][0]["id"] == 7 def test_put_product_falls_back_to_post_for_slash_sku_signature_issue(): oauth = _FakeOAuth( [ (401, {"message": "The signature is invalid. Verify and try again."}, "401"), (200, {"id": 1}, None), ] ) client = MagentoRestClient(oauth) status, body = client.put_product("ACH-BBC36/39L/R", {"sku": "ACH-BBC36/39L/R", "visibility": 1}) assert status == 200 assert body == {"id": 1} assert oauth.calls[0][0:2] == ("PUT", "products/ACH-BBC36%252F39L%252FR") assert oauth.calls[1][0:2] == ("POST", "products") def test_delete_product_tries_single_encode_then_confirms_absence_for_slash_sku(monkeypatch): oauth = _FakeOAuth( [ (404, {"message": "The product that was requested doesn't exist."}, "404"), (404, {"message": "The product that was requested doesn't exist."}, "404"), ] ) client = MagentoRestClient(oauth) monkeypatch.setattr(client, "product_exists_by_sku_search", lambda _sku: False) status, err = client.delete_product("RTA-MSW-BCOR36L/R") assert status == 204 assert err is None assert oauth.calls[0][0:2] == ("DELETE", "products/RTA-MSW-BCOR36L%252FR") assert oauth.calls[1][0:2] == ("DELETE", "products/RTA-MSW-BCOR36L%2FR") def test_delete_product_reports_failure_when_slash_sku_still_searchable(monkeypatch): oauth = _FakeOAuth( [ (404, {"message": "missing"}, "404"), (404, {"message": "missing"}, "404"), ] ) client = MagentoRestClient(oauth) monkeypatch.setattr(client, "product_exists_by_sku_search", lambda _sku: True) status, err = client.delete_product("RTA-MSW-BCOR36L/R") assert status == 404 assert err assert "missing" in err or "404" in err def test_build_magento_oauth_kwargs_prefers_public_api_by_default(): kwargs = build_magento_oauth_kwargs( { "store_base_url": "https://example.com", "magento_api_base_url": "https://example.com/rest", "internal_api_base_url": "http://127.0.0.1/rest", "internal_host_header": "example.com", "prefer_internal_api": False, "rest_base_path": "V1", "signature_method": "HMAC-SHA256", } ) assert kwargs["base_url"] == "https://example.com/rest" assert kwargs["default_headers"] == {} assert kwargs["verify_ssl"] is True assert kwargs["fallback_base_url"] == "http://127.0.0.1/rest" assert kwargs["fallback_headers"] == {"Host": "example.com"} assert kwargs["fallback_verify_ssl"] is False def test_build_magento_oauth_kwargs_uses_internal_api_and_host_header_when_enabled(): kwargs = build_magento_oauth_kwargs( { "store_base_url": "https://example.com", "magento_api_base_url": "https://example.com/rest", "internal_api_base_url": "http://127.0.0.1/rest", "internal_host_header": "example.com", "prefer_internal_api": True, "store_code": "us_eng", "rest_base_path": "V1", "signature_method": "HMAC-SHA256", } ) assert kwargs["base_url"] == "http://127.0.0.1/rest" assert kwargs["rest_base_path"] == "us_eng/V1" assert kwargs["default_headers"] == {"Host": "example.com"} assert kwargs["verify_ssl"] is False assert kwargs["fallback_base_url"] is None assert kwargs["fallback_verify_ssl"] is True def test_build_magento_oauth_kwargs_disables_verify_for_loopback_https(): kwargs = build_magento_oauth_kwargs( { "store_base_url": "https://127.0.0.1", "magento_api_base_url": "https://127.0.0.1/rest", "prefer_internal_api": False, "rest_base_path": "V1", "signature_method": "HMAC-SHA256", } ) assert kwargs["base_url"] == "https://127.0.0.1/rest" assert kwargs["verify_ssl"] is False def test_build_magento_oauth_kwargs_verify_ssl_override(): kwargs = build_magento_oauth_kwargs( { "store_base_url": "https://example.com", "magento_api_base_url": "https://example.com/rest", "prefer_internal_api": False, "rest_base_path": "V1", "signature_method": "HMAC-SHA256", }, verify_ssl=False, ) assert kwargs["verify_ssl"] is False def test_oauth_client_retries_413_via_internal_transport_without_ssl_verification(monkeypatch): calls = [] class _Resp: def __init__(self, status_code, url, text="", headers=None): self.status_code = status_code self.url = url self.text = text self.headers = headers or {} self.content = text.encode("utf-8") if text else b"{}" def json(self): return {"ok": True, "url": self.url} def _fake_request(method, url, **kwargs): calls.append({"method": method, "url": url, "verify": kwargs.get("verify"), "headers": kwargs.get("headers")}) if len(calls) == 1: return _Resp(413, url, text="too large", headers={"Server": "cloudflare"}) return _Resp(200, url) monkeypatch.setattr("magento.oauth_client.requests.request", _fake_request) client = MagentoOAuthClient( base_url="https://public.example/rest", consumer_key="ck", consumer_secret="cs", access_token="at", access_token_secret="ats", rest_base_path="V1", verify_ssl=True, fallback_base_url="https://10.0.0.5/rest", fallback_headers={"Host": "public.example"}, fallback_verify_ssl=False, ) status, body, err = client._request("POST", "piehs/collection-media", json={"filename": "spec.pdf"}) assert status == 200 assert err is None assert body == {"ok": True, "url": "https://10.0.0.5/rest/V1/piehs/collection-media"} assert calls[0]["url"] == "https://public.example/rest/V1/piehs/collection-media" assert calls[0]["verify"] is True assert calls[1]["url"] == "https://10.0.0.5/rest/V1/piehs/collection-media" assert calls[1]["verify"] is False assert calls[1]["headers"]["Host"] == "public.example" def test_oauth_client_suppresses_insecure_request_warning_when_verify_false(monkeypatch): class _Resp: status_code = 200 url = "http://127.0.0.1/rest/V1/test" text = "" headers = {} content = b"{}" def json(self): return {} def _fake_request(method, url, **kwargs): warnings.warn("unverified https request", InsecureRequestWarning) return _Resp() monkeypatch.setattr("magento.oauth_client.requests.request", _fake_request) client = MagentoOAuthClient( base_url="http://127.0.0.1/rest", consumer_key="ck", consumer_secret="cs", access_token="at", access_token_secret="ats", verify_ssl=False, ) with warnings.catch_warnings(record=True) as caught: warnings.simplefilter("always") status, body, err = client._request("GET", "test") assert status == 200 assert body == {} assert err is None assert not any(item.category is InsecureRequestWarning for item in caught) def test_oauth_client_recovers_after_server_down_rounds(monkeypatch): """After round-1 503s, sleep 5m then succeed on the next recovery round.""" sleeps: list[float] = [] calls = {"n": 0} class _Resp: def __init__(self, status_code): self.status_code = status_code self.url = "https://mage.example/rest/V1/categories/464" self.text = "Service Unavailable" if status_code == 503 else "{}" self.headers = {} self.content = self.text.encode("utf-8") def json(self): return {"id": 464} if self.status_code == 200 else {"message": "down"} def _fake_request(method, url, **kwargs): calls["n"] += 1 # First round: 3×503; second round first attempt succeeds. if calls["n"] <= _IMMEDIATE_RETRY_COUNT: return _Resp(503) return _Resp(200) monkeypatch.setattr("magento.oauth_client.requests.request", _fake_request) monkeypatch.setattr("magento.oauth_client.time.sleep", lambda s: sleeps.append(s)) client = MagentoOAuthClient( base_url="https://mage.example/rest", consumer_key="ck", consumer_secret="cs", access_token="at", access_token_secret="ats", ) status, body, err = client._request("GET", "categories/464") assert status == 200 assert body == {"id": 464} assert err is None assert calls["n"] == _IMMEDIATE_RETRY_COUNT + 1 # Within-round delays (2) + one 5-minute recovery wait before round 2 assert 300.0 in sleeps assert sleeps.count(30.0) == _IMMEDIATE_RETRY_COUNT - 1 def test_oauth_client_exhausts_all_server_down_recovery_rounds(monkeypatch): sleeps: list[float] = [] class _Resp: status_code = 503 url = "https://mage.example/rest/V1/categories/464" text = "Service Unavailable" headers = {} content = b"Service Unavailable" def json(self): return {"message": "down"} monkeypatch.setattr( "magento.oauth_client.requests.request", lambda *a, **k: _Resp(), ) monkeypatch.setattr("magento.oauth_client.time.sleep", lambda s: sleeps.append(s)) client = MagentoOAuthClient( base_url="https://mage.example/rest", consumer_key="ck", consumer_secret="cs", access_token="at", access_token_secret="ats", ) status, body, err = client._request("GET", "categories/464") assert status == 503 assert list(_SERVER_DOWN_RECOVERY_WAITS) == [300, 600, 1800] for wait in _SERVER_DOWN_RECOVERY_WAITS: assert wait in sleeps # 4 rounds × 2 intra-round delays = 8 × 30s, plus 3 recovery waits assert sleeps.count(30.0) == (1 + len(_SERVER_DOWN_RECOVERY_WAITS)) * (_IMMEDIATE_RETRY_COUNT - 1)