FS-4952 : delete endpoints for grants and applications

app/blueprints/fund/routes.py

@@ -11,7 +11,7 @@
 from app.blueprints.fund.forms import FundForm
 from app.blueprints.fund.services import build_fund_rows
 from app.db.models.fund import Fund, FundingType
-from app.db.queries.fund import add_fund, get_all_funds, get_fund_by_id, update_fund
+from app.db.queries.fund import add_fund, get_all_funds, get_fund_by_id, update_fund, delete_selected_fund
 from app.shared.helpers import flash_message
 from app.shared.table_pagination import GovUKTableAndPagination
 
@@ -44,13 +44,17 @@ def view_all_funds():
     return render_template("view_all_funds.html", **params)
 
 
-@fund_bp.route("/<uuid:fund_id>", methods=["GET"])
+@fund_bp.route("/<uuid:fund_id>", methods=["GET", "DELETE"])
 def view_fund_details(fund_id):
     """
     Renders grant details page
     """
     form = FundForm()
+    if request.method == "DELETE":
+        delete_selected_fund(fund_id)
+        return redirect(url_for("fund_bp.view_all_funds"))
     fund = get_fund_by_id(fund_id)
+    #TODO at this time we are not implementing the delete grant but later we have to implement
     return render_template("fund_details.html", form=form, fund=fund)
 
 

app/blueprints/round/routes.py

@@ -19,10 +19,11 @@
 )
 from app.db.queries.clone import clone_single_round
 from app.db.queries.fund import get_all_funds, get_fund_by_id
-from app.db.queries.round import get_all_rounds, get_round_by_id
+from app.db.queries.round import get_all_rounds, get_round_by_id, delete_selected_round
 from app.shared.forms import SelectFundForm
 from app.shared.helpers import flash_message
 from app.shared.table_pagination import GovUKTableAndPagination
+from config import Config
 
 INDEX_BP_DASHBOARD = "index_bp.dashboard"
 ROUND_DETAILS = "round_bp.round_details"
@@ -197,12 +198,16 @@ def clone_round(round_id):
     return redirect(url_for(ROUND_DETAILS, round_id=round_id))
 
 
-@round_bp.route("/<round_id>")
+@round_bp.route("/<round_id>", methods=["GET", "DELETE"])
 def round_details(round_id):
     fund_round = get_round_by_id(round_id)
     form = RoundForm(data={"fund_id": fund_round.fund_id})
-    cloned_form = CloneRoundForm(data={"fund_id": fund_round.fund_id})
     fund_form = FundForm()
+    if request.method == "DELETE":
+        delete_selected_round(round_id)
+        return redirect(url_for("round_bp.view_all_rounds"))
+    cloned_form = CloneRoundForm(data={"fund_id": fund_round.fund_id})
+    # TODO at this time we are not implementing the delete applications but later we have to implement
     return render_template(
-        "round_details.html", form=form, fund_form=fund_form, round=fund_round, cloned_form=cloned_form
-    )
+        "round_details.html", form=form, fund_form=fund_form, round=fund_round,
+        cloned_form=cloned_form)

app/db/queries/fund.py

@@ -1,8 +1,12 @@
+from f86018f3ab453f90c0c3__mypyc import init_djlint___src
 from flask import current_app
 from sqlalchemy import String, cast, select
+from sqlalchemy.orm import joinedload
 
 from app.db import db
+from app.db.models import Round, Component, Page, Form, Lizt
 from app.db.models.fund import Fund, Organisation
+from app.db.queries.util import delete_all_related_objects
 
 
 def add_organisation(organisation: Organisation) -> Organisation:
@@ -38,3 +42,37 @@ def get_fund_by_id(id: str) -> Fund:
 
 def get_fund_by_short_name(short_name: str) -> Fund:
     return db.session.query(Fund).filter_by(short_name=short_name).first()
+
+
+def _delete_sections_for_fund_round(fund: Fund):
+    for round_detail in fund.rounds:
+        for section in round_detail.sections:
+            if section:
+                lizt_ids = [component.list_id for form in section.forms for page in form.pages for component in
+                            page.components]
+                page_ids = [page.page_id for form in section.forms for page in form.pages]
+                form_ids = [form.form_id for form in section.forms]
+                section_ids = [section.section_id]
+
+                delete_all_related_objects(db=db, model=Component, column=Component.page_id, ids=page_ids)
+                delete_all_related_objects(db=db, model=Lizt, column=Lizt.list_id, ids=lizt_ids)
+                delete_all_related_objects(db=db, model=Page, column=Page.form_id, ids=form_ids)
+                delete_all_related_objects(db=db, model=Form, column=Form.section_id, ids=section_ids)
+
+                db.session.delete(section)
+                db.session.commit()
+
+
+def delete_selected_fund(fund_id):
+    fund: Fund = db.session.get(Fund, fund_id, options=[joinedload(Fund.rounds).joinedload(Round.sections)])
+    if not fund:
+        raise ValueError(f"Fund with id {fund_id} not found")
+    try:
+        _delete_sections_for_fund_round(fund)
+        delete_all_related_objects(db=db, model=Round, column=Round.round_id,
+                                   ids=[round_detail.round_id for round_detail in fund.rounds])
+        delete_all_related_objects(db=db, model=Fund, column=Fund.fund_id, ids=[fund_id])
+        db.session.commit()
+    except Exception as e:
+        db.session.rollback()
+        print(f"Failed to delete fund {fund_id} : Error {e}")

