FS-4952 : delete endpoints for grants and applications

app/db/queries/fund.py

@@ -1,8 +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 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 +41,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/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"""

tests/seed_test_data.py

@@ -39,7 +39,7 @@
     "project_name_field_id": 1,
     "prospectus_link": "https://www.gov.uk/government/organisations/ministry-of-housing-communities-local-government",
     "privacy_notice_link": "https://www.gov.uk/government/organisations/"
-    "ministry-of-housing-communities-local-government",
+                           "ministry-of-housing-communities-local-government",
     "contact_email": "help@fab.gov.uk",
     "instructions_json": {},
     "feedback_link": "https://www.gov.uk/government/organisations/ministry-of-housing-communities-local-government",
@@ -71,7 +71,7 @@
 page_five_id = uuid4()
 alt_page_id = uuid4()
 
-
+#NOSONAR Ignore since this data is related to unit tests
 def init_salmon_fishing_fund():
     organisation_uuid = uuid4()
     o: Organisation = Organisation(
@@ -355,7 +355,7 @@ def init_salmon_fishing_fund():
         "organisations": [o],
     }
 
-
+#NOSONAR Ignore since this data is related to unit tests
 def init_unit_test_data() -> dict:
     organisation_uuid = uuid4()
     o: Organisation = Organisation(
@@ -481,6 +481,142 @@ def init_unit_test_data() -> dict:
         "themes": [t1],
     }
 
+#NOSONAR Ignore since this data is related to unit tests
+def fund_without_assessment() -> dict:
+    organisation_uuid = uuid4()
+    o: Organisation = Organisation(
+        organisation_id=organisation_uuid,
+        name=f"Ministry of Testing - {str(organisation_uuid)[:5]}",
+        short_name=f"MoT-{str(organisation_uuid)[:5]}",
+        logo_uri="https://www.google.com",
+        audit_info={"user": "dummy_user", "timestamp": datetime.now().isoformat(), "action": "create"},
+    )
+
+    f2: Fund = Fund(
+        fund_id=uuid4(),
+        name_json={"en": "Unit Test Fund 2"},
+        title_json={"en": "funding to improve testing"},
+        description_json={"en": "A £10m fund to improve testing across the devolved nations."},
+        welsh_available=False,
+        short_name=f"UTF{randint(0, 999)}",
+        owner_organisation_id=o.organisation_id,
+        funding_type=FundingType.COMPETITIVE,
+        ggis_scheme_reference_number="G3-SCH-0000092414",
+    )
+
+    f2_r1: Round = Round(
+        round_id=uuid4(),
+        fund_id=f2.fund_id,
+        audit_info={"user": "dummy_user", "timestamp": datetime.now().isoformat(), "action": "create"},
+        title_json={"en": "round the first"},
+        short_name=f"UTR{randint(0, 999)}",
+        opens=datetime.now(),
+        deadline=datetime.now(),
+        assessment_start=datetime.now(),
+        reminder_date=datetime.now(),
+        assessment_deadline=datetime.now(),
+        prospectus_link="https://www.google.com",
+        privacy_notice_link="https://www.google.com",
+        contact_email="test@test.com",
+        feedback_link="https://www.google.com",
+        project_name_field_id="12312312312",
+        guidance_url="https://www.google.com",
+        feedback_survey_config={
+            "has_feedback_survey": False,
+            "has_section_feedback": False,
+            "has_research_survey": False,
+            "is_feedback_survey_optional": False,
+            "is_section_feedback_optional": False,
+            "is_research_survey_optional": False,
+        },
+        eligibility_config={"has_eligibility": False},
+        eoi_decision_schema={"en": {"valid": True}, "cy": {"valid": False}},
+    )
+
+    f2_r1_s1: Section = Section(
+        section_id=uuid4(), index=1, round_id=f2_r1.round_id, name_in_apply_json={"en": "Organisation Information 2"}
+    )
+
+    f2_r1_s2: Section = Section(
+        section_id=uuid4(), index=1, round_id=f2_r1.round_id, name_in_apply_json={"en": "Organisation Information 3"}
+    )
+
+    f2_r1_s1_f1: Form = Form(
+        form_id=uuid4(),
+        section_id=f2_r1_s1.section_id,
+        name_in_apply_json={"en": "About your organisation"},
+        section_index=1,
+        runner_publish_name="about-your-org",
+        template_name="About your organization template",
+    )
+
+    f2_r1_s1_f2: Form = Form(
+        form_id=uuid4(),
+        section_id=f2_r1_s2.section_id,
+        name_in_apply_json={"en": "About your organisation 2"},
+        section_index=1,
+        runner_publish_name="about-your-org",
+        template_name="About your organization template",
+    )
+
+    f2_r1_s1_f1_p1: Page = Page(
+        page_id=uuid4(),
+        form_id=f2_r1_s1_f1.form_id,
+        display_path="organisation-name",
+        name_in_apply_json={"en": "Organisation Name"},
+        form_index=1,
+        default_next_page_id=None,
+    )
+
+    f2_r1_s1_f1_p2: Page = Page(
+        page_id=uuid4(),
+        form_id=f2_r1_s1_f2.form_id,
+        display_path="organisation-name",
+        name_in_apply_json={"en": "Organisation Name"},
+        form_index=1,
+        default_next_page_id=None,
+    )
+
+    f2_r1_s1_f1_p1_c1: Component = Component(
+        component_id=uuid4(),
+        page_id=f2_r1_s1_f1_p1.page_id,
+        title="What is your organisation's name?",
+        hint_text="This must match the registered legal organisation name",
+        type=ComponentType.TEXT_FIELD,
+        page_index=1,
+        options={"hideTitle": False, "classes": ""},
+        runner_component_name="organisation_name",
+    )
+
+    l1: Lizt = Lizt(
+        list_id=uuid4(),
+        name="classifications_list",
+        type="string",
+        items=[{"text": "Charity", "value": "charity"}, {"text": "Public Limited Company", "value": "plc"}],
+        is_template=True,
+    )
+
+    f2_r1_s1_f1_p1_c2_with_list: Component = Component(
+        component_id=uuid4(),
+        page_id=f2_r1_s1_f1_p2.page_id,
+        title="How is your organisation classified?",
+        type=ComponentType.RADIOS_FIELD,
+        page_index=2,
+        options={"hideTitle": False, "classes": ""},
+        runner_component_name="organisation_classification",
+        list_id=l1.list_id,
+    )
+    return {
+        "lists": [l1],
+        "funds": [f2],
+        "organisations": [o],
+        "rounds": [f2_r1],
+        "sections": [f2_r1_s1, f2_r1_s2],
+        "forms": [f2_r1_s1_f1, f2_r1_s1_f2],
+        "pages": [f2_r1_s1_f1_p1, f2_r1_s1_f1_p2],
+        "components": [f2_r1_s1_f1_p1_c1, f2_r1_s1_f1_p1_c2_with_list]
+    }
+
 
 def add_default_page_paths(db, default_next_page_config):
     # set up the default paths

tests/test_db.py

@@ -4,8 +4,9 @@
 
 import pytest
 
-from app.db.models import Form, Fund, Organisation, Round, Section
+from app.db.models import Form, Fund, Organisation, Round, Section, Component, Lizt
 from app.db.models.fund import FundingType
+from sqlalchemy.orm import joinedload
 from app.db.queries.application import (
     delete_form_from_section,
     delete_section_from_round,
@@ -16,8 +17,8 @@
     move_section_up,
     swap_elements_in_list,
 )
-from app.db.queries.fund import add_fund, add_organisation, get_all_funds, get_fund_by_id
-from app.db.queries.round import add_round, get_round_by_id
+from app.db.queries.fund import add_fund, add_organisation, get_all_funds, get_fund_by_id, delete_selected_fund
+from app.db.queries.round import add_round, get_round_by_id, delete_selected_round
 from tests.seed_test_data import BASIC_FUND_INFO, BASIC_ROUND_INFO
 
 
@@ -568,3 +569,45 @@ def test_base_path_sequence_insert(seed_dynamic_data, _db):
     added_round_2 = add_round(new_round_2)
     assert added_round_2.section_base_path
     assert added_round_2.section_base_path > added_round_1.section_base_path
+
+
+def test_delete_grant(_db, seed_fund_without_assessment):
+    """Test that the delete endpoint redirects to grant table page"""
+    test_fund: Fund = seed_fund_without_assessment["funds"][0]
+    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"
+    delete_selected_fund(test_fund.fund_id)
+    _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"
+
+
+def test_delete_application(_db, seed_fund_without_assessment):
+    """Test that the delete endpoint redirects application table page"""
+    test_round: Round = seed_fund_without_assessment["rounds"][0]
+    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"
+    delete_selected_round(test_round.round_id)
+    _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"