"""M&E framework — Results Framework / Results Hierarchy. Covers the hierarchy shape the framework prescribes: Strategic Objective (the LogFrame container) → Impact (chain root, no parent) → Intermediate Outcome → Output (University / Secretariat) plus the draft → submit → approve workflow, multiple impacts, and the E1/E2 orphan exception handling (no Intermediate Outcome without an Impact, no Output without an Intermediate Outcome). Do NOT run under a parallel pytest — the test_iilmp DB collides. """ from __future__ import annotations import pytest from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError from django.urls import reverse from apps.core.audit import tasks as audit_tasks from apps.core.audit.models import AuditLog from apps.mel.indicators.forms import LogFrameRowForm from apps.mel.indicators.models import ( LogFrame, LogFrameApproval, LogFrameLevel, LogFrameRow, OutputType, ) from apps.mel.indicators.services import ( ResultsFrameworkIncomplete, approve_logframe, return_logframe_to_draft, submission_blockers, submit_logframe_for_approval, ) User = get_user_model() pytestmark = pytest.mark.django_db @pytest.fixture def officer(): return User.objects.create_user( email="rf-officer@example.com", password="x", role="mel_officer", ) @pytest.fixture def manager(): return User.objects.create_user( email="rf-manager@example.com", password="x", role="program_manager", ) @pytest.fixture def sync_audit(monkeypatch): """Run write_audit_log.delay() synchronously — tests aren't CELERY eager.""" def _eager_delay(**kwargs): return audit_tasks.persist_audit_log(**kwargs) monkeypatch.setattr(audit_tasks.write_audit_log, "delay", _eager_delay) def _full_framework(slug="rf-full", *, code=""): """A complete framework: the Strategic Objective (LogFrame) with a full Impact → Intermediate Outcome → University Output chain beneath it.""" lf = LogFrame.objects.create(name=f"RF {slug}", slug=slug, code=code) impact = LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.IMPACT, title="Impact 1", ) outcome = LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.OUTCOME, title="Intermediate Outcome 1", parent=impact, ) output = LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.OUTPUT, title="University Output 1", parent=outcome, output_type=OutputType.UNIVERSITY, ) return lf, impact, outcome, output # --------------------------------------------------------------------------- # Hierarchy shape — Impact is the chain root # --------------------------------------------------------------------------- def test_impact_is_the_root_with_no_parent(): lf, impact, *_ = _full_framework() impact.full_clean() # no parent — the Strategic Objective is the container def test_impact_cannot_have_a_parent(): lf, impact, outcome, output = _full_framework(slug="rf-impact-parent") rogue = LogFrameRow( logframe=lf, level=LogFrameLevel.IMPACT, title="Rogue impact", parent=impact, ) with pytest.raises(ValidationError): rogue.full_clean() def test_multiple_impacts_allowed_under_one_strategic_objective(): """A Strategic Objective may have one or many Impacts.""" lf, impact1, *_ = _full_framework() impact2 = LogFrameRow(logframe=lf, level=LogFrameLevel.IMPACT, title="Impact 2") impact2.full_clean() impact2.save() assert lf.rows.filter(level=LogFrameLevel.IMPACT).count() == 2 def test_intermediate_outcome_requires_impact_parent(): lf = LogFrame.objects.create(name="LF", slug="rf-orphan-oc") orphan = LogFrameRow(logframe=lf, level=LogFrameLevel.OUTCOME, title="Orphan outcome") with pytest.raises(ValidationError): orphan.full_clean() def test_output_requires_outcome_parent(): lf = LogFrame.objects.create(name="LF", slug="rf-orphan-out") orphan = LogFrameRow(logframe=lf, level=LogFrameLevel.OUTPUT, title="Orphan output") with pytest.raises(ValidationError): orphan.full_clean() # --------------------------------------------------------------------------- # University / Secretariat output typing # --------------------------------------------------------------------------- def test_output_type_only_on_output_rows(): lf, impact, outcome, output = _full_framework() outcome.output_type = OutputType.UNIVERSITY with pytest.raises(ValidationError): outcome.full_clean() def test_form_requires_output_type_for_output_rows(): lf, impact, outcome, output = _full_framework(slug="rf-form-out") form = LogFrameRowForm( data={ "logframe": lf.pk, "parent": outcome.pk, "level": LogFrameLevel.OUTPUT, "output_type": "", "title": "New output", "order": 0, }, logframe_scope=lf, ) assert not form.is_valid() assert "output_type" in form.errors def test_form_accepts_secretariat_output(): lf, impact, outcome, output = _full_framework(slug="rf-form-sec") form = LogFrameRowForm( data={ "logframe": lf.pk, "parent": outcome.pk, "level": LogFrameLevel.OUTPUT, "output_type": OutputType.SECRETARIAT, "title": "Secretariat output", "order": 0, }, logframe_scope=lf, ) assert form.is_valid(), form.errors row = form.save() assert row.is_secretariat_output # --------------------------------------------------------------------------- # Draft → submit → approve workflow # --------------------------------------------------------------------------- def test_incomplete_framework_blocks_submission(): """A Strategic Objective with only an Impact is missing its Outcome/Output.""" lf = LogFrame.objects.create(name="Incomplete", slug="rf-incomplete") LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.IMPACT, title="Impact only", ) blockers = submission_blockers(lf) assert blockers # missing outcome/output with pytest.raises(ResultsFrameworkIncomplete): submit_logframe_for_approval(lf, submitted_by=None) def test_complete_framework_submits_and_approves(officer, manager): lf, *_ = _full_framework(slug="rf-approve") assert lf.approval_status == LogFrameApproval.DRAFT assert submission_blockers(lf) == [] submit_logframe_for_approval(lf, submitted_by=officer) lf.refresh_from_db() assert lf.approval_status == LogFrameApproval.SUBMITTED assert lf.submitted_by_id == officer.pk approve_logframe(lf, approved_by=manager) lf.refresh_from_db() assert lf.approval_status == LogFrameApproval.APPROVED assert lf.approved_by_id == manager.pk def test_return_to_draft(officer): lf, *_ = _full_framework(slug="rf-return") submit_logframe_for_approval(lf, submitted_by=officer) return_logframe_to_draft(lf, actor=officer) lf.refresh_from_db() assert lf.approval_status == LogFrameApproval.DRAFT assert lf.submitted_by is None def test_submit_is_idempotent(officer): lf, *_ = _full_framework(slug="rf-idem") submit_logframe_for_approval(lf, submitted_by=officer) # Second call is a no-op, not an error. submit_logframe_for_approval(lf, submitted_by=officer) lf.refresh_from_db() assert lf.approval_status == LogFrameApproval.SUBMITTED # --------------------------------------------------------------------------- # Strategic Objective code uniqueness # --------------------------------------------------------------------------- def test_strategic_objective_code_unique_when_set(): _full_framework(slug="rf-code-a", code="SO-2024-01") dup = LogFrame(name="Dup", slug="rf-code-b", code="SO-2024-01") with pytest.raises(ValidationError): dup.full_clean() dup.validate_constraints() def test_blank_codes_do_not_collide(): LogFrame.objects.create(name="A", slug="rf-blank-a", code="") lf_b = LogFrame(name="B", slug="rf-blank-b", code="") lf_b.full_clean() # blank codes are exempt from the unique constraint # --------------------------------------------------------------------------- # Views — row edit + workflow + audit # --------------------------------------------------------------------------- def test_row_edit_view_updates_and_audits( client, officer, sync_audit, django_capture_on_commit_callbacks ): lf, impact, outcome, output = _full_framework(slug="rf-edit") client.force_login(officer) url = reverse("mel_indicators:logframe_row_edit", kwargs={"pk": output.pk}) # log_audit defers via transaction.on_commit → capture + execute so the # audit row is actually written inside the test transaction. with django_capture_on_commit_callbacks(execute=True): resp = client.post( url, { "logframe": lf.pk, "parent": outcome.pk, "level": LogFrameLevel.OUTPUT, "output_type": OutputType.SECRETARIAT, "title": "Renamed output", "order": 0, }, ) assert resp.status_code == 302 output.refresh_from_db() assert output.title == "Renamed output" assert output.output_type == OutputType.SECRETARIAT assert AuditLog.objects.filter( target_model="LogFrameRow", object_id=str(output.pk) ).exists() def test_submit_view_blocks_incomplete(client, officer): lf = LogFrame.objects.create(name="Incomplete", slug="rf-view-incomplete") LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.IMPACT, title="Impact only", ) client.force_login(officer) url = reverse("mel_indicators:logframe_submit", kwargs={"slug": lf.slug}) resp = client.post(url, follow=True) lf.refresh_from_db() assert lf.approval_status == LogFrameApproval.DRAFT # still draft — blocked # --------------------------------------------------------------------------- # Contextual authoring + strict chain # --------------------------------------------------------------------------- def test_submission_requires_full_impact_outcome_output_chain(): """A complete framework needs Impact → Intermediate Outcome → Output.""" lf = LogFrame.objects.create(name="Partial", slug="rf-partial-chain") impact = LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.IMPACT, title="I", ) LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.OUTCOME, title="O", parent=impact, ) blockers = submission_blockers(lf) assert any("Output" in b for b in blockers) def test_form_rejects_output_directly_under_impact(): """Strict chain: an Output sits under an Intermediate Outcome, not an Impact.""" lf, impact, outcome, output = _full_framework(slug="rf-strict-chain") form = LogFrameRowForm( data={ "logframe": lf.pk, "parent": impact.pk, "level": LogFrameLevel.OUTPUT, "output_type": OutputType.UNIVERSITY, "title": "Skip-level output", "order": 0, }, logframe_scope=lf, ) assert not form.is_valid() assert "parent" in form.errors def test_logframe_form_requires_code_and_planning_period(): """'Code, Title, and Planning Period are mandatory' for a Strategic Objective.""" from apps.mel.indicators.forms import LogFrameForm form = LogFrameForm( data={"name": "No code", "slug": "rf-no-code", "active": "on"} ) assert not form.is_valid() assert "code" in form.errors assert "planning_period" in form.errors def test_row_create_prefills_parent_level_and_output_type(client, officer): """Contextual add-child buttons hand over parent + level + output type.""" lf, impact, outcome, output = _full_framework(slug="rf-prefill") client.force_login(officer) url = ( reverse("mel_indicators:logframe_row_create") + f"?parent={outcome.pk}&level=output&output_type=secretariat" ) resp = client.get(url) assert resp.status_code == 200 form = resp.context["form"] assert form.initial["parent"] == outcome.pk assert form.initial["level"] == "output" assert form.initial["output_type"] == "secretariat" # The parent dropdown is scoped to the parent's own framework. assert set(form.fields["parent"].queryset.values_list("logframe_id", flat=True)) == {lf.pk} def test_detail_renders_children_directly_under_their_parent(client, officer): """Hierarchy view groups each Outcome under its own Impact (no interleaving).""" lf, impact1, outcome1, output1 = _full_framework(slug="rf-tree") impact2 = LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.IMPACT, title="Impact 2", ) outcome2 = LogFrameRow.objects.create( logframe=lf, level=LogFrameLevel.OUTCOME, title="Outcome 2", parent=impact2, ) client.force_login(officer) resp = client.get( reverse("mel_indicators:logframe_detail", kwargs={"slug": lf.slug}) ) assert resp.status_code == 200 rows = resp.context["rows"] pks = [r.pk for r in rows] # outcome1 must come after impact1 and BEFORE impact2 (no interleaving). assert pks.index(impact1.pk) < pks.index(outcome1.pk) < pks.index(impact2.pk) < pks.index(outcome2.pk) depths = {r.pk: r.tree_depth for r in rows} # Impact is the top of the chain (depth 0) — the Strategic Objective is the # framework header, not a row. assert depths[impact1.pk] == 0 assert depths[outcome1.pk] == 1 assert depths[output1.pk] == 2 body = resp.content.decode() assert "+ Add Impact" in body # framework-header root action assert "+ Add outcome" in body assert "+ University output" in body assert "+ Secretariat output" in body assert "+ Add indicator" in body # --------------------------------------------------------------------------- # Edit Strategic Objective header # --------------------------------------------------------------------------- def test_logframe_edit_view_updates_and_audits( client, officer, sync_audit, django_capture_on_commit_callbacks ): lf, *_ = _full_framework("rf-hdr-edit", code="SO-EDIT") client.force_login(officer) with django_capture_on_commit_callbacks(execute=True): resp = client.post( reverse("mel_indicators:logframe_edit", kwargs={"slug": lf.slug}), { "name": lf.name, "code": "SO-EDIT-2", "slug": lf.slug, "description": "updated", "programme": "", "planning_period": "2025–2030", "active": "on", }, ) assert resp.status_code in (302, 303) lf.refresh_from_db() assert lf.code == "SO-EDIT-2" assert lf.planning_period == "2025–2030" assert AuditLog.objects.filter( target_model="LogFrame", action=AuditLog.Action.UPDATE, object_id=str(lf.pk), ).exists() def test_logframe_create_view_writes_audit( client, officer, sync_audit, django_capture_on_commit_callbacks ): client.force_login(officer) with django_capture_on_commit_callbacks(execute=True): resp = client.post( reverse("mel_indicators:logframe_create"), { "name": "Brand New Strategic Objective", "code": "SO-NEW", "slug": "rf-brand-new", "description": "", "programme": "", "planning_period": "2026–2030", "active": "on", }, ) assert resp.status_code in (302, 303) lf = LogFrame.objects.get(slug="rf-brand-new") assert AuditLog.objects.filter( target_model="LogFrame", action=AuditLog.Action.CREATE, object_id=str(lf.pk), ).exists() # --------------------------------------------------------------------------- # Model backstop — strict one-level adjacency # --------------------------------------------------------------------------- def test_output_cannot_skip_directly_under_impact(): lf, impact, outcome, output = _full_framework("rf-skip") bad = LogFrameRow( logframe=lf, level=LogFrameLevel.OUTPUT, title="Skip output", parent=impact, output_type=OutputType.UNIVERSITY, ) with pytest.raises(ValidationError): bad.full_clean() # --------------------------------------------------------------------------- # Indicator name uniqueness within an output # --------------------------------------------------------------------------- def test_indicator_name_unique_within_output_case_insensitive(): from django.db import IntegrityError, transaction from apps.mel.indicators.models import Indicator, IndicatorFrequency lf, impact, outcome, output = _full_framework("rf-indname") Indicator.objects.create( logframe_row=output, code="in-1", name="Farmers trained", unit="x", calculation_method="c", data_source="m", frequency=IndicatorFrequency.MONTHLY, ) with pytest.raises(IntegrityError): with transaction.atomic(): Indicator.objects.create( logframe_row=output, code="in-2", name="farmers trained", unit="x", calculation_method="c", data_source="m", frequency=IndicatorFrequency.MONTHLY, ) # --------------------------------------------------------------------------- # Audit records are immutable # --------------------------------------------------------------------------- def test_auditlog_records_are_immutable(): row = AuditLog.objects.create( action=AuditLog.Action.CREATE, target_model="LogFrame", object_id="1", ) row.object_repr = "tampered" with pytest.raises(ValidationError): row.save() with pytest.raises(ValidationError): row.delete()