sources/azure: move pps handling out of _poll_imds() (#2075)

Pull out remaining PPS handling bits from _poll_imds() and add two
explicit methods for the overloaded path:
- _wait_for_pps_running_reuse() for running PPS logic.
- _wait_for_pps_unknown_reuse() for unknown and recovery PPS logic.

For consistency:
- Rename _wait_for_all_nics_ready() -> _wait_for_pps_savable_reuse().
- Move reporting ready logic into _wait_for_pps_os_disk_shutdown().

Drop several impacted tests as coverage already exists in
TestProvisioning, and update the rest to handle the +/- 1 DHCP attempt
due to varying assumptions around PPS state and DHCP.

2 files changed, 135 insertions, 304 deletions

diff --git a/cloudinit/sources/DataSourceAzure.py b/cloudinit/sources/DataSourceAzure.py
index 807c02c7..927e8cf0 100644
--- a/cloudinit/sources/DataSourceAzure.py
+++ b/cloudinit/sources/DataSourceAzure.py
@@ -9,6 +9,7 @@ import crypt
 import os
 import os.path
 import re
+import socket
 import xml.etree.ElementTree as ET
 from enum import Enum
 from pathlib import Path
@@ -571,11 +572,14 @@ class DataSourceAzure(sources.DataSource):
                 report_diagnostic_event(msg, logger_func=LOG.error)
                 raise sources.InvalidMetaDataException(msg)
 
-            if pps_type == PPSType.SAVABLE:
-                self._wait_for_all_nics_ready()
+            if pps_type == PPSType.RUNNING:
+                self._wait_for_pps_running_reuse()
+            elif pps_type == PPSType.SAVABLE:
+                self._wait_for_pps_savable_reuse()
             elif pps_type == PPSType.OS_DISK:
-                self._report_ready_for_pps(create_marker=False)
                 self._wait_for_pps_os_disk_shutdown()
+            else:
+                self._wait_for_pps_unknown_reuse()
 
             md, userdata_raw, cfg, files = self._reprovision()
             # fetch metadata again as it has changed after reprovisioning
@@ -974,15 +978,6 @@ class DataSourceAzure(sources.DataSource):
             self._create_report_ready_marker()
 
     @azure_ds_telemetry_reporter
-    def _wait_for_pps_os_disk_shutdown(self):
-        report_diagnostic_event(
-            "Waiting for host to shutdown VM...",
-            logger_func=LOG.info,
-        )
-        sleep(31536000)
-        raise BrokenAzureDataSource("Shutdown failure for PPS disk.")
-
-    @azure_ds_telemetry_reporter
     def _check_if_nic_is_primary(self, ifname: str) -> bool:
         """Check if a given interface is the primary nic or not."""
         # For now, only a VM's primary NIC can contact IMDS and WireServer. If
@@ -1060,16 +1055,71 @@ class DataSourceAzure(sources.DataSource):
             report_diagnostic_event(str(error), logger_func=LOG.error)
 
     @azure_ds_telemetry_reporter
-    def _wait_for_all_nics_ready(self):
-        """Wait for nic(s) to be hot-attached. There may be multiple nics
-        depending on the customer request.
-        But only primary nic would be able to communicate with wireserver
-        and IMDS. So we detect and save the primary nic to be used later.
-        """
+    def _create_bound_netlink_socket(self) -> socket.socket:
+        try:
+            return netlink.create_bound_netlink_socket()
+        except netlink.NetlinkCreateSocketError as error:
+            report_diagnostic_event(
+                f"Failed to create netlink socket: {error}",
+                logger_func=LOG.error,
+            )
+            raise
+
+    @azure_ds_telemetry_reporter
+    def _wait_for_pps_os_disk_shutdown(self):
+        """Report ready and wait for host to initiate shutdown."""
+        self._report_ready_for_pps(create_marker=False)
+
+        report_diagnostic_event(
+            "Waiting for host to shutdown VM...",
+            logger_func=LOG.info,
+        )
+        sleep(31536000)
+        raise BrokenAzureDataSource("Shutdown failure for PPS disk.")
+
+    @azure_ds_telemetry_reporter
+    def _wait_for_pps_running_reuse(self) -> None:
+        """Report ready and wait for nic link to switch upon re-use."""
+        nl_sock = self._create_bound_netlink_socket()
+
+        try:
+            if (
+                self._ephemeral_dhcp_ctx is None
+                or self._ephemeral_dhcp_ctx.iface is None
+            ):
+                raise RuntimeError("missing ephemeral context")
+
+            iface = self._ephemeral_dhcp_ctx.iface
+            self._report_ready_for_pps()
+
+            LOG.debug(
+                "Wait for vnetswitch to happen on %s",
+                iface,
+            )
+            with events.ReportEventStack(
+                name="wait-for-media-disconnect-connect",
+                description="wait for vnet switch",
+                parent=azure_ds_reporter,
+            ):
+                try:
+                    netlink.wait_for_media_disconnect_connect(nl_sock, iface)
+                except AssertionError as e:
+                    report_diagnostic_event(
+                        "Error while waiting for vnet switch: %s" % e,
+                        logger_func=LOG.error,
+                    )
+        finally:
+            nl_sock.close()
+
+        # Teardown source PPS network configuration.
+        self._teardown_ephemeral_networking()
+
+    @azure_ds_telemetry_reporter
+    def _wait_for_pps_savable_reuse(self):
+        """Report ready and wait for nic(s) to be hot-attached upon re-use."""
+        nl_sock = self._create_bound_netlink_socket()
 
-        nl_sock = None
         try:
-            nl_sock = netlink.create_bound_netlink_socket()
             self._report_ready_for_pps(expect_url_error=True)
             try:
                 self._teardown_ephemeral_networking()
@@ -1083,76 +1133,25 @@ class DataSourceAzure(sources.DataSource):
 
             self._wait_for_nic_detach(nl_sock)
             self._wait_for_hot_attached_primary_nic(nl_sock)
-        except netlink.NetlinkCreateSocketError as e:
-            report_diagnostic_event(str(e), logger_func=LOG.warning)
-            raise
         finally:
-            if nl_sock:
-                nl_sock.close()
+            nl_sock.close()
 
     @azure_ds_telemetry_reporter
-    def _poll_imds(self):
-        """Poll IMDS for the new provisioning data until we get a valid
-        response. Then return the returned JSON object."""
-        nl_sock = None
-        report_ready = bool(
-            not os.path.isfile(self._reported_ready_marker_file)
-        )
-        dhcp_attempts = 0
-
-        if report_ready:
-            try:
-                if (
-                    self._ephemeral_dhcp_ctx is None
-                    or self._ephemeral_dhcp_ctx.iface is None
-                ):
-                    raise RuntimeError("Missing ephemeral context")
-                iface = self._ephemeral_dhcp_ctx.iface
+    def _wait_for_pps_unknown_reuse(self):
+        """Report ready if needed for unknown/recovery PPS."""
+        if os.path.isfile(self._reported_ready_marker_file):
+            # Already reported ready, nothing to do.
+            return
 
-                nl_sock = netlink.create_bound_netlink_socket()
-                self._report_ready_for_pps()
+        self._report_ready_for_pps()
 
-                LOG.debug(
-                    "Wait for vnetswitch to happen on %s",
-                    iface,
-                )
-                with events.ReportEventStack(
-                    name="wait-for-media-disconnect-connect",
-                    description="wait for vnet switch",
-                    parent=azure_ds_reporter,
-                ):
-                    try:
-                        netlink.wait_for_media_disconnect_connect(
-                            nl_sock, iface
-                        )
-                    except AssertionError as e:
-                        report_diagnostic_event(
-                            "Error while waiting for vnet switch: %s" % e,
-                            logger_func=LOG.error,
-                        )
-            except netlink.NetlinkCreateSocketError as e:
-                report_diagnostic_event(
-                    "Failed to create bound netlink socket: %s" % e,
-                    logger_func=LOG.warning,
-                )
-                raise sources.InvalidMetaDataException(
-                    "Failed to report ready while in provisioning pool."
-                ) from e
-            except NoDHCPLeaseError as e:
-                report_diagnostic_event(
-                    "DHCP failed while in provisioning pool",
-                    logger_func=LOG.warning,
-                )
-                raise sources.InvalidMetaDataException(
-                    "Failed to report ready while in provisioning pool."
-                ) from e
-            finally:
-                if nl_sock:
-                    nl_sock.close()
-
-            # Teardown old network configuration.
-            self._teardown_ephemeral_networking()
+        # Teardown source PPS network configuration.
+        self._teardown_ephemeral_networking()
 
+    @azure_ds_telemetry_reporter
+    def _poll_imds(self) -> bytes:
+        """Poll IMDs for reprovisiondata XML document data."""
+        dhcp_attempts = 0
         reprovision_data = None
         while not reprovision_data:
             if not self._is_ephemeral_networking_up():
@@ -1177,7 +1176,6 @@ class DataSourceAzure(sources.DataSource):
             "attempted dhcp %d times after reuse" % dhcp_attempts,
             logger_func=LOG.debug,
         )
