import hashlib import hmac as hmac_mod import json import httpx import pytest from urllib.parse import urlparse, parse_qs from app.core.config import settings from app.services.integrations.shopify_oauth import ( SHOPIFY_API_VERSION, SHOPIFY_SCOPES, ShopQueryError, TokenExchangeError, build_authorize_url, build_hmac_message, exchange_code, fetch_shop_info, is_valid_shop_domain, normalise_shop_domain, redirect_uri, verify_hmac, shopify_credentials, ) @pytest.mark.parametrize("shop", [ "galaxiq-braexaal.myshopify.com", "a.myshopify.com", "store123.myshopify.com", ]) def test_valid_shop_domains_accepted(shop): assert is_valid_shop_domain(shop) is True @pytest.mark.parametrize("shop", [ None, "", "evil.com", "mystore.com", "evil.com/x.myshopify.com", "../../etc", "galaxiq.myshopify.com.attacker.com", "-leading-hyphen.myshopify.com", "has space.myshopify.com", "galaxiq.myshopify.com\n", "GALAXIQ.MYSHOPIFY.COM.evil.com", ]) def test_invalid_shop_domains_rejected(shop): assert is_valid_shop_domain(shop) is False @pytest.mark.parametrize("raw,expected", [ ("galaxiq-braexaal", "galaxiq-braexaal.myshopify.com"), ("galaxiq-braexaal.myshopify.com", "galaxiq-braexaal.myshopify.com"), ("https://galaxiq-braexaal.myshopify.com/admin", "galaxiq-braexaal.myshopify.com"), (" GALAXIQ-BRAEXAAL.MyShopify.com ", "galaxiq-braexaal.myshopify.com"), ("evil.com", None), ("", None), (None, None), ]) def test_normalise_shop_domain(raw, expected): assert normalise_shop_domain(raw) == expected def test_authorize_url_shape(): url = build_authorize_url("galaxiq-braexaal.myshopify.com", "nonce123") parsed = urlparse(url) assert parsed.scheme == "https" assert parsed.netloc == "galaxiq-braexaal.myshopify.com" assert parsed.path == "/admin/oauth/authorize" q = parse_qs(parsed.query) assert q["scope"] == [SHOPIFY_SCOPES] assert q["state"] == ["nonce123"] assert q["redirect_uri"] == [redirect_uri()] assert q["client_id"] == [settings.SHOPIFY_CLIENT_ID] def test_authorize_url_omits_grant_options(): # Presence of grant_options[] yields an ONLINE token that expires with the # merchant's admin session, breaking every scheduled sync. url = build_authorize_url("galaxiq-braexaal.myshopify.com", "nonce123") assert "grant_options" not in url def test_constants(): assert SHOPIFY_API_VERSION == "2026-07" assert SHOPIFY_SCOPES == "read_products,read_inventory" def test_redirect_uri_matches_registered_path(): assert redirect_uri().endswith("/api/shopify/callback") # Must match whatever secret verify_hmac actually uses, not a fixed literal. SECRET = settings.SHOPIFY_CLIENT_SECRET def _sign(params): """Produce the digest Shopify would send for these params.""" msg = build_hmac_message(params) return hmac_mod.new(SECRET.encode(), msg.encode(), hashlib.sha256).hexdigest() def test_message_excludes_hmac_and_signature_and_sorts(): msg = build_hmac_message({ "shop": "s.myshopify.com", "code": "abc", "hmac": "deadbeef", "signature": "ignored", "timestamp": "1700000000", }) assert msg == "code=abc&shop=s.myshopify.com×tamp=1700000000" def test_message_escapes_percent_first_then_amp_and_equals(): # % must be escaped before & and =, or "%" becomes "%2526". msg = build_hmac_message({"a": "100%", "b": "x&y", "c": "k=v"}) assert msg == "a=100%25&b=x%26y&c=k%3Dv" def test_valid_hmac_accepted(): params = { "code": "0907a61c0c8d55e99db179b68161bc00", "shop": "galaxiq-braexaal.myshopify.com", "state": "nonce123", "timestamp": "1786337376", } params["hmac"] = _sign(params) assert verify_hmac(params) is True def test_tampered_parameter_rejected(): params = { "code": "0907a61c0c8d55e99db179b68161bc00", "shop": "galaxiq-braexaal.myshopify.com", "state": "nonce123", "timestamp": "1786337376", } params["hmac"] = _sign(params) params["shop"] = "attacker.myshopify.com" # one field changed assert verify_hmac(params) is False def test_single_character_tamper_rejected(): params = {"code": "abc", "shop": "s.myshopify.com", "timestamp": "1"} params["hmac"] = _sign(params) params["code"] = "abd" assert verify_hmac(params) is False def test_missing_hmac_rejected(): assert verify_hmac({"shop": "s.myshopify.com"}) is False def test_short_hmac_returns_false_and_does_not_raise(): # compare_digest raises on differing lengths; that must not 500 the route. assert verify_hmac({"shop": "s.myshopify.com", "hmac": "ab"}) is False def test_non_hex_hmac_returns_false(): assert verify_hmac({"shop": "s.myshopify.com", "hmac": "zz" * 32}) is False def _client(handler): return httpx.AsyncClient(transport=httpx.MockTransport(handler)) @pytest.mark.asyncio async def test_exchange_code_returns_token_and_scope(): seen = {} def handler(request): seen["url"] = str(request.url) seen["body"] = json.loads(request.content) return httpx.Response(200, json={ "access_token": "shpat_realtoken", "scope": "read_products,read_inventory", }) async with _client(handler) as c: out = await exchange_code("s.myshopify.com", "the-code", client=c) assert out == {"access_token": "shpat_realtoken", "scope": "read_products,read_inventory"} assert seen["url"] == "https://s.myshopify.com/admin/oauth/access_token" assert seen["body"]["code"] == "the-code" assert seen["body"]["client_id"] == settings.SHOPIFY_CLIENT_ID assert seen["body"]["client_secret"] == settings.SHOPIFY_CLIENT_SECRET @pytest.mark.asyncio async def test_exchange_code_raises_on_non_200(): def handler(request): return httpx.Response(400, text="Oauth error invalid_request") async with _client(handler) as c: with pytest.raises(TokenExchangeError): await exchange_code("s.myshopify.com", "bad-code", client=c) @pytest.mark.asyncio async def test_exchange_code_raises_when_token_missing(): def handler(request): return httpx.Response(200, json={"scope": "read_products"}) async with _client(handler) as c: with pytest.raises(TokenExchangeError): await exchange_code("s.myshopify.com", "c", client=c) @pytest.mark.asyncio async def test_fetch_shop_info_parses_graphql(): seen = {} def handler(request): seen["url"] = str(request.url) seen["token"] = request.headers.get("X-Shopify-Access-Token") return httpx.Response(200, json={"data": {"shop": { "name": "Galaxiq dev", "currencyCode": "USD", "primaryDomain": {"url": "https://galaxiq-braexaal.myshopify.com"}, }}}) async with _client(handler) as c: out = await fetch_shop_info("s.myshopify.com", "shpat_x", client=c) assert out == {"name": "Galaxiq dev", "currency_code": "USD", "primary_domain": "https://galaxiq-braexaal.myshopify.com"} assert seen["url"] == "https://s.myshopify.com/admin/api/2026-07/graphql.json" assert seen["token"] == "shpat_x" @pytest.mark.asyncio async def test_fetch_shop_info_raises_on_401(): def handler(request): return httpx.Response(401, text="Invalid API key or access token") async with _client(handler) as c: with pytest.raises(ShopQueryError): await fetch_shop_info("s.myshopify.com", "bad", client=c) @pytest.mark.asyncio async def test_fetch_shop_info_raises_on_graphql_errors(): def handler(request): return httpx.Response(200, json={"errors": [{"message": "Access denied"}]}) async with _client(handler) as c: with pytest.raises(ShopQueryError): await fetch_shop_info("s.myshopify.com", "t", client=c) def test_credentials_prefer_platform_configs(monkeypatch): import app.services.integrations.shopify_oauth as mod monkeypatch.setattr(mod, "get_platform_config", lambda p: { "client_id": "from-platform", "client_secret": "secret-from-platform", "redirect_uri": None, "scope": None}) assert shopify_credentials() == ("from-platform", "secret-from-platform") def test_credentials_fall_back_to_env_when_no_row(monkeypatch): # A missing platform_configs row must not break a working connector. import app.services.integrations.shopify_oauth as mod monkeypatch.setattr(mod, "get_platform_config", lambda p: None) client_id, secret = shopify_credentials() assert client_id == settings.SHOPIFY_CLIENT_ID assert secret == settings.SHOPIFY_CLIENT_SECRET def test_credentials_fall_back_when_the_lookup_raises(monkeypatch): # The platform database being unreachable must not take the connector down. import app.services.integrations.shopify_oauth as mod def boom(platform): raise RuntimeError("master db unreachable") monkeypatch.setattr(mod, "get_platform_config", boom) assert shopify_credentials() == (settings.SHOPIFY_CLIENT_ID, settings.SHOPIFY_CLIENT_SECRET)