app/db/queries/round.py

@@ -1,9 +1,11 @@
 from flask import current_app
 from sqlalchemy import String, cast, select
+from sqlalchemy.orm import joinedload
 
 from app.db import db
-from app.db.models import Fund
+from app.db.models import Fund, Component, Page, Form, Lizt
 from app.db.models.round import Round
+from app.db.queries.util import delete_all_related_objects
 
 
 def add_round(round: Round) -> Round:
@@ -33,3 +35,33 @@ def get_round_by_short_name_and_fund_id(fund_id: str, short_name: str) -> Round:
 def get_all_rounds() -> list[Round]:
     stmt = select(Round).join(Round.fund).order_by(cast(Fund.title_json["en"], String))
     return db.session.scalars(stmt).all()
+
+
+def _delete_sections_for_round(round_detail: Round):
+    for section_detail in round_detail.sections:
+        lizt_ids = [component.list_id for form in section_detail.forms for page in form.pages for component in
+                    page.components]
+        page_ids = [page.page_id for form in section_detail.forms for page in form.pages]
+        form_ids = [form.form_id for form in section_detail.forms]
+        section_ids = [section_detail.section_id]
+
+        delete_all_related_objects(db=db, model=Component, column=Component.page_id, ids=page_ids)
+        delete_all_related_objects(db=db, model=Lizt, column=Lizt.list_id, ids=lizt_ids)
+        delete_all_related_objects(db=db, model=Page, column=Page.form_id, ids=form_ids)
+        delete_all_related_objects(db=db, model=Form, column=Form.section_id, ids=section_ids)
+
+        db.session.delete(section_detail)
+        db.session.commit()
+
+
+def delete_selected_round(round_id):
+    round_detail: Round = db.session.get(Round, round_id, options=[joinedload(Round.sections)])
+    if not round_detail:
+        raise ValueError(f"Round with id {round_id} not found")
+    try:
+        _delete_sections_for_round(round_detail)
+        delete_all_related_objects(db=db, model=Round, column=Round.round_id, ids=[round_id])
+        db.session.commit()
+    except Exception as e:
+        db.session.rollback()
+        print(f"Failed to delete round {round_id} : Error {e}")

app/db/queries/util.py

@@ -0,0 +1,9 @@
+from sqlalchemy import delete
+
+
+def delete_all_related_objects(db, model, column, ids):
+    ## Delete objects for a given object type and based on the filter id and data
+    if ids:
+        stmt = delete(model).filter(column.in_(ids))
+        db.session.execute(stmt)
+        db.session.commit()

tests/blueprints/fund/test_routes.py

@@ -1,8 +1,9 @@
 import pytest
 from bs4 import BeautifulSoup
 from flask import g
+from sqlalchemy.orm import joinedload
 
-from app.db.models import Fund
+from app.db.models import Fund, Round, Section, Component, Lizt
 from app.db.models.fund import FundingType
 from app.db.queries.fund import get_fund_by_id
 from tests.helpers import submit_form
@@ -329,8 +330,8 @@ def test_view_fund_details(flask_test_client, seed_dynamic_data):
     html = response.data.decode("utf-8")
     assert f'<h1 class="govuk-heading-l">{test_fund.name_json["en"]}</h1>' in html
     assert (
-        f'<a class="govuk-link govuk-link--no-visited-state" href="/grants/{test_fund.fund_id}/edit#name_en">Change'
-        f'<span class="govuk-visually-hidden"> Grant name</span></a>' in html  # noqa: E501
+            f'<a class="govuk-link govuk-link--no-visited-state" href="/grants/{test_fund.fund_id}/edit#name_en">Change'
+            f'<span class="govuk-visually-hidden"> Grant name</span></a>' in html  # noqa: E501
     )
     assert '<a href="/grants/" class="govuk-back-link">Back</a>' in html
 
@@ -356,3 +357,30 @@ def test_create_fund_welsh_error_messages(flask_test_client, seed_dynamic_data):
     assert b"Enter the Welsh grant name" in response.data  # Validation error message
     assert b"Enter the Welsh application name" in response.data  # Validation error message
     assert b"Enter the Welsh grant description" in response.data  # Validation error message