-
         return reprovision_data
 
     @azure_ds_telemetry_reporter
diff --git a/tests/unittests/sources/test_azure.py b/tests/unittests/sources/test_azure.py
index 6c3e9281..0648f08c 100644
--- a/tests/unittests/sources/test_azure.py
+++ b/tests/unittests/sources/test_azure.py
@@ -262,10 +262,11 @@ def mock_util_mount_cb():
 
 
 @pytest.fixture
-def mock_util_write_file():
-    with mock.patch(
-        MOCKPATH + "util.write_file",
-        autospec=True,
+def wrapped_util_write_file():
+    with mock.patch.object(
+        dsaz.util,
+        "write_file",
+        wraps=write_file,
     ) as m:
         yield m
 
@@ -1348,89 +1349,6 @@ scbus-1 on xpt0 bus 0
         dsrc.crawl_metadata()
         self.assertEqual(2, self.m_fetch.call_count)
 
-    @mock.patch("cloudinit.sources.DataSourceAzure.util.write_file")
-    @mock.patch(
-        "cloudinit.sources.DataSourceAzure.DataSourceAzure._report_ready"
-    )
-    @mock.patch("cloudinit.sources.DataSourceAzure.DataSourceAzure._poll_imds")
-    def test_crawl_metadata_on_reprovision_reports_ready(
-        self, poll_imds_func, m_report_ready, m_write
-    ):
-        """If reprovisioning, report ready at the end"""
-        ovfenv = construct_ovf_env(preprovisioned_vm=True)
-
-        data = {"ovfcontent": ovfenv, "sys_cfg": {}}
-        dsrc = self._get_ds(data)
-        poll_imds_func.return_value = ovfenv
-        dsrc.crawl_metadata()
-        self.assertEqual(1, m_report_ready.call_count)
-
-    @mock.patch("cloudinit.sources.DataSourceAzure.util.write_file")
-    @mock.patch(
-        "cloudinit.sources.DataSourceAzure.DataSourceAzure._report_ready"
-    )
-    @mock.patch("cloudinit.sources.DataSourceAzure.DataSourceAzure._poll_imds")
-    @mock.patch(
-        "cloudinit.sources.DataSourceAzure.DataSourceAzure."
-        "_wait_for_all_nics_ready"
-    )
-    def test_crawl_metadata_waits_for_nic_on_savable_vms(
-        self, detect_nics, poll_imds_func, report_ready_func, m_write
-    ):
-        """If reprovisioning, report ready at the end"""
-        ovfenv = construct_ovf_env(
-            preprovisioned_vm=True, preprovisioned_vm_type="Savable"
-        )
-
-        data = {"ovfcontent": ovfenv, "sys_cfg": {}}
-        dsrc = self._get_ds(data)
-        poll_imds_func.return_value = ovfenv
-        dsrc.crawl_metadata()
-        self.assertEqual(1, report_ready_func.call_count)
-        self.assertEqual(1, detect_nics.call_count)
-
-    @mock.patch("cloudinit.sources.DataSourceAzure.util.write_file")
-    @mock.patch(
-        "cloudinit.sources.helpers.netlink.wait_for_media_disconnect_connect"
-    )
-    @mock.patch(
-        "cloudinit.sources.DataSourceAzure.DataSourceAzure._report_ready",
-        return_value=True,
-    )
-    @mock.patch(
-        "cloudinit.sources.DataSourceAzure.imds.fetch_reprovision_data"
-    )
-    def test_crawl_metadata_on_reprovision_reports_ready_using_lease(
-        self, m_fetch_reprovision_data, m_report_ready, m_media_switch, m_write
-    ):
-        """If reprovisioning, report ready using the obtained lease"""
-        ovfenv = construct_ovf_env(preprovisioned_vm=True)
-
-        data = {"ovfcontent": ovfenv, "sys_cfg": {}}
-        dsrc = self._get_ds(data)
-
-        lease = {
-            "interface": "eth9",
-            "fixed-address": "192.168.2.9",
-            "routers": "192.168.2.1",
-            "subnet-mask": "255.255.255.0",
-            "unknown-245": "624c3620",
-        }
-        self.m_dhcp.return_value.obtain_lease.return_value = lease
-        m_media_switch.return_value = None
-
-        reprovision_ovfenv = construct_ovf_env()
-        m_fetch_reprovision_data.return_value = reprovision_ovfenv.encode(
-            "utf-8"
-        )
-
-        dsrc.crawl_metadata()
-
-        assert m_report_ready.mock_calls == [
-            mock.call(),
-            mock.call(pubkey_info=None),
-        ]
-
     def test_waagent_d_has_0700_perms(self):
         # we expect /var/lib/waagent to be created 0700
         dsrc = self._get_ds({"ovfcontent": construct_ovf_env()})
@@ -2790,7 +2708,7 @@ class TestPreprovisioningHotAttachNics(CiTestCase):
     ):
         """Report ready first and then wait for nic detach"""
         dsa = dsaz.DataSourceAzure({}, distro=None, paths=self.paths)
