From fcdb970958aef88ab72eb656a462679f793633f8 Mon Sep 17 00:00:00 2001
From: sr73318 <sr73318@netapp.com>
Date: Wed, 14 Oct 2026 16:23:04 +0530
Subject: [PATCH] CSTACKEX-207: Automation for snapshot and revert for VM and
 CS-volume

---
 test/integration/plugins/ontap/README.md      |  44 +-
 test/integration/plugins/ontap/TEST_CASES.md  |  58 +-
 test/integration/plugins/ontap/guest_ssh.py   | 216 +++++
 .../iscsi/instance/test_vm_volume_attach.py   | 796 +++++++++++++++++-
 .../nfs3/instance/test_vm_volume_attach.py    | 784 ++++++++++++++++-
 test/integration/plugins/ontap/ontap.cfg      |   2 +
 .../plugins/ontap/ontap_test_base.py          |  21 +
 test/integration/plugins/ontap/setup_env.sh   |   9 +-
 8 files changed, 1886 insertions(+), 44 deletions(-)
 create mode 100644 test/integration/plugins/ontap/guest_ssh.py

diff --git a/test/integration/plugins/ontap/README.md b/test/integration/plugins/ontap/README.md
index 2adbf449af28..92e799a36a6f 100644
--- a/test/integration/plugins/ontap/README.md
+++ b/test/integration/plugins/ontap/README.md
@@ -43,7 +43,9 @@ for the deployment, health-gate, and artifact contracts.
 test/integration/plugins/ontap/
 ├── ontap.cfg                     # Environment config (IPs, credentials, zone info)
 ├── ontap_test_base.py            # Shared base class and ONTAP REST client
-├── TEST_CASES.md                 # Full test case reference table (72 tests)
+├── guest_ssh.py                  # Guest SSH (paramiko) for the VM workflow suites
+├── setup_env.sh                  # venv: Marvin, pyvmomi, paramiko<4
+├── TEST_CASES.md                 # Full test case reference table (106 tests)
 ├── README.md                     # This file
 │
 ├── nfs3/
