opportunity assesment

Author: leoyliao001

README.md

@@ -4,6 +4,16 @@ A full-stack Migration Portal demo with a separated frontend and backend: Vue 3
 
 ---
 
+## Opportunity Assessment task scoping
+
+The assessment page reuses `refer doc/WPM_Pulse_OA_Task_Scoping.html`, including its complete 1,670-task catalogue and both scoping methods. Keep that file in the deployed repository: Vite bundles the UI, and the backend reads its catalogue to validate immutable task taxonomy. Restart the backend and rebuild the frontend together when changing the template.
+
+Apply migration `0056_opportunity_assessment_scope` before deploying the new page. `opportunity_assessment_scope` stores both method datasets, the active method and a revision per project. Saving also synchronizes the active method into `opportunity_assessment`; the other method remains in the project snapshot to avoid double counting in downstream consumers. New row fields, including country, migration decision and exception, are retained in `scope_details`. Existing task rows and their legacy details are retained on first load.
+
+Save and Reload use the authenticated project `opportunity-assessment/task-scope/` API, not browser local storage. Access remains restricted to the intake requestor and Super Admin. Stale revisions are rejected instead of silently overwriting another edit. CSV import/export uses the supplied template format and locks uploads to the project's migration request ID.
+
+FTE follows the supplied template: `volume * monthly frequency * minutes per unit / 60 / 140`. The 140 productive hours per month is the template's current assumption; confirm it before adopting a different organization-wide FTE standard.
+
 ## Prerequisites (Windows)
 
 This guide targets **Windows** with **Command Prompt (CMD)**. Install the following before you start:

backend/api/migrations/0056_opportunity_assessment_scope.py

@@ -0,0 +1,25 @@
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+    dependencies = [("api", "0055_business_impact_analysis")]
+
+    operations = [
+        migrations.AddField(
+            model_name="opportunityassessment", name="scope_details",
+            field=models.JSONField(blank=True, default=dict),
+        ),
+        migrations.CreateModel(
+            name="OpportunityAssessmentScope",
+            fields=[
+                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("state", models.JSONField(default=dict)),
+                ("revision", models.PositiveIntegerField(default=0)),
+                ("updated_by", models.CharField(blank=True, default="", max_length=254)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("project", models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name="task_scope", to="api.migrationintakesubmission")),
+            ],
+            options={"db_table": "opportunity_assessment_scope"},
+        ),
+    ]

backend/api/models.py

@@ -371,6 +371,7 @@ class OpportunityAssessment(models.Model):
     task_found_in_service_catalog = models.CharField(max_length=255, blank=True, default="")
     migratable_to_gsc = models.CharField(max_length=255, blank=True, default="")
     fte_calculation = models.CharField(max_length=64, blank=True, default="")
+    scope_details = models.JSONField(default=dict, blank=True)
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 
@@ -385,6 +386,17 @@ def __str__(self):
         return f"{self.migration_request_id} — {self.task_name or self.l4 or self.id}"
 
 
+class OpportunityAssessmentScope(models.Model):
+    project = models.OneToOneField(MigrationIntakeSubmission, on_delete=models.CASCADE, related_name="task_scope")
+    state = models.JSONField(default=dict)
+    revision = models.PositiveIntegerField(default=0)
+    updated_by = models.CharField(max_length=254, blank=True, default="")
+    updated_at = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        db_table = "opportunity_assessment_scope"
+
+
 class ApprovalInput(models.Model):
     """Activity Function / Product approval stakeholder email mapping."""
 

backend/api/routes/migration_dashboard.py

@@ -1,4 +1,5 @@
 from django.urls import path