-        dsa._wait_for_all_nics_ready()
+        dsa._wait_for_pps_savable_reuse()
         self.assertEqual(1, m_report_ready.call_count)
         self.assertEqual(1, m_wait_for_hot_attached_primary_nic.call_count)
         self.assertEqual(1, m_detach.call_count)
@@ -2849,7 +2767,7 @@ class TestPreprovisioningHotAttachNics(CiTestCase):
         m_dhcpv4.return_value = dhcp_ctx
         m_imds.side_effect = [md]
 
-        dsa._wait_for_all_nics_ready()
+        dsa._wait_for_pps_savable_reuse()
 
         self.assertEqual(1, m_detach.call_count)
         # only wait for primary nic
@@ -2869,7 +2787,7 @@ class TestPreprovisioningHotAttachNics(CiTestCase):
         m_imds.reset_mock()
         m_imds.side_effect = [{}, md]
         dsa = dsaz.DataSourceAzure({}, distro=distro, paths=self.paths)
-        dsa._wait_for_all_nics_ready()
+        dsa._wait_for_pps_savable_reuse()
         self.assertEqual(1, m_detach.call_count)
         self.assertEqual(2, m_attach.call_count)
         self.assertEqual(2, m_dhcpv4.call_count)
@@ -2974,9 +2892,8 @@ class TestPreprovisioningHotAttachNics(CiTestCase):
         dsa = dsaz.DataSourceAzure({}, distro=distro, paths=self.paths)
 
         self.assertRaises(
-            netlink.NetlinkCreateSocketError, dsa._wait_for_all_nics_ready
+            netlink.NetlinkCreateSocketError, dsa._wait_for_pps_savable_reuse
         )