+
+
+@pytest.mark.usefixtures("set_auth_cookie", "patch_validate_token_rs256_internal_user")
+def test_delete_fund_feature_enabled(_db, flask_test_client, seed_fund_without_assessment):
+    """Test that the delete endpoint redirects when a feature flag is enabled."""
+    test_fund: Fund = seed_fund_without_assessment["funds"][0]
+    flask_test_client.get(f"/grants/{test_fund.fund_id}")
+    with flask_test_client.session_transaction():
+        output: Fund = _db.session.get(Fund, test_fund.fund_id,
+                                       options=[joinedload(Fund.rounds).joinedload(Round.sections)])
+        assert output is not None, "No values present in the db"
+        response = flask_test_client.delete(f"/grants/{test_fund.fund_id}", data={
+            "csrf_token": g.csrf_token,
+        }, follow_redirects=True)
+        assert response.status_code == 200  # Assuming redirection to a valid page
+        _db.session.commit()
+        output_f = _db.session.get(Fund, test_fund.fund_id,
+                                   options=[joinedload(Fund.rounds).joinedload(Round.sections)])
+        assert output_f is None, "Grant delete did not happened"
+        output_r = _db.session.query(Round).all()
+        assert not output_r, "Round delete did not happened"
+        output_s = _db.session.query(Section).all()
+        assert not output_s, "Section delete did not happened"
+        output_c = _db.session.query(Component).all()
+        assert not output_c, "Component delete did not happened"
+        output_l = _db.session.query(Lizt).all()
+        assert not output_l, "Lizt delete did not happened"

tests/blueprints/round/test_routes.py

@@ -1,8 +1,9 @@
 import pytest
 from bs4 import BeautifulSoup
 from flask import g, url_for
+from sqlalchemy.orm import joinedload
 
-from app.db.models import Round
+from app.db.models import Round, Fund, Section, Component, Lizt
 from app.db.queries.round import get_round_by_id
 from tests.helpers import submit_form
 
@@ -330,3 +331,30 @@ def test_clone_round(flask_test_client, seed_dynamic_data):
         soup = BeautifulSoup(response.data, "html.parser")
         notification = soup.find("div", {"class": "govuk-notification-banner__content"})
         assert notification.text.strip() == "Error copying application"
+
+
+@pytest.mark.usefixtures("set_auth_cookie", "patch_validate_token_rs256_internal_user")
+def test_delete_fund_feature_enabled(_db, flask_test_client, seed_fund_without_assessment):
+    """Test that the delete endpoint redirects when a feature flag is enabled."""
+    test_round: Round = seed_fund_without_assessment["rounds"][0]
+    flask_test_client.get(f"/rounds/{test_round.round_id}")
+    with flask_test_client.session_transaction():
+        output: Fund = _db.session.get(Fund, test_round.fund_id,
+                                       options=[joinedload(Fund.rounds).joinedload(Round.sections)])
+        assert output is not None, "No values present in the db"
+        response = flask_test_client.delete(f"/rounds/{test_round.round_id}", data={
+            "csrf_token": g.csrf_token,
+        }, follow_redirects=True)
+        assert response.status_code == 200  # Assuming redirection to a valid page
+        _db.session.commit()
+        output_f = _db.session.get(Fund, test_round.fund_id,
+                                   options=[joinedload(Fund.rounds).joinedload(Round.sections)])
+        assert output_f is not None, "Grant deleted"
+        output_r = _db.session.query(Round).all()
+        assert not output_r, "Round delete did not happened"
+        output_s = _db.session.query(Section).all()
+        assert not output_s, "Section delete did not happened"
+        output_c = _db.session.query(Component).all()
+        assert not output_c, "Component delete did not happened"
+        output_l = _db.session.query(Lizt).all()
+        assert not output_l, "Lizt delete did not happened"

tests/conftest.py

@@ -11,7 +11,7 @@
 from app.create_app import create_app
 from app.import_config.load_form_json import load_form_jsons
 from config import Config
-from tests.seed_test_data import init_unit_test_data, insert_test_data
+from tests.seed_test_data import init_unit_test_data, insert_test_data, fund_without_assessment
 
 pytest_plugins = ["fsd_test_utils.fixtures.db_fixtures"]
 
@@ -47,6 +47,29 @@ def seed_dynamic_data(request, app, clear_test_data, _db, enable_preserve_test_d
     _db.session.commit()
 
 
+@pytest.fixture(scope="function")
+def seed_fund_without_assessment(request, app, clear_test_data, _db, enable_preserve_test_data):
+    marker = request.node.get_closest_marker("seed_config")
+
+    if marker is None:
+        fab_seed_data = fund_without_assessment()
+    else:
+        fab_seed_data = marker.args[0]
+    insert_test_data(db=_db, test_data=fab_seed_data)
+    yield fab_seed_data
+    # cleanup data after test
+    # rollback incase of any errors during test session
+    _db.session.rollback()
+    # disable foreign key checks
+    _db.session.execute(text("SET session_replication_role = replica"))
+    # delete all data from tables
+    for table in reversed(_db.metadata.sorted_tables):
+        _db.session.execute(table.delete())
+    # reset foreign key checks
+    _db.session.execute(text("SET session_replication_role = DEFAULT"))
+    _db.session.commit()
+
+
 @pytest.fixture(scope="function")
 def db_with_templates(app, _db):
     """Ensures a clean database but with templates already loaded"""