Fix creation of proposal in rollback cases.

.github/workflows/test/action.yaml

@@ -7,8 +7,8 @@ runs:
     - name: "Prepare"
       id: prep
       run: |
-        out=$(make venv/bin venv/bin/mypy venv/bin/ruff venv/bin/pytest venv/lib/*/site-packages/mock 2>&1)
-        ret=$?
+        ret=0
+        out=$(make venv/bin venv/bin/mypy venv/bin/ruff venv/bin/pytest venv/lib/*/site-packages/mock 2>&1) || ret=$?
         if [ $ret != 0 ]
         then
           echo "$out" >&2

.vscode/settings.json

@@ -1,5 +1,4 @@
 {
-    "python.linting.mypyEnabled": true,
     "editor.formatOnSave": true,
     "editor.codeActionsOnSave": {
         "source.organizeImports": "never",
@@ -17,11 +16,5 @@
         "-p",
         "test_*.py"
     ],
-    "python.analysis.diagnosticMode": "workspace",
-    "python.linting.mypyArgs": [
-        "--config=mypy.ini",
-        "--show-column-numbers",
-        "--no-pretty"
-    ],
-    "python.linting.mypyPath": "venv/bin/mypy"
+    "python.analysis.diagnosticMode": "workspace"
 }

bin/airflow

@@ -7,7 +7,7 @@ export PYTHONPATH=$(dirname "$SCRIPT_DIR")/shared
 
 if [ "$1" == "setup" ]
 then
-    python3 -m venv "$VENV_DIR"
+    python -m venv "$VENV_DIR"
     "$VENV_DIR"/bin/pip3 install "apache-airflow[celery]==2.9.1" \
         apache-airflow-providers-slack[common.sql] \
         apache-airflow-providers-google \
@@ -20,8 +20,6 @@ then
     pushd "$AIRFLOW_HOME"
     ln -sfT ../dags dags
     ln -sfT ../plugins plugins
-    sed -i 's/allowed_deserialization_classes_regex.*/allowed_deserialization_classes_regex = (airflow|dfinity)[.].*/' airflow.cfg
-    sed -i 's/^allowed_deserialization_classes /# allowed_deserialization_classes.../' airflow.cfg
     sed -i 's/reload_on_plugin_change.*/reload_on_plugin_change = True/' airflow.cfg
     sed -i 's/load_examples.*/load_examples = False/' airflow.cfg
     popd
@@ -52,5 +50,6 @@ then
     exit
 fi
 
+export AIRFLOW__CORE__ALLOWED_DESERIALIZATION_CLASSES_REGEXP="(dfinity|airflow).*"
 export PATH="$VENV_DIR/bin:$PATH"
 exec "$VENV_DIR"/bin/airflow "$@"

mypy.ini

@@ -4,3 +4,6 @@ exclude = plugins/dags
 ignore_missing_imports = True
 follow_imports = silent
 strict = True
+
+[mypy-bottle]
+follow_imports = skip

plugins/operators/ic_os_rollout.py

@@ -99,19 +99,17 @@ def execute(self, context: Context) -> dict[str, int | str | bool]:
             self.subnet_id, self.git_revision
         )
         runner = dre.DRE(network=self.network, subprocess_hook=SubprocessHook())
-        # https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/logging-tasks.html#grouping-of-log-lines
-        props = runner.get_ic_os_version_deployment_proposals_for_subnet_and_revision(
-            subnet_id=subnet_id,
-            git_revision=git_revision,
-        )
 
-        def per_status(
-            props: list[ic_types.AbbrevProposal], status: ic_types.ProposalStatus
-        ) -> list[ic_types.AbbrevProposal]:
-            return [p for p in props if p["status"] == status]
-
-        executeds = per_status(props, ic_types.ProposalStatus.PROPOSAL_STATUS_EXECUTED)
-        opens = per_status(props, ic_types.ProposalStatus.PROPOSAL_STATUS_OPEN)
+        # Get proposals sorted by proposal number.
+        props = sorted(
+            runner.get_ic_os_version_deployment_proposals_for_subnet(
+                subnet_id=subnet_id,
+            ),
+            key=lambda prop: -prop["proposal_id"],
+        )
+        props_for_git_revision = [
+            p for p in props if p["payload"]["replica_version_id"] == git_revision
+        ]
 
         if self.simulate_proposal:
             self.log.info(f"simulate_proposal={self.simulate_proposal}")