+from api.views.opportunity_task_scope import opportunity_task_scope
 from api.views.business_impact_analysis import business_impact_analysis
 from api.views.project_artefacts import project_artefact, resource_projects
 from api.views.business_case import (
@@ -42,6 +43,7 @@
 )
 
 urlpatterns = [
+    path("projects/<int:project_id>/opportunity-assessment/task-scope/", opportunity_task_scope, name="opportunity-task-scope"),
     path("projects/<int:project_id>/business-impact-analysis/", business_impact_analysis, name="business-impact-analysis"),
     path("resources/", resource_projects, name="resource-projects"),
     path("projects/<int:project_id>/resources/", project_artefact, {"kind": "resources"}, name="project-resources"),

backend/api/test_opportunity_task_scope.py

@@ -0,0 +1,116 @@
+from django.test import TestCase
+from rest_framework.test import APIClient
+
+from api.models import MigrationIntakeSubmission, OpportunityAssessment, OpportunityAssessmentScope
+from api.views.opportunity_task_scope import _catalogue
+
+
+class OpportunityTaskScopeTests(TestCase):
+    def setUp(self):
+        self.project = MigrationIntakeSubmission.objects.create(
+            migration_request_id="OA-SCOPE-TEST", requested_date="2026-11-02", requestor="owner@example.com",
+            project_name="Scope test", migration_type="1:1 Transfer", migration_type_value="1:1-transfer",
+            region="APA", areas=["SEA"], countries=["Singapore", "Malaysia"],
+            function_name="CX", products=["LCL"], proposed_scope="Test", fte_number="1",
+        )
+        self.client = APIClient()
+        self.client.credentials(HTTP_X_USER_EMAIL=self.project.requestor)
+        self.url = f"/api/migration-dashboard/projects/{self.project.pk}/opportunity-assessment/task-scope/"
+
+    def row(self, ref=0, row_id="t1"):
+        item = _catalogue()[ref]
+        return {"id": row_id, "ref": ref, "ctry": "Singapore; Malaysia", "loc": "Manila",
+                "p": item[0], "l1": item[1], "l2": item[2], "l3": item[3], "l4": item[4],
+                "task": "Task", "desc": "Details", "cx": "High", "sop": "Y", "freq": "Daily",
+                "uom": "Cases", "vol": "400", "tpu": "2", "mgsc": "Yes", "exc": "Language Barrier"}
+
+    def save(self, state, revision=0, **extra):
+        return self.client.post(self.url, {"migration_request_id": self.project.migration_request_id,
+                                         "state": state, "revision": revision, **extra}, format="json")
+
+    def test_round_trip_both_modes_and_sync_active_rows(self):
+        state = {"cat": [self.row(0, "c0")], "task": [self.row(1)], "seq": 2, "mode": "task"}
+        response = self.save(state)
+        self.assertEqual(response.status_code, 200, response.data)
+        row = OpportunityAssessment.objects.get()
+        self.assertEqual(row.fte_calculation, "2.0000")
+        self.assertEqual(row.scope_details["exc"], "Language Barrier")
+        self.assertEqual(row.scope_details["ctry"], "Singapore; Malaysia")
+        loaded = self.client.get(self.url).data
+        self.assertEqual(loaded["state"]["task"][0]["desc"], "Details")
+        self.assertEqual(len(loaded["state"]["cat"]), 1)
+        loaded["state"]["mode"] = "cat"
+        response = self.save(loaded["state"], revision=1)
+        self.assertEqual(response.status_code, 200, response.data)
+        self.assertEqual(OpportunityAssessment.objects.count(), 1)
+        self.assertEqual(OpportunityAssessment.objects.get().scope_details["method"], "cat")
+        self.assertEqual(len(response.data["state"]["task"]), 1)
+
+    def test_invalid_upload_is_atomic_and_wrong_project_is_rejected(self):
+        state = {"cat": None, "task": [self.row()], "seq": 2, "mode": "task"}
+        response = self.save(state, migration_request_id="ANOTHER-PROJECT")
+        self.assertEqual(response.status_code, 400)
+        state["task"][0]["vol"] = "NaN"
+        self.assertEqual(self.save(state).status_code, 400)
+        self.assertFalse(OpportunityAssessmentScope.objects.exists())
+        self.assertFalse(OpportunityAssessment.objects.exists())
+
+    def test_complete_catalogue_can_be_saved_without_truncation(self):
+        rows = [self.row(index, f"c{index}") for index in range(len(_catalogue()))]
+        state = {"cat": rows, "task": [], "seq": 1, "mode": "cat"}
+        response = self.save(state)
+        self.assertEqual(response.status_code, 200, response.data)
+        self.assertEqual(OpportunityAssessment.objects.count(), len(_catalogue()))
+        self.assertEqual(self.save(response.data["state"], revision=1).status_code, 200)
+        self.assertEqual(OpportunityAssessment.objects.count(), len(_catalogue()))
+
+    def test_intake_product_alias_preserves_display_name(self):
+        self.project.products = ["General LCL"]
+        self.project.save()
+        intake = self.client.get(self.url).data["intake"]
+        self.assertEqual(intake["product"], "General LCL")
+        self.assertEqual(intake["catalogue_products"], ["LCL"])
+
+    def test_country_and_duplicate_validation(self):
+        state = {"cat": None, "task": [self.row()], "seq": 2, "mode": "task"}
+        state["task"][0]["ctry"] = "Denmark"
+        self.assertEqual(self.save(state).status_code, 400)
+        state["task"] = [self.row(), self.row()]
+        self.assertEqual(self.save(state).status_code, 400)
+
+    def test_stale_save_cannot_overwrite_and_delete_is_persisted(self):
+        state = {"cat": None, "task": [self.row()], "seq": 2, "mode": "task"}
+        self.assertEqual(self.save(state).status_code, 200)
+        state["task"] = []
+        self.assertEqual(self.save(state).status_code, 409)
+        self.assertEqual(OpportunityAssessment.objects.count(), 1)
+        self.assertEqual(self.save(state, revision=1).status_code, 200)
+        self.assertEqual(OpportunityAssessment.objects.count(), 0)
+        self.assertEqual(self.client.get(self.url).data["state"]["task"], [])
+
+    def test_legacy_rows_are_preserved_on_first_load(self):
+        OpportunityAssessment.objects.create(migration_request_id=self.project.migration_request_id,
+                                             product="Legacy", task_name="Existing task", l3="Legacy L3", upstream="Original upstream", sop_iop_exists="Yes")
+        response = self.client.get(self.url)
+        self.assertEqual(response.data["state"]["mode"], "task")
+        self.assertEqual(response.data["state"]["task"][0]["sop"], "Y")
+        state = response.data["state"]
+        state["task"][0]["exc"] = "Legal Issues"
+        self.assertEqual(self.save(state).status_code, 200)
+        self.assertEqual(OpportunityAssessment.objects.get().task_name, "Existing task")
+        state["cat"] = [self.row(0, "c0")]
+        state["mode"] = "cat"
+        self.assertEqual(self.save(state, revision=1).status_code, 200)
+        state["mode"] = "task"
+        self.assertEqual(self.save(state, revision=2).status_code, 200)
+        self.assertEqual(OpportunityAssessment.objects.get().upstream, "Original upstream")
+
+    def test_access_control_and_project_isolation(self):
+        self.client.credentials(HTTP_X_USER_EMAIL="other@example.com")
+        self.assertEqual(self.client.get(self.url).status_code, 403)
+        self.assertEqual(self.save({}).status_code, 403)
+        self.client.credentials()
+        self.assertEqual(self.client.get(self.url).status_code, 401)
+        self.client.credentials(HTTP_X_USER_EMAIL=self.project.requestor)
+        OpportunityAssessment.objects.create(migration_request_id="OTHER", task_name="Other task")
+        self.assertEqual(self.client.get(self.url).data["state"]["task"], [])

backend/api/views/opportunity_assessment.py

@@ -60,7 +60,7 @@
     "migratable_to_gsc": "Migratable to GSC as per service catalogue",
     "fte_calculation": "FTE Calculation",
 }
-OA_SKIP_FIELDS = frozenset({"id", "created_at", "updated_at"})
+OA_SKIP_FIELDS = frozenset({"id", "created_at", "updated_at", "scope_details"})
 
 
 def _all_fields():