@@ -122,6 +124,39 @@ python3 -m pip install --user \
 python3 -c "import marvin; print('Marvin OK')"
 ```
 
+The VM workflow suites also need a venv with pinned dependencies. From the repo root:
+
+```bash
+bash test/integration/plugins/ontap/setup_env.sh
+```
+
+The script is safe to re-run. It creates `test/integration/plugins/ontap/.venv` when that directory is missing, then installs into it:
+
+- Marvin (from `MARVIN_ARCHIVE` when set, otherwise from `tools/marvin` after generating `commands.xml`)
+- `pyvmomi==8.0.2.0.1` (pyvmomi 9.x needs Python 3.10+)
+- `paramiko>=3.4,<4`
+
+`run_tests.sh` uses that venv automatically.
+
+#### Paramiko (guest SSH)
+
+`nfs3/instance/test_vm_volume_attach.py` and `iscsi/instance/test_vm_volume_attach.py` log into the guest through `guest_ssh.py`. The lab template is CentOS 5.5, and its sshd only offers SHA-1 key exchange (`diffie-hellman-group14-sha1`) and `ssh-rsa` host keys. Paramiko 4.x removed SHA-1, so those logins fail unless the venv has Paramiko 3.x.
+
+`setup_env.sh` installs the pin:
+
+```bash
+test/integration/plugins/ontap/.venv/bin/pip install "paramiko>=3.4,<4"
+```
+
+`guest_ssh.check_paramiko()` checks for `diffie-hellman-group14-sha1` before the first guest connection and tells you to re-run `setup_env.sh` when the active interpreter has Paramiko 4. Guest login itself comes from `ontap.cfg` (`cloudstack.templateUsername` / `cloudstack.templatePassword`).
+
+Confirm the venv before a VM workflow run:
+
+```bash
+test/integration/plugins/ontap/.venv/bin/python -c \
+  "import paramiko; print(paramiko.__version__); assert 'diffie-hellman-group14-sha1' in paramiko.Transport._kex_info"
+```
+
 ---
 
 ## Configuration — `ontap.cfg`
@@ -334,13 +369,13 @@ self.assertEqual(result.state, "Maintenance")
 | NFS3 Pool with Volumes | `nfs3/pool/test_pool_with_volumes.py` | 7 | Same + live volume present; negative delete guard |
 | NFS3 Zone-Scoped Pool | `nfs3/pool/test_zone_scoped_pool.py` | 4 | Zone scope — all hosts connected via `attachZone` |
 | NFS3 Volume Lifecycle | `nfs3/volume/test_volume_lifecycle.py` | 5 | Volume is metadata-only; FlexVol unchanged on delete |
-| NFS3 VM + Volume Attach | `nfs3/instance/test_vm_volume_attach.py` | 10 | Full VM lifecycle with hot-plug/detach; ROOT on tagged pool seeds/reuses template cache, which survives VM delete |
+| NFS3 VM + Volume Attach | `nfs3/instance/test_vm_volume_attach.py` | 26 | Full VM lifecycle with hot-plug/detach, guest SSH, volume and VM snapshots; ROOT on tagged pool seeds/reuses template cache, which survives VM delete |
 | NFS3 Template Cache Negative | `nfs3/template/test_template_cache_negative.py` | 3 | Tag mismatch; undersized pool; out-of-band cache delete |
 | iSCSI Pool Lifecycle | `iscsi/pool/test_pool_lifecycle.py` | 8 | Create, disable, enable, maintenance, delete + igroups |
 | iSCSI Pool with Volumes | `iscsi/pool/test_pool_with_volumes.py` | 7 | Same + live LUN present; negative delete guard |
 | iSCSI Zone-Scoped Pool | `iscsi/pool/test_zone_scoped_pool.py` | 4 | Zone scope |
 | iSCSI Volume Lifecycle | `iscsi/volume/test_volume_lifecycle.py` | 5 | LUN created per CS volume; LUN removed on delete |
-| iSCSI VM + Volume Attach | `iscsi/instance/test_vm_volume_attach.py` | 10 | Full VM lifecycle; LUN-maps on VM start/stop/detach; ROOT on tagged pool seeds/reuses `cs_tmpl_*` LUN cache |
+| iSCSI VM + Volume Attach | `iscsi/instance/test_vm_volume_attach.py` | 26 | Full VM lifecycle; LUN-maps on VM start/stop/detach; guest SSH, volume and VM snapshots; ROOT on tagged pool seeds/reuses `cs_tmpl_*` LUN cache |
 | iSCSI Template Cache Negative | `iscsi/template/test_template_cache_negative.py` | 3 | Tag mismatch; undersized pool; out-of-band cache delete |
 
 For the goal, dependencies, and exact success criteria of every individual test, see [TEST_CASES.md](TEST_CASES.md).
@@ -351,12 +386,13 @@ For the goal, dependencies, and exact success criteria of every individual test,
 
 | Symptom | Likely cause | Fix |
 |---------|-------------|-----|
+| `paramiko … has no SHA-1 key exchange` / guest SSH fails on CentOS 5.5 | Paramiko 4.x is installed; the guest sshd only offers SHA-1 | Re-run `bash test/integration/plugins/ontap/setup_env.sh` and run tests with `test/integration/plugins/ontap/.venv` (`paramiko>=3.4,<4`) |
 | `ModuleNotFoundError: No module named 'ontap_test_base'` | Missing `PYTHONPATH` prefix | Prefix every run with `PYTHONPATH=test/integration/plugins/ontap` |
 | `Marvin Init Failed` | CloudStack API unreachable | Check `mgtSvrIp:8096` is reachable; restart `cloudstack-management` |
 | `Lost connection to MySQL` | MySQL not accepting remote connections | Enable remote MySQL access (see Prerequisites §3) |
 | `sh: python: command not found` (repeated) | Marvin internal call — harmless on macOS | Ignore; Marvin Init still succeeds |
 | Pool state never reaches `Maintenance` | KVM agent not responding | Check `cloudstack-agent` on KVM host; verify host is connected in CloudStack UI |
-| iSCSI `test_07` error 530 | KVM guest does not ACK SCSI hot-unplug | Known environment limitation — see TEST_CASES.md Suite 10 note |
+| iSCSI `test_22` error 530 | KVM guest does not ACK SCSI hot-unplug | Detach of the data LUN from a running VM; see TEST_CASES.md Suite 10 `test_22_detach_volume_from_vm` |
 | ONTAP REST `401 Unauthorized` | Wrong credentials in `ontap.cfg` | Verify `username`/`password` under `ontap` section |
 | `ISCSI protocol is not enabled on SVM` / many pool tests EXCEPTION then FAIL | SVM iSCSI service off or no `data_iscsi` LIF | Enable iSCSI on the SVM and assign `default-data-iscsi` (or `default-data-blocks`) to at least one data LIF; `run_tests.sh` now fails this check before Marvin starts |
 | `No ready KVM user template available` | Template still downloading | Re-run `setup_zone` (step 12 waits for template readiness); or wait in CloudStack UI |
diff --git a/test/integration/plugins/ontap/TEST_CASES.md b/test/integration/plugins/ontap/TEST_CASES.md
index d3f2b1be58d1..665ccf41ea69 100644
--- a/test/integration/plugins/ontap/TEST_CASES.md
+++ b/test/integration/plugins/ontap/TEST_CASES.md
@@ -19,7 +19,7 @@
 
 # ONTAP Integration Test Cases
 
-Complete reference for all 72 test cases across 12 test suites.
+Complete reference for all 106 test cases across 12 test suites.
 Each suite is sequential — tests must run in numbered order; each step builds on state created by the previous step.
 
 ---
@@ -114,7 +114,7 @@ Each suite is sequential — tests must run in numbered order; each step builds
 **File:** `nfs3/instance/test_vm_volume_attach.py`
 **Class:** `TestOntapVMVolumeAttach`
 **Tag:** `vm_volume_workflow`
-**Total:** 10 tests | **Scope:** end-to-end — NFS3 pool, data volume, running VM (ROOT on the ONTAP pool via a tagged compute offering), primary template cache seed / reuse / survive VM delete, attach/detach lifecycle
+**Total:** 26 tests | **Scope:** end-to-end — NFS3 pool, data volume, running VM (ROOT on the ONTAP pool via a tagged compute offering), primary template cache seed / reuse / survive VM delete, attach/detach, guest SSH, volume snapshots, and VM snapshots. Guest login uses `guest_ssh.py` (paramiko) against the CentOS 5.5 template.
 
 | # | Test method | Goal | Depends on | CloudStack success criteria | ONTAP success criteria | Type |
 |---|-------------|------|------------|-----------------------------|------------------------|------|
@@ -126,8 +126,24 @@ Each suite is sequential — tests must run in numbered order; each step builds
 | 04 | `test_04_attach_volume_to_vm` | Attach the ONTAP data volume to the running VM (hot-plug) | test_03 (`vm`, `volume`) | `volume.virtualmachineid == vm.id`; `attachVolume` job succeeds | FlexVol `online`; after attach, a data file matching volume UUID present in FlexVol (`list_files_in_volume`) | positive |
 | 05 | `test_05_stop_vm_export_retained` | Stop the running VM with volume attached | test_04 | `vm.state == "Stopped"` | FlexVol still `online`; NFS export policy still present | positive |
 | 06 | `test_06_start_vm_volume_accessible` | Start the stopped VM | test_05 | `vm.state == "Running"` | FlexVol still `online` | positive |
-| 07 | `test_07_detach_volume_from_vm` | Hot-detach the ONTAP volume from the running VM (TDS Detach NFS3) | test_06 (`vm`, `volume`) | `volume.virtualmachineid` cleared; `volume.state == "Ready"` | FlexVol still `online`; data file **still present** (NFS3: file persists until `deleteVolume`, not on detach) | positive |
-| 08 | `test_08_destroy_vm_and_cleanup` | Destroy VM (expunge), delete volume, enter maintenance, force-delete pool | test_07 | VM no longer listed; spool_ref still Ready after VM expunge; volume no longer listed; pool no longer listed | Cache file present after VM expunge; FlexVol deleted; export policy deleted | cleanup |
+| 07 | `test_07_open_ssh_and_write_root_data_files` | Port-forward SSH to the guest and write ROOT and DATA marker files | test_06 (`vm`, `volume`, guest network) | ROOT and DATA both on the ONTAP pool; DATA still attached; SSH login succeeds; both markers read back | FlexVol still `online` | positive |
+| 08 | `test_08_root_volume_snapshot_create` | VS-01: snapshot the ROOT volume of the running VM | test_07 | Snapshot `BackedUp`; VM stays `Running` | One new FlexVol snapshot | positive |
+| 09 | `test_09_data_volume_snapshot_create` | VS-03: snapshot the attached DATA volume | test_08 | Snapshot `BackedUp`; VM stays `Running` | One new FlexVol snapshot | positive |
+| 10 | `test_10_volume_revert_running_vm_rejected` | VS-19: revert a volume snapshot while the VM is Running | test_09 | Revert is rejected | FlexVol unchanged | negative |
+| 11 | `test_11_root_volume_revert` | VS-11: write A, snapshot ROOT, write B, stop, revert, start | test_10 | After revert and start, ROOT marker is A | New FlexVol snapshot for the snapshot; revert snapshot removed in tearDown | positive |
+| 12 | `test_12_data_volume_revert` | Write A, snapshot DATA, write B, stop, revert, start | test_11 | After revert and start, DATA marker is A | Same as test_11 for the DATA volume | positive |
+| 13 | `test_13_vm_snapshot_create_running` | VMS-01: VM snapshot of the running VM (kept for test_15) | test_12 | VM snapshot `Ready`; VM stays `Running` | One new FlexVol snapshot | positive |
+| 14 | `test_14_vm_snapshot_create_stopped` | VMS-03: stop, VM snapshot, start (kept for test_15) | test_13 | VM snapshot `Ready` | One new FlexVol snapshot | positive |
+| 15 | `test_15_vm_snapshot_s1_running_s2_stopped_revert` | VMS-09: revert to the test_13 snapshot, then to the test_14 snapshot | test_13, test_14 | Each revert finishes; guest file matches that snapshot | FlexVol snapshots for both VM snapshots until tearDown | positive |
+| 16 | `test_16_vm_snapshot_duplicate_requests` | VMS-53: two VM snapshot requests for the same VM at once | test_15 | Both snapshots `Ready` (a rejected request is retried) | Two new FlexVol snapshots | positive |
+| 17 | `test_17_deploy_second_vm` | Deploy VM-B whose ROOT disk is on the same ONTAP pool | test_16 | VM-B `Running`; ROOT `storageid` = pool | ROOT file present on the shared FlexVol | positive |
+| 18 | `test_18_volume_revert_leaves_other_vm_untouched` | VS-28: revert VM-A's ROOT | test_17 | VM-A's ROOT marker is restored; VM-B's file is unchanged | FlexVol stays `online` | positive |
+| 19 | `test_19_concurrent_vm_snapshots_two_vms` | VMS-54: VM snapshots of both VMs at the same time | test_18 | Both VM snapshots `Ready` | Two new FlexVol snapshots | positive |
+| 20 | `test_20_concurrent_volume_snapshot_and_revert` | Volume snapshot of VM-A's ROOT while VM-B's ROOT reverts | test_19 | Snapshot `BackedUp` and revert both finish (retried on error 530) | New FlexVol snapshot for the volume snapshot | positive |
+| 21 | `test_21_concurrent_vm_snapshot_and_revert` | VM snapshot of VM-A while VM-B reverts to its VM snapshot | test_20 | Both operations finish; VM-A snapshot `Ready` | New FlexVol snapshot for VM-A | positive |
+| 22 | `test_22_detach_volume_from_vm` | Hot-detach the ONTAP volume from the running VM (TDS Detach NFS3) | test_21 (`vm`, `volume`) | `volume.virtualmachineid` cleared; `volume.state == "Ready"` | FlexVol still `online`; data file **still present** (NFS3: file persists until `deleteVolume`, not on detach) | positive |
+| 23 | `test_23_destroy_vm_with_vm_snapshot` | VMS-38: expunge VM-B while it still has a VM snapshot | test_21 (VM-B snapshot) | VM-B no longer listed | VM-B's FlexVol snapshot removed | positive |
+| 24 | `test_24_destroy_vm_and_cleanup` | Destroy VM (expunge), delete volume, enter maintenance, force-delete pool | test_22, test_23 | VM no longer listed; spool_ref still Ready after VM expunge; volume no longer listed; pool no longer listed | Cache file present after VM expunge; FlexVol deleted; export policy deleted | cleanup |
 
 ---
 
@@ -208,7 +224,7 @@ Each suite is sequential — tests must run in numbered order; each step builds
 **File:** `iscsi/instance/test_vm_volume_attach.py`
 **Class:** `TestOntapVMVolumeAttachISCSI`
 **Tag:** `iscsi_vm_workflow`
-**Total:** 10 tests | **Scope:** end-to-end — iSCSI pool, data volume (LUN), running VM (ROOT on the ONTAP pool via a tagged compute offering), `cs_tmpl_<templateId>` LUN cache seed / reuse / survive VM delete, attach/stop/start/detach lifecycle
+**Total:** 26 tests | **Scope:** end-to-end — iSCSI pool, data volume (LUN), running VM (ROOT on the ONTAP pool via a tagged compute offering), `cs_tmpl_<templateId>` LUN cache seed / reuse / survive VM delete, attach/stop/start/detach, guest SSH, volume snapshots, and VM snapshots. LUN-map checks are limited to the data volume's LUN. Guest login uses `guest_ssh.py` (paramiko) against the CentOS 5.5 template.
 
 | # | Test method | Goal | Depends on | CloudStack success criteria | ONTAP success criteria | Type |
 |---|-------------|------|------------|-----------------------------|------------------------|------|
@@ -218,12 +234,26 @@ Each suite is sequential — tests must run in numbered order; each step builds
 | 03a | `test_03a_deploy_second_vm_reuses_template_cache` | Deploy VM-2 — reuse cache | test_03 | VM-2 Running; ROOT on pool; still one spool_ref | Still one `cs_tmpl_*`; non-cache LUN count +1 | positive |
 | 03b | `test_03b_expunge_second_vm_template_cache_survives` | Expunge VM-2 — cache LUN remains | test_03a | spool_ref still Ready | VM-2 ROOT LUN gone (non-cache count back to baseline); `cs_tmpl_*` still present | positive |
 | 04 | `test_04_attach_volume_to_vm` | Hot-attach the ONTAP iSCSI volume to the running VM — a LUN-map is created (TDS SN 27) | test_03 (`vm`, `volume`) | `volume.virtualmachineid == vm.id` | ≥1 LUN-map linking the LUN to the host's igroup | positive |
-| 05 | `test_05_stop_vm_lun_unmapped` | Stop VM — LUN-maps must be removed (TDS VM Stop iSCSI) | test_04 | `vm.state == "Stopped"` | 0 LUN-maps; LUN itself **still present** in FlexVol | positive |
-| 06 | `test_06_start_vm_lun_remapped` | Start VM — LUN-maps must be re-created (TDS VM Start iSCSI) | test_05 | `vm.state == "Running"` | ≥1 LUN-map re-created | positive |
-| 07 | `test_07_detach_volume_from_vm` | Hot-detach the iSCSI volume from the running VM (TDS Detach iSCSI) | test_06 (`vm`, `volume`) | `volume.virtualmachineid` cleared | 0 LUN-maps; LUN still in FlexVol | positive ⚠️ |
-| 08 | `test_08_destroy_vm_and_cleanup` | Destroy VM (expunge), delete volume, enter maintenance, delete pool | test_07 | VM gone; spool_ref still Ready after VM expunge; volume gone; pool gone | `cs_tmpl_*` present after VM expunge; FlexVol deleted; all LUNs and igroups deleted | cleanup |
-
-> ⚠️ **test_07 known status:** iSCSI hot-detach from a running VM relies on the KVM guest acknowledging the SCSI device removal. On this environment the guest does not acknowledge in time, causing CloudStack error 530. This is a KVM-host-level or guest-template limitation, not a test code defect. All other 61 tests pass.
+| 05 | `test_05_stop_vm_lun_unmapped` | Stop VM — LUN-maps must be removed (TDS VM Stop iSCSI) | test_04 | `vm.state == "Stopped"` | 0 LUN-maps for the data LUN; LUN itself **still present** in FlexVol | positive |
+| 06 | `test_06_start_vm_lun_remapped` | Start VM — LUN-maps must be re-created (TDS VM Start iSCSI) | test_05 | `vm.state == "Running"` | ≥1 LUN-map re-created for the data LUN | positive |
+| 07 | `test_07_open_ssh_and_write_root_data_files` | Port-forward SSH to the guest and write ROOT and DATA marker files | test_06 (`vm`, `volume`, guest network) | ROOT and DATA both on the ONTAP pool; DATA still attached; SSH login succeeds; both markers read back | Data LUN still mapped | positive |
+| 08 | `test_08_root_volume_snapshot_create` | VS-01: snapshot the ROOT volume of the running VM | test_07 | Snapshot `BackedUp`; VM stays `Running` | One new FlexVol snapshot | positive |
+| 09 | `test_09_data_volume_snapshot_create` | VS-03: snapshot the attached DATA volume | test_08 | Snapshot `BackedUp`; VM stays `Running` | One new FlexVol snapshot | positive |
+| 10 | `test_10_volume_revert_running_vm_rejected` | VS-20: revert a volume snapshot while the VM is Running | test_09 | Revert is rejected | FlexVol unchanged | negative |
+| 11 | `test_11_root_volume_revert` | VS-12: write A, snapshot ROOT, write B, stop, revert, start | test_10 | After revert and start, ROOT marker is A | New FlexVol snapshot for the snapshot; revert snapshot removed in tearDown | positive |
+| 12 | `test_12_data_volume_revert` | Write A, snapshot DATA, write B, stop, revert, start | test_11 | After revert and start, DATA marker is A | Same as test_11 for the DATA LUN | positive |
+| 13 | `test_13_vm_snapshot_create_running` | VM snapshot of the running VM, ROOT + DATA LUNs (kept for test_15) | test_12 | VM snapshot `Ready`; VM stays `Running` | One new FlexVol snapshot | positive |
+| 14 | `test_14_vm_snapshot_create_stopped` | Stop, VM snapshot, start (kept for test_15) | test_13 | VM snapshot `Ready` | One new FlexVol snapshot | positive |
+| 15 | `test_15_vm_snapshot_s1_running_s2_stopped_revert` | VMS-09: revert to the test_13 snapshot, then to the test_14 snapshot | test_13, test_14 | Each revert finishes; guest file matches that snapshot | FlexVol snapshots for both VM snapshots until tearDown | positive |
+| 16 | `test_16_vm_snapshot_duplicate_requests` | VMS-53: two VM snapshot requests for the same VM at once | test_15 | Both snapshots `Ready` (a rejected request is retried) | Two new FlexVol snapshots | positive |
+| 17 | `test_17_deploy_second_vm` | Deploy VM-B whose ROOT LUN is on the same ONTAP pool | test_16 | VM-B `Running`; ROOT on pool | One more non-cache LUN; `cs_tmpl_*` still one | positive |
+| 18 | `test_18_volume_revert_leaves_other_vm_untouched` | VS-28: revert VM-A's ROOT | test_17 | VM-A's ROOT marker is restored; VM-B's file is unchanged | FlexVol stays `online`; both ROOT LUNs remain | positive |
+| 19 | `test_19_concurrent_vm_snapshots_two_vms` | VMS-55: VM snapshots of both VMs at the same time | test_18 | Both VM snapshots `Ready` | Two new FlexVol snapshots | positive |
+| 20 | `test_20_concurrent_volume_snapshot_and_revert` | Volume snapshot of VM-A's ROOT while VM-B's ROOT reverts | test_19 | Snapshot `BackedUp` and revert both finish (retried on error 530) | New FlexVol snapshot for the volume snapshot | positive |
+| 21 | `test_21_concurrent_vm_snapshot_and_revert` | VM snapshot of VM-A while VM-B reverts to its VM snapshot | test_20 | Both operations finish; VM-A snapshot `Ready` | New FlexVol snapshot for VM-A | positive |
+| 22 | `test_22_detach_volume_from_vm` | Hot-detach the iSCSI volume from the running VM (TDS Detach iSCSI) | test_21 (`vm`, `volume`) | `volume.virtualmachineid` cleared | 0 LUN-maps for the data LUN; LUN still in FlexVol | positive |
+| 23 | `test_23_destroy_vm_with_vm_snapshot` | Expunge VM-B while it still has a VM snapshot | test_21 (VM-B snapshot) | VM-B no longer listed | VM-B ROOT LUN gone; `cs_tmpl_*` still present | positive |
+| 24 | `test_24_destroy_vm_and_cleanup` | Destroy VM (expunge), delete volume, enter maintenance, delete pool | test_22, test_23 | VM gone; spool_ref still Ready after VM expunge; volume gone; pool gone | `cs_tmpl_*` present after VM expunge; FlexVol deleted; all LUNs and igroups deleted | cleanup |
 
 ---
 
@@ -265,12 +295,12 @@ Each suite is sequential — tests must run in numbered order; each step builds
 | NFS3 Pool with Volumes | NFS3 | Cluster | 7 | ✅ |
 | NFS3 Zone-Scoped Pool | NFS3 | Zone | 4 | ✅ |
 | NFS3 Volume Lifecycle | NFS3 | Cluster | 5 | ✅ |
-| NFS3 VM + Volume Attach | NFS3 | Cluster | 10 | 🆕 +2 template cache |
+| NFS3 VM + Volume Attach | NFS3 | Cluster | 26 | ✅ includes guest SSH, volume snapshots, VM snapshots |
 | NFS3 Template Cache Negative | NFS3 | Cluster | 3 | 🆕 |
 | iSCSI Pool Lifecycle | iSCSI | Cluster | 8 | ✅ |
 | iSCSI Pool with Volumes | iSCSI | Cluster | 7 | ✅ |
 | iSCSI Zone-Scoped Pool | iSCSI | Zone | 4 | ✅ |
 | iSCSI Volume Lifecycle | iSCSI | Cluster | 5 | ✅ |
-| iSCSI VM + Volume Attach | iSCSI | Cluster | 10 | ⚠️ 7/8 + 🆕 2 template cache |
+| iSCSI VM + Volume Attach | iSCSI | Cluster | 26 | ✅ includes guest SSH, volume snapshots, VM snapshots |
 | iSCSI Template Cache Negative | iSCSI | Cluster | 3 | 🆕 |
-| **Total** | | | **72** | |
+| **Total** | | | **106** | |
diff --git a/test/integration/plugins/ontap/guest_ssh.py b/test/integration/plugins/ontap/guest_ssh.py
new file mode 100644
index 000000000000..749e918214e8
--- /dev/null
+++ b/test/integration/plugins/ontap/guest_ssh.py
@@ -0,0 +1,216 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+SSH access to the guest VMs deployed by the ONTAP VM workflow tests
+(``*/instance/test_vm_volume_attach.py``).
+
+The template is CentOS 5.5; its login comes from ontap.cfg
+(``cloudstack.templateUsername`` / ``cloudstack.templatePassword``, default
+root/password).  Guests sit on an isolated network, so each VM gets a
+port-forwarding rule (public port -> 22) on the network's source-NAT IP.
+Markers are plain files written and read with paramiko.
+"""
+
+import logging
+import random
+import time
+import uuid
+
+import paramiko
+
+from marvin.cloudstackAPI import (
+    createPortForwardingRule as createPortForwardingRuleAPI,
+    deletePortForwardingRule as deletePortForwardingRuleAPI,
+    listPublicIpAddresses as listPublicIpAddressesAPI,
+)
+
+logger = logging.getLogger("OntapGuestSsh")
+
+ROOT_MARKER = "/root/ontap-marker"
+DATA_MOUNT = "/mnt/data"
+DATA_MARKER = DATA_MOUNT + "/ontap-marker"
+
+_DATA_MOUNT_CMD = (
+    "mkdir -p {m}; "
+    "mountpoint -q {m} 2>/dev/null || grep -q ' {m} ' /proc/mounts || "
+    "mount /dev/vdb {m} 2>/dev/null || mount /dev/sdb {m} 2>/dev/null"
+).format(m=DATA_MOUNT)
+
+_DATA_FORMAT_CMD = (
+    "mkdir -p {m}; "
+    "grep -q ' {m} ' /proc/mounts || "
+    "mount /dev/vdb {m} 2>/dev/null || mount /dev/sdb {m} 2>/dev/null || "
+    "{{ if [ -b /dev/vdb ]; then d=/dev/vdb; else d=/dev/sdb; fi; "
+    "mkfs.ext3 -F -q $d && mount $d {m}; }}"
+).format(m=DATA_MOUNT)
+
+
+def check_paramiko():
+    """Fail fast when paramiko cannot talk to the CentOS 5.5 guest sshd."""
+    if "diffie-hellman-group14-sha1" not in paramiko.Transport._kex_info:
+        raise GuestSshError(
+            "paramiko %s has no SHA-1 key exchange, which the CentOS 5.5 guest "
+            "requires. Run test/integration/plugins/ontap/setup_env.sh and use "
+            "its .venv (it pins paramiko<4)." % paramiko.__version__
+        )
+
+
+def guest_credentials(config):
+    """Return (username, password) for the template from ontap.cfg."""
+    cs_cfg = config.get("cloudstack", {})
+    return (cs_cfg.get("templateUsername", "root"),
+            cs_cfg.get("templatePassword", "password"))
+
+
+def new_token(label="snapshot-test"):
+    """Return readable, unique text to write into a snapshot test file."""
+    readable_label = "".join(
+        char if char.isalnum() or char in ("-", "_") else "-"
+        for char in label
+    ).strip("-")
+    return "%s-%s" % (readable_label or "snapshot-test",
+                      uuid.uuid4().hex[:12])
+
+
+class GuestSshError(Exception):
+    """SSH to the guest or a guest command failed."""
+
+
+class GuestSsh(object):
+    """Run shell commands on a guest as root over SSH."""
+
+    def __init__(self, host, port=22, username="root", password="password",
+                 connect_timeout=600, retry_interval=10):
+        self.host = host
+        self.port = int(port)
+        self.username = username
+        self.password = password
+        self.connect_timeout = connect_timeout
+        self.retry_interval = retry_interval
+        self._ssh = None
+
+    def __enter__(self):
+        self.connect()
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+        return False
+
+    def connect(self):
+        """Connect, retrying until the guest's sshd is up after boot."""
+        check_paramiko()
+        deadline = time.time() + self.connect_timeout
+        last_error = None
+        while time.time() < deadline:
+            ssh = paramiko.SSHClient()
+            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+            try:
+                ssh.connect(self.host, port=self.port, username=self.username,
+                            password=self.password, timeout=15,
+                            allow_agent=False, look_for_keys=False)
+                self._ssh = ssh
+                logger.info("SSH connected to %s:%d", self.host, self.port)
+                return
+            except Exception as exc:
+                last_error = exc
+                ssh.close()
+                time.sleep(self.retry_interval)
+        raise GuestSshError("SSH to %s:%d failed within %ds: %s"
+                            % (self.host, self.port, self.connect_timeout, last_error))
+
+    def close(self):
+        if self._ssh is not None:
+            self._ssh.close()
+            self._ssh = None
+
+    def run(self, command, check=True, timeout=300):
+        """Run a command; return stdout.  Raise on non-zero exit if check."""
+        if self._ssh is None:
+            self.connect()
+        _, stdout, stderr = self._ssh.exec_command(command, timeout=timeout)
+        status = stdout.channel.recv_exit_status()
+        out = stdout.read().decode("utf-8", "replace")
+        err = stderr.read().decode("utf-8", "replace")
+        if check and status != 0:
+            raise GuestSshError("'%s' on %s exited %d: %s"
+                                % (command, self.host, status, err.strip()))
+        return out
+
+    def write_root_marker(self, token):
+        self.run("echo '%s' > %s && sync" % (token, ROOT_MARKER))
+
+    def read_root_marker(self):
+        return self.run("cat %s 2>/dev/null; true" % ROOT_MARKER).strip()
+
+    def write_data_marker(self, token):
+        self.run(_DATA_FORMAT_CMD)
+        self.run("echo '%s' > %s && sync" % (token, DATA_MARKER))
+
+    def read_data_marker(self):
+        self.run(_DATA_MOUNT_CMD, check=False)
+        return self.run("cat %s 2>/dev/null; true" % DATA_MARKER).strip()
+
+
+# ---------------------------------------------------------------------------
+# Port forwarding
+# ---------------------------------------------------------------------------
+
+def source_nat_ip(apiclient, network_id):
+    """Return the source-NAT public IP record for an isolated network."""
+    cmd = listPublicIpAddressesAPI.listPublicIpAddressesCmd()
+    cmd.associatednetworkid = network_id
+    cmd.issourcenat = True
+    cmd.listall = True
+    ips = apiclient.listPublicIpAddresses(cmd) or []
+    if not ips:
+        raise GuestSshError("No source-NAT IP found for network %s" % network_id)
+    return ips[0]
+
+
+def open_ssh_port(apiclient, vm_id, network_id, public_port=None):
+    """Forward a public port on the source-NAT IP to the VM's port 22.
+
+    Returns (public_ip, public_port, rule_id).
+    """
+    ip = source_nat_ip(apiclient, network_id)
+    last_error = None
+    for _ in range(10):
+        port = public_port or random.randint(20000, 29999)
+        cmd = createPortForwardingRuleAPI.createPortForwardingRuleCmd()
+        cmd.ipaddressid = ip.id
+        cmd.privateport = 22
+        cmd.publicport = port
+        cmd.protocol = "TCP"
+        cmd.virtualmachineid = vm_id
+        cmd.openfirewall = True
+        try:
+            rule = apiclient.createPortForwardingRule(cmd)
+            return ip.ipaddress, port, rule.id
+        except Exception as exc:
+            last_error = exc
+            if public_port:
+                break
+    raise GuestSshError("Could not create SSH port forward for VM %s: %s"
+                        % (vm_id, last_error))
+
+
+def close_ssh_port(apiclient, rule_id):
+    cmd = deletePortForwardingRuleAPI.deletePortForwardingRuleCmd()
+    cmd.id = rule_id
+    apiclient.deletePortForwardingRule(cmd)
diff --git a/test/integration/plugins/ontap/iscsi/instance/test_vm_volume_attach.py b/test/integration/plugins/ontap/iscsi/instance/test_vm_volume_attach.py
index 2e247c5e2f87..8797701bfd32 100644
--- a/test/integration/plugins/ontap/iscsi/instance/test_vm_volume_attach.py
+++ b/test/integration/plugins/ontap/iscsi/instance/test_vm_volume_attach.py
@@ -35,15 +35,15 @@
   - detachVolume → LUN-map is removed
 
 Tests must run in order (test_01, test_02, test_03, test_03a, test_03b,
-test_04 ... test_08).  Each step builds on the shared state established by the
+test_04 ... test_24).  Each step builds on the shared state established by the
 previous step.
 
 The VM is deployed with a compute offering whose storage tags match the pool,
 so its ROOT volume lands on the ONTAP pool and exercises the primary
 template cache (``cs_tmpl_<templateId>`` LUN: seed / reuse / survive VM delete)
-alongside the data-volume workflow.  LUN-map assertions are filtered to the
-data volume's LUN, so the ROOT and cache LUNs in the same FlexVol do not
-affect them.
+alongside the data-volume and snapshot workflow.  LUN-map assertions are
+filtered to the data volume's LUN, so the ROOT and cache LUNs in the same
+FlexVol do not affect them.
 
 Workflow:
   01  Create iSCSI primary storage pool on ONTAP (tagged <storagePoolTags>-tmpl-cache)
@@ -54,8 +54,33 @@
   04  Attach iSCSI data volume to running VM (LUN-map created)
   05  Stop VM — LUN-map for attached volume is removed from ONTAP
   06  Start VM — LUN-map is re-created on ONTAP
-  07  Detach data volume from running VM — LUN-map removed
-  08  Destroy VM (cache LUN still survives), delete data volume, delete pool
+  07  Open an SSH port forward to the VM; write ROOT and DATA files
+  08  Snapshot the ROOT volume of the running VM
+  09  Snapshot the DATA volume of the running VM
+  10  Revert a volume snapshot while the VM is Running — rejected
+  11  ROOT volume revert: write before-snapshot, snapshot, write
+      after-snapshot, revert, file is back to before-snapshot
+  12  DATA volume revert: same write, snapshot, write, revert check
+  13  Write before-s1; VM snapshot of the running VM (kept for step 15)
+  14  Write before-s2, stop, VM snapshot (kept for step 15)
+  15  Revert to the step-13 snapshot, then to the step-14 snapshot
+  16  Two VM snapshot requests for the same VM at once — both succeed
+  17  Deploy a second VM whose ROOT LUN is on the same pool
+  18  Revert VM-A's ROOT; VM-B's file is unchanged
+  19  VM snapshots of both VMs at the same time
+  20  Volume snapshot of VM-A while VM-B's ROOT reverts
+  21  VM snapshot of VM-A while VM-B reverts to its VM snapshot
+  22  Detach data volume from running VM — LUN-map removed
+  23  Destroy the second VM while it still has a VM snapshot
+  24  Destroy VM (cache LUN still survives), delete data volume, delete pool
+
+Snapshot steps only create and verify; tearDown deletes every snapshot the
+step created and waits for its FlexVol snapshot to go.  Steps 13 and 14 keep
+their VM snapshots so step 15 can revert to them; step 15's tearDown deletes
+both.  Step 23 is last so more snapshot cases can be added before it.
+
+Guest login for the CentOS 5.5 template comes from ontap.cfg
+(cloudstack.templateUsername / templatePassword); see guest_ssh.py.
 
 Prerequisites:
   - CloudStack management server with the NetApp ONTAP plugin deployed
@@ -74,6 +99,7 @@
 import logging
 import random
 import re
+import threading
 import time
 import unittest
 
@@ -96,9 +122,10 @@
     startVirtualMachine as startVirtualMachineAPI,
     stopVirtualMachine as stopVirtualMachineAPI,
 )
-from marvin.lib.base import ServiceOffering, StoragePool
+from marvin.lib.base import ServiceOffering, Snapshot, StoragePool, VmSnapshot, Volume
 from marvin.lib.common import list_storage_pools
 
+from guest_ssh import GuestSsh, guest_credentials, new_token, open_ssh_port
 from ontap_test_base import OntapRestClient, OntapTestBase, get_datacenter_config
 from helpers import template_cache_util as tcu
 from helpers.template_cache_base import (
@@ -110,6 +137,8 @@
 
 logger = logging.getLogger("TestOntapVMVolumeAttachISCSI")
 
+TAG = "iscsi_vm_workflow"
+
 
 # ---------------------------------------------------------------------------
 # Utility functions
@@ -134,6 +163,83 @@ def _wait_for_vm_state(api_client, vm_id, target_state, timeout=300,
     return None
 
 
+def _run_concurrent(*callables):
+    """Start all callables at the same moment; return [(result, exception)]."""
+    results = [(None, None)] * len(callables)
+    barrier = threading.Barrier(len(callables))
+
+    def worker(index, fn):
+        try:
+            barrier.wait(timeout=60)
+            results[index] = (fn(), None)
+        except Exception as exc:
+            results[index] = (None, exc)
+
+    threads = [threading.Thread(target=worker, args=(i, fn))
+               for i, fn in enumerate(callables)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    return results
+
+
+def _is_rejected(exc):
+    """True when CloudStack or ONTAP refused the call because another one was in progress."""
+    text = str(exc).lower()
+    return any(token in text for token in (
+        "internal error",
+        "failed to create snapshot",
+        "failed to create instance snapshot",
+        "failed to revert",
+        "another operation",
+        "other active",
+        "try again later",
+        "in progress",
+        "busy",
+    ))
+
+
+def _snapshot_and_revert_with_retry(snapshot_fn, revert_fn, attempts=3, pause=15):
+    """Run a snapshot and a revert together.
+
+    A call that is rejected is retried on its own. A call that already
+    succeeded is not repeated. Returns the snapshot from snapshot_fn.
+    """
+    snap = None
+    reverted = False
+    last_errors = []
+    for attempt in range(1, attempts + 1):
+        jobs = []
+        if snap is None:
+            jobs.append(("snapshot", snapshot_fn))
+        if not reverted:
+            jobs.append(("revert", revert_fn))
+        results = _run_concurrent(*(fn for _, fn in jobs))
+        last_errors = []
+        retryable = True
+        for (kind, _), (result, exc) in zip(jobs, results):
+            if exc is None:
+                if kind == "snapshot":
+                    snap = result
+                else:
+                    reverted = True
+                continue
+            last_errors.append("%s: %s" % (kind, exc))
+            if not _is_rejected(exc):
+                retryable = False
+        if snap is not None and reverted:
+            return snap
+        if not retryable or attempt == attempts:
+            break
+        logger.warning("snapshot/revert rejected, retry %d/%d: %s",
+                       attempt, attempts, last_errors)
+        time.sleep(pause)
+    raise AssertionError(
+        "Snapshot-while-revert failed after %d attempt(s): %s"
+        % (attempt, last_errors))
+
+
 # ---------------------------------------------------------------------------
 # Test data
 # ---------------------------------------------------------------------------
@@ -153,12 +259,14 @@ class TestData:
     DETAIL_STORAGE_IP = "storageIP"
 
     ONTAP_MIN_VOLUME_SIZE = 1677721600
+    # ROOT + DATA LUNs of two VMs plus FlexVol snapshot reserve
+    DEFAULT_POOL_SIZE = 50 * 1024 ** 3
 
     def __init__(self, storage_ip, svm_name, username, password,
                  scope="CLUSTER", provider="NetApp ONTAP",
                  tags="ontap-iscsi", capacitybytes=None):
         if capacitybytes is None:
-            capacitybytes = TestData.ONTAP_MIN_VOLUME_SIZE * 2
+            capacitybytes = TestData.DEFAULT_POOL_SIZE
         encoded_password = base64.b64encode(password.encode()).decode()
         self.testdata = {
             TestData.ontap: {
@@ -215,6 +323,15 @@ class TestOntapVMVolumeAttachISCSI(OntapTestBase):
     service_offering_id = None
     network_id = None
     _created_network_id = None  # network created by this suite for Advanced zones
+    vm_ssh = None               # (public_ip, public_port) forwarded to vm:22
+    vm2_ssh = None
+    vm_snap_running = None      # step 13 snapshot, reused by step 15
+    vm_snap_stopped = None      # step 14 snapshot, reused by step 15
+    token_s1 = None
+    token_s2 = None
+    kept_ontap_snaps = None
+    guest_user = "root"
+    guest_password = "password"
 
     _vol_name_prefix = "OntapISCSIVM"
 
@@ -260,6 +377,8 @@ def setUpClass(cls):
 
         cls._setup_cloudstack_resources(config, cls.testdata[TestData.account])
 
+        cls.guest_user, cls.guest_password = guest_credentials(config)
+
         # Discover a ready user KVM template (exclude SYSTEM type)
         template = find_ready_kvm_template(cls.apiClient, cls.zone.id)
         cls.template_id = template.id if template is not None else None
@@ -277,7 +396,6 @@ def setUpClass(cls):
         )
         cls._cleanup.append(service_offering)
         cls.service_offering_id = service_offering.id
-
         # Network ID for VM deployment
         cls.network_id = None
         zone_type = getattr(cls.zone, "networktype", "Basic")
@@ -320,9 +438,10 @@ def setUpClass(cls):
     def tearDownClass(cls):
         """
         Safety-net cleanup: destroy VMs if still alive, delete the guest
-        network created for Advanced zones (if not already deleted by test_08),
+        network created for Advanced zones (if not already deleted by test_24),
         then delegate pool/volume/account cleanup to the base class.
         """
+        cls._delete_kept_vm_snapshots()
         for vm in (cls.vm2, cls.vm):
             if vm is None:
                 continue
@@ -343,7 +462,7 @@ def tearDownClass(cls):
         cls.vm = None
 
         # Delete the guest network created for this account in Advanced zones.
-        # test_08 deletes it on the happy path; this is the fallback for
+        # test_24 deletes it on the happy path; this is the fallback for
         # mid-suite failures.
         if cls._created_network_id is not None:
             try:
@@ -456,6 +575,256 @@ def _data_luns(self, volume=None):
                     self.svm_name, self.__class__.pool.name)
                 if lun.get("name") == lun_name]
 
+    # ---- snapshot step helpers (steps 07-21, 23) ---------------------------
+
+    def setUp(self):
+        super(TestOntapVMVolumeAttachISCSI, self).setUp()
+        self._volume_snapshot_ids = []
+        self._vm_snapshot_ids = []
+        self._extra_volume_ids = []
+        self._created_ontap_snaps = set()
+
+    def tearDown(self):
+        """
+        Delete the snapshots and extra volumes a step created, then check
+        their FlexVol snapshots are gone from ONTAP.
+        """
+        for snap_id in self._vm_snapshot_ids:
+            try:
+                VmSnapshot.deleteVMSnapshot(self.apiClient, snap_id)
+            except Exception as exc:
+                logger.warning("tearDown: VM snapshot %s: %s" % (snap_id, exc))
+        for snap_id in self._volume_snapshot_ids:
+            try:
+                Snapshot({"id": snap_id}).delete(self.apiClient)
+            except Exception as exc:
+                logger.warning("tearDown: volume snapshot %s: %s" % (snap_id, exc))
+        for vol_id in self._extra_volume_ids:
+            try:
+                cmd = deleteVolumeAPI.deleteVolumeCmd()
+                cmd.id = vol_id
+                self.apiClient.deleteVolume(cmd)
+            except Exception as exc:
+                logger.warning("tearDown: volume %s: %s" % (vol_id, exc))
+        if self._created_ontap_snaps and not self._wait_ontap_snaps_gone(
+                self._created_ontap_snaps, fail=False):
+            logger.warning("tearDown: FlexVol snapshots still present: %s"
+                           % sorted(self._created_ontap_snaps))
+        super(TestOntapVMVolumeAttachISCSI, self).tearDown()
+
+    @classmethod
+    def _destroy_second_vm(cls):
+        if cls.vm2 is None:
+            return
+        try:
+            cmd = destroyVirtualMachineAPI.destroyVirtualMachineCmd()
+            cmd.id = cls.vm2.id
+            cmd.expunge = True
+            cls.apiClient.destroyVirtualMachine(cmd)
+        except Exception as exc:
+            logger.warning("Could not destroy second VM %s: %s" % (cls.vm2.id, exc))
+        cls.vm2 = None
+        cls.vm2_ssh = None
+
+    @classmethod
+    def _delete_kept_vm_snapshots(cls):
+        """Delete the step 13/14 snapshots when step 15 did not take them over."""
+        for snap in (cls.vm_snap_running, cls.vm_snap_stopped):
+            if snap is None:
+                continue
+            try:
+                VmSnapshot.deleteVMSnapshot(cls.apiClient, snap.id)
+            except Exception as exc:
+                logger.warning("tearDownClass: VM snapshot %s: %s" % (snap.id, exc))
+        cls.vm_snap_running = None
+        cls.vm_snap_stopped = None
+        cls.token_s1 = None
+        cls.token_s2 = None
+        cls.kept_ontap_snaps = set()
+
+    def _keep_vm_snapshot(self, snap, ontap_names, which):
+        """Leave this VM snapshot in place for step 15."""
+        if snap.id in self._vm_snapshot_ids:
+            self._vm_snapshot_ids.remove(snap.id)
+        names = set(ontap_names)
+        self._created_ontap_snaps -= names
+        kept = self.__class__.kept_ontap_snaps
+        if kept is None:
+            kept = set()
+            self.__class__.kept_ontap_snaps = kept
+        kept.update(names)
+        setattr(self.__class__, which, snap)
+
+    def _adopt_kept_vm_snapshots(self):
+        """Hand the step 13 and 14 snapshots to this step's tearDown."""
+        s1 = self.__class__.vm_snap_running
+        s2 = self.__class__.vm_snap_stopped
+        token_a = self.__class__.token_s1
+        token_b = self.__class__.token_s2
+        if s1 is None or s2 is None or not token_a or not token_b:
+            self.skipTest("VM snapshots from test_13 and test_14 are required")
+        ontap_names = set(self.__class__.kept_ontap_snaps or [])
+        self._vm_snapshot_ids.extend([s1.id, s2.id])
+        self._created_ontap_snaps |= ontap_names
+        self.__class__.vm_snap_running = None
+        self.__class__.vm_snap_stopped = None
+        self.__class__.token_s1 = None
+        self.__class__.token_s2 = None
+        self.__class__.kept_ontap_snaps = set()
+        return s1, s2, token_a, token_b, ontap_names
+
+    def _require_snapshot_vm(self, ssh=True):
+        if self.__class__.vm is None:
+            self.skipTest("VM not deployed — test_03 was skipped (no ready template)")
+        if ssh and self.__class__.vm_ssh is None:
+            self.skipTest("Guest SSH not set up — test_07 must pass first")
+        self._start_vm(self.__class__.vm.id)
+        return self.__class__.vm
+
+    def _require_second_vm(self):
+        self._require_snapshot_vm()
+        if self.__class__.vm2 is None or self.__class__.vm2_ssh is None:
+            self.skipTest("Second VM not deployed — test_17 must pass first")
+        self._start_vm(self.__class__.vm2.id)
+        return self.__class__.vm2
+
+    def _vm_state(self, vm_id):
+        vms = self.apiClient.listVirtualMachines(_list_vms_cmd(vm_id))
+        return vms[0].state if vms else None
+
+    def _stop_vm(self, vm_id):
+        if (self._vm_state(vm_id) or "").lower() == "stopped":
+            return
+        cmd = stopVirtualMachineAPI.stopVirtualMachineCmd()
+        cmd.id = vm_id
+        cmd.forced = True
+        self.apiClient.stopVirtualMachine(cmd)
+        self._poll_vm_state(vm_id, "Stopped", timeout=600)
+
+    def _start_vm(self, vm_id):
+        if (self._vm_state(vm_id) or "").lower() == "running":
+            return
+        cmd = startVirtualMachineAPI.startVirtualMachineCmd()
+        cmd.id = vm_id
+        self.apiClient.startVirtualMachine(cmd)
+        self._poll_vm_state(vm_id, "Running", timeout=900)
+
+    def _root_volume(self, vm_id):
+        cmd = listVolumesAPI.listVolumesCmd()
+        cmd.virtualmachineid = vm_id
+        cmd.type = "ROOT"
+        cmd.listall = True
+        vols = self.apiClient.listVolumes(cmd) or []
+        self.assertTrue(vols, "No ROOT volume found for VM %s" % vm_id)
+        return vols[0]
+
+    def _guest(self, which="vm"):
+        host, port = getattr(self.__class__, "%s_ssh" % which)
+        return GuestSsh(host, port, self.guest_user, self.guest_password)
+
+    def _write_marker(self, token, which="vm", disks=("ROOT",)):
+        with self._guest(which) as guest:
+            if "ROOT" in disks:
+                guest.write_root_marker(token)
+            if "DATA" in disks:
+                guest.write_data_marker(token)
+
+    def _assert_marker(self, token, which="vm", disk="ROOT", msg=""):
+        with self._guest(which) as guest:
+            found = guest.read_data_marker() if disk == "DATA" else guest.read_root_marker()
+        self.assertEqual(found, token, "%s: %s marker on %s is %r, expected %r"
+                         % (msg, disk, which, found, token))
+
+    def _ontap_snap_names(self):
+        return {s.get("name") for s in self.ontap.list_snapshots(self.__class__.pool.name)}
+
+    def _assert_new_ontap_snaps(self, before, snaps, kind, msg):
+        """Return the FlexVol snapshot created for each CloudStack snapshot.
+
+        A volume snapshot is stored as ``..._cs<dbId>`` and a VM snapshot as
+        ``..._vm<dbId>``. ONTAP snapshots from its own schedule, such as
+        ``hourly.*`` or ``Anti_ransomware_*``, are ignored.
+        """
+        new = self._ontap_snap_names() - before
+        prefix = "cs" if kind == "volume" else "vm"
+        table = "snapshots" if kind == "volume" else "vm_snapshots"
+        matched = set()
+        missing = []
+        for snap in snaps:
+            db_id = tcu.get_db_id(self.dbConnection, table, snap.id)
+            suffix = "_%s%s" % (prefix, db_id)
+            hits = sorted(name for name in new if name.endswith(suffix))
+            if len(hits) != 1:
+                missing.append("%s (suffix %s, matches %s)" % (snap.id, suffix, hits))
+            else:
+                matched.add(hits[0])
+        self.assertFalse(
+            missing,
+            "%s: FlexVol snapshot missing for %s; new FlexVol snapshots: %s"
+            % (msg, missing, sorted(new)))
+        ignored = new - matched
+        if ignored:
+            logger.info("%s: ignoring FlexVol snapshots not created by this step: %s",
+                        msg, sorted(ignored))
+        self._created_ontap_snaps |= matched
+        return matched
+
+    def _assert_ontap_snaps_present(self, names, msg):
+        missing = set(names) - self._ontap_snap_names()
+        self.assertFalse(missing, "%s: FlexVol snapshots missing: %s" % (msg, sorted(missing)))
+
+    def _wait_ontap_snaps_gone(self, names, timeout=180, fail=True):
+        names = set(names)
+        deadline = time.time() + timeout
+        left = names
+        while time.time() < deadline:
+            left = names & self._ontap_snap_names()
+            if not left:
+                return True
+            time.sleep(5)
+        if fail:
+            self.fail("FlexVol snapshots %s still present after %ds" % (sorted(left), timeout))
+        return False
+
+    def _take_volume_snapshot(self, volume_id):
+        snap = Snapshot.create(self.apiClient, volume_id,
+                               account=self.account.name, domainid=self.domain.id)
+        self._volume_snapshot_ids.append(snap.id)
+        self.assertEqual(snap.state, "BackedUp",
+                         "Volume snapshot %s is %s, expected BackedUp" % (snap.id, snap.state))
+        return snap
+
+    def _take_vm_snapshot(self, vm_id):
+        snap = VmSnapshot.create(self.apiClient, vm_id, snapshotmemory="false",
+                                 name="ontap_%d" % random.randint(0, 999999))
+        self._vm_snapshot_ids.append(snap.id)
+        self.assertEqual(snap.state, "Ready",
+                         "VM snapshot %s is %s, expected Ready" % (snap.id, snap.state))
+        return snap
+
+    def _vm_snapshots_of(self, vm_id):
+        return VmSnapshot.list(self.apiClient, virtualmachineid=vm_id, listall=True) or []
+
+    def _volume_revert_flow(self, disk):
+        """Write A, snapshot, write B, stop, revert, start; expect A."""
+        vm = self._require_snapshot_vm()
+        volume_id = (self.__class__.volume.id if disk == "DATA"
+                     else self._root_volume(vm.id).id)
+        token_a = new_token("before-snapshot")
+        token_b = new_token("after-snapshot")
+        self._write_marker(token_a, disks=(disk,))
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(volume_id)
+        new = self._assert_new_ontap_snaps(before, [snap], "volume", "%s volume snapshot" % disk)
+        self._write_marker(token_b, disks=(disk,))
+
+        self._stop_vm(vm.id)
+        Volume.revertToSnapshot(self.apiClient, snap.id)
+        self._assert_ontap_snaps_present(new, "after revert")
+        self._start_vm(vm.id)
+        self._assert_marker(token_a, disk=disk,
+                            msg="after revert (write after snapshot was %s)" % token_b)
+
     # ==================================================================
     # Test steps
     # ==================================================================
@@ -764,11 +1133,375 @@ def test_06_start_vm_lun_remapped(self):
         )
 
     # ------------------------------------------------------------------
-    # Step 07 — Detach volume from running VM
+    # Step 07 — Guest SSH access for snapshot checks
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_07_open_ssh_and_write_root_data_files(self):
+        """
+        Prepare the running VM for snapshot checks.
+        Verifies:
+          - ROOT and DATA LUNs are both on the ONTAP test pool
+          - SSH login (ontap.cfg templateUsername/templatePassword) works
+            through a port forward on the network's source-NAT IP
+          - The DATA disk can be formatted, mounted and written
+        """
+        vm = self._require_snapshot_vm(ssh=False)
+        if not self.__class__.network_id:
+            self.skipTest("Snapshot steps need an Advanced zone guest network for SSH")
+        pool_id = self.__class__.pool.id
+        root = self._root_volume(vm.id)
+        self.assertEqual(getattr(root, "storageid", None), pool_id,
+                         "ROOT volume is on '%s', not the ONTAP test pool"
+                         % getattr(root, "storage", None))
+        vol_cmd = listVolumesAPI.listVolumesCmd()
+        vol_cmd.id = self.__class__.volume.id
+        vols = self.apiClient.listVolumes(vol_cmd) or []
+        self.assertTrue(vols and getattr(vols[0], "virtualmachineid", None) == vm.id,
+                        "DATA volume must be attached before the snapshot steps")
+
+        host, port, _ = open_ssh_port(self.apiClient, vm.id, self.__class__.network_id)
+        self.__class__.vm_ssh = (host, port)
+        token = new_token("ssh-access-test")
+        self._write_marker(token, disks=("ROOT", "DATA"))
+        self._assert_marker(token, disk="ROOT", msg="guest access")
+        self._assert_marker(token, disk="DATA", msg="guest access")
+
+    # ------------------------------------------------------------------
+    # Steps 08-12 — Volume snapshots
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_08_root_volume_snapshot_create(self):
+        """
+        VS-01: snapshot the ROOT volume of the running VM.
+        Verifies: state BackedUp; VM stays Running; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(self._root_volume(vm.id).id)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        self._assert_new_ontap_snaps(before, [snap], "volume", "ROOT volume snapshot")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_09_data_volume_snapshot_create(self):
+        """
+        VS-03: snapshot the attached DATA volume.
+        Verifies: state BackedUp; VM stays Running; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(self.__class__.volume.id)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        self._assert_new_ontap_snaps(before, [snap], "volume", "DATA volume snapshot")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_10_volume_revert_running_vm_rejected(self):
+        """
+        VS-20: reverting a volume snapshot while the VM is Running fails.
+        Verifies: error says the Instance is not shut down; VM stays Running;
+        the FlexVol snapshot remains.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(self._root_volume(vm.id).id)
+        new = self._assert_new_ontap_snaps(before, [snap], "volume", "ROOT volume snapshot")
+        with self.assertRaises(Exception) as ctx:
+            Volume.revertToSnapshot(self.apiClient, snap.id)
+        self.assertIn("shutdown", str(ctx.exception).lower(),
+                      "Unexpected error: %s" % ctx.exception)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        self._assert_ontap_snaps_present(new, "after rejected revert")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_11_root_volume_revert(self):
+        """
+        VS-12: write A, snapshot ROOT, write B, stop, revert, start.
+        Verifies: ROOT marker is A; the FlexVol snapshot survives the revert.
+        """
+        self._volume_revert_flow("ROOT")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_12_data_volume_revert(self):
+        """
+        Write A, snapshot DATA, write B, stop, revert, start.
+        Verifies: DATA marker is A; the FlexVol snapshot survives the revert.
+        """
+        self._volume_revert_flow("DATA")
+
+    # ------------------------------------------------------------------
+    # Steps 13-16 — VM snapshots
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_13_vm_snapshot_create_running(self):
+        """
+        VM snapshot of the running VM (ROOT + DATA LUNs on one FlexVol).
+        Writes before-s1, then snapshots. The snapshot is kept for step 15.
+        Verifies: state Ready; VM stays Running; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        self.__class__.token_s1 = new_token("before-s1")
+        self._write_marker(self.__class__.token_s1, disks=("ROOT", "DATA"))
+        before = self._ontap_snap_names()
+        snap = self._take_vm_snapshot(vm.id)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        new = self._assert_new_ontap_snaps(before, [snap], "vm", "VM snapshot")
+        self._keep_vm_snapshot(snap, new, "vm_snap_running")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_14_vm_snapshot_create_stopped(self):
+        """
+        VM snapshot of the stopped VM.
+        Writes before-s2 while Running, then stops and snapshots.
+        The snapshot is kept for step 15.
+        Verifies: state Ready; VM stays Stopped; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        self.__class__.token_s2 = new_token("before-s2")
+        self._write_marker(self.__class__.token_s2, disks=("ROOT", "DATA"))
+        self._stop_vm(vm.id)
+        before = self._ontap_snap_names()
+        snap = self._take_vm_snapshot(vm.id)
+        self.assertEqual(self._vm_state(vm.id), "Stopped")
+        new = self._assert_new_ontap_snaps(before, [snap], "vm", "VM snapshot (Stopped)")
+        self._keep_vm_snapshot(snap, new, "vm_snap_stopped")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_15_vm_snapshot_s1_running_s2_stopped_revert(self):
+        """
+        VMS-09 (iSCSI ROOT + iSCSI DATA on the same pool): revert to the
+        running snapshot from step 13, then to the stopped snapshot from
+        step 14.
+        Verifies:
+          - After revert to S1: ROOT and DATA hold before-s1
+          - After revert to S2: ROOT and DATA hold before-s2
+          - Both FlexVol snapshots survive the reverts
+        """
+        if self.__class__.vm is None:
+            self.skipTest("VM not deployed — test_03 was skipped (no ready template)")
+        if self.__class__.vm_ssh is None:
+            self.skipTest("Guest SSH not set up — test_07 must pass first")
+        s1, s2, token_a, token_b, ontap_names = self._adopt_kept_vm_snapshots()
+        vm = self.__class__.vm
+
+        self._stop_vm(vm.id)
+        VmSnapshot.revertToSnapshot(self.apiClient, s1.id)
+        self._start_vm(vm.id)
+        self._assert_marker(token_a, disk="ROOT", msg="after revert to S1")
+        self._assert_marker(token_a, disk="DATA", msg="after revert to S1")
+        self.assertEqual(len(self._vm_snapshots_of(vm.id)), 2, "S2 should still be listed")
+
+        self._stop_vm(vm.id)
+        VmSnapshot.revertToSnapshot(self.apiClient, s2.id)
+        self._start_vm(vm.id)
+        self._assert_marker(token_b, disk="ROOT", msg="after revert to S2")
+        self._assert_marker(token_b, disk="DATA", msg="after revert to S2")
+        self._assert_ontap_snaps_present(ontap_names, "after both reverts")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_16_vm_snapshot_duplicate_requests(self):
+        """
+        VMS-53: two VM snapshot requests for the same VM at the same time.
+        A rejected request is retried on its own; one that already succeeded
+        is not repeated.
+        Verifies: both succeed and are Ready; two new FlexVol snapshots.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        pending = [
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_dup_a_%d" % random.randint(0, 99999)),
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_dup_b_%d" % random.randint(0, 99999)),
+        ]
+        ok = []
+        last_errors = []
+        attempts = 3
+        for attempt in range(1, attempts + 1):
+            results = _run_concurrent(*pending)
+            last_errors = []
+            still_pending = []
+            retryable = True
+            for fn, (result, exc) in zip(pending, results):
+                if exc is None and result is not None:
+                    ok.append(result)
+                    self._vm_snapshot_ids.append(result.id)
+                    continue
+                last_errors.append(exc)
+                if exc is None or not _is_rejected(exc):
+                    retryable = False
+                still_pending.append(fn)
+            pending = still_pending
+            if not pending:
+                break
+            if not retryable or attempt == attempts:
+                break
+            logger.warning("duplicate VM snapshot rejected, retry %d/%d: %s",
+                           attempt, attempts, last_errors)
+            time.sleep(15)
+        self.assertFalse(pending,
+                         "Duplicate VM snapshot requests failed: %s" % last_errors)
+        self.assertEqual(len(ok), 2, "Expected both VM snapshots to succeed; "
+                         "got %d" % len(ok))
+        for snap in ok:
+            self.assertEqual(snap.state, "Ready")
+            self.assertEqual(snap.virtualmachineid, vm.id)
+        self._assert_new_ontap_snaps(before, ok, "vm", "duplicate VM snapshot requests")
+
+    # ------------------------------------------------------------------
+    # Steps 17-21 — Second VM on the same pool: isolation and concurrency
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_17_deploy_second_vm(self):
+        """
+        Deploy a second VM whose ROOT LUN is on the same ONTAP pool.
+        Verifies: VM Running; ROOT on the test pool; SSH works.
+        """
+        self._require_snapshot_vm()
+        cmd = deployVirtualMachineAPI.deployVirtualMachineCmd()
+        cmd.zoneid = self.zone.id
+        cmd.templateid = self.__class__.template_id
+        cmd.serviceofferingid = self.__class__.service_offering_id
+        cmd.account = self.account.name
+        cmd.domainid = self.domain.id
+        cmd.networkids = self.__class__.network_id
+        vm2 = self.apiClient.deployVirtualMachine(cmd)
+        self.__class__.vm2 = vm2
+        self._poll_vm_state(vm2.id, "Running", timeout=900)
+        root = self._root_volume(vm2.id)
+        self.assertEqual(getattr(root, "storageid", None), self.__class__.pool.id,
+                         "Second VM ROOT is not on the ONTAP test pool")
+        host, port, _ = open_ssh_port(self.apiClient, vm2.id, self.__class__.network_id)
+        self.__class__.vm2_ssh = (host, port)
+        token = new_token("second-vm-ssh-access-test")
+        self._write_marker(token, which="vm2")
+        self._assert_marker(token, which="vm2", msg="second VM access")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_18_volume_revert_leaves_other_vm_untouched(self):
+        """
+        VS-28: two VMs on one FlexVol; revert VM-A's ROOT to its snapshot.
+        Verifies: VM-A is back at its snapshot; VM-B keeps data written after it.
+        """
+        vm = self._require_snapshot_vm()
+        self._require_second_vm()
+        a1 = new_token("vm-a-before-snapshot")
+        a2 = new_token("vm-a-after-snapshot")
+        b1 = new_token("vm-b-before-snapshot")
+        b2 = new_token("vm-b-after-snapshot")
+        self._write_marker(a1)
+        self._write_marker(b1, which="vm2")
+        snap = self._take_volume_snapshot(self._root_volume(vm.id).id)
+        self._write_marker(a2)
+        self._write_marker(b2, which="vm2")
+
+        self._stop_vm(vm.id)
+        Volume.revertToSnapshot(self.apiClient, snap.id)
+        self._start_vm(vm.id)
+        self._assert_marker(a1, msg="VM-A after revert")
+        self._assert_marker(b2, which="vm2", msg="VM-B after VM-A's revert")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_19_concurrent_vm_snapshots_two_vms(self):
+        """
+        VMS-55: VM snapshots on two VMs of the same iSCSI pool at the same time.
+        Verifies: both Ready and owned by their VMs; two distinct new FlexVol
+        snapshots.
+        """
+        vm = self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        before = self._ontap_snap_names()
+        results = _run_concurrent(
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_cc_a_%d" % random.randint(0, 99999)),
+            lambda: VmSnapshot.create(self.apiClient, vm2.id, snapshotmemory="false",
+                                      name="ontap_cc_b_%d" % random.randint(0, 99999)),
+        )
+        self._vm_snapshot_ids.extend(s.id for s, _ in results if s is not None)
+        errors = [e for _, e in results if e is not None]
+        self.assertFalse(errors, "Concurrent VM snapshots failed: %s" % errors)
+        snap_a, snap_b = results[0][0], results[1][0]
+        self.assertEqual(snap_a.state, "Ready")
+        self.assertEqual(snap_b.state, "Ready")
+        self.assertEqual(snap_a.virtualmachineid, vm.id)
+        self.assertEqual(snap_b.virtualmachineid, vm2.id)
+        self._assert_new_ontap_snaps(before, [snap_a, snap_b], "vm", "concurrent VM snapshots")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_20_concurrent_volume_snapshot_and_revert(self):
+        """
+        Volume snapshot of VM-A's ROOT while VM-B's ROOT reverts, at the same time.
+        A rejected snapshot or revert is retried on its own.
+        Verifies: VM-A snapshot BackedUp with its own FlexVol snapshot; VM-B is
+        back at its snapshot; VM-A data unchanged.
+        """
+        vm = self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        a1 = new_token("vm-a-current-data")
+        b1 = new_token("vm-b-before-snapshot")
+        b2 = new_token("vm-b-after-snapshot")
+        self._write_marker(a1)
+        self._write_marker(b1, which="vm2")
+        snap_b = self._take_volume_snapshot(self._root_volume(vm2.id).id)
+        self._write_marker(b2, which="vm2")
+        self._stop_vm(vm2.id)
+        before = self._ontap_snap_names()
+        root_a = self._root_volume(vm.id).id
+
+        snap_a = _snapshot_and_revert_with_retry(
+            lambda: Snapshot.create(self.apiClient, root_a,
+                                    account=self.account.name, domainid=self.domain.id),
+            lambda: Volume.revertToSnapshot(self.apiClient, snap_b.id),
+        )
+        self._volume_snapshot_ids.append(snap_a.id)
+        self.assertEqual(snap_a.state, "BackedUp")
+        self._assert_new_ontap_snaps(before, [snap_a], "volume", "VM-A snapshot during VM-B revert")
+
+        self._start_vm(vm2.id)
+        self._assert_marker(b1, which="vm2", msg="VM-B after revert")
+        self._assert_marker(a1, msg="VM-A during VM-B revert")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_21_concurrent_vm_snapshot_and_revert(self):
+        """
+        VM snapshot of VM-A while VM-B reverts to its VM snapshot, at the same time.
+        A rejected snapshot or revert is retried on its own.
+        Verifies: VM-A snapshot Ready with its own FlexVol snapshot; VM-B is
+        back at its snapshot; VM-A data unchanged.
+        """
+        vm = self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        a1 = new_token("vm-a-current-data")
+        b1 = new_token("vm-b-before-snapshot")
+        b2 = new_token("vm-b-after-snapshot")
+        self._write_marker(a1)
+        self._write_marker(b1, which="vm2")
+        snap_b = self._take_vm_snapshot(vm2.id)
+        self._write_marker(b2, which="vm2")
+        self._stop_vm(vm2.id)
+        before = self._ontap_snap_names()
+
+        snap_a = _snapshot_and_revert_with_retry(
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_ccr_%d" % random.randint(0, 99999)),
+            lambda: VmSnapshot.revertToSnapshot(self.apiClient, snap_b.id),
+        )
+        self._vm_snapshot_ids.append(snap_a.id)
+        self.assertEqual(snap_a.state, "Ready")
+        self._assert_new_ontap_snaps(before, [snap_a], "vm", "VM-A snapshot during VM-B revert")
+
+        self._start_vm(vm2.id)
+        self._assert_marker(b1, which="vm2", msg="VM-B after revert")
+        self._assert_marker(a1, msg="VM-A during VM-B revert")
+
+    # ------------------------------------------------------------------
+    # Step 22 — Detach volume from running VM
     # ------------------------------------------------------------------
 
     @attr(tags=["iscsi_vm_workflow"], required_hardware=True)
-    def test_07_detach_volume_from_vm(self):
+    def test_22_detach_volume_from_vm(self):
         """
         Detach the iSCSI data volume from the running VM.
         Verifies:
@@ -827,11 +1560,43 @@ def test_07_detach_volume_from_vm(self):
         )
 
     # ------------------------------------------------------------------
-    # Step 08 — Destroy VM and clean up pool
+    # Step 23 — Destroy the second VM while it still has a VM snapshot
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_23_destroy_vm_with_vm_snapshot(self):
+        """
+        Destroy (expunge) the second VM while it still has a VM snapshot.
+        Verifies: destroy succeeds; the VM snapshot is no longer listed; its
+        FlexVol snapshot is removed from ONTAP.
+        """
+        self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_vm_snapshot(vm2.id)
+        new = self._assert_new_ontap_snaps(before, [snap], "vm", "second VM snapshot")
+
+        cmd = destroyVirtualMachineAPI.destroyVirtualMachineCmd()
+        cmd.id = vm2.id
+        cmd.expunge = True
+        self.apiClient.destroyVirtualMachine(cmd)
+        self.__class__.vm2 = None
+        self.__class__.vm2_ssh = None
+        self._vm_snapshot_ids.remove(snap.id)
+
+        deadline = time.time() + 300
+        while time.time() < deadline and self._vm_snapshots_of(vm2.id):
+            time.sleep(10)
+        self.assertFalse(self._vm_snapshots_of(vm2.id),
+                         "VM snapshot still listed after destroying its VM")
+        self._wait_ontap_snaps_gone(new)
+
+    # ------------------------------------------------------------------
+    # Step 24 — Destroy VM and clean up pool
     # ------------------------------------------------------------------
 
     @attr(tags=["iscsi_vm_workflow"], required_hardware=True)
-    def test_08_destroy_vm_and_cleanup(self):
+    def test_24_destroy_vm_and_cleanup(self):
         """
         Destroy the VM (with expunge), delete the data volume, force-delete
         the ONTAP pool, and delete the guest network created for this suite.
@@ -846,6 +1611,7 @@ def test_08_destroy_vm_and_cleanup(self):
           - ONTAP: all per-host igroups deleted
           - CloudStack: guest network deleted (Advanced zones only)
         """
+        self._destroy_second_vm()
         pool = self.__class__.pool
         vol = self.__class__.volume
 
diff --git a/test/integration/plugins/ontap/nfs3/instance/test_vm_volume_attach.py b/test/integration/plugins/ontap/nfs3/instance/test_vm_volume_attach.py
index dbd6b210ce85..60285433728a 100644
--- a/test/integration/plugins/ontap/nfs3/instance/test_vm_volume_attach.py
+++ b/test/integration/plugins/ontap/nfs3/instance/test_vm_volume_attach.py
@@ -20,13 +20,13 @@
 with a running virtual machine.
 
 Tests must run in order (test_01, test_02, test_03, test_03a, test_03b,
-test_04 ... test_08).  Each step builds on the shared state established by the
+test_04 ... test_24).  Each step builds on the shared state established by the
 previous step.
 
 The VM is deployed with a compute offering whose storage tags match the pool,
 so its ROOT volume lands on the ONTAP pool and exercises the primary
 template cache (seed / reuse / survive VM delete) alongside the data-volume
-workflow.
+and snapshot workflow.
 
 Workflow:
   01  Create NFS3 primary storage pool on ONTAP (tagged <storagePoolTags>-tmpl-cache)
@@ -38,10 +38,35 @@
   04  Attach the ONTAP data volume to the running VM
   05  Stop the VM — export policy stays; volume remains attached in CS
   06  Start the VM — VM Running; volume still attached; FlexVol online
-  07  Detach the ONTAP data volume from the VM
-  08  Destroy VM (cache still survives); delete ONTAP volume; enter
+  07  Open an SSH port forward to the VM; write ROOT and DATA files
+  08  Snapshot the ROOT volume of the running VM
+  09  Snapshot the DATA volume of the running VM
+  10  Revert a volume snapshot while the VM is Running — rejected
+  11  ROOT volume revert: write before-snapshot, snapshot, write
+      after-snapshot, revert, file is back to before-snapshot
+  12  DATA volume revert: same write, snapshot, write, revert check
+  13  Write before-s1; VM snapshot of the running VM (kept for step 15)
+  14  Write before-s2, stop, VM snapshot (kept for step 15)
+  15  Revert to the step-13 snapshot, then to the step-14 snapshot
+  16  Two VM snapshot requests for the same VM at once — both succeed
+  17  Deploy a second VM whose ROOT disk is on the same pool
+  18  Revert VM-A's ROOT; VM-B's file is unchanged
+  19  VM snapshots of both VMs at the same time
+  20  Volume snapshot of VM-A while VM-B's ROOT reverts
+  21  VM snapshot of VM-A while VM-B reverts to its VM snapshot
+  22  Detach the ONTAP data volume from the VM
+  23  Destroy the second VM while it still has a VM snapshot
+  24  Destroy VM (cache still survives); delete ONTAP volume; enter
       maintenance; force-delete pool
 
+Snapshot steps only create and verify; tearDown deletes every snapshot the
+step created and waits for its FlexVol snapshot to go.  Steps 13 and 14 keep
+their VM snapshots so step 15 can revert to them; step 15's tearDown deletes
+both.  Step 23 is last so more snapshot cases can be added before it.
+
+Guest login for the CentOS 5.5 template comes from ontap.cfg
+(cloudstack.templateUsername / templatePassword); see guest_ssh.py.
+
 Prerequisites:
   - CloudStack management server with the NetApp ONTAP plugin deployed
   - KVM cluster registered in CloudStack with at least one executable template
@@ -59,6 +84,7 @@
 import base64
 import logging
 import random
+import threading
 import time
 import unittest
 
@@ -82,9 +108,10 @@
     stopVirtualMachine as stopVirtualMachineAPI,
     updateStoragePool as updateStoragePoolAPI,
 )
-from marvin.lib.base import ServiceOffering, StoragePool
+from marvin.lib.base import ServiceOffering, Snapshot, StoragePool, VmSnapshot, Volume
 from marvin.lib.common import list_storage_pools
 
+from guest_ssh import GuestSsh, guest_credentials, new_token, open_ssh_port
 from ontap_test_base import OntapRestClient, OntapTestBase, _parse_pool_details, get_datacenter_config
 from helpers import template_cache_util as tcu
 from helpers.template_cache_base import (
@@ -96,6 +123,8 @@
 
 logger = logging.getLogger("TestOntapVMVolumeAttach")
 
+TAG = "vm_volume_workflow"
+
 
 # ---------------------------------------------------------------------------
 # Test data
@@ -116,12 +145,14 @@ class TestData:
     DETAIL_STORAGE_IP = "storageIP"
 
     ONTAP_MIN_VOLUME_SIZE = 1677721600
+    # ROOT + DATA disks of two VMs plus FlexVol snapshot reserve
+    DEFAULT_POOL_SIZE = 50 * 1024 ** 3
 
     def __init__(self, storage_ip, svm_name, username, password,
                  protocol="NFS3", scope="CLUSTER", provider="NetApp ONTAP",
                  tags="ontap-nfs3", capacitybytes=None):
         if capacitybytes is None:
-            capacitybytes = TestData.ONTAP_MIN_VOLUME_SIZE * 2
+            capacitybytes = TestData.DEFAULT_POOL_SIZE
         encoded_password = base64.b64encode(password.encode()).decode()
         self.testdata = {
             TestData.ontap: {
@@ -178,6 +209,15 @@ class TestOntapVMVolumeAttach(OntapTestBase):
     service_offering_id = None
     network_id = None      # None for Basic zones
     _created_network_id = None  # network created by this suite for Advanced zones
+    vm_ssh = None          # (public_ip, public_port) forwarded to vm:22
+    vm2_ssh = None
+    vm_snap_running = None  # step 13 snapshot, reused by step 15
+    vm_snap_stopped = None  # step 14 snapshot, reused by step 15
+    token_s1 = None
+    token_s2 = None
+    kept_ontap_snaps = None
+    guest_user = "root"
+    guest_password = "password"
 
     _vol_name_prefix = "OntapVMVol"
 
@@ -224,6 +264,8 @@ def setUpClass(cls):
 
         cls._setup_cloudstack_resources(config, cls.testdata[TestData.account])
 
+        cls.guest_user, cls.guest_password = guest_credentials(config)
+
         # Discover a suitable user KVM template in the zone (must be fully
         # downloaded; system-type templates are excluded as they cannot be
         # deployed as user VMs).
@@ -288,6 +330,7 @@ def setUpClass(cls):
     @classmethod
     def tearDownClass(cls):
         """Destroy the VMs first, then delegate pool/volume cleanup to super."""
+        cls._delete_kept_vm_snapshots()
         for vm in (cls.vm2, cls.vm):
             if vm is None:
                 continue
@@ -375,6 +418,256 @@ def _volume_state(self, vol_id):
         vols = self.apiClient.listVolumes(cmd)
         return vols[0].state if vols else "unknown"
 
+    # ---- snapshot step helpers (steps 07-21, 23) ---------------------------
+
+    def setUp(self):
+        super(TestOntapVMVolumeAttach, self).setUp()
+        self._volume_snapshot_ids = []
+        self._vm_snapshot_ids = []
+        self._extra_volume_ids = []
+        self._created_ontap_snaps = set()
+
+    def tearDown(self):
+        """
+        Delete the snapshots and extra volumes a step created, then check
+        their FlexVol snapshots are gone from ONTAP.
+        """
+        for snap_id in self._vm_snapshot_ids:
+            try:
+                VmSnapshot.deleteVMSnapshot(self.apiClient, snap_id)
+            except Exception as exc:
+                logger.warning("tearDown: VM snapshot %s: %s" % (snap_id, exc))
+        for snap_id in self._volume_snapshot_ids:
+            try:
+                Snapshot({"id": snap_id}).delete(self.apiClient)
+            except Exception as exc:
+                logger.warning("tearDown: volume snapshot %s: %s" % (snap_id, exc))
+        for vol_id in self._extra_volume_ids:
+            try:
+                cmd = deleteVolumeAPI.deleteVolumeCmd()
+                cmd.id = vol_id
+                self.apiClient.deleteVolume(cmd)
+            except Exception as exc:
+                logger.warning("tearDown: volume %s: %s" % (vol_id, exc))
+        if self._created_ontap_snaps and not self._wait_ontap_snaps_gone(
+                self._created_ontap_snaps, fail=False):
+            logger.warning("tearDown: FlexVol snapshots still present: %s"
+                           % sorted(self._created_ontap_snaps))
+        super(TestOntapVMVolumeAttach, self).tearDown()
+
+    @classmethod
+    def _destroy_second_vm(cls):
+        if cls.vm2 is None:
+            return
+        try:
+            cmd = destroyVirtualMachineAPI.destroyVirtualMachineCmd()
+            cmd.id = cls.vm2.id
+            cmd.expunge = True
+            cls.apiClient.destroyVirtualMachine(cmd)
+        except Exception as exc:
+            logger.warning("Could not destroy second VM %s: %s" % (cls.vm2.id, exc))
+        cls.vm2 = None
+        cls.vm2_ssh = None
+
+    @classmethod
+    def _delete_kept_vm_snapshots(cls):
+        """Delete the step 13/14 snapshots when step 15 did not take them over."""
+        for snap in (cls.vm_snap_running, cls.vm_snap_stopped):
+            if snap is None:
+                continue
+            try:
+                VmSnapshot.deleteVMSnapshot(cls.apiClient, snap.id)
+            except Exception as exc:
+                logger.warning("tearDownClass: VM snapshot %s: %s" % (snap.id, exc))
+        cls.vm_snap_running = None
+        cls.vm_snap_stopped = None
+        cls.token_s1 = None
+        cls.token_s2 = None
+        cls.kept_ontap_snaps = set()
+
+    def _keep_vm_snapshot(self, snap, ontap_names, which):
+        """Leave this VM snapshot in place for step 15."""
+        if snap.id in self._vm_snapshot_ids:
+            self._vm_snapshot_ids.remove(snap.id)
+        names = set(ontap_names)
+        self._created_ontap_snaps -= names
+        kept = self.__class__.kept_ontap_snaps
+        if kept is None:
+            kept = set()
+            self.__class__.kept_ontap_snaps = kept
+        kept.update(names)
+        setattr(self.__class__, which, snap)
+
+    def _adopt_kept_vm_snapshots(self):
+        """Hand the step 13 and 14 snapshots to this step's tearDown."""
+        s1 = self.__class__.vm_snap_running
+        s2 = self.__class__.vm_snap_stopped
+        token_a = self.__class__.token_s1
+        token_b = self.__class__.token_s2
+        if s1 is None or s2 is None or not token_a or not token_b:
+            self.skipTest("VM snapshots from test_13 and test_14 are required")
+        ontap_names = set(self.__class__.kept_ontap_snaps or [])
+        self._vm_snapshot_ids.extend([s1.id, s2.id])
+        self._created_ontap_snaps |= ontap_names
+        self.__class__.vm_snap_running = None
+        self.__class__.vm_snap_stopped = None
+        self.__class__.token_s1 = None
+        self.__class__.token_s2 = None
+        self.__class__.kept_ontap_snaps = set()
+        return s1, s2, token_a, token_b, ontap_names
+
+    def _require_snapshot_vm(self, ssh=True):
+        if self.__class__.vm is None:
+            self.skipTest("VM not deployed — test_03 was skipped (no ready template)")
+        if ssh and self.__class__.vm_ssh is None:
+            self.skipTest("Guest SSH not set up — test_07 must pass first")
+        self._start_vm(self.__class__.vm.id)
+        return self.__class__.vm
+
+    def _require_second_vm(self):
+        self._require_snapshot_vm()
+        if self.__class__.vm2 is None or self.__class__.vm2_ssh is None:
+            self.skipTest("Second VM not deployed — test_17 must pass first")
+        self._start_vm(self.__class__.vm2.id)
+        return self.__class__.vm2
+
+    def _vm_state(self, vm_id):
+        vms = self.apiClient.listVirtualMachines(_list_vms_cmd(vm_id))
+        return vms[0].state if vms else None
+
+    def _stop_vm(self, vm_id):
+        if (self._vm_state(vm_id) or "").lower() == "stopped":
+            return
+        cmd = stopVirtualMachineAPI.stopVirtualMachineCmd()
+        cmd.id = vm_id
+        cmd.forced = True
+        self.apiClient.stopVirtualMachine(cmd)
+        self._poll_vm_state(vm_id, "Stopped", timeout=600)
+
+    def _start_vm(self, vm_id):
+        if (self._vm_state(vm_id) or "").lower() == "running":
+            return
+        cmd = startVirtualMachineAPI.startVirtualMachineCmd()
+        cmd.id = vm_id
+        self.apiClient.startVirtualMachine(cmd)
+        self._poll_vm_state(vm_id, "Running", timeout=900)
+
+    def _root_volume(self, vm_id):
+        cmd = listVolumesAPI.listVolumesCmd()
+        cmd.virtualmachineid = vm_id
+        cmd.type = "ROOT"
+        cmd.listall = True
+        vols = self.apiClient.listVolumes(cmd) or []
+        self.assertTrue(vols, "No ROOT volume found for VM %s" % vm_id)
+        return vols[0]
+
+    def _guest(self, which="vm"):
+        host, port = getattr(self.__class__, "%s_ssh" % which)
+        return GuestSsh(host, port, self.guest_user, self.guest_password)
+
+    def _write_marker(self, token, which="vm", disks=("ROOT",)):
+        with self._guest(which) as guest:
+            if "ROOT" in disks:
+                guest.write_root_marker(token)
+            if "DATA" in disks:
+                guest.write_data_marker(token)
+
+    def _assert_marker(self, token, which="vm", disk="ROOT", msg=""):
+        with self._guest(which) as guest:
+            found = guest.read_data_marker() if disk == "DATA" else guest.read_root_marker()
+        self.assertEqual(found, token, "%s: %s marker on %s is %r, expected %r"
+                         % (msg, disk, which, found, token))
+
+    def _ontap_snap_names(self):
+        return {s.get("name") for s in self.ontap.list_snapshots(self.__class__.pool.name)}
+
+    def _assert_new_ontap_snaps(self, before, snaps, kind, msg):
+        """Return the FlexVol snapshot created for each CloudStack snapshot.
+
+        A volume snapshot is stored as ``..._cs<dbId>`` and a VM snapshot as
+        ``..._vm<dbId>``. ONTAP snapshots from its own schedule, such as
+        ``hourly.*`` or ``Anti_ransomware_*``, are ignored.
+        """
+        new = self._ontap_snap_names() - before
+        prefix = "cs" if kind == "volume" else "vm"
+        table = "snapshots" if kind == "volume" else "vm_snapshots"
+        matched = set()
+        missing = []
+        for snap in snaps:
+            db_id = tcu.get_db_id(self.dbConnection, table, snap.id)
+            suffix = "_%s%s" % (prefix, db_id)
+            hits = sorted(name for name in new if name.endswith(suffix))
+            if len(hits) != 1:
+                missing.append("%s (suffix %s, matches %s)" % (snap.id, suffix, hits))
+            else:
+                matched.add(hits[0])
+        self.assertFalse(
+            missing,
+            "%s: FlexVol snapshot missing for %s; new FlexVol snapshots: %s"
+            % (msg, missing, sorted(new)))
+        ignored = new - matched
+        if ignored:
+            logger.info("%s: ignoring FlexVol snapshots not created by this step: %s",
+                        msg, sorted(ignored))
+        self._created_ontap_snaps |= matched
+        return matched
+
+    def _assert_ontap_snaps_present(self, names, msg):
+        missing = set(names) - self._ontap_snap_names()
+        self.assertFalse(missing, "%s: FlexVol snapshots missing: %s" % (msg, sorted(missing)))
+
+    def _wait_ontap_snaps_gone(self, names, timeout=180, fail=True):
+        names = set(names)
+        deadline = time.time() + timeout
+        left = names
+        while time.time() < deadline:
+            left = names & self._ontap_snap_names()
+            if not left:
+                return True
+            time.sleep(5)
+        if fail:
+            self.fail("FlexVol snapshots %s still present after %ds" % (sorted(left), timeout))
+        return False
+
+    def _take_volume_snapshot(self, volume_id):
+        snap = Snapshot.create(self.apiClient, volume_id,
+                               account=self.account.name, domainid=self.domain.id)
+        self._volume_snapshot_ids.append(snap.id)
+        self.assertEqual(snap.state, "BackedUp",
+                         "Volume snapshot %s is %s, expected BackedUp" % (snap.id, snap.state))
+        return snap
+
+    def _take_vm_snapshot(self, vm_id):
+        snap = VmSnapshot.create(self.apiClient, vm_id, snapshotmemory="false",
+                                 name="ontap_%d" % random.randint(0, 999999))
+        self._vm_snapshot_ids.append(snap.id)
+        self.assertEqual(snap.state, "Ready",
+                         "VM snapshot %s is %s, expected Ready" % (snap.id, snap.state))
+        return snap
+
+    def _vm_snapshots_of(self, vm_id):
+        return VmSnapshot.list(self.apiClient, virtualmachineid=vm_id, listall=True) or []
+
+    def _volume_revert_flow(self, disk):
+        """Write A, snapshot, write B, stop, revert, start; expect A."""
+        vm = self._require_snapshot_vm()
+        volume_id = (self.__class__.volume.id if disk == "DATA"
+                     else self._root_volume(vm.id).id)
+        token_a = new_token("before-snapshot")
+        token_b = new_token("after-snapshot")
+        self._write_marker(token_a, disks=(disk,))
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(volume_id)
+        new = self._assert_new_ontap_snaps(before, [snap], "volume", "%s volume snapshot" % disk)
+        self._write_marker(token_b, disks=(disk,))
+
+        self._stop_vm(vm.id)
+        Volume.revertToSnapshot(self.apiClient, snap.id)
+        self._assert_ontap_snaps_present(new, "after revert")
+        self._start_vm(vm.id)
+        self._assert_marker(token_a, disk=disk,
+                            msg="after revert (write after snapshot was %s)" % token_b)
+
     # ==================================================================
     # Test steps
     # ==================================================================
@@ -744,11 +1037,372 @@ def test_06_start_vm_volume_accessible(self):
         )
 
     # ------------------------------------------------------------------
-    # Step 07 - Detach ONTAP data volume from the VM
+    # Step 07 - Guest SSH access for snapshot checks
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_07_open_ssh_and_write_root_data_files(self):
+        """
+        Prepare the running VM for snapshot checks.
+        Verifies:
+          - ROOT and DATA volumes are both on the ONTAP test pool
+          - SSH login (ontap.cfg templateUsername/templatePassword) works
+            through a port forward on the network's source-NAT IP
+          - The DATA disk can be formatted, mounted and written
+        """
+        vm = self._require_snapshot_vm(ssh=False)
+        if not self.__class__.network_id:
+            self.skipTest("Snapshot steps need an Advanced zone guest network for SSH")
+        pool_id = self.__class__.pool.id
+        root = self._root_volume(vm.id)
+        self.assertEqual(getattr(root, "storageid", None), pool_id,
+                         "ROOT volume is on '%s', not the ONTAP test pool"
+                         % getattr(root, "storage", None))
+        vols = self.apiClient.listVolumes(_list_vols_cmd(self.__class__.volume.id)) or []
+        self.assertTrue(vols and getattr(vols[0], "virtualmachineid", None) == vm.id,
+                        "DATA volume must be attached before the snapshot steps")
+
+        host, port, _ = open_ssh_port(self.apiClient, vm.id, self.__class__.network_id)
+        self.__class__.vm_ssh = (host, port)
+        token = new_token("ssh-access-test")
+        self._write_marker(token, disks=("ROOT", "DATA"))
+        self._assert_marker(token, disk="ROOT", msg="guest access")
+        self._assert_marker(token, disk="DATA", msg="guest access")
+
+    # ------------------------------------------------------------------
+    # Steps 08-12 - Volume snapshots
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_08_root_volume_snapshot_create(self):
+        """
+        VS-01: snapshot the ROOT volume of the running VM.
+        Verifies: state BackedUp; VM stays Running; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(self._root_volume(vm.id).id)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        self._assert_new_ontap_snaps(before, [snap], "volume", "ROOT volume snapshot")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_09_data_volume_snapshot_create(self):
+        """
+        VS-03: snapshot the attached DATA volume.
+        Verifies: state BackedUp; VM stays Running; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(self.__class__.volume.id)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        self._assert_new_ontap_snaps(before, [snap], "volume", "DATA volume snapshot")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_10_volume_revert_running_vm_rejected(self):
+        """
+        VS-19: reverting a volume snapshot while the VM is Running fails.
+        Verifies: error says the Instance is not shut down; VM stays Running;
+        the FlexVol snapshot remains.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_volume_snapshot(self._root_volume(vm.id).id)
+        new = self._assert_new_ontap_snaps(before, [snap], "volume", "ROOT volume snapshot")
+        with self.assertRaises(Exception) as ctx:
+            Volume.revertToSnapshot(self.apiClient, snap.id)
+        self.assertIn("shutdown", str(ctx.exception).lower(),
+                      "Unexpected error: %s" % ctx.exception)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        self._assert_ontap_snaps_present(new, "after rejected revert")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_11_root_volume_revert(self):
+        """
+        VS-11: write A, snapshot ROOT, write B, stop, revert, start.
+        Verifies: ROOT marker is A; the FlexVol snapshot survives the revert.
+        """
+        self._volume_revert_flow("ROOT")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_12_data_volume_revert(self):
+        """
+        Write A, snapshot DATA, write B, stop, revert, start.
+        Verifies: DATA marker is A; the FlexVol snapshot survives the revert.
+        """
+        self._volume_revert_flow("DATA")
+
+    # ------------------------------------------------------------------
+    # Steps 13-16 - VM snapshots
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_13_vm_snapshot_create_running(self):
+        """
+        VMS-01: VM snapshot of the running VM (ROOT + DATA on one FlexVol).
+        Writes before-s1, then snapshots. The snapshot is kept for step 15.
+        Verifies: state Ready; VM stays Running; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        self.__class__.token_s1 = new_token("before-s1")
+        self._write_marker(self.__class__.token_s1, disks=("ROOT", "DATA"))
+        before = self._ontap_snap_names()
+        snap = self._take_vm_snapshot(vm.id)
+        self.assertEqual(self._vm_state(vm.id), "Running")
+        new = self._assert_new_ontap_snaps(before, [snap], "vm", "VM snapshot")
+        self._keep_vm_snapshot(snap, new, "vm_snap_running")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_14_vm_snapshot_create_stopped(self):
+        """
+        VMS-03: VM snapshot of the stopped VM.
+        Writes before-s2 while Running, then stops and snapshots.
+        The snapshot is kept for step 15.
+        Verifies: state Ready; VM stays Stopped; one new FlexVol snapshot.
+        """
+        vm = self._require_snapshot_vm()
+        self.__class__.token_s2 = new_token("before-s2")
+        self._write_marker(self.__class__.token_s2, disks=("ROOT", "DATA"))
+        self._stop_vm(vm.id)
+        before = self._ontap_snap_names()
+        snap = self._take_vm_snapshot(vm.id)
+        self.assertEqual(self._vm_state(vm.id), "Stopped")
+        new = self._assert_new_ontap_snaps(before, [snap], "vm", "VM snapshot (Stopped)")
+        self._keep_vm_snapshot(snap, new, "vm_snap_stopped")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_15_vm_snapshot_s1_running_s2_stopped_revert(self):
+        """
+        VMS-09 (NFS ROOT + NFS DATA on the same pool): revert to the running
+        snapshot from step 13, then to the stopped snapshot from step 14.
+        Verifies:
+          - After revert to S1: ROOT and DATA hold before-s1
+          - After revert to S2: ROOT and DATA hold before-s2
+          - Both FlexVol snapshots survive the reverts
+        """
+        if self.__class__.vm is None:
+            self.skipTest("VM not deployed — test_03 was skipped (no ready template)")
+        if self.__class__.vm_ssh is None:
+            self.skipTest("Guest SSH not set up — test_07 must pass first")
+        s1, s2, token_a, token_b, ontap_names = self._adopt_kept_vm_snapshots()
+        vm = self.__class__.vm
+
+        self._stop_vm(vm.id)
+        VmSnapshot.revertToSnapshot(self.apiClient, s1.id)
+        self._start_vm(vm.id)
+        self._assert_marker(token_a, disk="ROOT", msg="after revert to S1")
+        self._assert_marker(token_a, disk="DATA", msg="after revert to S1")
+        self.assertEqual(len(self._vm_snapshots_of(vm.id)), 2, "S2 should still be listed")
+
+        self._stop_vm(vm.id)
+        VmSnapshot.revertToSnapshot(self.apiClient, s2.id)
+        self._start_vm(vm.id)
+        self._assert_marker(token_b, disk="ROOT", msg="after revert to S2")
+        self._assert_marker(token_b, disk="DATA", msg="after revert to S2")
+        self._assert_ontap_snaps_present(ontap_names, "after both reverts")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_16_vm_snapshot_duplicate_requests(self):
+        """
+        VMS-53: two VM snapshot requests for the same VM at the same time.
+        A rejected request is retried on its own; one that already succeeded
+        is not repeated.
+        Verifies: both succeed and are Ready; two new FlexVol snapshots.
+        """
+        vm = self._require_snapshot_vm()
+        before = self._ontap_snap_names()
+        pending = [
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_dup_a_%d" % random.randint(0, 99999)),
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_dup_b_%d" % random.randint(0, 99999)),
+        ]
+        ok = []
+        last_errors = []
+        attempts = 3
+        for attempt in range(1, attempts + 1):
+            results = _run_concurrent(*pending)
+            last_errors = []
+            still_pending = []
+            retryable = True
+            for fn, (result, exc) in zip(pending, results):
+                if exc is None and result is not None:
+                    ok.append(result)
+                    self._vm_snapshot_ids.append(result.id)
+                    continue
+                last_errors.append(exc)
+                if exc is None or not _is_rejected(exc):
+                    retryable = False
+                still_pending.append(fn)
+            pending = still_pending
+            if not pending:
+                break
+            if not retryable or attempt == attempts:
+                break
+            logger.warning("duplicate VM snapshot rejected, retry %d/%d: %s",
+                           attempt, attempts, last_errors)
+            time.sleep(15)
+        self.assertFalse(pending,
+                         "Duplicate VM snapshot requests failed: %s" % last_errors)
+        self.assertEqual(len(ok), 2, "Expected both VM snapshots to succeed; "
+                         "got %d" % len(ok))
+        for snap in ok:
+            self.assertEqual(snap.state, "Ready")
+            self.assertEqual(snap.virtualmachineid, vm.id)
+        self._assert_new_ontap_snaps(before, ok, "vm", "duplicate VM snapshot requests")
+
+    # ------------------------------------------------------------------
+    # Steps 17-21 - Second VM on the same pool: isolation and concurrency
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_17_deploy_second_vm(self):
+        """
+        Deploy a second VM whose ROOT disk is on the same ONTAP pool.
+        Verifies: VM Running; ROOT on the test pool; SSH works.
+        """
+        self._require_snapshot_vm()
+        cmd = deployVirtualMachineAPI.deployVirtualMachineCmd()
+        cmd.zoneid = self.zone.id
+        cmd.templateid = self.__class__.template_id
+        cmd.serviceofferingid = self.__class__.service_offering_id
+        cmd.account = self.account.name
+        cmd.domainid = self.domain.id
+        cmd.networkids = self.__class__.network_id
+        vm2 = self.apiClient.deployVirtualMachine(cmd)
+        self.__class__.vm2 = vm2
+        self._poll_vm_state(vm2.id, "Running", timeout=900)
+        root = self._root_volume(vm2.id)
+        self.assertEqual(getattr(root, "storageid", None), self.__class__.pool.id,
+                         "Second VM ROOT is not on the ONTAP test pool")
+        host, port, _ = open_ssh_port(self.apiClient, vm2.id, self.__class__.network_id)
+        self.__class__.vm2_ssh = (host, port)
+        token = new_token("second-vm-ssh-access-test")
+        self._write_marker(token, which="vm2")
+        self._assert_marker(token, which="vm2", msg="second VM access")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_18_volume_revert_leaves_other_vm_untouched(self):
+        """
+        VS-28: two VMs on one FlexVol; revert VM-A's ROOT to its snapshot.
+        Verifies: VM-A is back at its snapshot; VM-B keeps data written after it.
+        """
+        vm = self._require_snapshot_vm()
+        self._require_second_vm()
+        a1 = new_token("vm-a-before-snapshot")
+        a2 = new_token("vm-a-after-snapshot")
+        b1 = new_token("vm-b-before-snapshot")
+        b2 = new_token("vm-b-after-snapshot")
+        self._write_marker(a1)
+        self._write_marker(b1, which="vm2")
+        snap = self._take_volume_snapshot(self._root_volume(vm.id).id)
+        self._write_marker(a2)
+        self._write_marker(b2, which="vm2")
+
+        self._stop_vm(vm.id)
+        Volume.revertToSnapshot(self.apiClient, snap.id)
+        self._start_vm(vm.id)
+        self._assert_marker(a1, msg="VM-A after revert")
+        self._assert_marker(b2, which="vm2", msg="VM-B after VM-A's revert")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_19_concurrent_vm_snapshots_two_vms(self):
+        """
+        VMS-54: VM snapshots on two VMs of the same NFS pool at the same time.
+        Verifies: both Ready and owned by their VMs; two distinct new FlexVol
+        snapshots.
+        """
+        vm = self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        before = self._ontap_snap_names()
+        results = _run_concurrent(
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_cc_a_%d" % random.randint(0, 99999)),
+            lambda: VmSnapshot.create(self.apiClient, vm2.id, snapshotmemory="false",
+                                      name="ontap_cc_b_%d" % random.randint(0, 99999)),
+        )
+        self._vm_snapshot_ids.extend(s.id for s, _ in results if s is not None)
+        errors = [e for _, e in results if e is not None]
+        self.assertFalse(errors, "Concurrent VM snapshots failed: %s" % errors)
+        snap_a, snap_b = results[0][0], results[1][0]
+        self.assertEqual(snap_a.state, "Ready")
+        self.assertEqual(snap_b.state, "Ready")
+        self.assertEqual(snap_a.virtualmachineid, vm.id)
+        self.assertEqual(snap_b.virtualmachineid, vm2.id)
+        self._assert_new_ontap_snaps(before, [snap_a, snap_b], "vm", "concurrent VM snapshots")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_20_concurrent_volume_snapshot_and_revert(self):
+        """
+        Volume snapshot of VM-A's ROOT while VM-B's ROOT reverts, at the same time.
+        A rejected snapshot or revert is retried on its own.
+        Verifies: VM-A snapshot BackedUp with its own FlexVol snapshot; VM-B is
+        back at its snapshot; VM-A data unchanged.
+        """
+        vm = self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        a1 = new_token("vm-a-current-data")
+        b1 = new_token("vm-b-before-snapshot")
+        b2 = new_token("vm-b-after-snapshot")
+        self._write_marker(a1)
+        self._write_marker(b1, which="vm2")
+        snap_b = self._take_volume_snapshot(self._root_volume(vm2.id).id)
+        self._write_marker(b2, which="vm2")
+        self._stop_vm(vm2.id)
+        before = self._ontap_snap_names()
+        root_a = self._root_volume(vm.id).id
+
+        snap_a = _snapshot_and_revert_with_retry(
+            lambda: Snapshot.create(self.apiClient, root_a,
+                                    account=self.account.name, domainid=self.domain.id),
+            lambda: Volume.revertToSnapshot(self.apiClient, snap_b.id),
+        )
+        self._volume_snapshot_ids.append(snap_a.id)
+        self.assertEqual(snap_a.state, "BackedUp")
+        self._assert_new_ontap_snaps(before, [snap_a], "volume", "VM-A snapshot during VM-B revert")
+
+        self._start_vm(vm2.id)
+        self._assert_marker(b1, which="vm2", msg="VM-B after revert")
+        self._assert_marker(a1, msg="VM-A during VM-B revert")
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_21_concurrent_vm_snapshot_and_revert(self):
+        """
+        VM snapshot of VM-A while VM-B reverts to its VM snapshot, at the same time.
+        A rejected snapshot or revert is retried on its own.
+        Verifies: VM-A snapshot Ready with its own FlexVol snapshot; VM-B is
+        back at its snapshot; VM-A data unchanged.
+        """
+        vm = self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        a1 = new_token("vm-a-current-data")
+        b1 = new_token("vm-b-before-snapshot")
+        b2 = new_token("vm-b-after-snapshot")
+        self._write_marker(a1)
+        self._write_marker(b1, which="vm2")
+        snap_b = self._take_vm_snapshot(vm2.id)
+        self._write_marker(b2, which="vm2")
+        self._stop_vm(vm2.id)
+        before = self._ontap_snap_names()
+
+        snap_a = _snapshot_and_revert_with_retry(
+            lambda: VmSnapshot.create(self.apiClient, vm.id, snapshotmemory="false",
+                                      name="ontap_ccr_%d" % random.randint(0, 99999)),
+            lambda: VmSnapshot.revertToSnapshot(self.apiClient, snap_b.id),
+        )
+        self._vm_snapshot_ids.append(snap_a.id)
+        self.assertEqual(snap_a.state, "Ready")
+        self._assert_new_ontap_snaps(before, [snap_a], "vm", "VM-A snapshot during VM-B revert")
+
+        self._start_vm(vm2.id)
+        self._assert_marker(b1, which="vm2", msg="VM-B after revert")
+        self._assert_marker(a1, msg="VM-A during VM-B revert")
+
+    # ------------------------------------------------------------------
+    # Step 22 - Detach ONTAP data volume from the VM
     # ------------------------------------------------------------------
 
     @attr(tags=["vm_volume_workflow"], required_hardware=True)
-    def test_07_detach_volume_from_vm(self):
+    def test_22_detach_volume_from_vm(self):
         """
         Detach the ONTAP data volume from the running VM.
         Verifies:
@@ -831,11 +1485,43 @@ def test_07_detach_volume_from_vm(self):
         )
 
     # ------------------------------------------------------------------
-    # Step 08 - Destroy VM, delete volume, delete pool
+    # Step 23 - Destroy the second VM while it still has a VM snapshot
+    # ------------------------------------------------------------------
+
+    @attr(tags=[TAG], required_hardware=True)
+    def test_23_destroy_vm_with_vm_snapshot(self):
+        """
+        VMS-38: destroy (expunge) the second VM while it still has a VM snapshot.
+        Verifies: destroy succeeds; the VM snapshot is no longer listed; its
+        FlexVol snapshot is removed from ONTAP.
+        """
+        self._require_snapshot_vm()
+        vm2 = self._require_second_vm()
+        before = self._ontap_snap_names()
+        snap = self._take_vm_snapshot(vm2.id)
+        new = self._assert_new_ontap_snaps(before, [snap], "vm", "second VM snapshot")
+
+        cmd = destroyVirtualMachineAPI.destroyVirtualMachineCmd()
+        cmd.id = vm2.id
+        cmd.expunge = True
+        self.apiClient.destroyVirtualMachine(cmd)
+        self.__class__.vm2 = None
+        self.__class__.vm2_ssh = None
+        self._vm_snapshot_ids.remove(snap.id)
+
+        deadline = time.time() + 300
+        while time.time() < deadline and self._vm_snapshots_of(vm2.id):
+            time.sleep(10)
+        self.assertFalse(self._vm_snapshots_of(vm2.id),
+                         "VM snapshot still listed after destroying its VM")
+        self._wait_ontap_snaps_gone(new)
+
+    # ------------------------------------------------------------------
+    # Step 24 - Destroy VM, delete volume, delete pool
     # ------------------------------------------------------------------
 
     @attr(tags=["vm_volume_workflow"], required_hardware=True)
-    def test_08_destroy_vm_and_cleanup(self):
+    def test_24_destroy_vm_and_cleanup(self):
         """
         Destroy the VM, delete the ONTAP data volume, enter maintenance,
         then force-delete the pool.
@@ -852,6 +1538,7 @@ def test_08_destroy_vm_and_cleanup(self):
         self.assertIsNotNone(self.__class__.pool,
                              "Pool absent — test_01 must pass first")
 
+        self._destroy_second_vm()
         vm = self.__class__.vm
         vol = self.__class__.volume
         pool = self.__class__.pool
@@ -950,6 +1637,83 @@ def _list_vols_cmd(vol_id):
     return cmd
 
 
+def _run_concurrent(*callables):
+    """Start all callables at the same moment; return [(result, exception)]."""
+    results = [(None, None)] * len(callables)
+    barrier = threading.Barrier(len(callables))
+
+    def worker(index, fn):
+        try:
+            barrier.wait(timeout=60)
+            results[index] = (fn(), None)
+        except Exception as exc:
+            results[index] = (None, exc)
+
+    threads = [threading.Thread(target=worker, args=(i, fn))
+               for i, fn in enumerate(callables)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    return results
+
+
+def _is_rejected(exc):
+    """True when CloudStack or ONTAP refused the call because another one was in progress."""
+    text = str(exc).lower()
+    return any(token in text for token in (
+        "internal error",
+        "failed to create snapshot",
+        "failed to create instance snapshot",
+        "failed to revert",
+        "another operation",
+        "other active",
+        "try again later",
+        "in progress",
+        "busy",
+    ))
+
+
+def _snapshot_and_revert_with_retry(snapshot_fn, revert_fn, attempts=3, pause=15):
+    """Run a snapshot and a revert together.
+
+    A call that is rejected is retried on its own. A call that already
+    succeeded is not repeated. Returns the snapshot from snapshot_fn.
+    """
+    snap = None
+    reverted = False
+    last_errors = []
+    for attempt in range(1, attempts + 1):
+        jobs = []
+        if snap is None:
+            jobs.append(("snapshot", snapshot_fn))
+        if not reverted:
+            jobs.append(("revert", revert_fn))
+        results = _run_concurrent(*(fn for _, fn in jobs))
+        last_errors = []
+        retryable = True
+        for (kind, _), (result, exc) in zip(jobs, results):
+            if exc is None:
+                if kind == "snapshot":
+                    snap = result
+                else:
+                    reverted = True
+                continue
+            last_errors.append("%s: %s" % (kind, exc))
+            if not _is_rejected(exc):
+                retryable = False
+        if snap is not None and reverted:
+            return snap
+        if not retryable or attempt == attempts:
+            break
+        logger.warning("snapshot/revert rejected, retry %d/%d: %s",
+                       attempt, attempts, last_errors)
+        time.sleep(pause)
+    raise AssertionError(
+        "Snapshot-while-revert failed after %d attempt(s): %s"
+        % (attempt, last_errors))
+
+
 def _wait_for_vm_state(api_client, vm_id, target_state, timeout=120,
                        interval=5):
     """Blocking wait for a VM to reach target_state (used in tearDownClass)."""
diff --git a/test/integration/plugins/ontap/ontap.cfg b/test/integration/plugins/ontap/ontap.cfg
index 64ee5174d379..47879ae4d340 100644
--- a/test/integration/plugins/ontap/ontap.cfg
+++ b/test/integration/plugins/ontap/ontap.cfg
@@ -121,6 +121,8 @@
     "clusterName": null,
     "domainName": "ROOT",
     "templateName": "CentOS 5.5(64-bit) no GUI (KVM)",
+    "templateUsername": "root",
+    "templatePassword": "<<password>>",
     "systemVmTimeoutSec": 3600,
     "templateReadyTimeoutSec": 3600,
     "pollIntervalSec": 60
diff --git a/test/integration/plugins/ontap/ontap_test_base.py b/test/integration/plugins/ontap/ontap_test_base.py
index 17bec2c1251c..27387051f6b0 100644
--- a/test/integration/plugins/ontap/ontap_test_base.py
+++ b/test/integration/plugins/ontap/ontap_test_base.py
@@ -433,6 +433,27 @@ def delete_file_in_volume(self, vol_name, file_path):
         self._delete("/storage/volumes/%s/files/%s" % (vol_uuid, encoded_path))
         return True
 
+    def list_snapshots(self, vol_name):
+        """Return FlexVol snapshot records (name, uuid) for the named volume.
+
+        Uses ``GET /api/storage/volumes/{uuid}/snapshots``.  An empty list means
+        the volume does not exist or has no snapshots.
+        """
+        vol = self.get_volume(vol_name)
+        if not vol:
+            return []
+        vol_uuid = vol.get("uuid", "")
+        if not vol_uuid:
+            return []
+        try:
+            data = self._get(
+                "/storage/volumes/%s/snapshots" % vol_uuid,
+                params={"fields": "name,uuid", "max_records": "1000"},
+            )
+        except Exception:
+            return []
+        return list(data.get("records", []) or [])
+
 
 # ---------------------------------------------------------------------------
 # Base test class
diff --git a/test/integration/plugins/ontap/setup_env.sh b/test/integration/plugins/ontap/setup_env.sh
index 2200e7c1c6f0..77b01bbe601f 100755
--- a/test/integration/plugins/ontap/setup_env.sh
+++ b/test/integration/plugins/ontap/setup_env.sh
@@ -95,6 +95,12 @@ if ! "${PIP}" install "pyvmomi==8.0.2.0.1"; then
     exit 1
 fi
 
+echo "==> Pinning paramiko<4 (CentOS 5.5 guest sshd only offers SHA-1 kex and ssh-rsa)"
+if ! "${PIP}" install "paramiko>=3.4,<4"; then
+    echo "Unable to install paramiko; verify access to the configured Python package index" >&2
+    exit 1
+fi
+
 echo ""
 echo "==> Verifying installation"
 if ! "${PYTHON}" -c "import marvin; print('Marvin OK')"; then
@@ -125,7 +131,8 @@ fi
 echo "Nose test discovery compatibility OK"
 
 echo ""
-echo "Done. Activate the venv with:"
+echo "Done. Guest SSH uses paramiko>=3.4,<4 (CentOS 5.5 sshd is SHA-1 only)."
+echo "Activate the venv with:"
 echo "  source test/integration/plugins/ontap/.venv/bin/activate"
 echo ""
 echo "Run zone setup tests:"