added scope as configurable option, defaults to CATALOG_SCOPE

added scope as configurable option, defaults to `CATALOG_SCOPE`

resolve conflicts.

change to constant

style fix

pyiceberg/catalog/rest.py

@@ -294,7 +294,11 @@ def _fetch_access_token(self, session: Session, credential: str) -> str:
             client_id, client_secret = credential.split(SEMICOLON)
         else:
             client_id, client_secret = None, credential
-        data = {GRANT_TYPE: CLIENT_CREDENTIALS, CLIENT_ID: client_id, CLIENT_SECRET: client_secret, SCOPE: CATALOG_SCOPE}
+
+        # take scope from properties or use default CATALOG_SCOPE
+        scope = self.properties.get(SCOPE) or CATALOG_SCOPE
+
+        data = {GRANT_TYPE: CLIENT_CREDENTIALS, CLIENT_ID: client_id, CLIENT_SECRET: client_secret, SCOPE: scope}
         response = session.post(
             url=self.auth_url, data=data, headers={**session.headers, "Content-type": "application/x-www-form-urlencoded"}
         )

tests/catalog/test_rest.py

@@ -46,6 +46,7 @@
 TEST_CREDENTIALS = "client:secret"
 TEST_AUTH_URL = "https://auth-endpoint/"
 TEST_TOKEN = "some_jwt_token"
+TEST_SCOPE = "openid_offline_corpds_ds_profile"
 TEST_HEADERS = {
     "Content-type": "application/json",
     "X-Client-Version": "0.14.1",
@@ -136,6 +137,43 @@ def test_token_200_without_optional_fields(rest_mock: Mocker) -> None:
     )
 
 
+def test_token_with_default_scope(rest_mock: Mocker) -> None:
+    mock_request = rest_mock.post(
+        f"{TEST_URI}v1/oauth/tokens",
+        json={
+            "access_token": TEST_TOKEN,
+            "token_type": "Bearer",
+            "expires_in": 86400,
+            "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        },
+        status_code=200,
+        request_headers=OAUTH_TEST_HEADERS,
+    )
+    assert (
+        RestCatalog("rest", uri=TEST_URI, credential=TEST_CREDENTIALS)._session.headers["Authorization"] == f"Bearer {TEST_TOKEN}"
+    )
+    assert "catalog" in mock_request.last_request.text
+
+
+def test_token_with_custom_scope(rest_mock: Mocker) -> None:
+    mock_request = rest_mock.post(
+        f"{TEST_URI}v1/oauth/tokens",
+        json={
+            "access_token": TEST_TOKEN,
+            "token_type": "Bearer",
+            "expires_in": 86400,
+            "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        },
+        status_code=200,
+        request_headers=OAUTH_TEST_HEADERS,
+    )
+    assert (
+        RestCatalog("rest", uri=TEST_URI, credential=TEST_CREDENTIALS, scope=TEST_SCOPE)._session.headers["Authorization"]
+        == f"Bearer {TEST_TOKEN}"
+    )
+    assert TEST_SCOPE in mock_request.last_request.text
+
+
 def test_token_200_w_auth_url(rest_mock: Mocker) -> None:
     rest_mock.post(
         TEST_AUTH_URL,