-        # dsa._wait_for_all_nics_ready()
 
 
 @mock.patch("cloudinit.net.find_fallback_nic", return_value="eth9")
@@ -2986,7 +2903,6 @@ class TestPreprovisioningHotAttachNics(CiTestCase):
     "cloudinit.sources.helpers.netlink.wait_for_media_disconnect_connect"
 )
 @mock.patch(MOCKPATH + "imds.fetch_reprovision_data")
-@mock.patch(MOCKPATH + "DataSourceAzure._report_ready", return_value=True)
 class TestPreprovisioningPollIMDS(CiTestCase):
     def setUp(self):
         super(TestPreprovisioningPollIMDS, self).setUp()
@@ -2998,7 +2914,6 @@ class TestPreprovisioningPollIMDS(CiTestCase):
     @mock.patch("time.sleep", mock.MagicMock())
     def test_poll_imds_re_dhcp_on_timeout(
         self,
-        m_report_ready,
         m_fetch_reprovisiondata,
         m_media_switch,
         m_dhcp,
@@ -3010,7 +2925,6 @@ class TestPreprovisioningPollIMDS(CiTestCase):
             url_helper.UrlError(requests.Timeout("Fake connection timeout")),
             b"ovf data",
         ]
-        report_file = self.tmp_path("report_marker", self.tmp)
         lease = {
             "interface": "eth9",
             "fixed-address": "192.168.2.9",
@@ -3026,21 +2940,15 @@ class TestPreprovisioningPollIMDS(CiTestCase):
 
         dsa = dsaz.DataSourceAzure({}, distro=mock.Mock(), paths=self.paths)
         dsa._ephemeral_dhcp_ctx = dhcp_ctx
-        with mock.patch.object(
-            dsa, "_reported_ready_marker_file", report_file
-        ):
-            dsa._poll_imds()
-
-        assert m_report_ready.mock_calls == [mock.call()]
+        dsa._poll_imds()
 
-        self.assertEqual(2, m_dhcp.call_count, "Expected 2 DHCP calls")
+        self.assertEqual(1, m_dhcp.call_count, "Expected 1 DHCP calls")
         assert m_fetch_reprovisiondata.call_count == 2
 
     @mock.patch("os.path.isfile")
     def test_poll_imds_skips_dhcp_if_ctx_present(
         self,
         m_isfile,
-        report_ready_func,
         m_fetch_reprovisiondata,
         m_media_switch,
         m_dhcp,
@@ -3052,14 +2960,10 @@ class TestPreprovisioningPollIMDS(CiTestCase):
         polling for reprovisiondata. Note that if this ctx is set when
         _poll_imds is called, then it is not expected to be waiting for
         media_disconnect_connect either."""
-        report_file = self.tmp_path("report_marker", self.tmp)
         m_isfile.return_value = True
         dsa = dsaz.DataSourceAzure({}, distro=None, paths=self.paths)
         dsa._ephemeral_dhcp_ctx = mock.Mock(lease={})
-        with mock.patch.object(
-            dsa, "_reported_ready_marker_file", report_file
-        ):
-            dsa._poll_imds()
+        dsa._poll_imds()
         self.assertEqual(0, m_dhcp.call_count)
         self.assertEqual(0, m_media_switch.call_count)
 
@@ -3069,7 +2973,6 @@ class TestPreprovisioningPollIMDS(CiTestCase):
         self,
         m_ephemeral_dhcpv4,
         m_isfile,
-        report_ready_func,
         m_fetch_reprovisiondata,
         m_media_switch,
         m_dhcp,
@@ -3090,14 +2993,11 @@ class TestPreprovisioningPollIMDS(CiTestCase):
             ),
             b"ovf data",
         ]
-        report_file = self.tmp_path("report_marker", self.tmp)
         m_isfile.return_value = True
         distro = mock.MagicMock()
         distro.get_tmp_exec_path = self.tmp_dir
         dsa = dsaz.DataSourceAzure({}, distro=distro, paths=self.paths)
-        with mock.patch.object(
-            dsa, "_reported_ready_marker_file", report_file
-        ), mock.patch.object(dsa, "_ephemeral_dhcp_ctx") as m_dhcp_ctx:
+        with mock.patch.object(dsa, "_ephemeral_dhcp_ctx") as m_dhcp_ctx:
             m_dhcp_ctx.obtain_lease.return_value = "Dummy lease"
             dsa._ephemeral_dhcp_ctx = m_dhcp_ctx
             dsa._poll_imds()
@@ -3106,107 +3006,6 @@ class TestPreprovisioningPollIMDS(CiTestCase):
         self.assertEqual(0, m_media_switch.call_count)
         self.assertEqual(2, m_fetch_reprovisiondata.call_count)
 
-    def test_does_not_poll_imds_report_ready_when_marker_file_exists(
-        self,
-        m_report_ready,
-        m_fetch_reprovisiondata,
-        m_media_switch,
-        m_dhcp,
-        m_net,
-        m_fallback,
-    ):
-        """poll_imds should not call report ready when the reported ready
-        marker file exists"""
-        report_file = self.tmp_path("report_marker", self.tmp)
-        write_file(report_file, content="dont run report_ready :)")
-        m_dhcp.return_value = [
-            {
-                "interface": "eth9",
-                "fixed-address": "192.168.2.9",
-                "routers": "192.168.2.1",
-                "subnet-mask": "255.255.255.0",
-                "unknown-245": "624c3620",
-            }
-        ]
-        m_media_switch.return_value = None
-        dsa = dsaz.DataSourceAzure({}, distro=mock.Mock(), paths=self.paths)
-        with mock.patch.object(
-            dsa, "_reported_ready_marker_file", report_file
-        ):
-            dsa._poll_imds()
-        self.assertEqual(m_report_ready.call_count, 0)
-
-    @mock.patch(MOCKPATH + "imds.fetch_metadata_with_api_fallback")
-    def test_poll_imds_report_ready_success_writes_marker_file(
-        self,
-        m_fetch,
-        m_report_ready,
-        m_fetch_reprovisiondata,
-        m_media_switch,
-        m_dhcp,
-        m_net,
-        m_fallback,
-    ):
-        """poll_imds should write the report_ready marker file if
-        reporting ready succeeds"""
-        report_file = self.tmp_path("report_marker", self.tmp)
-        m_dhcp.return_value = [
-            {
-                "interface": "eth9",
-                "fixed-address": "192.168.2.9",
-                "routers": "192.168.2.1",
-                "subnet-mask": "255.255.255.0",
-                "unknown-245": "624c3620",
-            }
-        ]
-        m_media_switch.return_value = None
-        distro = mock.MagicMock()
-        distro.get_tmp_exec_path = self.tmp_dir
-        dsa = dsaz.DataSourceAzure({}, distro=distro, paths=self.paths)
-        self.assertFalse(os.path.exists(report_file))
-        dsa._ephemeral_dhcp_ctx = mock.Mock(interface="eth9")
-        with mock.patch.object(
-            dsa, "_reported_ready_marker_file", report_file
-        ):
-            dsa._poll_imds()
-        self.assertEqual(m_report_ready.call_count, 1)
-        self.assertTrue(os.path.exists(report_file))
-
-    def test_poll_imds_report_ready_failure_raises_exc_and_doesnt_write_marker(
-        self,
-        m_report_ready,
-        m_fetch_reprovisiondata,
-        m_media_switch,
-        m_dhcp,
-        m_net,
-        m_fallback,
-    ):
-        """poll_imds should write the report_ready marker file if
-        reporting ready succeeds"""
-        report_file = self.tmp_path("report_marker", self.tmp)
-        m_dhcp.return_value = [
-            {
-                "interface": "eth9",
-                "fixed-address": "192.168.2.9",
-                "routers": "192.168.2.1",
-                "subnet-mask": "255.255.255.0",
-                "unknown-245": "624c3620",
-            }
-        ]
-        m_media_switch.return_value = None
-        m_report_ready.side_effect = [Exception("fail")]
-        distro = mock.MagicMock()
-        distro.get_tmp_exec_path = self.tmp_dir
-        dsa = dsaz.DataSourceAzure({}, distro=distro, paths=self.paths)
-        self.assertFalse(os.path.exists(report_file))
-        dsa._ephemeral_dhcp_ctx = mock.Mock(interface="eth9")
-        with mock.patch.object(
-            dsa, "_reported_ready_marker_file", report_file
-        ):
-            self.assertRaises(InvalidMetaDataException, dsa._poll_imds)
-        self.assertEqual(m_report_ready.call_count, 1)
-        self.assertFalse(os.path.exists(report_file))
-
 
 @mock.patch(MOCKPATH + "DataSourceAzure._report_ready", mock.MagicMock())
 @mock.patch(MOCKPATH + "subp.subp", mock.MagicMock())
@@ -3241,7 +3040,7 @@ class TestAzureDataSourcePreprovisioning(CiTestCase):
             }
         ]
         dsa = dsaz.DataSourceAzure({}, distro=mock.Mock(), paths=self.paths)
-        dsa._ephemeral_dhcp_ctx = mock.Mock(interface="eth9")
+        dsa._ephemeral_dhcp_ctx = None
         self.assertTrue(len(dsa._poll_imds()) > 0)
         self.assertEqual(m_dhcp.call_count, 1)
         m_net.assert_any_call(
@@ -3273,7 +3072,7 @@ class TestAzureDataSourcePreprovisioning(CiTestCase):
         content = construct_ovf_env(username=username, hostname=hostname)
         m_fetch_reprovisiondata.side_effect = [content]
         dsa = dsaz.DataSourceAzure({}, distro=mock.Mock(), paths=self.paths)
-        dsa._ephemeral_dhcp_ctx = mock.Mock(interface="eth9")
+        dsa._ephemeral_dhcp_ctx = None
         md, _ud, cfg, _d = dsa._reprovision()
         self.assertEqual(md["local-hostname"], hostname)
         self.assertEqual(cfg["system_info"]["default_user"]["name"], username)
@@ -3645,7 +3444,9 @@ class TestProvisioning:
         mock_util_find_devs_with,
         mock_util_load_file,
         mock_util_mount_cb,
+        wrapped_util_write_file,
         mock_wrapping_setup_ephemeral_networking,
+        patched_data_dir_path,
         patched_reported_ready_marker_path,
     ):
         self.azure_ds = azure_ds
@@ -3671,9 +3472,11 @@ class TestProvisioning:
         self.mock_util_find_devs_with = mock_util_find_devs_with
         self.mock_util_load_file = mock_util_load_file
         self.mock_util_mount_cb = mock_util_mount_cb
+        self.wrapped_util_write_file = wrapped_util_write_file
         self.mock_wrapping_setup_ephemeral_networking = (
             mock_wrapping_setup_ephemeral_networking
         )
+        self.patched_data_dir_path = patched_data_dir_path
         self.patched_reported_ready_marker_path = (
             patched_reported_ready_marker_path
         )
@@ -3757,6 +3560,10 @@ class TestProvisioning:
         # Verify netlink.
         assert self.mock_netlink.mock_calls == []
 
+        # Verify no reported_ready marker written.
+        assert self.wrapped_util_write_file.mock_calls == []
+        assert self.patched_reported_ready_marker_path.exists() is False
+
     def test_running_pps(self):
         self.imds_md["extended"]["compute"]["ppsType"] = "Running"
 
@@ -3849,10 +3656,15 @@ class TestProvisioning:
         assert self.mock_netlink.mock_calls == [
             mock.call.create_bound_netlink_socket(),
             mock.call.wait_for_media_disconnect_connect(mock.ANY, "ethBoot0"),
-            mock.call.create_bound_netlink_socket().__bool__(),
             mock.call.create_bound_netlink_socket().close(),
         ]
 
+        # Verify reported_ready marker written and cleaned up.
+        assert self.wrapped_util_write_file.mock_calls[0] == mock.call(
+            self.patched_reported_ready_marker_path.as_posix(), mock.ANY
+        )
+        assert self.patched_reported_ready_marker_path.exists() is False
+
     def test_savable_pps(self):
         self.imds_md["extended"]["compute"]["ppsType"] = "Savable"
 
@@ -3961,10 +3773,15 @@ class TestProvisioning:
             mock.call.create_bound_netlink_socket(),
             mock.call.wait_for_nic_detach_event(nl_sock),
             mock.call.wait_for_nic_attach_event(nl_sock, ["ethAttached1"]),
-            mock.call.create_bound_netlink_socket().__bool__(),
             mock.call.create_bound_netlink_socket().close(),
         ]
 
+        # Verify reported_ready marker written and cleaned up.
+        assert self.wrapped_util_write_file.mock_calls[0] == mock.call(
+            self.patched_reported_ready_marker_path.as_posix(), mock.ANY
+        )
+        assert self.patched_reported_ready_marker_path.exists() is False
+
     @pytest.mark.parametrize(
         "fabric_side_effect",
         [
@@ -4110,10 +3927,15 @@ class TestProvisioning:
             mock.call.create_bound_netlink_socket(),
             mock.call.wait_for_nic_detach_event(nl_sock),
             mock.call.wait_for_nic_attach_event(nl_sock, ["ethAttached1"]),
-            mock.call.create_bound_netlink_socket().__bool__(),
             mock.call.create_bound_netlink_socket().close(),
         ]
 
+        # Verify reported_ready marker written and cleaned up.
+        assert self.wrapped_util_write_file.mock_calls[0] == mock.call(
+            self.patched_reported_ready_marker_path.as_posix(), mock.ANY
+        )
+        assert self.patched_reported_ready_marker_path.exists() is False
+
     @pytest.mark.parametrize("pps_type", ["Savable", "Running", "None"])
     def test_recovery_pps(self, pps_type):
         self.patched_reported_ready_marker_path.write_text("")
@@ -4186,6 +4008,16 @@ class TestProvisioning:
         # Verify no netlink operations for recovering PPS.
         assert self.mock_netlink.mock_calls == []
 
+        # Verify reported_ready marker not written.
+        assert self.wrapped_util_write_file.mock_calls == [
+            mock.call(
+                filename=str(self.patched_data_dir_path / "ovf-env.xml"),
+                content=mock.ANY,
+                mode=mock.ANY,
+            )
+        ]
+        assert self.patched_reported_ready_marker_path.exists() is False
+
     @pytest.mark.parametrize("pps_type", ["Savable", "Running", "Unknown"])
     def test_source_pps_fails_initial_dhcp(self, pps_type):
         self.imds_md["extended"]["compute"]["ppsType"] = pps_type
@@ -4281,6 +4113,7 @@ class TestProvisioning:
         # Ensure no reported ready marker is left behind as the VM's next
         # boot will behave like a typical provisioning boot.
         assert self.patched_reported_ready_marker_path.exists() is False
+        assert self.wrapped_util_write_file.mock_calls == []
 
 
 class TestValidateIMDSMetadata: