add more

Author: fangnx

src/confluent_kafka/schema_registry/_async/schema_registry_client.py

@@ -1307,6 +1307,30 @@ async def set_config(
             'config/{}'.format(_urlencode(subject_name)), body=config.to_dict()
         )
 
+    async def delete_config(self, subject_name: Optional[str] = None) -> 'ServerConfig':
+        """
+        Delete the specified subject-level compatibility level config and revert to the global default.
+
+        Args:
+            subject_name (str, optional): Subject name. Deletes global config
+                if left unset.
+
+        Returns:
+            ServerConfig: The old deleted config
+
+        Raises:
+            SchemaRegistryError: if the request was unsuccessful.
+
+        See Also:
+            `DELETE Subject Config API Reference <https://docs.confluent.io/current/schema-registry/develop/api.html#delete--config-(string- subject)>`_
+        """
+        if subject_name is not None:
+            url = 'config/{}'.format(_urlencode(subject_name))
+        else:
+            url = 'config'
+        result = await self._rest_client.delete(url)
+        return ServerConfig.from_dict(result)
+
     async def get_config(self, subject_name: Optional[str] = None) -> 'ServerConfig':
         """
         Get the current config.
@@ -1430,6 +1454,23 @@ async def update_global_mode(self, mode: str, force: bool = False) -> str:
         """  # noqa: E501
         result = await self._rest_client.put('mode?force={}'.format(force), body={'mode': mode})
         return result['mode']
+    async def get_contexts(self, offset: int = 0, limit: int = -1) -> List[str]:
+        """
+        Retrieves a list of contexts.
+
+        Args:
+            offset (int): Pagination offset for results.
+            limit (int): Pagination size for results. Ignored if negative.
+
+        Returns:
+            List[str]: List of contexts.
+
+        Raises:
+            SchemaRegistryError: if the request was unsuccessful.
+        """  # noqa: E501
+
+        result = await self._rest_client.get('contexts', query={'offset': offset, 'limit': limit})
+        return result
 
     def clear_latest_caches(self):
         self._latest_version_cache.clear()

src/confluent_kafka/schema_registry/_sync/schema_registry_client.py

@@ -1307,6 +1307,30 @@ def set_config(
             'config/{}'.format(_urlencode(subject_name)), body=config.to_dict()
         )
 
+    def delete_config(self, subject_name: Optional[str] = None) -> 'ServerConfig':
+        """
+        Delete the specified subject-level compatibility level config and revert to the global default.
+
+        Args:
+            subject_name (str, optional): Subject name. Deletes global config
+                if left unset.
+
+        Returns:
+            ServerConfig: The old deleted config
+
+        Raises:
+            SchemaRegistryError: if the request was unsuccessful.
+
+        See Also:
+            `DELETE Subject Config API Reference <https://docs.confluent.io/current/schema-registry/develop/api.html#delete--config-(string- subject)>`_
+        """
+        if subject_name is not None:
+            url = 'config/{}'.format(_urlencode(subject_name))
+        else:
+            url = 'config'
+        result = self._rest_client.delete(url)
+        return ServerConfig.from_dict(result)
+
     def get_config(self, subject_name: Optional[str] = None) -> 'ServerConfig':
         """
         Get the current config.
@@ -1430,6 +1454,23 @@ def update_global_mode(self, mode: str, force: bool = False) -> str:
         """  # noqa: E501
         result = self._rest_client.put('mode?force={}'.format(force), body={'mode': mode})
         return result['mode']
+    def get_contexts(self, offset: int = 0, limit: int = -1) -> List[str]:
+        """
+        Retrieves a list of contexts.
+
+        Args:
+            offset (int): Pagination offset for results.
+            limit (int): Pagination size for results. Ignored if negative.
+
+        Returns:
+            List[str]: List of contexts.
+
+        Raises:
+            SchemaRegistryError: if the request was unsuccessful.
+        """  # noqa: E501
+
+        result = self._rest_client.get('contexts', query={'offset': offset, 'limit': limit})
+        return result
 
     def clear_latest_caches(self):
         self._latest_version_cache.clear()

tests/schema_registry/_async/test_api_client.py

@@ -426,6 +426,24 @@ async def test_set_compatibility_invalid(mock_schema_registry):
     e.value.error_code = 42203
 
 
+async def test_delete_config(mock_schema_registry):
+    conf = {'url': TEST_URL}
+    sr = AsyncSchemaRegistryClient(conf)
+
+    result = await sr.delete_config()
+    assert result.compatibility == 'FULL'
+
+
+async def test_delete_config_subject_not_found(mock_schema_registry):
+    conf = {'url': TEST_URL}
+    sr = AsyncSchemaRegistryClient(conf)
+
+    with pytest.raises(SchemaRegistryError, match="Subject not found") as e:
+        await sr.delete_config("notfound")
+    assert e.value.http_status_code == 404
+    assert e.value.error_code == 40401
+
+
 async def test_get_compatibility_subject_not_found(mock_schema_registry):
     conf = {'url': TEST_URL}
     sr = AsyncSchemaRegistryClient(conf)
@@ -436,6 +454,14 @@ async def test_get_compatibility_subject_not_found(mock_schema_registry):
     assert e.value.error_code == 40401
 
 
+async def test_get_contexts(mock_schema_registry):
+    conf = {'url': TEST_URL}
+    sr = AsyncSchemaRegistryClient(conf)
+
+    result = await sr.get_contexts()
+    assert result == ['context1', 'context2']
+
+
 async def test_schema_equivilence(load_avsc):
     schema_str1 = load_avsc('basic_schema.avsc')
     schema_str2 = load_avsc('basic_schema.avsc')

tests/schema_registry/_sync/test_api_client.py

@@ -426,6 +426,24 @@ def test_set_compatibility_invalid(mock_schema_registry):
     e.value.error_code = 42203
 
 
+def test_delete_config(mock_schema_registry):
+    conf = {'url': TEST_URL}
+    sr = SchemaRegistryClient(conf)
+
+    result = sr.delete_config()
+    assert result.compatibility == 'FULL'
+
+
+def test_delete_config_subject_not_found(mock_schema_registry):
+    conf = {'url': TEST_URL}
+    sr = SchemaRegistryClient(conf)
+
+    with pytest.raises(SchemaRegistryError, match="Subject not found") as e:
+        sr.delete_config("notfound")
+    assert e.value.http_status_code == 404
+    assert e.value.error_code == 40401
+
+
 def test_get_compatibility_subject_not_found(mock_schema_registry):
     conf = {'url': TEST_URL}
     sr = SchemaRegistryClient(conf)
@@ -436,6 +454,14 @@ def test_get_compatibility_subject_not_found(mock_schema_registry):
     assert e.value.error_code == 40401
 
 
+def test_get_contexts(mock_schema_registry):
+    conf = {'url': TEST_URL}
+    sr = SchemaRegistryClient(conf)
+
+    result = sr.get_contexts()
+    assert result == ['context1', 'context2']
+
+
 def test_schema_equivilence(load_avsc):
     schema_str1 = load_avsc('basic_schema.avsc')
     schema_str2 = load_avsc('basic_schema.avsc')

tests/schema_registry/conftest.py

@@ -97,10 +97,12 @@
 +--------+-------------------------------------------------+-------+------------------------------+
 | DELETE | /subjects/notfound/versions/422                 | 42202 | Invalid version              |
 +--------+-------------------------------------------------+-------+------------------------------+
-| GET    | /config/notconfig                               | 40401 | Subject not found            |
+| GET    | /config/notfound                                | 40401 | Subject not found            |
 +--------+-------------------------------------------------+-------+------------------------------+
 | PUT    | /config**                                       | 42203 | Invalid compatibility level  |
 +--------+-------------------------------------------------+-------+------------------------------+
