added reset / reload

src/geoserverx/_async/gsx.py

@@ -341,6 +341,32 @@ async def get_all_layer_groups(
             results = self.response_recognise(responses.status_code)
             return results
 
+    # Reset geoserver
+    async def reset_geoserver(self) -> GSResponse:
+        """
+        Resets all authentication, store, raster, and schema caches. This operation is used to force GeoServer to drop all caches and store connections and reconnect to each of them the next time they are needed by a request. This is useful in case the stores themselves cache some information about the data structures they manage that may have changed in the meantime.
+        """
+        Client = self.http_client
+        responses = await Client.put(
+            "/reset",
+            headers=self.head,
+        )
+        results = self.response_recognise(responses.status_code)
+        return results
+
+    # Reload geoserver
+    async def reload_geoserver(self) -> GSResponse:
+        """
+        Reloads the GeoServer catalog and configuration from disk. This operation is used in cases where an external tool has modified the on-disk configuration. This operation will also force GeoServer to drop any internal caches and reconnect to all data stores.
+        """
+        Client = self.http_client
+        responses = await Client.put(
+            "/reload",
+            headers=self.head,
+        )
+        results = self.response_recognise(responses.status_code)
+        return results
+
     # Get all geofence rules
     async def get_all_geofence_rules(self) -> Union[RulesResponse, GSResponse]:
         Client = self.http_client

src/geoserverx/_sync/gsx.py

@@ -464,6 +464,34 @@ def get_all_layer_groups(
             results = self.response_recognise(responses.status_code)
             return results
 
+    # Reset geoserver
+    @exception_handler
+    def reset_geoserver(self) -> GSResponse:
+        """
+        Resets all authentication, store, raster, and schema caches. This operation is used to force GeoServer to drop all caches and store connections and reconnect to each of them the next time they are needed by a request. This is useful in case the stores themselves cache some information about the data structures they manage that may have changed in the meantime.
+        """
+        Client = self.http_client
+        responses = Client.put(
+            "/reset",
+            headers=self.head,
+        )
+        results = self.response_recognise(responses.status_code)
+        return results
+
+    # Reload geoserver
+    @exception_handler
+    def reload_geoserver(self) -> GSResponse:
+        """
+        Reloads the GeoServer catalog and configuration from disk. This operation is used in cases where an external tool has modified the on-disk configuration. This operation will also force GeoServer to drop any internal caches and reconnect to all data stores.
+        """
+        Client = self.http_client
+        responses = Client.put(
+            "/reload",
+            headers=self.head,
+        )
+        results = self.response_recognise(responses.status_code)
+        return results
+
     # Get all geofence rules
     @exception_handler
     def get_all_geofence_rules(self) -> Union[RulesResponse, GSResponse]:

src/geoserverx/cli/cli.py

@@ -447,6 +447,28 @@ def geofence_rules(
         typer.echo("Async support will be shortly")
 
 
+# Reset geoserver
+@SyncGeoServerX.exception_handler
+@app.command(help="Reset geoserver settings")
+def reset(
+    request: requestEnum = requestEnum._sync,
+    url: str = typer.Option(
+        "http://127.0.0.1:8080/geoserver/rest/", help="Geoserver REST URL"
+    ),
+    password: str = typer.Option("geoserver", help="Geoserver Password"),
+    username: str = typer.Option("admin", help="Geoserver username"),
+):
+    """
+    Resets all authentication, store, raster, and schema caches. This operation is used to force GeoServer to drop all caches and store connections and reconnect to each of them the next time they are needed by a request. This is useful in case the stores themselves cache some information about the data structures they manage that may have changed in the meantime.
+    """
+    if request.value == "sync":
+        client = SyncGeoServerX(username, password, url)
+        result = client.reset_geoserver()
+        typer.secho(result, fg=typer.colors.GREEN)
+    else:
+        typer.echo("Async support will be shortly")
+
+
 # get geofence rule
 @SyncGeoServerX.exception_handler
 @app.command(help="Get geofence rule in the Geoserver")
@@ -471,3 +493,25 @@ def geofence_rule(
             print(result)
     else:
         typer.echo("Async support will be shortly")
+
+
+# Reload geoserver
+@SyncGeoServerX.exception_handler
+@app.command(help="Reload geoserver settings")
+def reload(
+    request: requestEnum = requestEnum._sync,
+    url: str = typer.Option(
+        "http://127.0.0.1:8080/geoserver/rest/", help="Geoserver REST URL"
+    ),
+    password: str = typer.Option("geoserver", help="Geoserver Password"),
+    username: str = typer.Option("admin", help="Geoserver username"),
+):
+    """
+    Reloads the GeoServer catalog and configuration from disk. This operation is used in cases where an external tool has modified the on-disk configuration. This operation will also force GeoServer to drop any internal caches and reconnect to all data stores.
+    """
+    if request.value == "sync":
+        client = SyncGeoServerX(username, password, url)
+        result = client.reload_geoserver()
+        typer.secho(result, fg=typer.colors.GREEN)
+    else:
+        typer.echo("Async support will be shortly")

tests/_async/test_gsx.py

@@ -468,6 +468,18 @@ async def test_get_all_layer_groups_NetworkError(create_a_client, respx_mock):
         assert response.response == "Error in connecting to Geoserver"
 
 
+# Test - reset_geoserver
+@pytest.mark.asyncio
+async def test_reset_geoserver_validation(
+    create_a_client, respx_mock, bad_reset_geoserver_connection
+):
+    respx_mock.put(f"{baseUrl}reset").mock(
+        return_value=httpx.Response(404, json=bad_reset_geoserver_connection)
+    )
+    response = await create_a_client.reset_geoserver()
+    assert response.code == 404
+
+
 # Test - all_geofence_rules
 @pytest.mark.asyncio
 async def test_all_geofence_rules_validation(
@@ -485,6 +497,56 @@ async def test_all_geofence_rules_validation(
     assert response.code == 404
 
 
+@pytest.mark.asyncio
+async def test_reset_geoserver_success(
+    create_a_client, respx_mock, good_reset_geoserver_connection
+):
+    respx_mock.put(f"{baseUrl}reset").mock(
+        return_value=httpx.Response(200, json=good_reset_geoserver_connection)
+    )
+    response = await create_a_client.reset_geoserver()
+    assert response.response == "Executed successfully"
+
+
+@pytest.mark.asyncio
+async def test_reset_geoserver_NetworkError(create_a_client, respx_mock):
+    respx.put(f"{baseUrl}reset").mock(side_effect=httpx.ConnectError)
+    with pytest.raises(httpx.ConnectError):
+        response = await create_a_client.reset_geoserver()
+        assert response.response == "Error in connecting to Geoserver"
+
+
+# Test - reload_geoserver
+@pytest.mark.asyncio
+async def test_reload_geoserver_validation(
+    create_a_client, respx_mock, bad_reload_geoserver_connection
+):
+    respx_mock.put(f"{baseUrl}reload").mock(
+        return_value=httpx.Response(404, json=bad_reload_geoserver_connection)
+    )
+    response = await create_a_client.reload_geoserver()
+    assert response.code == 404
+
+
+@pytest.mark.asyncio
+async def test_reload_geoserver_success(
+    create_a_client, respx_mock, good_reload_geoserver_connection
+):
+    respx_mock.put(f"{baseUrl}reload").mock(
+        return_value=httpx.Response(200, json=good_reload_geoserver_connection)
+    )
+    response = await create_a_client.reload_geoserver()
+    assert response.response == "Executed successfully"
+
+
+@pytest.mark.asyncio
+async def test_reload_geoserver_NetworkError(create_a_client, respx_mock):
+    respx.put(f"{baseUrl}reload").mock(side_effect=httpx.ConnectError)
+    with pytest.raises(httpx.ConnectError):
+        response = await create_a_client.reload_geoserver()
+        assert response.response == "Error in connecting to Geoserver"
+
+
 @pytest.mark.asyncio
 async def test_all_geofence_rules_success(
     create_a_client, respx_mock, good_all_geofence_rules_connection

tests/_sync/test_gsx.py

@@ -413,6 +413,60 @@ def test_get_all_layer_groups_ConnectError(client: SyncGeoServerX, respx_mock):
     assert response.response == "Error in connecting to Geoserver"
 
 
+# Test - reset_geoserver
+def test_reset_geoserver_validation(
+    client: SyncGeoServerX, bad_reset_geoserver_connection, respx_mock
+):
+    respx_mock.put(f"{baseUrl}reset").mock(
+        return_value=httpx.Response(404, json=bad_reset_geoserver_connection)
+    )
+    response = client.reset_geoserver()
+    assert response.response == "Result not found"
+
+
+def test_reset_geoserver_success(
+    client: SyncGeoServerX, good_reset_geoserver_connection, respx_mock
+):
+    respx_mock.put(f"{baseUrl}reset").mock(
+        return_value=httpx.Response(200, json=good_reset_geoserver_connection)
+    )
+    response = client.reset_geoserver()
+    assert response.response == "Executed successfully"
+
+
+def test_reset_geoserver_ConnectError(client: SyncGeoServerX, respx_mock):
+    respx_mock.put(f"{baseUrl}reset").mock(side_effect=httpx.ConnectError)
+    response = client.reset_geoserver()
+    assert response.response == "Error in connecting to Geoserver"
+
+
+# Test - reload_geoserver
+def test_reload_geoserver_validation(
+    client: SyncGeoServerX, bad_reload_geoserver_connection, respx_mock
+):
+    respx_mock.put(f"{baseUrl}reload").mock(
+        return_value=httpx.Response(404, json=bad_reload_geoserver_connection)
+    )
+    response = client.reload_geoserver()
+    assert response.response == "Result not found"
+
+
+def test_reload_geoserver_success(
+    client: SyncGeoServerX, good_reload_geoserver_connection, respx_mock
+):
+    respx_mock.put(f"{baseUrl}reload").mock(
+        return_value=httpx.Response(200, json=good_reload_geoserver_connection)
+    )
+    response = client.reload_geoserver()
+    assert response.response == "Executed successfully"
+
+
+def test_reload_geoserver_ConnectError(client: SyncGeoServerX, respx_mock):
+    respx_mock.put(f"{baseUrl}reload").mock(side_effect=httpx.ConnectError)
+    response = client.reload_geoserver()
+    assert response.response == "Error in connecting to Geoserver"
+
+
 # Test - all_geofence_rules
 def test_all_geofence_rules_validation(
     client: SyncGeoServerX, bad_all_geofence_rules_connection, respx_mock

tests/cli/test_cli.py

@@ -390,6 +390,52 @@ def test_get_all_layer_groups_NetworkError(respx_mock):
     assert "Error in connecting to Geoserver" in result.stdout
 
 
+# Test - reset_geoserver
+def test_reset_geoserver_validation(bad_reset_geoserver_connection, respx_mock):
+    respx_mock.put(f"{baseUrl}reset").mock(
+        return_value=httpx.Response(404, json=bad_reset_geoserver_connection)
+    )
+    result = runner.invoke(app, ["reset"])
+    assert "404" in result.stdout
+
+
+def test_reset_geoserver_success(good_reset_geoserver_connection, respx_mock):
+    respx_mock.put(f"{baseUrl}reset").mock(
+        return_value=httpx.Response(200, json=good_reset_geoserver_connection)
+    )
+    result = runner.invoke(app, ["reset"])
+    assert "200" in result.stdout
+
+
+def test_reset_geoserver_NetworkError(respx_mock):
+    respx_mock.put(f"{baseUrl}reset").mock(side_effect=httpx.ConnectError)
+    result = runner.invoke(app, ["reset"])
+    assert "Error in connecting to Geoserver" in result.stdout
+
+
+# Test - reload_geoserver
+def test_reload_geoserver_validation(bad_reload_geoserver_connection, respx_mock):
+    respx_mock.put(f"{baseUrl}reload").mock(
+        return_value=httpx.Response(404, json=bad_reload_geoserver_connection)
+    )
+    result = runner.invoke(app, ["reload"])
+    assert "404" in result.stdout
+
+
+def test_reload_geoserver_success(good_reload_geoserver_connection, respx_mock):
+    respx_mock.put(f"{baseUrl}reload").mock(
+        return_value=httpx.Response(200, json=good_reload_geoserver_connection)
+    )
+    result = runner.invoke(app, ["reload"])
+    assert "200" in result.stdout
+
+
+def test_reload_geoserver_NetworkError(respx_mock):
+    respx_mock.put(f"{baseUrl}reload").mock(side_effect=httpx.ConnectError)
+    result = runner.invoke(app, ["reload"])
+    assert "Error in connecting to Geoserver" in result.stdout
+
+
 # Test - all_geofence_rules
 def test_all_geofence_rules_validation(bad_all_geofence_rules_connection, respx_mock):
     respx_mock.get(f"{baseUrl}about/status.json").mock(

tests/conftest.py

@@ -672,6 +672,12 @@ def networkbad_layer_groups_connection() -> dict:
     return item
 
 
+@pytest.fixture
+def good_reset_geoserver_connection() -> dict:
+    item = {"code": 200, "response": "Executed successfully'"}
+    return item
+
+
 @pytest.fixture
 def good_all_geofence_rules_connection() -> dict:
     item = {
@@ -809,18 +815,36 @@ def good_all_geofence_rules_connection() -> dict:
     return item
 
 
+@pytest.fixture
+def bad_reset_geoserver_connection() -> dict:
+    item = {"code": 404, "response": "Result not found"}
+    return item
+
+
 @pytest.fixture
 def bad_all_geofence_rules_connection() -> dict:
     item = {"code": 404, "response": "Result not found"}
     return item
 
 
+@pytest.fixture
+def invalid_reset_geoserver_connection() -> dict:
+    item = {"code": 404, "response": "Result not found"}
+    return item
+
+
 @pytest.fixture
 def networkbad_all_geofence_rules_connection() -> dict:
     item = {"code": 503, "response": "Geoserver unavailable"}
     return item
 
 
+@pytest.fixture
+def good_reload_geoserver_connection() -> dict:
+    item = {"code": 200, "response": "Executed successfully'"}
+    return item
+
+
 @pytest.fixture
 def good_new_geofence_rule_connection() -> dict:
     item = {
@@ -842,12 +866,24 @@ def good_new_geofence_rule_connection() -> dict:
     return item
 
 
+@pytest.fixture
+def bad_reload_geoserver_connection() -> dict:
+    item = {"code": 404, "response": "Result not found"}
+    return item
+
+
 @pytest.fixture
 def bad_new_geofence_rule_connection() -> dict:
     item = {"Rule": ""}
     return item
 
 
+@pytest.fixture
+def invalid_reload_geoserver_connection() -> dict:
+    item = {"code": 404, "response": "Result not found"}
+    return item
+
+
 @pytest.fixture
 def invalid_new_geofence_rule_connection() -> dict:
     item = {"code": 404, "response": "Result not found"}