@@ -134,41 +132,52 @@ def per_status(
         except IndexError:
             raise RuntimeError(f"No replicas have been found with subnet {subnet_id}")
 
-        if executeds:
-            url = f"{self.network.proposal_display_url}/{executeds[0]['proposal_id']}"
+        if not props:
+            self.log.info("No proposals for subnet.  Will create.")
+        elif not props_for_git_revision:
             self.log.info(
-                "Proposal " + url + f" titled {executeds[0]['title']}"
-                f" has executed.  No need to do anything."
+                "No proposals with revision %s for subnet.  Will create.", git_revision
             )
-            return {
-                "proposal_id": int(executeds[0]["proposal_id"]),
-                "proposal_url": url,
-                "needs_vote": False,
-            }
-
-        if opens:
-            url = f"{self.network.proposal_display_url}/{opens[0]['proposal_id']}"
+        elif props_for_git_revision[0]["proposal_id"] < props[0]["proposal_id"]:
             self.log.info(
-                "Proposal " + url + f" titled {opens[0]['title']}"
-                " is open.  Continuing to next step until proposal has executed."
+                "Proposal %s with git revision %s for subnet "
+                "is not the last (%s).  Will create.",
+                props_for_git_revision[0]["proposal_id"],
+                git_revision,
+                props[0]["proposal_id"],
             )
-            return {
-                "proposal_id": int(opens[0]["proposal_id"]),
-                "proposal_url": url,
-                "needs_vote": True,
-            }
-
-        if not props:
+        elif props_for_git_revision[0]["status"] not in (
+            ic_types.ProposalStatus.PROPOSAL_STATUS_OPEN,
+            ic_types.ProposalStatus.PROPOSAL_STATUS_ADOPTED,
+            ic_types.ProposalStatus.PROPOSAL_STATUS_EXECUTED,
+        ):
             self.log.info(
-                f"No proposals for subnet ID {subnet_id} to "
-                + f"adopt revision {git_revision}."
+                "Proposal %s with git revision %s for subnet "
+                "is in state %s and must be created again.  Will create.",
+                props_for_git_revision[0]["proposal_id"],
+                git_revision,
+                props_for_git_revision[0]["status"],
             )
         else:
-            self.log.info("The following proposals neither open nor executed exist:")
-            for p in props:
-                self.log.info(
-                    f"* {self.network.proposal_display_url}/{p['proposal_id']}"
-                )
+            prop = props_for_git_revision[0]
+            self.log.info(
+                "Proposal %s with git revision %s for subnet "
+                "is in state %s and does not need to be created.",
+                ["proposal_id"],
+                git_revision,
+                prop["status"],
+            )
+            url = f"{self.network.proposal_display_url}/{prop['proposal_id']}"
+            self.log.info(
+                "Proposal " + url + f" titled {prop['title']}"
+                f" has executed.  No need to do anything."
+            )
+            return {
+                "proposal_id": int(prop["proposal_id"]),
+                "proposal_url": url,
+                "needs_vote": prop["status"]
+                == ic_types.ProposalStatus.PROPOSAL_STATUS_OPEN,
+            }
 
         self.log.info(
             f"Creating proposal for subnet ID {subnet_id} to "

shared/dfinity/dre.py

@@ -140,6 +140,8 @@ def run(
                 if dry_run:
                     cmd.append("--dry-run")
                 if yes and not dry_run:
+                    # In dry-run mode, this kicks in, but cmd.index raises
+                    # ValueError when it cannot find the value.
                     try:
                         pos = cmd.index("propose")
                         cmd.insert(pos + 1, "--yes")
@@ -196,7 +198,7 @@ def str_to_status(t: str) -> ic_types.ProposalStatus:
         data = json.loads(r.output)
         results: list[ic_types.AbbrevProposal] = []
         for d in data:
-            d["proposal_id"] = d["id"]
+            d["proposal_id"] = int(d["id"])
             del d["id"]
             d["status"] = str_to_status("PROPOSAL_STATUS_" + d["status"].upper())
             d["topic"] = re.sub("([A-Z])", "_\\1", d["topic"])
@@ -229,6 +231,21 @@ def get_ic_os_version_deployment_proposals_for_subnet_and_revision(
             and r["payload"].get("replica_version_id") == git_revision
         ]
 
+    def get_ic_os_version_deployment_proposals_for_subnet(
+        self,
+        subnet_id: str,
+        limit: int = 1000,
+    ) -> list[ic_types.AbbrevProposal]:
+        return [
+            r
+            for r in self.get_proposals(
+                topic=ic_types.ProposalTopic.TOPIC_IC_OS_VERSION_DEPLOYMENT,
+                limit=limit,
+            )
+            if r["payload"].get("subnet_id") == subnet_id
+            and r["payload"].get("replica_version_id") is not None
+        ]
+
     def get_subnet_list(self) -> list[str]:
         r = self.run("get", "subnet-list", "--json", full_stdout=True)
         if r.exit_code != 0:

tests/test_ic_os_rollout_operators.py

@@ -1,6 +1,8 @@
+import contextlib
 import unittest
 
-from dfinity.ic_types import IC_NETWORKS
+import mock
+from dfinity.ic_types import IC_NETWORKS, AbbrevProposal, ProposalStatus, ProposalTopic
 from operators.ic_os_rollout import CreateProposalIdempotently
 
 
@@ -15,3 +17,115 @@ def test_instantiation(self) -> None:
             simulate_proposal=True,
             network=IC_NETWORKS["mainnet"],
         )
+
+
+class TestCreateProposal(unittest.TestCase):
+    network = IC_NETWORKS["mainnet"]
+
+    def _exercise(self) -> CreateProposalIdempotently:
+        def fake_xcom_push(key, value, context=None):  # type:ignore
+            return
+
+        k = CreateProposalIdempotently(
+            task_id="x",
+            subnet_id="yinp6-35cfo-wgcd2",
+            git_revision="d5eb7683e144acb0f8850fedb29011f34bfbe4ac",
+            simulate_proposal=True,
+            network=IC_NETWORKS["mainnet"],
+        )
+        k.xcom_push = fake_xcom_push
+        return k
+
+    def _prop(
+        self,
+        proposal_id: int,
+        proposal_status: ProposalStatus,
+        git_revision: str = "d5eb7683e144acb0f8850fedb29011f34bfbe4ac",
+    ) -> AbbrevProposal:
+        return {
+            "proposal_id": proposal_id,
+            "payload": {
+                "replica_version_id": git_revision,
+                "subnet_id": ("yinp6-35cfo-wgcd2"),
+            },
+            "proposer": "80",
+            "status": proposal_status,
+            "summary": "Update subnet "
+            "yinp6-35cfo-wgcd2"
+            "to replica version "
+            f"[{git_revision}](https:##dashboard.internetcomputer.org/release/{git_revision})\n",
+            "title": f"Update subnet yinp6 to replica version {git_revision}",
+            "topic": ProposalTopic.TOPIC_IC_OS_VERSION_DEPLOYMENT,
+        }
+
+    @contextlib.contextmanager
+    def _ctx(self):  # type:ignore
+        with mock.patch(
+            "dfinity.dre.DRE.get_ic_os_version_deployment_proposals_for_subnet"
+        ) as m, mock.patch(
+            "dfinity.prom_api.query_prometheus_servers"
+        ) as n, mock.patch(
+            "dfinity.dre.AuthenticatedDRE.propose_to_update_subnet_replica_version"
+        ) as p, mock.patch(
+            "airflow.models.variable.Variable.get"
+        ) as v:
+            n.return_value = [{"value": 13}]
+            v.return_value = "FAKE CERT"
+            yield m, p
+
+    def test_no_proposals(self) -> None:
+        with self._ctx() as (m, p):
+            m.return_value = []
+            p.return_value = -123456
+            res = self._exercise().execute({})
+            assert res["needs_vote"] is True
+            assert res["proposal_id"] == p.return_value
+
+    def test_latest_proposal_matches(self) -> None:
+        with self._ctx() as (m, p):
+            props: list[AbbrevProposal] = [
+                self._prop(123456, ProposalStatus.PROPOSAL_STATUS_OPEN),
+            ]
+            m.return_value = props
+            p.return_value = -123456
+            res = self._exercise().execute({})
+            assert res["needs_vote"] is True
+            assert res["proposal_id"] == props[0]["proposal_id"]
+
+            # Now let's test with proposal status executed.
+            props[0]["status"] = ProposalStatus.PROPOSAL_STATUS_EXECUTED
+            res = self._exercise().execute({})
+            assert res["needs_vote"] is False
+            assert res["proposal_id"] == props[0]["proposal_id"]
+
+    def test_match_exists_but_not_latest(self) -> None:
+        with self._ctx() as (m, p):
+            props: list[AbbrevProposal] = [
+                self._prop(123456, ProposalStatus.PROPOSAL_STATUS_EXECUTED),
+                self._prop(
+                    123457,
+                    ProposalStatus.PROPOSAL_STATUS_OPEN,
+                    git_revision="some other shit",
+                ),
+            ]
+            m.return_value = props
+            p.return_value = -123456
+            res = self._exercise().execute({})
+            assert res["needs_vote"] is True
+            assert res["proposal_id"] == p.return_value
+
+    def test_match_exists_is_latest(self) -> None:
+        with self._ctx() as (m, p):
+            props: list[AbbrevProposal] = [
+                self._prop(123458, ProposalStatus.PROPOSAL_STATUS_EXECUTED),
+                self._prop(
+                    123457,
+                    ProposalStatus.PROPOSAL_STATUS_OPEN,
+                    git_revision="some other shit",
+                ),
+            ]
+            m.return_value = props
+            p.return_value = -123456
+            res = self._exercise().execute({})
+            assert res["needs_vote"] is False
+            assert res["proposal_id"] == props[0]["proposal_id"]