+| DELETE | /config/notfound                                | 40401 | Subject not found            |
++--------+-------------------------------------------------+-------+------------------------------+
 | POST   | /compatibility/subjects/notfound/versions/[0-9] | 40401 | Subject not found            |
 +--------+-------------------------------------------------+-------+------------------------------+
 | POST   | /compatibility/subjects/invalid/versions/[0-9]  | 42201 | Invalid Schema               |
@@ -138,8 +140,11 @@ def mock_schema_registry():
         respx_mock.post(COMPATIBILITY_SUBJECTS_ALL_VERSIONS_RE).mock(
             side_effect=post_compatibility_subjects_all_versions_callback)
 
-        respx_mock.get(COMPATIBILITY_RE).mock(side_effect=get_compatibility_callback)
-        respx_mock.put(COMPATIBILITY_RE).mock(side_effect=put_compatibility_callback)
+        respx_mock.get(CONFIG_RE).mock(side_effect=get_config_callback)
+        respx_mock.put(CONFIG_RE).mock(side_effect=put_config_callback)
+        respx_mock.delete(CONFIG_RE).mock(side_effect=delete_config_callback)
+
+        respx_mock.get(CONTEXTS_RE).mock(side_effect=get_contexts_callback)
 
         respx_mock.get(MODE_GLOBAL_RE).mock(side_effect=get_global_mode_callback)
         respx_mock.put(MODE_GLOBAL_RE).mock(side_effect=put_global_mode_callback)
@@ -175,13 +180,16 @@ def mock_schema_registry():
 SUBJECTS_VERSIONS_SCHEMA_RE = re.compile(r"/subjects/(.*)/versions/(.*)/schema(\?.*)?$")
 SUBJECTS_VERSIONS_REFERENCED_BY_RE = re.compile(r"/subjects/(.*)/versions/(.*)/referencedby(\?.*)?$")
 
-COMPATIBILITY_RE = re.compile("/config/?(.*)$")
+CONFIG_RE = re.compile("/config/?(.*)$")
+
 COMPATIBILITY_SUBJECTS_VERSIONS_RE = re.compile("/compatibility/subjects/(.*)/versions/?(.*)$")
 COMPATIBILITY_SUBJECTS_ALL_VERSIONS_RE = re.compile("/compatibility/subjects/(.*)/versions")
 
 MODE_GLOBAL_RE = re.compile(r"/mode(\?.*)?$")
 MODE_RE = re.compile("/mode/(.*)$")
 
+CONTEXTS_RE = re.compile(r"/contexts(\?.*)?$")
+
 # constants
 SCHEMA_ID = 47
 VERSION = 3
@@ -222,10 +230,10 @@ def _load_avsc(name) -> str:
         return fd.read()
 
 
-def get_compatibility_callback(request, route):
+def get_config_callback(request, route):
     COUNTER['GET'][request.url.path] += 1
 
-    path_match = re.match(COMPATIBILITY_RE, request.url.path)
+    path_match = re.match(CONFIG_RE, request.url.path)
     subject = path_match.group(1)
 
     if subject == "notfound":
@@ -235,7 +243,7 @@ def get_compatibility_callback(request, route):
     return Response(200, json={'compatibility': 'FULL'})
 
 
-def put_compatibility_callback(request, route):
+def put_config_callback(request, route):
     COUNTER['PUT'][request.url.path] += 1
 
     body = json.loads(request.content.decode('utf-8'))
@@ -248,6 +256,24 @@ def put_compatibility_callback(request, route):
     return Response(200, json=body)
 
 
+def delete_config_callback(request, route):
+    COUNTER['DELETE'][request.url.path] += 1
+
+    path_match = re.match(CONFIG_RE, request.url.path)
+    subject = path_match.group(1)
+
+    if subject == "notfound":
+        return Response(404, json={'error_code': 40401,
+                                   'message': "Subject not found"})
+
+    return Response(200, json={'compatibility': 'FULL'})
+
+
+def get_contexts_callback(request, route):
+    COUNTER['GET'][request.url.path] += 1
+    return Response(200, json=['context1', 'context2'])
+
+
 def delete_subject_callback(request, route):
     COUNTER['DELETE'][request.url.path] += 1